fix(ssh): join authenticated stream teardown and validate package closure

.agents/notes/implemented/architecture/2026-09-11-posix-ssh-runtime.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 .agents/notes/implemented/architecture/2026-09-11-posix-ssh-runtime.md
-2026-09-11-posix-ssh-runtime.md: 252677358f4ecf26d85d3eede81e3d2faf34c17f
-2026-09-11-posix-ssh-runtime.zh.md: 7a5468644927fa22630ea6c82e939b08efac8e03
+2026-09-11-posix-ssh-runtime.md: f3fe3ac4aa3279d2821e075ec858cc6bfc7e9b27
+2026-09-11-posix-ssh-runtime.zh.md: 27bee34c8e1cc4a5c6f8cd34f603ba7a3eb27398

.agents/notes/implemented/architecture/2026-09-11-posix-ssh-runtime.md

@@ -18,7 +18,7 @@ Confinement is asynchronous and cancellable: the running helper resolves each po
 
 Private administrative RPC and each program stream use independent SSH channels. A remote reservation creates the requested stream endpoints before the payload starts. Stdout and stderr cannot carry administrative replies, and pausing one output channel does not consume fd 7’s flow-control window. This trades a larger remote helper for explicit binary transport and bounded stream retention.
 
-Each stream receives a fresh 256-bit TLS pre-shared key through private administrative RPC. TLS 1.2 with `PSK-AES256-GCM-SHA384` authenticates both endpoints and protects subsequent bytes before the stream can publish. The key never travels as a stream preface. Socket permissions alone are insufficient: file-effect confinement can permit same-user connections or pathname replacement in writable temporary directories. No administrative Unix listener or stream secret is exposed to the payload.
+Each stream receives a fresh 256-bit TLS pre-shared key through private administrative RPC. TLS 1.2 with `PSK-AES256-GCM-SHA384` authenticates both endpoints and protects subsequent bytes before the stream can publish. The key never travels as a stream preface. Socket permissions alone are insufficient: file-effect confinement can permit same-user connections or pathname replacement in writable temporary directories. No administrative Unix listener or stream secret is exposed to the payload. Cancellation transfers to the TLS wrapper after wrapping; cleanup closes that wrapper before the underlying socket so native TLS reads cannot outlive their transport.
 
 Collected stdout and stderr carry bounded tail snapshots through handlers that only update output observations. Capture continues while a snapshot is waiting for transport. Final snapshots preserve raw-byte offsets, and completed spill files use the local provider’s retained-output storage after connection disposal.
 

.agents/notes/implemented/architecture/2026-09-11-posix-ssh-runtime.zh.md

@@ -18,7 +18,7 @@ SSH 提供经过认证的字节通道及逐通道流量控制，但普通 exec
 
 私有管理 RPC 与各条程序流使用独立 SSH 通道。远端预留操作在程序启动前创建请求的流端点。stdout 与 stderr 无法承载管理回复，暂停一条输出通道也不会占用 fd 7 的流控窗口。这用更大的远端辅助程序换取显式二进制传输与有界流保留。
 
-每条流通过私有管理 RPC 获得新的 256 位 TLS 预共享密钥。TLS 1.2 配合 `PSK-AES256-GCM-SHA384` 在流发布前认证两端并保护后续字节。密钥绝不作为流前缀传输。仅靠套接字权限不足：文件效果限制可能允许同用户连接，或允许在可写临时目录中替换路径名。程序不会获得管理 Unix 监听端点或流密钥。
+每条流通过私有管理 RPC 获得新的 256 位 TLS 预共享密钥。TLS 1.2 配合 `PSK-AES256-GCM-SHA384` 在流发布前认证两端并保护后续字节。密钥绝不作为流前缀传输。仅靠套接字权限不足：文件效果限制可能允许同用户连接，或允许在可写临时目录中替换路径名。程序不会获得管理 Unix 监听端点或流密钥。包装完成后，取消责任转移到 TLS 包装流；清理先关闭该流，再关闭底层套接字，确保原生 TLS 读取不会超过传输层的生命周期。
 
 收集的 stdout 与 stderr 通过仅更新输出观测的处理器传递有界尾部快照。快照等待传输时，输出捕获继续进行。最终快照保留原始字节偏移，已完成的 spill 文件在连接释放后使用本地提供方的保留输出存储。
 

docs/config-catalog.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 docs/config-catalog.md
-config-catalog.md: 01c41d7447ff6a770a7533300a6efbe727fa2a49
-config-catalog.zh.md: 68c87ddbde5a0b3382e7903926c90a90b16b52fa
+config-catalog.md: 5a67956f9b7b18c37105f9526553817b1e4305fb
+config-catalog.zh.md: 9755afb73078cacbc67b4f6998f21db0c820208c

docs/config-catalog.md

@@ -2286,14 +2286,14 @@ export interface Config {
   requestTimeoutMs?: number
   /** Maximum JSON payload bytes per helper request or response. */
   maxFrameBytes?: number
-  /** Maximum outstanding administrative requests. */
+  /** Maximum ordinary requests; heartbeat and bounded resource cleanup have reserved capacity. */
   maxPending?: number
   /** Remote helper lease; loss of heartbeats starts remote managed cleanup. */
   leaseMs?: number
 }
 ```
 
-Source: [`packages/ssh/ssh/src/index.ts:19`](../packages/ssh/ssh/src/index.ts)
+Source: [`packages/ssh/ssh/src/index.ts:17`](../packages/ssh/ssh/src/index.ts)
 
 <a id="deepseek-aidsh-storage-domain"></a>
 

docs/config-catalog.zh.md

@@ -2288,14 +2288,14 @@ export interface Config {
   requestTimeoutMs?: number
   /** Maximum JSON payload bytes per helper request or response. */
   maxFrameBytes?: number
-  /** Maximum outstanding administrative requests. */
+  /** Maximum ordinary requests; heartbeat and bounded resource cleanup have reserved capacity. */
   maxPending?: number
   /** Remote helper lease; loss of heartbeats starts remote managed cleanup. */
   leaseMs?: number
 }
 ```
 
-来源： [`packages/ssh/ssh/src/index.ts:19`](../packages/ssh/ssh/src/index.ts)
+来源： [`packages/ssh/ssh/src/index.ts:17`](../packages/ssh/ssh/src/index.ts)
 
 <a id="deepseek-aidsh-storage-domain"></a>
 

docs/subsystems/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 docs/subsystems/README.md
-README.md: 9afbdda338ed797279b719eaf577be45a84349e2
-README.zh.md: 78f799d99221d8458ad1639dffbdda69b6783ef6
+README.md: 55777b67508469dcaa0139cddd73b5101616b902
+README.zh.md: f58eef405fa22bcebce4996ec392d5ffc6e87b1b

docs/subsystems/README.md

@@ -30,6 +30,7 @@ One page per subsystem of the DeepSeek Harness: what it is, the data structures
 | [attachment.md](attachment.md) | durable image identity and metadata, validation inputs, verified reads, and the `AttachmentStore` seam |
 | [shell.md](shell.md) | the bash executor seam: `ShellExecRequest`/`Spec`, `ShellRunResult`, background `ShellProcess` handles |
 | [subprocess.md](subprocess.md) | the subprocess seam: fully-explicit `SubprocessSpawnSpec`, offset-based output readers, unclassified `SubprocessOutcome`, and the managed `DSH_*` environment vocabulary |
+| [ssh.md](ssh.md) | the POSIX SSH connection and remote filesystem, subprocess and sandbox providers |
 | [terminal.md](terminal.md) | persistent terminal ids, backend/session contracts, send readiness, bounded reads, and owner-visible snapshots |
 | [sandbox.md](sandbox.md) | per-session policy resolution and the process-confinement seam: file-effect modes, execution/provider policies, `ConfinedArgv`, enforcement and fail-closed errors |
 | [code-runtime.md](code-runtime.md) | the code-execution seam: `CodeRunRequest`/`Result`, binding namespaces, captured logs, the `CodeRunFailure` taxonomy |

docs/subsystems/README.zh.md

@@ -30,6 +30,7 @@
 | [attachment.md](attachment.zh.md) | 持久图片标识与元数据、校验输入、经校验读取，以及 `AttachmentStore` seam |
 | [shell.md](shell.zh.md) | bash 执行器 seam：`ShellExecRequest`/`Spec`、`ShellRunResult`、后台 `ShellProcess` 句柄 |
 | [subprocess.md](subprocess.zh.md) | 子进程 seam：完全显式的 `SubprocessSpawnSpec`、基于偏移的输出读取器、不含分类的 `SubprocessOutcome`，以及受管 `DSH_*` 环境词汇 |
+| [ssh.md](ssh.zh.md) | POSIX SSH 连接与远程文件系统、子进程和沙箱提供方 |
 | [terminal.md](terminal.zh.md) | 持久化终端 ID、后端/会话约定、发送就绪状态、有界读取与 owner 可见快照 |
 | [sandbox.md](sandbox.zh.md) | 每会话策略解析与进程约束 seam：文件效果模式、执行/提供方策略、`ConfinedArgv`、强制执行与故障关闭错误 |
 | [code-runtime.md](code-runtime.zh.md) | 代码执行 seam：`CodeRunRequest`/`Result`、绑定命名空间、捕获日志、`CodeRunFailure` 分类体系 |

docs/subsystems/ssh.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 docs/subsystems/ssh.md
-ssh.md: 427a5b9181cb43e9fc859b879794217fa0637de5
-ssh.zh.md: 572d4729e7e83221d31b99fb4c48a8dfc95895d8
+ssh.md: e7b37a0edc9f00748dcfaa820e0d2739b59d788e
+ssh.zh.md: ec076ffb16d26e2121415c49f5a47a662867db6a

docs/subsystems/ssh.md

@@ -128,7 +128,7 @@ async request<T>(method: string, params: unknown, result: z.ZodType<T>, signal?:
 async connectStream(endpoint: SshStreamEndpoint, signal?: AbortSignal): Promise<Socket>
 
 /** Tear down the helper's remote managed ranges before releasing the SSH master when reachable. */
-async dispose(): Promise<void>
+dispose(): Promise<void>
 ```
 
 Source: [`packages/ssh/ssh/src/index.ts`](../../packages/ssh/ssh/src/index.ts)

docs/subsystems/ssh.zh.md

@@ -128,7 +128,7 @@ async request<T>(method: string, params: unknown, result: z.ZodType<T>, signal?:
 async connectStream(endpoint: SshStreamEndpoint, signal?: AbortSignal): Promise<Socket>
 
 /** Tear down the helper's remote managed ranges before releasing the SSH master when reachable. */
-async dispose(): Promise<void>
+dispose(): Promise<void>
 ```
 
 Source: [`packages/ssh/ssh/src/index.ts`](../../packages/ssh/ssh/src/index.ts)

packages/extensions/tool-cordis/src/api-catalog.ts

@@ -2208,11 +2208,6 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
         description: 'Verified remote helper coordinates; callers must await this before launch.',
         parameters: [],
       },
-      {
-        signature: 'readonly helperPath: string',
-        description: 'Installed helper entry in the remote filesystem.',
-        parameters: [],
-      },
       {
         signature: 'async request<T>(method: string, params: unknown, result: z.ZodType<T>, signal?: AbortSignal, wait: boolean = false): Promise<T>',
         description: 'Send a helper operation; cancellation never replays an ambiguous mutation.',
@@ -2226,7 +2221,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
         returns: 'a paused socket; attach a consumer before resuming it.',
       },
       {
-        signature: 'async dispose(): Promise<void>',
+        signature: 'dispose(): Promise<void>',
         description: 'Tear down the helper\'s remote managed ranges before releasing the SSH master when reachable.',
         parameters: [],
       },

packages/ssh/fs-ssh/package.json

@@ -22,7 +22,7 @@
     "./package.json": "./package.json"
   },
   "files": [
-    "lib/*.js",
+    "lib/index.js",
     "lib/types/**/*.d.ts"
   ],
   "license": "MIT",

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

@@ -6,7 +6,7 @@ import type { SandboxExecutionPolicy, SandboxMode } from '@deepseek-ai/dsh-sandb
 import type {} from '@deepseek-ai/dsh-sandbox-policy'
 import type {} from '@deepseek-ai/dsh-ssh'
 import { RemoteOperationError } from '@deepseek-ai/dsh-ssh/protocol'
-import { editResultSchema, entriesSchema, infoSchema, pathInfoSchema, targetSchema, writeResultSchema } from '@deepseek-ai/dsh-ssh/schemas'
+import { editResultSchema, entriesSchema, infoSchema, pathInfoSchema, targetSchema, textStreamIdSchema, writeResultSchema } from '@deepseek-ai/dsh-ssh/schemas'
 import { z } from 'zod'
 
 const errorCodes: Record<FsErrorCode, true> = {
@@ -51,7 +51,7 @@ export class SshFileSystem extends FileSystem {
   }
 
   override async streamText(target: FsTarget, signal?: AbortSignal): Promise<AsyncIterable<string>> {
-    const id = await this.call('fs.stream', { target }, z.string().uuid(), signal)
+    const id = await this.call('fs.stream', { target }, textStreamIdSchema, signal)
     const call = this.call.bind(this)
     return (async function* () {
       let ended = false
@@ -69,11 +69,11 @@ export class SshFileSystem extends FileSystem {
   }
 
   override async readBytes(target: FsTarget, signal: AbortSignal | undefined, maxBytes: number): Promise<Uint8Array> {
-    return Buffer.from(await this.call('fs.readBytes', { target, maxBytes }, z.string().base64(), signal), 'base64')
+    return Buffer.from(await this.call('fs.readBytes', { target, maxBytes }, z.base64(), signal), 'base64')
   }
 
   override async readByteRange(target: FsTarget, range: { offset: number; length: number }, signal?: AbortSignal): Promise<Uint8Array> {
-    return Buffer.from(await this.call('fs.readRange', { target, ...range }, z.string().base64(), signal), 'base64')
+    return Buffer.from(await this.call('fs.readRange', { target, ...range }, z.base64(), signal), 'base64')
   }
 
   override async listDir(target: FsTarget, signal?: AbortSignal): Promise<FsDirEntry[]> {

packages/ssh/sandbox-ssh/package.json

@@ -22,7 +22,7 @@
     "./package.json": "./package.json"
   },
   "files": [
-    "lib/*.js",
+    "lib/index.js",
     "lib/types/**/*.d.ts"
   ],
   "license": "MIT",

packages/ssh/sandbox-ssh/tests/provider.spec.ts

@@ -79,7 +79,7 @@ describe('SSH sandbox provider', () => {
     const controller = new AbortController()
     const reason = new Error('cancel during confinement')
     state.dispatch.mockImplementationOnce((_method, _params, signal) => new Promise((_resolve, reject) => {
-      signal?.addEventListener('abort', () => { reject(signal.reason) }, { once: true })
+      signal?.addEventListener('abort', () => { reject(reason) }, { once: true })
     }))
     const pending = state.ctx.sandbox.confine(['true'], policy, controller.signal)
     const rejected = expect(pending).rejects.toBe(reason)

packages/ssh/ssh/package.json

@@ -34,7 +34,13 @@
     }
   },
   "files": [
-    "lib/*.js",
+    "lib/index.js",
+    "lib/helper.js",
+    "lib/protocol.js",
+    "lib/schemas.js",
+    "lib/protocol-*.js",
+    "lib/schemas-*.js",
+    "lib/stream-security-*.js",
     "lib/types/**/*.d.ts"
   ],
   "license": "MIT",
@@ -48,7 +54,8 @@
     "@deepseek-ai/dsh-sandbox": "workspace:^",
     "@deepseek-ai/dsh-sandbox-local": "workspace:^",
     "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
-    "@deepseek-ai/dsh-session-projection": "workspace:^"
+    "@deepseek-ai/dsh-session-projection": "workspace:^",
+    "@deepseek-ai/dsh-brand": "workspace:^"
   },
   "dependencies": {
     "@deepseek-ai/schemastery": "workspace:^",
@@ -70,6 +77,7 @@
     "@deepseek-ai/dsh-subprocess-ssh": "workspace:^",
     "@deepseek-ai/dsh-sandbox-ssh": "workspace:^",
     "@deepseek-ai/dsh-lsp": "workspace:^",
-    "@deepseek-ai/dsh-lsp-stdio": "workspace:^"
+    "@deepseek-ai/dsh-lsp-stdio": "workspace:^",
+    "@deepseek-ai/dsh-brand": "workspace:^"
   }
 }

packages/ssh/ssh/src/helper-processes.ts

@@ -7,9 +7,9 @@ import { createServer, type Server } from 'node:tls'
 import type { Duplex, Readable, Writable } from 'node:stream'
 import { finished, pipeline } from 'node:stream/promises'
 import type { Context } from '@deepseek-ai/cordis'
-import type { SubprocessHandle, SubprocessSpawnSpec, SubprocessTerminalHandle, SubprocessTerminalSignal } from '@deepseek-ai/dsh-subprocess'
+import type { SubprocessHandle, SubprocessSpawnSpec, SubprocessTerminalHandle } from '@deepseek-ai/dsh-subprocess'
 import { OutputCollector, prepareManagedProcessBinding } from '@deepseek-ai/dsh-subprocess-local/output'
-import { doneSchema, outputSnapshotFrameLimit, spawnSchema, type SshStreamEndpoint } from './schemas.ts'
+import { doneSchema, outputSnapshotFrameLimit, spawnSchema, type SshProcessId, type SshStreamEndpoint } from './schemas.ts'
 import { z } from 'zod'
 import { SSH_STREAM_TLS_OPTIONS } from './stream-security.ts'
 import { SshRpcPeer } from './protocol.ts'
@@ -37,6 +37,18 @@ interface ProcessRecord {
   expiry: NodeJS.Timeout
 }
 
+/** Join TLS, underlying socket, and listener closure before removing their directory. */
+async function closeEndpoint(endpoint: Endpoint): Promise<void> {
+  const sockets = [...new Set([...(endpoint.socket === undefined ? [] : [endpoint.socket]), ...endpoint.pending])]
+  const closed = sockets.map(socket => new Promise<void>((resolve) => {
+    if (socket.closed) resolve()
+    else socket.once('close', () => { resolve() })
+  }))
+  const listenerClosed = new Promise<void>((resolve) => { endpoint.server.close(() => { resolve() }) })
+  for (const socket of sockets) socket.destroy()
+  await Promise.all([listenerClosed, ...closed])
+}
+
 /** Coalesce live tail updates while capture continues independently of network readers. */
 class CollectedOutputForwarder {
   private readonly peer: SshRpcPeer
@@ -53,18 +65,18 @@ class CollectedOutputForwarder {
     if (this.stopped) return
     this.dirty = true
     if (this.running !== undefined) return
-    this.running = this.flush().catch(() => { this.stopped = true; this.peer.close() }).finally(() => {
-      this.running = undefined
-      if (this.dirty && !this.stopped) this.offer()
-    })
+    this.running = this.flush()
   }
 
   private async flush(): Promise<void> {
-    while (this.dirty && !this.stopped) {
-      this.dirty = false
-      const snapshot = this.collector.snapshot()
-      await this.peer.request('snapshot', { tail: snapshot.bytes.toString('base64'), totalBytes: snapshot.totalBytes }, z.null())
-    }
+    try {
+      while (this.dirty && !this.stopped) {
+        this.dirty = false
+        const snapshot = this.collector.snapshot()
+        await this.peer.request('snapshot', { tail: snapshot.bytes.toString('base64'), totalBytes: snapshot.totalBytes }, z.null())
+      }
+    } catch { this.stopped = true; this.peer.close() }
+    finally { this.running = undefined }
   }
 
   async finish(): Promise<void> {
@@ -76,8 +88,8 @@ class CollectedOutputForwarder {
 
 /** Owns remote launch reservations through final process-range quiescence. */
 export class RemoteProcesses {
-  private readonly records = new Map<string, ProcessRecord>()
-  private readonly completed = new Map<string, unknown>()
+  private readonly records = new Map<SshProcessId, ProcessRecord>()
+  private readonly completed = new Map<SshProcessId, unknown>()
   private closing = false
 
   constructor(
@@ -90,10 +102,10 @@ export class RemoteProcesses {
    * @param raw - untrusted process request received over SSH.
    * @returns the reservation id and authenticated stream coordinates.
    */
-  async prepare(raw: unknown): Promise<{ id: string; streams: Partial<Record<Channel, SshStreamEndpoint>> }> {
+  async prepare(raw: unknown): Promise<{ id: SshProcessId; streams: Partial<Record<Channel, SshStreamEndpoint>> }> {
     if (this.closing || this.records.size >= this.limit) throw new Error('SSH process capacity unavailable')
     const request = spawnSchema.parse(raw)
-    const id = randomUUID()
+    const id = randomUUID() as SshProcessId
     const directory = join(this.root, id)
     const record: ProcessRecord = {
       request, directory, endpoints: {}, controller: new AbortController(),
@@ -130,7 +142,7 @@ export class RemoteProcesses {
    * @param signal - cancellation of pending process publication.
    * @returns the terminal pid when the request owns a PTY.
    */
-  async start(id: string, signal?: AbortSignal): Promise<{ pid?: number }> {
+  async start(id: SshProcessId, signal?: AbortSignal): Promise<{ pid?: number }> {
     const record = this.record(id)
     if (record.start !== undefined) throw new Error('SSH process launch was already requested')
     signal?.throwIfAborted()
@@ -141,7 +153,7 @@ export class RemoteProcesses {
     return record.terminal === undefined ? {} : { pid: record.terminal.pid }
   }
 
-  private async startOnce(id: string, record: ProcessRecord): Promise<void> {
+  private async startOnce(id: SshProcessId, record: ProcessRecord): Promise<void> {
     await Promise.all(Object.values(record.endpoints).map(endpoint => endpoint.connected))
     clearTimeout(record.expiry)
     if (this.closing) throw new Error('SSH helper is closing')
@@ -253,7 +265,7 @@ export class RemoteProcesses {
    * @param id - the started process reservation.
    * @returns the exit observation and remote spill paths.
    */
-  async done(id: string): Promise<unknown> {
+  async done(id: SshProcessId): Promise<unknown> {
     if (this.completed.has(id)) return this.completed.get(id)
     const record = this.record(id)
     if (record.start === undefined) throw new Error('SSH process has not started')
@@ -267,7 +279,7 @@ export class RemoteProcesses {
    * @param signal - cancellation of this observation, leaving ownership intact.
    * @returns whether the owned process range is empty.
    */
-  async wait(id: string, signal?: AbortSignal): Promise<boolean> {
+  async wait(id: SshProcessId, signal?: AbortSignal): Promise<boolean> {
     if (this.completed.has(id)) return true
     const record = this.record(id)
     await record.start
@@ -280,7 +292,7 @@ export class RemoteProcesses {
    * Terminate and await the managed range independently of output readers.
    * @param id - the process reservation to stop.
    */
-  async terminate(id: string): Promise<void> {
+  async terminate(id: SshProcessId): Promise<void> {
     if (this.completed.has(id)) return
     const record = this.record(id)
     record.controller.abort(new Error('SSH process termination requested'))
@@ -297,12 +309,12 @@ export class RemoteProcesses {
    * @param value - input bytes as text or the signal name.
    * @returns the operation's wire result.
    */
-  async terminal(id: string, operation: 'write' | 'inspect' | 'signal', value?: string): Promise<unknown> {
+  async terminal(id: SshProcessId, operation: 'write' | 'inspect' | 'signal', value?: string): Promise<unknown> {
     const terminal = this.record(id).terminal
     if (terminal === undefined) throw new Error('SSH handle does not own a terminal')
     if (operation === 'write') { await terminal.write(z.string().parse(value)); return null }
     if (operation === 'inspect') return await terminal.inspectForeground() ?? null
-    return terminal.signalForeground(z.enum(['SIGINT', 'SIGTERM', 'SIGKILL', 'SIGTSTP', 'SIGHUP']).parse(value) as SubprocessTerminalSignal)
+    return terminal.signalForeground(z.enum(['SIGINT', 'SIGTERM', 'SIGKILL', 'SIGTSTP', 'SIGHUP']).parse(value))
   }
 
   /** Stop every owned process on lease expiry or disconnect. */
@@ -314,18 +326,14 @@ export class RemoteProcesses {
     if (errors.length > 0) throw new AggregateError(errors, 'SSH remote process cleanup failed')
   }
 
-  private async release(id: string): Promise<void> {
+  private async release(id: SshProcessId): Promise<void> {
     const record = this.records.get(id)
     if (record === undefined) return
     record.release ??= (async () => {
       clearTimeout(record.expiry)
       record.controller.abort(new Error('SSH process reservation closed'))
       await record.preparing?.catch(() => {})
-      for (const endpoint of Object.values(record.endpoints)) {
-        for (const socket of endpoint.pending) socket.destroy()
-        endpoint.socket?.destroy()
-        endpoint.server.close()
-      }
+      await Promise.all(Object.values(record.endpoints).map(closeEndpoint))
       await record.start?.catch(() => {})
       record.ordinary?.terminate()
       if (record.ordinary !== undefined) await record.ordinary.waitForExit()
@@ -336,22 +344,18 @@ export class RemoteProcesses {
     await record.release
   }
 
-  private record(id: string): ProcessRecord {
+  private record(id: SshProcessId): ProcessRecord {
     const record = this.records.get(id)
     if (record === undefined) throw new Error('Unknown or expired SSH process handle')
     return record
   }
 
-  private async rememberCompleted(id: string, record: ProcessRecord, result: unknown): Promise<void> {
+  private async rememberCompleted(id: SshProcessId, record: ProcessRecord, result: unknown): Promise<void> {
     if (this.records.get(id) !== record) return
     this.records.delete(id)
     this.completed.set(id, result)
-    if (this.completed.size > this.limit * 4) this.completed.delete(this.completed.keys().next().value as string)
-    for (const endpoint of Object.values(record.endpoints)) {
-      endpoint.server.close()
-      for (const socket of endpoint.pending) socket.destroy()
-      endpoint.socket?.destroy()
-    }
+    if (this.completed.size > this.limit * 4) this.completed.delete(this.completed.keys().next().value as SshProcessId)
+    await Promise.all(Object.values(record.endpoints).map(closeEndpoint))
     await rm(record.directory, { recursive: true, force: true })
   }
 
@@ -365,7 +369,7 @@ export class RemoteProcesses {
     const endpoint: Endpoint = { path, capability: capability.toString('hex'), server, connected: connected.promise, pending: new Set() }
     server.maxConnections = 8
     void connected.promise.catch(() => {})
-    server.on('connection', (socket) => {
+    server.on('connection', (socket: Socket) => {
       endpoint.pending.add(socket)
       socket.once('close', () => { endpoint.pending.delete(socket) })
       socket.on('error', () => {})
@@ -387,9 +391,7 @@ export class RemoteProcesses {
       await chmod(path, 0o600)
       return endpoint
     } catch (error) {
-      for (const socket of endpoint.pending) socket.destroy()
-      endpoint.socket?.destroy()
-      await new Promise<void>((resolve) => { server.close(() => { resolve() }) })
+      await closeEndpoint(endpoint)
       throw error
     }
   }

packages/ssh/ssh/src/helper.ts

@@ -4,7 +4,7 @@ import { readFileSync } from 'node:fs'
 import { mkdtemp, rm } from 'node:fs/promises'
 import type { Readable, Writable } from 'node:stream'
 import { Context } from '@deepseek-ai/cordis'
-import { FsError, type FsTarget, type FsEditRequest, type FsWriteIntent, type FsVersion } from '@deepseek-ai/dsh-fs'
+import { FsError, type FsTarget, type FsWriteIntent, type FsVersion } from '@deepseek-ai/dsh-fs'
 import { SandboxedFileSystem } from '@deepseek-ai/dsh-fs-sandbox'
 import { LocalSubprocessRuntime } from '@deepseek-ai/dsh-subprocess-local'
 import { LocalSandboxProvider } from '@deepseek-ai/dsh-sandbox-local'
@@ -14,12 +14,14 @@ import type { SandboxExecutionPolicy, SandboxPolicy } from '@deepseek-ai/dsh-san
 import { z } from 'zod'
 import { SshRpcPeer, SSH_MAX_PROCESS_HANDLES, SSH_MAX_TEXT_STREAMS, SSH_PROTOCOL_VERSION } from './protocol.ts'
 import { RemoteProcesses } from './helper-processes.ts'
-import { editSchema, environmentSchema, intentSchema, policySchema, remotePath, targetSchema } from './schemas.ts'
+import { editSchema, environmentSchema, intentSchema, policySchema, processIdSchema, remotePath, targetSchema, textStreamIdSchema } from './schemas.ts'
+import type { SshTextStreamId } from './schemas.ts'
 
 const MAX_FRAME_BYTES = 64 * 1024 * 1024
 const MAX_TEXT_BYTES = 8 * 1024 * 1024
 const object = z.object({}).strict()
-const idRequest = z.object({ id: z.string().uuid() }).strict()
+const processIdRequest = z.object({ id: processIdSchema }).strict()
+const textStreamIdRequest = z.object({ id: textStreamIdSchema }).strict()
 
 async function services() {
   const ctx = new Context()
@@ -55,7 +57,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
   const root = await mkdtemp('/tmp/dsh-ssh-')
   const processes = new RemoteProcesses(ctx, root, SSH_MAX_PROCESS_HANDLES, 30_000)
   const lifetime = new AbortController()
-  const iterators = new Map<string, { iterator: AsyncIterator<string>; controller: AbortController }>()
+  const iterators = new Map<SshTextStreamId, { iterator: AsyncIterator<string>; controller: AbortController }>()
   let lease: NodeJS.Timeout | undefined
   let leaseMs = 30_000
   let initialized = false
@@ -66,7 +68,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
       if (lease !== undefined) clearTimeout(lease)
       lifetime.abort(new Error('SSH helper is closing'))
       for (const record of iterators.values()) record.controller.abort(lifetime.signal.reason)
-      await Promise.allSettled([...iterators.values()].map(record => record.iterator.return?.()))
+      await Promise.allSettled([...iterators.values()].map(async record => record.iterator.return?.()))
       iterators.clear()
       try { await processes.close() }
       finally {
@@ -108,12 +110,12 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
     if (method === 'heartbeat') { object.parse(raw); touchLease(); return null }
     if (method === 'close') { object.parse(raw); await close(); return null }
     if (method === 'process.prepare') return processes.prepare(raw)
-    if (method === 'process.start') return processes.start(idRequest.parse(raw).id, signal)
-    if (method === 'process.done') return processes.done(idRequest.parse(raw).id)
-    if (method === 'process.wait') return processes.wait(idRequest.parse(raw).id, signal)
-    if (method === 'process.terminate') { await processes.terminate(idRequest.parse(raw).id); return null }
+    if (method === 'process.start') return processes.start(processIdRequest.parse(raw).id, signal)
+    if (method === 'process.done') return processes.done(processIdRequest.parse(raw).id)
+    if (method === 'process.wait') return processes.wait(processIdRequest.parse(raw).id, signal)
+    if (method === 'process.terminate') { await processes.terminate(processIdRequest.parse(raw).id); return null }
     if (method === 'terminal.write' || method === 'terminal.inspect' || method === 'terminal.signal') {
-      const input = z.object({ id: z.string().uuid(), value: z.string().optional() }).strict().parse(raw)
+      const input = z.object({ id: processIdSchema, value: z.string().optional() }).strict().parse(raw)
       return processes.terminal(input.id, method === 'terminal.write' ? 'write' : method === 'terminal.inspect' ? 'inspect' : 'signal', input.value)
     }
     if (method === 'executable') {
@@ -145,7 +147,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
         try {
           signal.throwIfAborted()
           if (iterators.size >= SSH_MAX_TEXT_STREAMS) throw new Error('SSH text stream limit reached')
-          const id = randomUUID()
+          const id = randomUUID() as SshTextStreamId
           iterators.set(id, { iterator, controller })
           return id
         } catch (error) {
@@ -165,7 +167,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
       return text
     }
     if (method === 'fs.next' || method === 'fs.streamClose') {
-      const { id } = idRequest.parse(raw)
+      const { id } = textStreamIdRequest.parse(raw)
       const record = iterators.get(id)
       if (record === undefined) throw new Error('Unknown SSH text stream')
       if (method === 'fs.streamClose') {
@@ -180,7 +182,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
         signal.throwIfAborted()
         const next = await record.iterator.next()
         if (next.done) iterators.delete(id)
-        return { done: next.done ?? false, value: next.value ?? '' }
+        return { done: next.done ?? false, value: next.done ? '' : next.value }
       } catch (error) {
         iterators.delete(id)
         record.controller.abort(error)
@@ -214,7 +216,7 @@ export async function runSshHelper(transport: HelperTransport): Promise<void> {
         input.expected === undefined ? undefined : intentSchema.parse(input.expected) as FsWriteIntent, signal, resolved,
       )
       return ctx.fs.editText(
-        target, editSchema.parse(input.edit) as FsEditRequest,
+        target, editSchema.parse(input.edit),
         input.expected === undefined
           ? undefined : z.object({ version: z.string() }).strict().parse(input.expected) as { version: FsVersion },
         signal, resolved,

packages/ssh/ssh/src/index.ts

@@ -83,7 +83,8 @@ export class SshConnection extends Service {
     }).refine(value => (value.bootstrapPath === undefined) === (value.bootstrapHash === undefined), 'bootstrapPath and bootstrapHash must be paired')
       .parse(config) as typeof this.config
     this.ready = this.start()
-    void this.ready.catch((error: unknown) => { this.fail(error instanceof Error ? error : new Error(String(error))) })
+    // Startup uses Node I/O, local validation, and Error-valued RPC failures.
+    void this.ready.catch((error: unknown) => { this.fail(error as Error) })
     ctx.effect(() => () => this.dispose())
   }
 
@@ -149,7 +150,7 @@ export class SshConnection extends Service {
       await this.controlCommand(['-O', 'forward', '-o', 'ExitOnForwardFailure=yes', '-L', forward], signal)
     } catch (error) { await cancelForward(); throw error }
     signal.throwIfAborted()
-    const socket = createConnection({ path: local, allowHalfOpen: true, signal })
+    const socket = createConnection({ path: local, allowHalfOpen: true })
     this.sockets.add(socket)
     socket.once('close', () => {
       this.sockets.delete(socket)
@@ -157,6 +158,7 @@ export class SshConnection extends Service {
     })
     await new Promise<void>((resolve, reject) => {
       const cleanup = (): void => {
+        signal.removeEventListener('abort', aborted)
         socket.off('connect', connected)
         socket.off('error', failed)
         socket.off('close', closed)
@@ -164,11 +166,13 @@ export class SshConnection extends Service {
       const connected = (): void => { cleanup(); resolve() }
       const failed = (error: Error): void => { cleanup(); reject(error) }
       const closed = (): void => { failed(new Error('SSH connection closed before stream establishment')) }
+      const aborted = (): void => { socket.destroy(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason))) }
       socket.once('connect', connected)
       socket.once('error', failed)
       socket.once('close', closed)
+      signal.addEventListener('abort', aborted, { once: true })
     })
-    const authenticated = await authenticateStream(socket, endpoint.capability, this.config.requestTimeoutMs)
+    const authenticated = await authenticateStream(socket, endpoint.capability, this.config.requestTimeoutMs, signal)
     this.sockets.add(authenticated)
     authenticated.on('error', () => { authenticated.destroy() })
     authenticated.once('close', () => { this.sockets.delete(authenticated) })
@@ -190,7 +194,8 @@ export class SshConnection extends Service {
       if (this.failure === undefined) await this.rpc?.request('close', {}, z.null(), AbortSignal.timeout(this.config.requestTimeoutMs))
     } finally {
       this.rpc?.close()
-      const socketClosures = [...this.sockets].map(socket => new Promise<void>((resolve) => {
+      // TLS wrappers release their reads before their underlying sockets close.
+      const socketClosures = [...this.sockets].reverse().map(socket => new Promise<void>((resolve) => {
         if (socket.closed) resolve()
         else { socket.once('close', () => { resolve() }); socket.destroy() }
       }))
@@ -232,7 +237,6 @@ export class SshConnection extends Service {
       force.unref()
     }
     combined.addEventListener('abort', escalate, { once: true })
-    if (combined.aborted) escalate()
     try { await result.promise }
     finally {
       await closed
@@ -247,7 +251,7 @@ export class SshConnection extends Service {
     this.lifetime.abort(error)
     if (this.heartbeat !== undefined) clearInterval(this.heartbeat)
     this.rpc?.close(error)
-    for (const socket of this.sockets) socket.destroy(error)
+    for (const socket of [...this.sockets].reverse()) socket.destroy(error)
     this.child?.kill('SIGTERM')
   }
 
@@ -279,7 +283,7 @@ export class SshConnection extends Service {
     let heartbeatPending: Promise<unknown> | undefined
     this.heartbeat = setInterval(() => {
       heartbeatPending ??= rpc.request('heartbeat', {}, z.null(), AbortSignal.timeout(this.config.leaseMs / 2))
-        .catch((error: unknown) => { this.fail(error instanceof Error ? error : new Error(String(error))) })
+        .catch((error: unknown) => { this.fail(error as Error) })
         .finally(() => { heartbeatPending = undefined })
     }, Math.floor(this.config.leaseMs / 3))
     this.heartbeat.unref()

packages/ssh/ssh/src/protocol.ts

@@ -4,6 +4,7 @@ import { randomUUID } from 'node:crypto'
 import { EventEmitter } from 'node:events'
 import type { Readable, Writable } from 'node:stream'
 import { z } from 'zod'
+import type { Branded } from '@deepseek-ai/dsh-brand'
 
 /** Wire version shared by the installed helper and client package. */
 export const SSH_PROTOCOL_VERSION = 1
@@ -30,11 +31,13 @@ function requestClass(method: string): RequestClass {
 }
 
 const errorSchema = z.object({ name: z.string(), message: z.string(), code: z.string().optional() }).strict()
+type SshRpcRequestId = Branded<'SshRpcRequestId'>
+const requestIdSchema = z.string().transform((value): SshRpcRequestId => value as SshRpcRequestId)
 const frameSchema = z.discriminatedUnion('type', [
-  z.object({ type: z.literal('request'), id: z.string(), method: z.string(), params: z.unknown() }).strict(),
-  z.object({ type: z.literal('result'), id: z.string(), value: z.unknown() }).strict(),
-  z.object({ type: z.literal('error'), id: z.string(), error: errorSchema }).strict(),
-  z.object({ type: z.literal('cancel'), id: z.string() }).strict(),
+  z.object({ type: z.literal('request'), id: requestIdSchema, method: z.string(), params: z.unknown() }).strict(),
+  z.object({ type: z.literal('result'), id: requestIdSchema, value: z.unknown() }).strict(),
+  z.object({ type: z.literal('error'), id: requestIdSchema, error: errorSchema }).strict(),
+  z.object({ type: z.literal('cancel'), id: requestIdSchema }).strict(),
 ])
 type Frame = z.infer<typeof frameSchema>
 type RequestHandler = (method: string, params: unknown, signal: AbortSignal) => Promise<unknown>
@@ -53,8 +56,12 @@ export class RemoteOperationError extends Error {
 
 /** The peer owns pending calls and rejects ambiguous operations on connection loss; it never replays requests. */
 export class SshRpcPeer extends EventEmitter {
-  private readonly pending = new Map<string, { resolve(value: unknown): void; reject(error: Error): void; requestClass: RequestClass }>()
-  private readonly active = new Map<string, { controller: AbortController; requestClass: RequestClass }>()
+  private readonly pending = new Map<SshRpcRequestId, {
+    resolve(value: unknown): void
+    reject(error: Error): void
+    requestClass: RequestClass
+  }>()
+  private readonly active = new Map<SshRpcRequestId, { controller: AbortController; requestClass: RequestClass }>()
   private writeTail = Promise.resolve()
   private queuedBytes = 0
   private failure: Error | undefined
@@ -86,7 +93,7 @@ export class SshRpcPeer extends EventEmitter {
     if (this.failure !== undefined) throw this.failure
     const kind = requestClass(method)
     if (this.atCapacity(kind, this.pending.values())) throw new Error('SSH helper pending request limit reached')
-    const id = randomUUID()
+    const id = randomUUID() as SshRpcRequestId
     const result = Promise.withResolvers<unknown>()
     void result.promise.catch(() => {})
     this.pending.set(id, { ...result, requestClass: kind })

packages/ssh/ssh/src/schemas.ts

@@ -1,5 +1,15 @@
 /** Strict JSON validation for SSH helper requests and remote observations. */
 import { z } from 'zod'
+import type { Branded } from '@deepseek-ai/dsh-brand'
+
+/** Identity of one prepared or running process in its owning SSH helper. */
+export type SshProcessId = Branded<'SshProcessId'>
+/** Identity of one open text iterator in its owning SSH helper. */
+export type SshTextStreamId = Branded<'SshTextStreamId'>
+/** Admit a process identity from the private helper protocol. */
+export const processIdSchema = z.uuid().transform((value): SshProcessId => value as SshProcessId)
+/** Admit a text iterator identity from the private helper protocol. */
+export const textStreamIdSchema = z.uuid().transform((value): SshTextStreamId => value as SshTextStreamId)
 
 /** A remote POSIX absolute path; spelling is preserved until remote canonicalization. */
 export const remotePath = z.string().min(1).refine(value => value.startsWith('/') && !value.includes('\0'), 'expected an absolute POSIX path')
@@ -47,11 +57,11 @@ export const streamEndpointSchema = z.object({ path: remotePath, capability: z.s
 /** A stream capability reaches only the authenticated SSH client and its helper. */
 export type SshStreamEndpoint = z.infer<typeof streamEndpointSchema>
 /** Prepared process and its independently authenticated stream endpoints. */
-export const preparedSchema = z.object({ id: z.string().uuid(), streams: z.partialRecord(z.enum(['stdin', 'stdout', 'stderr', 'control', 'terminal']), streamEndpointSchema) }).strict()
+export const preparedSchema = z.object({ id: processIdSchema, streams: z.partialRecord(z.enum(['stdin', 'stdout', 'stderr', 'control', 'terminal']), streamEndpointSchema) }).strict()
 /** Direct process exit facts. */
 export const outcomeSchema = z.object({ exitCode: z.number().int().nullable(), signal: z.string().nullable() }).strict()
 /** A bounded raw tail positioned in whole-stream byte coordinates. */
-export const outputSnapshotSchema = z.object({ tail: z.string().base64(), totalBytes: z.number().int().nonnegative() }).strict()
+export const outputSnapshotSchema = z.object({ tail: z.base64(), totalBytes: z.number().int().nonnegative() }).strict()
 /**
  * Bound one encoded tail and its RPC envelope by the declared collection budget.
  * @param maxBytes - the collector's retained raw-byte limit.

packages/ssh/ssh/src/stream-security.ts

@@ -12,15 +12,20 @@ export const SSH_STREAM_TLS_OPTIONS = {
  * @param socket - the connected OpenSSH forwarding socket.
  * @param capability - the per-stream 256-bit key encoded as hexadecimal.
  * @param timeoutMs - deadline for completing TLS authentication.
+ * @param signal - cancellation of authentication and the resulting TLS stream.
  * @returns an authenticated paused stream; the key is never transmitted as data.
  */
-export async function authenticateStream(socket: Socket, capability: string, timeoutMs: number): Promise<TLSSocket> {
+export async function authenticateStream(socket: Socket, capability: string, timeoutMs: number, signal?: AbortSignal): Promise<TLSSocket> {
+  if (signal?.aborted) { socket.destroy(); signal.throwIfAborted() }
   const stream = connect({
     ...SSH_STREAM_TLS_OPTIONS, socket, rejectUnauthorized: true,
     pskCallback: () => ({ psk: Buffer.from(capability, 'hex'), identity: 'dsh-stream' }),
     // PSK proves peer identity without an X.509 certificate or hostname.
     checkServerIdentity: () => undefined,
   })
+  const abort = (): void => { stream.destroy(signal?.reason instanceof Error ? signal.reason : new Error(String(signal?.reason))) }
+  signal?.addEventListener('abort', abort, { once: true })
+  stream.once('close', () => { signal?.removeEventListener('abort', abort) })
   try {
     await new Promise<void>((resolve, reject) => {
       const cleanup = (): void => {

packages/ssh/ssh/tests/connection-disposal.spec.ts

@@ -225,7 +225,7 @@ describe.skipIf(process.platform === 'win32')('SSH stream establishment disposal
     socket.destroy()
     const result = await observed
     expect(result).toHaveProperty('error')
-    expect(String('error' in result ? result.error : '')).toMatch(/closed|lost|refused/)
+    expect(String('error' in result ? result.error : '')).toMatch(/clos(?:ed|ing)|lost|refused/)
     expect(errors).toEqual([])
     expect(socket.listenerCount('connect')).toBe(0)
     expect(socket.listenerCount('data')).toBe(0)
@@ -238,7 +238,7 @@ describe.skipIf(process.platform === 'win32')('SSH stream establishment disposal
     await service.dispose()
     const result = await observed
     expect(result).toHaveProperty('error')
-    expect(String('error' in result ? result.error : '')).toMatch(/closed|lost|refused/)
+    expect(String('error' in result ? result.error : '')).toMatch(/clos(?:ed|ing)|lost|refused/)
     expect(socket.destroyed).toBe(true)
     expect(socket.listenerCount('connect')).toBe(0)
     expect(socket.listenerCount('data')).toBe(0)

packages/ssh/ssh/tests/fixtures/helper.ts

@@ -23,7 +23,7 @@ export async function createHelperHarness(handshake = true, leaseMs = 30_000) {
   let closing: Promise<void> | undefined
   const close = () => {
     closing ??= (async () => {
-      for (const socket of sockets) socket.destroy()
+      for (const socket of [...sockets].reverse()) socket.destroy()
       client.close()
       controller.abort()
       try { await serving } finally { await rm(root, { recursive: true, force: true }) }

packages/ssh/ssh/tests/helper-processes.spec.ts

@@ -0,0 +1,345 @@
+/** Process reservations retain native cleanup ownership across stream and allocation failures. */
+import { once } from 'node:events'
+import { chmod, mkdir, mkdtemp, readdir, rm } from 'node:fs/promises'
+import { createConnection, type Socket } from 'node:net'
+import { PassThrough } from 'node:stream'
+import { Context } from '@deepseek-ai/cordis'
+import type { SubprocessHandle, SubprocessOutcome, SubprocessTerminalHandle } from '@deepseek-ai/dsh-subprocess'
+import { describe, expect, it, vi } from 'vitest'
+import { RemoteProcesses } from '../src/helper-processes.ts'
+import type { SshProcessId, SshStreamEndpoint } from '../src/schemas.ts'
+import { authenticateStream } from '../src/stream-security.ts'
+import { SshRpcPeer } from '../src/protocol.ts'
+import { outputSnapshotSchema } from '../src/schemas.ts'
+
+vi.mock('node:fs/promises', async (original) => {
+  const actual = await original<typeof import('node:fs/promises')>()
+  return { ...actual, chmod: vi.fn(actual.chmod), rm: vi.fn(actual.rm) }
+})
+
+const outcome = { exitCode: 0, signal: null }
+const ordinaryRequest = { argv: ['target'], cwd: '/tmp', graceMs: 20, stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' } }
+const terminalRequest = { argv: ['target'], cwd: '/tmp', graceMs: 20, terminal: { rows: 24, cols: 80 } }
+
+async function harness() {
+  const root = await mkdtemp('/tmp/dsh-ssh-owner-')
+  const ctx = new Context()
+  const resolve = vi.fn(async () => ({ targetKey: root, displayPath: root }))
+  const spawn = vi.fn<(spec: unknown) => SubprocessHandle>()
+  const spawnTerminal = vi.fn<(spec: unknown) => Promise<SubprocessTerminalHandle>>()
+  ctx.provide('fs', { resolve, processPath: () => root } as never)
+  ctx.provide('subprocess', { spawn, spawnTerminal } as never)
+  const owner = new RemoteProcesses(ctx, root, 1, 5000)
+  const sockets = new Set<Socket>()
+  const connect = async (endpoint: SshStreamEndpoint) => {
+    const raw = createConnection({ path: endpoint.path, allowHalfOpen: true })
+    sockets.add(raw)
+    raw.on('error', () => {})
+    await once(raw, 'connect')
+    const socket = await authenticateStream(raw, endpoint.capability, 5000)
+    sockets.add(socket)
+    socket.on('error', () => {})
+    return socket
+  }
+  const prepare = async (request: unknown) => {
+    const prepared = await owner.prepare(request)
+    const entries = await Promise.all(Object.entries(prepared.streams).map(async ([name, endpoint]) => [name, await connect(endpoint)]))
+    const channels = Object.fromEntries(entries) as Record<string, Socket>
+    return { id: prepared.id, channels }
+  }
+  const close = async () => {
+    for (const socket of [...sockets].reverse()) socket.destroy()
+    try { await owner.close() } finally { await rm(root, { recursive: true, force: true }) }
+  }
+  return { root, owner, resolve, spawn, spawnTerminal, prepare, close }
+}
+
+function ordinary(control = false) {
+  const completion = Promise.withResolvers<SubprocessOutcome>()
+  const stdin = new PassThrough()
+  const stdout = new PassThrough()
+  const stderr = new PassThrough()
+  const channel = control ? new PassThrough() : undefined
+  const finish = () => { stdout.end(); stderr.end(); channel?.end(); completion.resolve(outcome) }
+  const terminate = vi.fn(finish)
+  const waitForExit = vi.fn(async () => { await completion.promise; return true })
+  const handle: SubprocessHandle = {
+    stdin, stdout, stderr, control: channel, collected: {}, done: completion.promise, terminate, waitForExit,
+  }
+  return { handle, finish, completion, stdin, stdout, stderr, channel, terminate, waitForExit }
+}
+
+function terminal() {
+  const output = new PassThrough()
+  const completion = Promise.withResolvers<SubprocessOutcome>()
+  const write = vi.fn(async (_data: string) => {})
+  const inspectForeground = vi.fn<SubprocessTerminalHandle['inspectForeground']>(async () => undefined)
+  const signalForeground = vi.fn(async () => 42)
+  const terminate = vi.fn(async () => { output.end(); completion.resolve(outcome) })
+  const handle: SubprocessTerminalHandle = {
+    pid: 42, output, done: completion.promise, write, inspectForeground, signalForeground, terminate,
+  }
+  return { handle, completion, output, write, inspectForeground, signalForeground, terminate }
+}
+
+describe.skipIf(process.platform === 'win32')('SSH helper process settlement', () => {
+  it('forwards terminal operations and retains a bounded completed-result cache', async () => {
+    const test = await harness()
+    const ids: SshProcessId[] = []
+    try {
+      for (let index = 0; index < 5; index++) {
+        const child = terminal()
+        test.spawnTerminal.mockResolvedValueOnce(child.handle)
+        const run = await test.prepare({ ...terminalRequest, env: { PRESENT: 'value', REMOVED: null } })
+        ids.push(run.id)
+        run.channels.terminal!.end()
+        run.channels.terminal!.resume()
+        expect(await test.owner.start(run.id)).toEqual({ pid: 42 })
+        expect(test.spawnTerminal).toHaveBeenLastCalledWith(expect.objectContaining({ env: { PRESENT: 'value' }, rows: 24, cols: 80 }))
+        await expect(test.owner.start(run.id)).rejects.toThrow('already requested')
+        await test.owner.terminal(run.id, 'write', 'input')
+        expect(child.write).toHaveBeenCalledWith('input')
+        expect(await test.owner.terminal(run.id, 'inspect')).toBeNull()
+        child.inspectForeground.mockResolvedValueOnce({ processGroupId: 42, inputWaiting: true })
+        expect(await test.owner.terminal(run.id, 'inspect')).toEqual({ processGroupId: 42, inputWaiting: true })
+        expect(await test.owner.terminal(run.id, 'signal', 'SIGINT')).toBe(42)
+        expect(child.signalForeground).toHaveBeenCalledWith('SIGINT')
+        expect(await test.owner.wait(run.id)).toBe(true)
+        await expect(test.owner.done(run.id)).resolves.toEqual({ outcome, spills: {}, collected: {} })
+        await vi.waitFor(async () => { expect(await readdir(test.root)).toEqual([]) })
+        expect(await test.owner.done(run.id)).toEqual({ outcome, spills: {}, collected: {} })
+        expect(await test.owner.wait(run.id)).toBe(true)
+        await test.owner.terminate(run.id)
+      }
+      await expect(test.owner.done(ids[0]!)).rejects.toThrow('Unknown or expired')
+    } finally { await test.close() }
+  })
+
+  it('joins a terminal allocated after its start request is cancelled', async () => {
+    const test = await harness()
+    const allocated = Promise.withResolvers<SubprocessTerminalHandle>()
+    const entered = Promise.withResolvers<undefined>()
+    const child = terminal()
+    test.spawnTerminal.mockImplementation(() => { entered.resolve(undefined); return allocated.promise })
+    try {
+      const run = await test.prepare(terminalRequest)
+      const controller = new AbortController()
+      const started = test.owner.start(run.id, controller.signal)
+      const rejected = expect(started).rejects.toThrow('cancelled allocation')
+      await entered.promise
+      controller.abort(new Error('cancelled allocation'))
+      allocated.resolve(child.handle)
+      await rejected
+      expect(child.terminate).toHaveBeenCalledOnce()
+      await test.owner.terminate(run.id)
+      expect(child.terminate).toHaveBeenCalledTimes(2)
+    } finally { await test.close() }
+  })
+
+  it('pipes stdin bytes and distinguishes an unstarted reservation from a completed process', async () => {
+    const test = await harness()
+    const child = ordinary()
+    test.spawn.mockReturnValue(child.handle)
+    try {
+      const run = await test.prepare({ ...ordinaryRequest, env: { VALUE: 'kept', REMOVED: null }, stdio: { ...ordinaryRequest.stdio, stdin: 'pipe' } })
+      await expect(test.owner.done(run.id)).rejects.toThrow('has not started')
+      await expect(test.owner.wait(run.id)).rejects.toThrow('was not started')
+      const bytes: Buffer[] = []
+      child.stdin.on('data', (chunk: Buffer) => { bytes.push(Buffer.from(chunk)) })
+      const ended = once(child.stdin, 'end')
+      await test.owner.start(run.id)
+      expect(test.spawn).toHaveBeenCalledWith(expect.objectContaining({ env: { VALUE: 'kept', REMOVED: undefined } }))
+      run.channels.stdout!.end(); run.channels.stdout!.resume()
+      run.channels.stderr!.end(); run.channels.stderr!.resume()
+      run.channels.stdin!.end('stdin bytes')
+      await ended
+      expect(Buffer.concat(bytes).toString()).toBe('stdin bytes')
+      await test.owner.terminate(run.id)
+      expect(await test.owner.done(run.id)).toEqual({ outcome, spills: {}, collected: {} })
+      expect(child.waitForExit).toHaveBeenCalled()
+    } finally { await test.close() }
+  })
+
+  it('rejects a missing native control descriptor and still reaps the native process', async () => {
+    const test = await harness()
+    const child = ordinary()
+    test.spawn.mockReturnValue(child.handle)
+    try {
+      const run = await test.prepare({ ...ordinaryRequest, stdio: { ...ordinaryRequest.stdio, control: 'pipe' } })
+      await expect(test.owner.start(run.id)).rejects.toThrow('did not establish fd 7')
+      await test.owner.terminate(run.id)
+      expect(child.terminate).toHaveBeenCalled()
+      expect(child.waitForExit).toHaveBeenCalled()
+    } finally { await test.close() }
+  })
+
+  it('closes an unpublished listener when filesystem permissions cannot be established', async () => {
+    const test = await harness()
+    try {
+      vi.mocked(chmod).mockRejectedValueOnce(Object.assign(new Error('listener chmod denied'), { code: 'EACCES' }))
+      await expect(test.owner.prepare(ordinaryRequest)).rejects.toThrow('listener chmod denied')
+      expect(await readdir(test.root)).toEqual([])
+      const run = await test.prepare(ordinaryRequest)
+      await test.owner.terminate(run.id)
+      expect(await readdir(test.root)).toEqual([])
+    } finally { await test.close() }
+  })
+
+  it('rolls back a listener path that exceeds the operating system socket limit', async () => {
+    const root = await mkdtemp('/tmp/dsh-ssh-long-')
+    const longRoot = `${root}/${'x'.repeat(140)}`
+    await mkdir(longRoot)
+    const owner = new RemoteProcesses(new Context(), longRoot, 1, 5000)
+    try {
+      await expect(owner.prepare(ordinaryRequest)).rejects.toThrow()
+    } finally { await owner.close(); await rm(root, { recursive: true, force: true }) }
+  })
+
+  it('continues collecting while a live snapshot waits for acknowledgement', async () => {
+    const test = await harness()
+    const child = ordinary()
+    test.spawn.mockReturnValue(child.handle)
+    const entered = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    let peer: SshRpcPeer | undefined
+    try {
+      const run = await test.prepare({ ...ordinaryRequest, stdio: { ...ordinaryRequest.stdio, stdout: { maxBytes: 64 } } })
+      peer = new SshRpcPeer(run.channels.stdout!, run.channels.stdout!, 4096, 1, async (method, raw) => {
+        expect(method).toBe('snapshot')
+        outputSnapshotSchema.parse(raw)
+        entered.resolve(undefined)
+        await release.promise
+        return null
+      })
+      run.channels.stderr!.end(); run.channels.stderr!.resume()
+      await test.owner.start(run.id)
+      child.stdout.write('first')
+      await entered.promise
+      child.stdout.write('second')
+      child.finish()
+      expect(await test.owner.done(run.id)).toMatchObject({ collected: { stdout: { totalBytes: 11, tail: Buffer.from('firstsecond').toString('base64') } } })
+      release.resolve(undefined)
+      await vi.waitFor(async () => { expect(await readdir(test.root)).toEqual([]) })
+    } finally { release.resolve(undefined); peer?.close(); await test.close() }
+  })
+
+  it('keeps final capture available after the live snapshot transport closes', async () => {
+    const test = await harness()
+    const child = ordinary()
+    test.spawn.mockReturnValue(child.handle)
+    const remoteClosed = Promise.withResolvers<undefined>()
+    // The spy invokes this saved implementation with its original receiver.
+    // eslint-disable-next-line @typescript-eslint/unbound-method
+    const originalClose = SshRpcPeer.prototype.close
+    let peer: SshRpcPeer | undefined
+    const observeClose = vi.spyOn(SshRpcPeer.prototype, 'close').mockImplementation(function (this: SshRpcPeer, error?: Error) {
+      originalClose.call(this, error)
+      if (this !== peer) remoteClosed.resolve(undefined)
+    })
+    try {
+      const run = await test.prepare({ ...ordinaryRequest, stdio: { ...ordinaryRequest.stdio, stdout: { maxBytes: 64 } } })
+      peer = new SshRpcPeer(run.channels.stdout!, run.channels.stdout!, 4096, 1, async () => null)
+      run.channels.stderr!.end(); run.channels.stderr!.resume()
+      await test.owner.start(run.id)
+      peer.close()
+      await remoteClosed.promise
+      child.stdout.write('after channel loss')
+      child.finish()
+      expect(await test.owner.done(run.id)).toMatchObject({ collected: { stdout: {
+        totalBytes: 18, tail: Buffer.from('after channel loss').toString('base64'),
+      } } })
+    } finally { peer?.close(); observeClose.mockRestore(); await test.close() }
+  })
+
+  it('reports a native cleanup failure instead of claiming quiescence', async () => {
+    const test = await harness()
+    const child = ordinary()
+    child.waitForExit.mockRejectedValue(new Error('native range no longer observable'))
+    test.spawn.mockReturnValue(child.handle)
+    try {
+      const run = await test.prepare(ordinaryRequest)
+      await test.owner.start(run.id)
+      await expect(test.owner.close()).rejects.toThrow('SSH remote process cleanup failed')
+      expect(child.terminate).toHaveBeenCalled()
+    } finally { await test.close() }
+  })
+
+  it('expires an unused reservation and prevents launch during helper shutdown', async () => {
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const test = await harness()
+    try {
+      const unused = await test.owner.prepare(ordinaryRequest)
+      await vi.advanceTimersByTimeAsync(5000)
+      await vi.waitFor(async () => { expect(await readdir(test.root)).toEqual([]) })
+      await expect(test.owner.start(unused.id)).rejects.toThrow('Unknown or expired')
+      const run = await test.prepare(ordinaryRequest)
+      const started = test.owner.start(run.id)
+      const rejected = expect(started).rejects.toThrow('helper is closing')
+      await test.owner.close()
+      await rejected
+      expect(test.spawn).not.toHaveBeenCalled()
+    } finally { await test.close(); vi.useRealTimers() }
+  })
+
+  it('reaps a terminal after its driver rejects completion', async () => {
+    const test = await harness()
+    const child = terminal()
+    test.spawnTerminal.mockResolvedValue(child.handle)
+    try {
+      const run = await test.prepare(terminalRequest)
+      await test.owner.start(run.id)
+      const done = expect(test.owner.done(run.id)).rejects.toThrow('terminal driver disconnected')
+      child.completion.reject(new Error('terminal driver disconnected'))
+      child.output.destroy(new Error('terminal stream closed'))
+      await done
+      await test.owner.terminate(run.id)
+      expect(child.terminate).toHaveBeenCalled()
+    } finally { await test.close() }
+  })
+
+  it('preserves an ordinary driver failure through direct-result and cleanup observations', async () => {
+    const test = await harness()
+    const child = ordinary()
+    test.spawn.mockReturnValue(child.handle)
+    try {
+      const run = await test.prepare(ordinaryRequest)
+      await test.owner.start(run.id)
+      const failed = expect(test.owner.done(run.id)).rejects.toThrow('ordinary driver disconnected')
+      child.completion.reject(new Error('ordinary driver disconnected'))
+      child.stdout.end(); child.stderr.end()
+      await failed
+      await expect(test.owner.close()).rejects.toThrow('SSH remote process cleanup failed')
+    } finally { await test.close() }
+  })
+
+  it('observes an expired reservation cleanup error without an unhandled rejection', async () => {
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const test = await harness()
+    try {
+      const run = await test.owner.prepare(ordinaryRequest)
+      vi.mocked(rm).mockClear().mockRejectedValueOnce(new Error('temporary directory became unavailable'))
+      await vi.advanceTimersByTimeAsync(5000)
+      await vi.waitFor(() => { expect(rm).toHaveBeenCalledOnce() })
+      await expect(test.owner.start(run.id)).rejects.toThrow('Unknown or expired')
+    } finally { await test.close(); vi.useRealTimers() }
+  })
+
+  it.each(['control', 'socket'] as const)('destroys the other control endpoint after a %s failure', async (failed) => {
+    const test = await harness()
+    const child = ordinary(true)
+    test.spawn.mockReturnValue(child.handle)
+    try {
+      const run = await test.prepare({ ...ordinaryRequest, stdio: { ...ordinaryRequest.stdio, stdin: 'pipe', control: 'pipe' } })
+      await test.owner.start(run.id)
+      run.channels.stdout!.end(); run.channels.stdout!.resume()
+      run.channels.stderr!.end(); run.channels.stderr!.resume()
+      child.stdin.destroy(new Error('native input pipe closed'))
+      child.stdout.destroy(new Error('native output pipe closed'))
+      if (failed === 'control') child.channel!.destroy(new Error('native control failed'))
+      else run.channels.control!.destroy(new Error('forwarded control failed'))
+      await vi.waitFor(() => { expect(child.channel!.destroyed).toBe(true) })
+      await test.owner.terminate(run.id)
+      await expect(test.owner.done(run.id)).resolves.toMatchObject({ outcome })
+    } finally { await test.close() }
+  })
+})

packages/ssh/ssh/tests/helper-runtime.spec.ts

@@ -89,7 +89,7 @@ describe.skipIf(process.platform === 'win32')('SSH helper runtime', () => {
       await expect(test.client.request('sandbox', { argv: ['true'], policy: { mode: 'danger-full-access', workspaceRoot: test.root } }, z.unknown())).rejects.toThrow('does not need')
       expect(await test.client.request('executable', { command: process.execPath, env: { REMOVED: null } }, z.string())).toBe(process.execPath)
       expect(await test.client.request('executable', { command: process.execPath }, z.string())).toBe(process.execPath)
-      const wrapped = await test.client.request('sandbox', { argv: ['true'], policy: policy(test.root) }, z.object({ argv: z.array(z.string()), enforcement: z.enum(['full', 'partial']) }).passthrough())
+      const wrapped = await test.client.request('sandbox', { argv: ['true'], policy: policy(test.root) }, z.looseObject({ argv: z.array(z.string()), enforcement: z.enum(['full', 'partial']) }))
       expect(wrapped.argv.at(-1)).toBe('true')
     } finally { await test.close() }
   })

packages/ssh/ssh/tests/identifiers.spec.ts

@@ -0,0 +1,19 @@
+import { expect, expectTypeOf, it } from 'vitest'
+import { preparedSchema, processIdSchema, textStreamIdSchema } from '../src/schemas.ts'
+import type { SshProcessId, SshTextStreamId } from '../src/schemas.ts'
+
+it('keeps process and text-stream identities distinct while retaining their UUID wire values', () => {
+  const wire = 'a25daf3e-c6dd-4bb7-9f39-14f09eb2d155'
+  const process = processIdSchema.parse(wire)
+  const stream = textStreamIdSchema.parse(wire)
+  expectTypeOf(process).toEqualTypeOf<SshProcessId>()
+  expectTypeOf(stream).toEqualTypeOf<SshTextStreamId>()
+  expectTypeOf(stream).not.toExtend<SshProcessId>()
+  expectTypeOf(process).not.toExtend<SshTextStreamId>()
+  expectTypeOf<string>().not.toExtend<SshProcessId>()
+  expectTypeOf<string>().not.toExtend<SshTextStreamId>()
+  expect(JSON.stringify({ process, stream })).toBe(JSON.stringify({ process: wire, stream: wire }))
+  expect(preparedSchema.parse({ id: wire, streams: {} }).id).toBe(process)
+  expect(processIdSchema.safeParse('invalid').success).toBe(false)
+  expect(textStreamIdSchema.safeParse('invalid').success).toBe(false)
+})

packages/ssh/ssh/tests/live.e2e.ts

@@ -146,7 +146,7 @@ describe.skipIf(!enabled)('POSIX SSH runtime acceptance', () => {
       })
       const bytes = Buffer.from(Array.from({ length: 128 * 1024 }, (_, index) => index % 256))
       const chunks: Buffer[] = []
-      const receiving = (async () => { for await (const chunk of handle.control!) chunks.push(Buffer.from(chunk)) })()
+      const receiving = (async () => { for await (const chunk of handle.control!) chunks.push(Buffer.from(chunk as Uint8Array)) })()
       handle.control!.end(bytes)
       const result = await processResult(handle)
       await receiving
@@ -218,7 +218,7 @@ describe.skipIf(!enabled)('POSIX SSH runtime acceptance', () => {
         argv: [victim.hello.node, '-e', code], cwd: victim.root,
         stdio: { stdin: 'ignore', stdout: 'pipe', stderr: { maxBytes: 4096 } }, graceMs: 500,
       })
-      const [bytes] = await once(handle.stdout!, 'data')
+      const [bytes] = await once(handle.stdout!, 'data') as [Buffer]
       const scope = `/sys/fs/cgroup${Buffer.from(bytes).toString().trim().split('::')[1]}`
       expect(scope).toMatch(/dsh-subprocess-[^\n]+\.scope$/)
       const scopeTarget = await observer.ctx.fs.resolve(scope)

packages/ssh/ssh/tests/protocol-boundaries.spec.ts

@@ -35,7 +35,7 @@ describe('SSH protocol wire and allocation bounds', () => {
     const controller = new AbortController()
     controller.abort('caller cancelled before allocation')
     const chunks: Buffer[] = []
-    output.on('data', (chunk) => { chunks.push(chunk) })
+    output.on('data', (chunk: Buffer) => { chunks.push(chunk) })
     await expect(connection.request('read', {}, z.null(), controller.signal)).rejects.toBe('caller cancelled before allocation')
     expect(chunks).toEqual([])
   })
@@ -44,7 +44,7 @@ describe('SSH protocol wire and allocation bounds', () => {
     const { connection, input, output } = peer({ pending: 1 })
     const sent = once(output, 'data')
     const first = connection.request('read', {}, z.string())
-    const [chunk] = await sent
+    const [chunk] = await sent as [Buffer]
     const request = JSON.parse(Buffer.from(chunk).subarray(4).toString()) as { id: string }
     await expect(connection.request('read', {}, z.string())).rejects.toThrow('pending request limit')
     input.write(frame({ type: 'result', id: request.id, value: 'first response' }))

packages/ssh/ssh/tests/startup-behavior.spec.ts

@@ -0,0 +1,339 @@
+/** SSH startup identity and request lifetimes over the actual administrative protocol. */
+import { EventEmitter } from 'node:events'
+import { Duplex, PassThrough } from 'node:stream'
+import { Context, Service } from '@deepseek-ai/cordis'
+import { describe, expect, it, onTestFinished, vi } from 'vitest'
+import { z } from 'zod'
+import { SshRpcPeer } from '../src/protocol.ts'
+import { SshConnection } from '../src/index.ts'
+import type { Config } from '../src/index.ts'
+
+const transport = vi.hoisted(() => ({ spawn: vi.fn(), exec: vi.fn(), connect: vi.fn(), tls: vi.fn(), directory: vi.fn(), remove: vi.fn() }))
+vi.mock('node:child_process', async original => ({ ...await original<typeof import('node:child_process')>(), spawn: transport.spawn, execFile: transport.exec }))
+vi.mock('node:net', async original => ({ ...await original<typeof import('node:net')>(), createConnection: transport.connect }))
+vi.mock('node:tls', async original => ({ ...await original<typeof import('node:tls')>(), connect: transport.tls }))
+vi.mock('node:fs/promises', () => ({ mkdtemp: transport.directory, rm: transport.remove }))
+
+class Stream extends Duplex {
+  finishDestroy: ((error: Error | null) => void) | undefined
+  holdDestroy = false
+  override _read(): void {}
+  override _write(_bytes: Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void { callback() }
+  override _destroy(error: Error | null, callback: (error: Error | null) => void): void {
+    if (this.holdDestroy) this.finishDestroy = callback
+    else callback(error)
+  }
+  releaseDestroy(): void {
+    const callback = this.finishDestroy
+    this.finishDestroy = undefined
+    callback?.(null)
+  }
+  disableRenegotiation(): void {}
+}
+
+class Child extends EventEmitter {
+  readonly stdin = new PassThrough()
+  readonly stdout = new PassThrough()
+  readonly stderr = new PassThrough()
+  readonly signals: string[] = []
+  readonly terminating = Promise.withResolvers<undefined>()
+  ignoreTerm = false
+  closed = false
+  kill(signal = 'SIGTERM'): boolean {
+    this.signals.push(signal)
+    this.terminating.resolve(undefined)
+    if (this.ignoreTerm && signal !== 'SIGKILL') return true
+    if (!this.closed) {
+      this.closed = true
+      queueMicrotask(() => { this.emit('close', 0, null) })
+    }
+    return true
+  }
+}
+
+const config: Config = {
+  host: 'test-alias', node: '/remote/node', helper: '/remote/helper.js', helperHash: 'a'.repeat(64), workspace: '/remote/workspace',
+  requestTimeoutMs: 1000, maxFrameBytes: 4096, maxPending: 8, leaseMs: 30_000,
+}
+const hello = {
+  protocol: 1, hash: 'a'.repeat(64), platform: 'linux', nodeVersion: 'v24.19.0',
+  node: '/canonical/node', root: '/tmp/remote-helper', workspace: '/canonical/workspace',
+}
+
+function setup(options: {
+  config?: Partial<Config>
+  hello?: Record<string, unknown>
+  holdHello?: boolean
+  holdConnect?: boolean
+  forwardFailure?: Error
+  heartbeatFailure?: Error
+  directoryFailure?: Error
+} = {}) {
+  const ctx = new Context()
+  const child = new Child()
+  const raw = new Stream()
+  const secure = new Stream()
+  const releaseHello = Promise.withResolvers<undefined>()
+  const helloEntered = Promise.withResolvers<undefined>()
+  const requestEntered = Promise.withResolvers<AbortSignal>()
+  const released = Promise.withResolvers<null>()
+  const calls: { method: string; params: unknown }[] = []
+  const helper = new SshRpcPeer(child.stdin, child.stdout, 4096, 8, async (method, params, signal) => {
+    calls.push({ method, params })
+    if (method === 'hello') {
+      helloEntered.resolve(undefined)
+      if (options.holdHello) await releaseHello.promise
+      return { ...hello, ...options.hello }
+    }
+    if (method === 'held') { requestEntered.resolve(signal); return released.promise }
+    if (method === 'echo') return params
+    if (method === 'heartbeat' && options.heartbeatFailure !== undefined) throw options.heartbeatFailure
+    if (method === 'heartbeat' || method === 'close') return null
+    throw new Error(`unexpected helper request ${method}`)
+  })
+  transport.directory.mockResolvedValue('/virtual/ssh-startup')
+  if (options.directoryFailure !== undefined) transport.directory.mockRejectedValueOnce(options.directoryFailure)
+  transport.remove.mockResolvedValue(undefined)
+  transport.spawn.mockReturnValue(child)
+  transport.exec.mockImplementation((_file: string, args: string[], _options: unknown, callback: (error: Error | null) => void) => {
+    const command = new Child()
+    queueMicrotask(() => { callback(args.includes('forward') ? options.forwardFailure ?? null : null); command.kill() })
+    return command
+  })
+  transport.connect.mockImplementation(() => {
+    if (!options.holdConnect) queueMicrotask(() => { raw.emit('connect') })
+    return raw
+  })
+  transport.tls.mockImplementation(() => {
+    raw.on('error', (error) => { secure.destroy(error) })
+    raw.once('close', () => { secure.destroy() })
+    secure.once('close', () => { raw.destroy() })
+    queueMicrotask(() => { secure.emit('secureConnect') })
+    return secure
+  })
+  // oxlint-disable-next-line eslint/prefer-const -- Failed construction still runs the pre-registered fixture cleanup.
+  let service: SshConnection | undefined
+  onTestFinished(async () => {
+    releaseHello.resolve(undefined)
+    released.resolve(null)
+    raw.releaseDestroy()
+    secure.releaseDestroy()
+    try { await service?.dispose() }
+    finally {
+      raw.destroy(); secure.destroy(); helper.close(); child.stderr.destroy(); child.kill()
+      await ctx.fiber.dispose()
+      vi.restoreAllMocks()
+      for (const mock of Object.values(transport)) mock.mockReset()
+    }
+  })
+  service = new SshConnection(ctx, { ...config, ...options.config })
+  return { service, child, helper, raw, secure, calls, requestEntered: requestEntered.promise,
+    helloEntered: helloEntered.promise, releaseHello: () => { releaseHello.resolve(undefined) }, release: () => { released.resolve(null) } }
+}
+
+describe.skipIf(process.platform === 'win32')('SSH connection startup', () => {
+  it.each([
+    { host: '-option' }, { host: 'alias; command' }, { node: 'relative' }, { helperHash: 'bad' },
+    { bootstrapPath: '/remote/bootstrap.js' }, { bootstrapHash: 'b'.repeat(64) },
+    { requestTimeoutMs: 0 }, { maxFrameBytes: 64 * 1024 * 1024 + 1 }, { maxPending: 129 }, { leaseMs: 2999 },
+  ])('rejects invalid deployment configuration before SSH starts: %j', (invalid) => {
+    expect(() => setup({ config: invalid })).toThrow()
+    expect(transport.spawn).not.toHaveBeenCalled()
+  })
+
+  it('rejects a non-POSIX client before starting SSH', () => {
+    const platform = vi.spyOn(process, 'platform', 'get').mockReturnValue('win32')
+    try { expect(() => setup()).toThrow('POSIX client') }
+    finally { platform.mockRestore() }
+    expect(transport.spawn).not.toHaveBeenCalled()
+  })
+
+  it('publishes only verified remote coordinates and quotes the configured executable paths', async () => {
+    const test = setup({ holdHello: true, config: { helper: "/remote/helper's file.js", bootstrapPath: '/remote/process.js', bootstrapHash: 'b'.repeat(64) }, hello: { bootstrapHash: 'b'.repeat(64) } })
+    expect(() => test.service.nodeExecutable).toThrow('not ready')
+    expect(() => test.service.bootstrapPath).toThrow('verified bootstrapPath')
+    await test.helloEntered
+    test.releaseHello()
+    await test.service[Service.init]()
+    expect(test.service.nodeExecutable).toBe('/canonical/node')
+    expect(test.service.bootstrapPath).toBe('/remote/process.js')
+    const argv = transport.spawn.mock.calls[0]?.[1] as string[]
+    expect(argv).toContain('StrictHostKeyChecking=yes')
+    expect(argv).toContain('ForwardAgent=no')
+    expect(argv.at(-1)).toBe("'/remote/node' '/remote/helper'\\''s file.js'")
+    expect(test.calls[0]?.params).toEqual({ protocol: 1, workspace: '/remote/workspace', leaseMs: 30_000, bootstrapPath: '/remote/process.js' })
+  })
+
+  it('permits filesystem-only deployments and refuses an unconfigured PTC bootstrap getter', async () => {
+    const test = setup()
+    await test.service.ready
+    expect(test.service.nodeExecutable).toBe('/canonical/node')
+    expect(() => test.service.bootstrapPath).toThrow('verified bootstrapPath')
+    expect(await test.service.request('echo', { value: 42 }, z.object({ value: z.number() }))).toEqual({ value: 42 })
+  })
+
+  it.each([
+    { hello: { hash: 'c'.repeat(64) }, error: 'helper digest' },
+    { hello: { bootstrapHash: 'c'.repeat(64) }, error: 'bootstrap digest' },
+    { hello: { protocol: 2 }, error: 'Invalid' },
+  ])('refuses a mismatched helper identity before readiness: $error', async ({ hello, error }) => {
+    const test = setup({ hello })
+    await expect(test.service.ready).rejects.toThrow(error)
+    expect(() => test.service.nodeExecutable).toThrow('not ready')
+    await expect(test.service.request('echo', {}, z.unknown())).rejects.toThrow()
+  })
+
+  it('joins disposal requested before directory allocation publishes a child', async () => {
+    const test = setup()
+    const rejected = expect(test.service.ready).rejects.toThrow('closed before startup')
+    const disposal = test.service.dispose()
+    await Promise.all([rejected, disposal])
+    expect(transport.spawn).not.toHaveBeenCalled()
+    expect(transport.remove).toHaveBeenCalledWith('/virtual/ssh-startup', { recursive: true, force: true })
+  })
+
+  it('preserves the first SSH process error and closes later requests', async () => {
+    const test = setup()
+    await test.service.ready
+    const failure = new Error('SSH transport refused')
+    test.child.emit('error', failure)
+    await Promise.resolve()
+    await expect(test.service.request('echo', {}, z.unknown())).rejects.toBe(failure)
+    expect(test.child.signals).toContain('SIGTERM')
+  })
+
+  it('cancels bounded administrative work and allows explicit process waits past that deadline', async () => {
+    const test = setup()
+    await test.service.ready
+    const deadline = new AbortController()
+    const timeout = vi.spyOn(AbortSignal, 'timeout').mockReturnValue(deadline.signal)
+    try {
+      const pending = test.service.request('held', {}, z.null())
+      const rejected = expect(pending).rejects.toThrow('cancelled')
+      const remoteSignal = await test.requestEntered
+      deadline.abort()
+      await rejected
+      await expect.poll(() => remoteSignal.aborted).toBe(true)
+      expect(timeout).toHaveBeenCalledWith(1000)
+      test.release()
+      expect(await test.service.request('echo', 42, z.number(), undefined, true)).toBe(42)
+      const caller = AbortSignal.abort(new Error('caller cancelled'))
+      await expect(test.service.request('echo', {}, z.unknown(), caller)).rejects.toThrow('caller cancelled')
+    } finally { timeout.mockRestore() }
+  })
+
+  it('returns authenticated paused streams and rejects paths outside the helper root', async () => {
+    const test = setup()
+    await test.service.ready
+    for (const path of ['/tmp/other/control', '/tmp/remote-helper/control:extra', '/tmp/remote-helper/control\n']) {
+      await expect(test.service.connectStream({ path, capability: 'b'.repeat(64) })).rejects.toThrow('invalid stream path')
+    }
+    expect(transport.exec).not.toHaveBeenCalled()
+    const socket = await test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) })
+    expect(socket).toBe(test.secure)
+    expect(socket.readableFlowing).toBe(false)
+    socket.destroy(new Error('data channel closed'))
+    await expect.poll(() => transport.exec.mock.calls.length).toBe(2)
+    expect(transport.exec.mock.calls[1]?.[1]).toContain('cancel')
+  })
+
+  it('cancels a failed forwarding reservation before surfacing its error', async () => {
+    const failure = new Error('forwarding refused')
+    const test = setup({ forwardFailure: failure })
+    await test.service.ready
+    await expect(test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) })).rejects.toBe(failure)
+    expect(transport.exec.mock.calls[1]?.[1]).toContain('cancel')
+    expect(transport.connect).not.toHaveBeenCalled()
+  })
+
+  it('keeps caller cancellation attached while a forwarded socket is connecting', async () => {
+    const test = setup({ holdConnect: true })
+    await test.service.ready
+    const controller = new AbortController()
+    const pending = test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) }, controller.signal)
+    const rejected = expect(pending).rejects.toThrow('caller stopped')
+    await expect.poll(() => transport.connect.mock.calls.length).toBe(1)
+    controller.abort('caller stopped')
+    await rejected
+    expect(test.raw.destroyed).toBe(true)
+  })
+
+  it('reports directory creation failure without starting a helper process', async () => {
+    const failure = new Error('temporary directory unavailable')
+    const test = setup({ directoryFailure: failure })
+    await expect(test.service.ready).rejects.toBe(failure)
+    await test.service.dispose()
+    expect(transport.spawn).not.toHaveBeenCalled()
+    expect(transport.remove).not.toHaveBeenCalled()
+  })
+
+  it('invalidates requests when a heartbeat reports a failed helper', async () => {
+    vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] })
+    const test = setup({ heartbeatFailure: new Error('helper unhealthy') })
+    try {
+      await test.service.ready
+      await vi.advanceTimersByTimeAsync(10_000)
+      await expect(test.service.request('echo', {}, z.unknown())).rejects.toThrow('helper unhealthy')
+      expect(test.child.signals).toContain('SIGTERM')
+    } finally { await test.service.dispose(); vi.useRealTimers() }
+  })
+
+  it('force-kills a helper process that does not exit after termination', async () => {
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const test = setup()
+    try {
+      await test.service.ready
+      test.child.ignoreTerm = true
+      const disposal = test.service.dispose()
+      await test.child.terminating.promise
+      await vi.advanceTimersByTimeAsync(1000)
+      await disposal
+      expect(test.child.signals).toContain('SIGKILL')
+    } finally {
+      test.child.ignoreTerm = false
+      test.child.kill()
+      await test.service.dispose()
+      vi.useRealTimers()
+    }
+  })
+
+  it('joins authenticated sockets whose native close callbacks are delayed', async () => {
+    const test = setup()
+    await test.service.ready
+    await test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) })
+    test.raw.holdDestroy = true
+    test.secure.holdDestroy = true
+    let disposed = false
+    const disposal = test.service.dispose().then(() => { disposed = true })
+    await test.child.terminating.promise
+    expect(disposed).toBe(false)
+    expect(test.raw.finishDestroy).toBeTypeOf('function')
+    expect(test.secure.finishDestroy).toBeTypeOf('function')
+    test.secure.releaseDestroy()
+    test.raw.releaseDestroy()
+    await disposal
+    expect(test.raw.closed).toBe(true)
+    expect(test.secure.closed).toBe(true)
+  })
+
+  it('closes authenticated sockets when the SSH process fails', async () => {
+    const test = setup()
+    await test.service.ready
+    await test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) })
+    test.child.emit('error', new Error('transport lost'))
+    await test.service.dispose()
+    expect(test.raw.destroyed).toBe(true)
+    expect(test.secure.destroyed).toBe(true)
+  })
+
+  it('contains a failed local forwarding-path removal and still disposes the connection', async () => {
+    const test = setup()
+    await test.service.ready
+    const socket = await test.service.connectStream({ path: '/tmp/remote-helper/control', capability: 'b'.repeat(64) })
+    transport.remove.mockRejectedValueOnce(new Error('forwarding path already removed'))
+    socket.destroy()
+    await expect.poll(() => transport.remove.mock.calls.length).toBe(1)
+    await test.service.dispose()
+    expect(transport.remove).toHaveBeenLastCalledWith('/virtual/ssh-startup', { recursive: true, force: true })
+  })
+})

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

@@ -2,10 +2,11 @@ import { mkdtemp, rm } from 'node:fs/promises'
 import { createConnection, type Socket } from 'node:net'
 import { once } from 'node:events'
 import { Context } from '@deepseek-ai/cordis'
-import { describe, expect, it } from 'vitest'
+import { connect as tlsConnect } from 'node:tls'
+import { describe, expect, it, vi } from 'vitest'
 import { RemoteProcesses } from '../src/helper-processes.ts'
 import type { SshStreamEndpoint } from '../src/schemas.ts'
-import { authenticateStream } from '../src/stream-security.ts'
+import { authenticateStream, SSH_STREAM_TLS_OPTIONS } from '../src/stream-security.ts'
 
 async function connect(endpoint: SshStreamEndpoint, capability: string): Promise<Socket> {
   const socket = createConnection(endpoint.path)
@@ -15,6 +16,47 @@ async function connect(endpoint: SshStreamEndpoint, capability: string): Promise
 }
 
 describe.skipIf(process.platform === 'win32')('SSH stream capabilities', () => {
+  it('refuses an unknown TLS identity without consuming a legitimate reservation', async () => {
+    const root = await mkdtemp('/tmp/dsh-ssh-identity-')
+    const owner = new RemoteProcesses(new Context(), root, 1, 5000)
+    let stream: Socket | undefined
+    try {
+      const prepared = await owner.prepare({ argv: ['true'], cwd: root, graceMs: 100, terminal: { rows: 24, cols: 80 } })
+      const endpoint = prepared.streams.terminal!
+      const raw = createConnection(endpoint.path)
+      await once(raw, 'connect')
+      stream = tlsConnect({ ...SSH_STREAM_TLS_OPTIONS, socket: raw,
+        pskCallback: () => ({ identity: 'unknown-client', psk: Buffer.from(endpoint.capability, 'hex') }),
+      })
+      await expect(once(stream, 'secureConnect')).rejects.toThrow()
+      stream.destroy()
+      stream = await connect(endpoint, endpoint.capability)
+      expect(stream.destroyed).toBe(false)
+    } finally { stream?.destroy(); await owner.close(); await rm(root, { recursive: true, force: true }) }
+  })
+
+  it('publishes only one of two already-connected clients for the same reservation', async () => {
+    const root = await mkdtemp('/tmp/dsh-ssh-duplicate-')
+    const owner = new RemoteProcesses(new Context(), root, 1, 5000)
+    const sockets: Socket[] = []
+    try {
+      const prepared = await owner.prepare({ argv: ['true'], cwd: root, graceMs: 100, terminal: { rows: 24, cols: 80 } })
+      const endpoint = prepared.streams.terminal!
+      const raw = [createConnection(endpoint.path), createConnection(endpoint.path)]
+      await Promise.all(raw.map(socket => once(socket, 'connect')))
+      const clients = await Promise.allSettled(raw.map(socket => authenticateStream(socket, endpoint.capability, 5000)))
+      for (const result of clients) if (result.status === 'fulfilled') {
+        result.value.on('error', () => {})
+        sockets.push(result.value)
+      }
+      await vi.waitFor(() => { expect(sockets.filter(socket => !socket.destroyed)).toHaveLength(1) })
+    } finally {
+      for (const socket of sockets) socket.destroy()
+      await owner.close()
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
   it('refuses another endpoint capability without consuming the legitimate reservation', async () => {
     const root = await mkdtemp('/tmp/dsh-ssh-auth-')
     const processes = new RemoteProcesses(new Context(), root, 4, 5000)

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

@@ -0,0 +1,65 @@
+/** Cancellation stays with TLS after it takes ownership of the underlying socket. */
+import { once } from 'node:events'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { createConnection, createServer, type Socket } from 'node:net'
+import { Context } from '@deepseek-ai/cordis'
+import { describe, expect, it } from 'vitest'
+import { RemoteProcesses } from '../src/helper-processes.ts'
+import { authenticateStream } from '../src/stream-security.ts'
+
+describe('SSH TLS cancellation ownership', () => {
+  it('cancels an authenticated stream before closing its underlying socket', async () => {
+    const root = await mkdtemp('/tmp/dsh-ssh-tls-life-')
+    const owner = new RemoteProcesses(new Context(), root, 1, 5000)
+    try {
+      const prepared = await owner.prepare({ argv: ['true'], cwd: root, graceMs: 100, terminal: { rows: 24, cols: 80 } })
+      const endpoint = prepared.streams.terminal!
+      const raw = createConnection({ path: endpoint.path, allowHalfOpen: true })
+      raw.on('error', () => {})
+      await once(raw, 'connect')
+      const controller = new AbortController()
+      const stream = await authenticateStream(raw, endpoint.capability, 5000, controller.signal)
+      stream.on('error', () => {})
+      const closed = new Promise<void>((resolve) => { stream.once('close', () => { resolve() }) })
+      const rawClosed = new Promise<void>((resolve) => { raw.once('close', () => { resolve() }) })
+      controller.abort(new Error('stream lifetime ended'))
+      await Promise.all([closed, rawClosed])
+      expect(stream.destroyed).toBe(true)
+      expect(raw.destroyed).toBe(true)
+    } finally { await owner.close(); await rm(root, { recursive: true, force: true }) }
+  })
+
+  it.each([false, true])('cancels TLS authentication with an arbitrary caller reason (already aborted: %s)', async (alreadyAborted) => {
+    const peers = new Set<Socket>()
+    const server = createServer((socket) => {
+      peers.add(socket)
+      socket.on('error', () => {})
+      socket.once('close', () => { peers.delete(socket) })
+      socket.resume()
+    })
+    server.listen(0, '127.0.0.1')
+    await once(server, 'listening')
+    const address = server.address()
+    if (address === null || typeof address === 'string') throw new Error('missing test listener')
+    const raw = createConnection(address.port, '127.0.0.1')
+    raw.on('error', () => {})
+    try {
+      await once(raw, 'connect')
+      const closed = new Promise<void>((resolve) => { raw.once('close', () => { resolve() }) })
+      const controller = new AbortController()
+      if (alreadyAborted) controller.abort('caller stopped')
+      const result = authenticateStream(raw, 'a'.repeat(64), 5000, controller.signal)
+      const rejected = alreadyAborted
+        ? expect(result).rejects.toBe('caller stopped')
+        : expect(result).rejects.toThrow('caller stopped')
+      if (!alreadyAborted) controller.abort('caller stopped')
+      await rejected
+      await closed
+      expect(raw.destroyed).toBe(true)
+    } finally {
+      raw.destroy()
+      for (const socket of peers) socket.destroy()
+      await new Promise<void>((resolve) => { server.close(() => { resolve() }) })
+    }
+  })
+})

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

@@ -39,7 +39,7 @@ describe.skipIf(process.platform === 'win32')('SSH stream pathname replacement',
         argv: confined.argv, cwd: root, graceMs: 500,
         stdio: { stdin: 'ignore', stdout: 'pipe', stderr: { maxBytes: 4096 } },
       })
-      const [ready] = await once(attacker.stdout!, 'data')
+      const [ready] = await once(attacker.stdout!, 'data') as [Buffer]
       expect(Buffer.from(ready).toString()).toBe('rebound\n')
       let control: Socket | undefined
       for (const [name, value] of Object.entries(prepared.streams)) {
@@ -52,7 +52,7 @@ describe.skipIf(process.platform === 'win32')('SSH stream pathname replacement',
       }
       const secret = `private-process-payload-${randomUUID()}`
       const received: Buffer[] = []
-      const reading = (async () => { for await (const bytes of control!) received.push(Buffer.from(bytes)) })()
+      const reading = (async () => { for await (const bytes of control!) received.push(Buffer.from(bytes as Uint8Array)) })()
       await owner.start(prepared.id)
       control!.end(secret)
       await reading

packages/ssh/ssh/tsconfig.json

@@ -40,6 +40,9 @@
     },
     {
       "path": "../../session/session-projection"
+    },
+    {
+      "path": "../../util/brand"
     }
   ]
 }

packages/ssh/subprocess-ssh/package.json

@@ -22,7 +22,7 @@
     "./package.json": "./package.json"
   },
   "files": [
-    "lib/*.js",
+    "lib/index.js",
     "lib/types/**/*.d.ts"
   ],
   "license": "MIT",

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

@@ -7,6 +7,7 @@ import type { SubprocessCollectedOutputs, SubprocessHandle, SubprocessOutcome, S
 import { OutputCollector } from '@deepseek-ai/dsh-subprocess-local/output'
 import type { SshConnection } from '@deepseek-ai/dsh-ssh'
 import { doneSchema, foregroundSchema, outputSnapshotFrameLimit, outputSnapshotSchema, preparedSchema, remotePath, streamEndpointSchema } from '@deepseek-ai/dsh-ssh/schemas'
+import type { SshProcessId } from '@deepseek-ai/dsh-ssh/schemas'
 import { SshRpcPeer } from '@deepseek-ai/dsh-ssh/protocol'
 import { z } from 'zod'
 
@@ -32,11 +33,10 @@ class RemoteProcess implements SubprocessHandle {
   private readonly fromControl = new PassThrough()
   private readonly controller = new AbortController()
   private readonly started: Promise<void>
-  private id: string | undefined
+  private id: SshProcessId | undefined
   private sockets: Socket[] = []
   private quiescent = false
   private committed = false
-  private startRequested = false
   private termination: Promise<void> | undefined
   private readonly detachAbort: () => void
   private spills: { stdout?: string | undefined; stderr?: string | undefined } = {}
@@ -125,11 +125,11 @@ class RemoteProcess implements SubprocessHandle {
         if (name === 'stdout' || name === 'stderr') {
           const mode = this.spec.stdio[name]
           if (typeof mode === 'object') {
-            new SshRpcPeer(socket, socket, outputSnapshotFrameLimit(mode.maxBytes), 1, async (method, raw) => {
+            new SshRpcPeer(socket, socket, outputSnapshotFrameLimit(mode.maxBytes), 1, (method, raw) => Promise.resolve().then(() => {
               if (method !== 'snapshot') throw new Error('Unexpected SSH output-stream operation')
               this.updateCollection[name]?.(outputSnapshotSchema.parse(raw), false)
               return null
-            })
+            }))
           } else {
             socket.end()
             socket.pipe(name === 'stdout' ? this.out : this.err)
@@ -139,7 +139,6 @@ class RemoteProcess implements SubprocessHandle {
         if (name === 'control') { this.toControl.pipe(socket); socket.pipe(this.fromControl) }
       }
       this.controller.signal.throwIfAborted()
-      this.startRequested = true
       await this.ssh.request('process.start', { id: this.id }, z.object({}).strict(), this.controller.signal)
       this.committed = true
     } catch (error) {
@@ -173,14 +172,12 @@ class RemoteProcess implements SubprocessHandle {
 
   async waitForExit(signal?: AbortSignal): Promise<boolean> {
     if (this.quiescent) return true
-    try { await this.started } catch (error) {
+    try { await this.started } catch {
+      // Startup errors remain on done; a prepared process must first finish termination.
       if (this.termination !== undefined) await this.termination
-      if (this.quiescent || !this.startRequested) {
-        this.quiescent = true
-        this.detachAbort()
-        return true
-      }
-      throw error
+      this.quiescent = true
+      this.detachAbort()
+      return true
     }
     if (this.termination !== undefined) { await this.termination; return true }
     const result = await this.ssh.request('process.wait', { id: this.id }, z.boolean(), signal, true)
@@ -207,7 +204,7 @@ class RemoteProcess implements SubprocessHandle {
         new Promise<void>((resolve) => { timer = setTimeout(resolve, this.spec.graceMs) }),
       ])
     } finally {
-      if (timer !== undefined) clearTimeout(timer)
+      clearTimeout(timer)
       for (const dispose of disposers) dispose()
     }
   }

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

@@ -0,0 +1,293 @@
+/** Remote process transport, output observations, and managed cleanup through the public provider. */
+import { duplexPair, type Duplex } from 'node:stream'
+import { Context } from '@deepseek-ai/cordis'
+import { describe, expect, it, onTestFinished, vi } from 'vitest'
+import { z } from 'zod'
+import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { SshRpcPeer } from '../../ssh/src/protocol.ts'
+import { outputSnapshotFrameLimit } from '../../ssh/src/schemas.ts'
+import { SshSubprocessRuntime } from '../src/index.ts'
+
+const id = 'fd897b7b-1b7e-4cd0-9b8d-06b354062d91'
+const spec: SubprocessSpawnSpec = {
+  argv: ['/usr/bin/node', '-e', ''], cwd: '/workspace', graceMs: 5,
+  stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' },
+}
+interface FinalOutput {
+  outcome: { exitCode: number | null; signal: string | null }
+  spills: { stdout?: string; stderr?: string }
+  collected: { stdout?: { tail: string; totalBytes: number }; stderr?: { tail: string; totalBytes: number } }
+}
+const emptyResult: FinalOutput = { outcome: { exitCode: 0, signal: null }, spills: {}, collected: {} }
+
+async function setup(options: { pause?: 'prepare' | 'connect' | 'start'; failPrepare?: Error; failStart?: unknown; failTerminate?: Error; wait?: boolean } = {}) {
+  const ctx = new Context()
+  const disposalErrors: unknown[] = []
+  ctx.logger.error = ((error: unknown) => { disposalErrors.push(error) }) as typeof ctx.logger.error
+  const finished = Promise.withResolvers<FinalOutput>()
+  const gate = Promise.withResolvers<undefined>()
+  const entered = Promise.withResolvers<undefined>()
+  const started = Promise.withResolvers<undefined>()
+  const peers = new Map<string, { host: Duplex; remote: Duplex }>()
+  const rpcPeers: SshRpcPeer[] = []
+  const calls: { method: string; params: unknown }[] = []
+  let preparedSpec: SubprocessSpawnSpec | undefined
+  const waitStage = async (stage: string) => {
+    if (stage === options.pause) { entered.resolve(undefined); await gate.promise }
+  }
+  const connection = {
+    dispose: vi.fn(async () => {}),
+    request: async <T>(method: string, params: unknown, schema: z.ZodType<T>): Promise<T> => {
+      calls.push({ method, params })
+      let value: unknown
+      if (method === 'process.prepare') {
+        if (options.failPrepare !== undefined) throw options.failPrepare
+        preparedSpec = params as SubprocessSpawnSpec
+        await waitStage('prepare')
+        const names = ['stdout', 'stderr', ...(preparedSpec.stdio.stdin === 'pipe' ? ['stdin'] : []),
+          ...(preparedSpec.stdio.control === 'pipe' ? ['control'] : [])]
+        value = { id, streams: Object.fromEntries(names.map(name => [name, { path: `/tmp/test-${name}`, capability: 'a'.repeat(64) }])) }
+      } else if (method === 'process.start') {
+        started.resolve(undefined)
+        await waitStage('start')
+        if (options.failStart !== undefined) throw options.failStart
+        value = {}
+      } else if (method === 'process.done') value = await finished.promise
+      else if (method === 'process.wait') value = options.wait ?? true
+      else if (method === 'process.terminate') {
+        if (options.failTerminate !== undefined) throw options.failTerminate
+        value = null
+      } else if (method === 'executable') value = '/remote/bin/node'
+      else throw new Error(`Unexpected SSH operation ${method}`)
+      return schema.parse(value)
+    },
+    connectStream: async (endpoint: { path: string }, signal?: AbortSignal): Promise<Duplex> => {
+      await waitStage('connect')
+      signal?.throwIfAborted()
+      const name = endpoint.path.slice('/tmp/test-'.length)
+      const [host, remote] = duplexPair({ allowHalfOpen: true })
+      host.on('error', () => {})
+      remote.on('error', () => {})
+      peers.set(name, { host, remote })
+      return host
+    },
+  }
+  ctx.provide('ssh', connection as never)
+  const fiber = await ctx.plugin(SshSubprocessRuntime)
+  const close = async () => {
+    gate.resolve(undefined)
+    finished.resolve(emptyResult)
+    for (const peer of rpcPeers) peer.close()
+    for (const peer of peers.values()) { peer.host.destroy(); peer.remote.destroy() }
+    return fiber.dispose()
+  }
+  onTestFinished(async () => {
+    await close()
+    if (options.failTerminate === undefined) expect(disposalErrors).toEqual([])
+  })
+  const stream = (name: string): Duplex => {
+    const peer = peers.get(name)
+    if (peer === undefined) throw new Error(`stream ${name} has not been connected`)
+    return peer.remote
+  }
+  const snapshots = (name: string, maxBytes: number) => {
+    const socket = stream(name)
+    const peer = new SshRpcPeer(socket, socket, outputSnapshotFrameLimit(maxBytes), 1)
+    rpcPeers.push(peer)
+    return (tail: string, totalBytes: number, method = 'snapshot') => peer.request(method, {
+      tail: Buffer.from(tail).toString('base64'), totalBytes,
+    }, z.null())
+  }
+  const closeStream = async (name: string): Promise<void> => {
+    const peer = peers.get(name)
+    if (peer === undefined) throw new Error(`stream ${name} has not been connected`)
+    const closed = new Promise<void>((resolve) => { peer.host.once('close', () => { resolve() }) })
+    peer.host.destroy()
+    peer.remote.destroy()
+    await closed
+  }
+  return { runtime: ctx.subprocess, fiber, calls, connection, disposalErrors, finished, entered: entered.promise, started: started.promise,
+    release: () => { gate.resolve(undefined) }, stream, closeStream, snapshots, close }
+}
+
+async function readAll(stream: NodeJS.ReadableStream): Promise<string> {
+  const chunks: Buffer[] = []
+  for await (const value of stream) chunks.push(Buffer.from(value as Uint8Array))
+  return Buffer.concat(chunks).toString()
+}
+
+describe('SSH ordinary process behavior', () => {
+  it('forwards inherited output without ending the host standard streams', async () => {
+    const test = await setup()
+    const stdout: unknown[] = []
+    const stderr: unknown[] = []
+    const out = vi.spyOn(process.stdout, 'write').mockImplementation((chunk: unknown) => { stdout.push(chunk); return true })
+    const err = vi.spyOn(process.stderr, 'write').mockImplementation((chunk: unknown) => { stderr.push(chunk); return true })
+    try {
+      const handle = test.runtime.spawn({ ...spec, stdio: { stdin: 'ignore', stdout: 'inherit', stderr: 'inherit' } })
+      await test.started
+      test.stream('stdout').end('inherited output')
+      test.stream('stderr').end('inherited diagnostics')
+      test.finished.resolve(emptyResult)
+      await handle.done
+      expect(stdout.map(String).join('')).toContain('inherited output')
+      expect(stderr.map(String).join('')).toContain('inherited diagnostics')
+      expect(process.stdout.writableEnded).toBe(false)
+      expect(process.stderr.writableEnded).toBe(false)
+      expect(handle.stdout).toBeUndefined()
+      expect(handle.collected).toEqual({})
+    } finally { out.mockRestore(); err.mockRestore() }
+  })
+
+  it('bounds output draining after process completion even when the output peer remains open', async () => {
+    const test = await setup()
+    const handle = test.runtime.spawn(spec)
+    await test.started
+    test.finished.resolve(emptyResult)
+    expect(await handle.done).toEqual({ exitCode: 0, signal: null })
+    expect(test.stream('stdout').destroyed).toBe(false)
+    expect(await handle.waitForExit()).toBe(true)
+  })
+
+  it('observes output transports that closed before the start acknowledgement arrives', async () => {
+    const test = await setup({ pause: 'start' })
+    const handle = test.runtime.spawn(spec)
+    await test.entered
+    await test.closeStream('stdout')
+    await test.closeStream('stderr')
+    test.release()
+    test.finished.resolve(emptyResult)
+    expect(await handle.done).toEqual({ exitCode: 0, signal: null })
+    expect(await handle.waitForExit()).toBe(true)
+    await test.close()
+    expect(test.disposalErrors).toEqual([])
+  })
+
+  it('keeps stdin, stdout, stderr and control independent and forwards explicit environment removals', async () => {
+    const test = await setup()
+    expect(await test.runtime.resolveExecutable('node', { PATH: '/remote/bin' })).toBe('/remote/bin/node')
+    const handle = test.runtime.spawn({ ...spec, env: { REMOVE: undefined, KEEP: 'value' },
+      stdio: { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', control: 'pipe' } })
+    const stdout = readAll(handle.stdout!)
+    const stderr = readAll(handle.stderr!)
+    const control = readAll(handle.control!)
+    handle.stdin!.end('ordinary input')
+    handle.control!.end('private input')
+    await test.started
+    expect(await readAll(test.stream('stdin'))).toBe('ordinary input')
+    expect(await readAll(test.stream('control'))).toBe('private input')
+    test.stream('stdout').end('ordinary output')
+    test.stream('stderr').end('diagnostic output')
+    test.stream('control').end('private output')
+    expect(await Promise.all([stdout, stderr, control])).toEqual(['ordinary output', 'diagnostic output', 'private output'])
+    test.finished.resolve(emptyResult)
+    expect(await handle.done).toEqual({ exitCode: 0, signal: null })
+    expect(await handle.waitForExit()).toBe(true)
+    expect(await handle.waitForExit()).toBe(true)
+    expect(test.calls.find(call => call.method === 'process.prepare')?.params).toMatchObject({ env: { REMOVE: null, KEEP: 'value' } })
+    handle.terminate()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(0)
+  })
+
+  it('retains authoritative collected byte offsets when older stream snapshots arrive after completion', async () => {
+    const test = await setup()
+    const handle = test.runtime.spawn({ ...spec, stdio: { stdin: 'ignore', stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } } })
+    await test.started
+    const snapshot = test.snapshots('stdout', 4)
+    await snapshot('cdef', 6)
+    expect(handle.collected.stdout!.readFrom(0)).toEqual({ text: 'cdef', nextOffset: 6, lossy: true })
+    test.finished.resolve({ ...emptyResult, spills: { stdout: '/remote/spill' }, collected: {
+      stdout: { tail: Buffer.from('ghij').toString('base64'), totalBytes: 10 }, stderr: { tail: '', totalBytes: 0 },
+    } })
+    await handle.done
+    await snapshot('x', 1)
+    expect(handle.collected.stdout!.readFrom(8)).toEqual({ text: 'ij', nextOffset: 10, lossy: false, spillPath: '/remote/spill' })
+  })
+
+  it.each([
+    { tail: '12345', total: 5, error: 'invalid collected output coordinates' },
+    { tail: 'abcd', total: 3, error: 'invalid collected output coordinates' },
+    { tail: 'a', total: 1, error: 'rewound collected output' },
+  ])('rejects a remote snapshot with $error', async ({ tail, total, error }) => {
+    const test = await setup()
+    const handle = test.runtime.spawn({ ...spec, stdio: { stdin: 'ignore', stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } } })
+    await test.started
+    const snapshot = test.snapshots('stdout', 4)
+    await snapshot('abcd', 4)
+    await expect(snapshot(tail, total)).rejects.toThrow(error)
+    await expect(snapshot('a', 1, 'unexpected')).rejects.toThrow('Unexpected SSH output-stream operation')
+    expect(handle.collected.stdout!.readFrom(0).text).toBe('abcd')
+    test.finished.resolve({ ...emptyResult, collected: { stdout: { tail: 'YWJjZA==', totalBytes: 4 }, stderr: { tail: '', totalBytes: 0 } } })
+    await handle.done
+  })
+
+  it('rejects a final output mode mismatch and terminates its remote process', async () => {
+    const test = await setup()
+    const handle = test.runtime.spawn({ ...spec, stdio: { stdin: 'ignore', stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } } })
+    await test.started
+    test.finished.resolve(emptyResult)
+    await expect(handle.done).rejects.toThrow('mismatched output collection modes')
+    expect(await handle.waitForExit()).toBe(true)
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+  })
+
+  it.each(['prepare', 'connect', 'start'] as const)('joins cancellation during %s and sends at most one termination', async (pause) => {
+    const test = await setup({ pause })
+    const controller = new AbortController()
+    const handle = test.runtime.spawn({ ...spec, signal: controller.signal })
+    await test.entered
+    controller.abort(new Error('caller stopped'))
+    handle.terminate()
+    const quiescent = handle.waitForExit()
+    test.release()
+    if (pause === 'start') test.finished.resolve(emptyResult)
+    await Promise.allSettled([handle.done])
+    expect(await quiescent).toBe(true)
+    handle.terminate()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+    if (pause !== 'start') expect(test.calls.some(call => call.method === 'process.start')).toBe(false)
+  })
+
+  it('reports range completion when preparation fails before any process can start', async () => {
+    const test = await setup({ failPrepare: new Error('prepare refused') })
+    const handle = test.runtime.spawn(spec)
+    const quiescent = handle.waitForExit()
+    await expect(handle.done).rejects.toThrow('prepare refused')
+    expect(await quiescent).toBe(true)
+    expect(test.calls.some(call => call.method === 'process.start' || call.method === 'process.terminate')).toBe(false)
+  })
+
+  it('reports an unconfirmed cleanup and closes the SSH connection', async () => {
+    const test = await setup({ failStart: new Error('start lost'), failTerminate: new Error('cleanup lost') })
+    test.connection.dispose.mockRejectedValue(new Error('connection cleanup failed'))
+    const handle = test.runtime.spawn(spec)
+    await expect(handle.done).rejects.toThrow('start lost')
+    await expect(handle.waitForExit()).rejects.toThrow('cleanup lost')
+    await test.close()
+    expect(test.disposalErrors).toEqual([expect.objectContaining({ message: 'SSH process cleanup could not be confirmed' })])
+    expect(test.connection.dispose).toHaveBeenCalled()
+  })
+
+  it('normalizes a non-Error transport rejection for stream consumers', async () => {
+    const test = await setup()
+    const handle = test.runtime.spawn(spec)
+    const error = new Promise<Error>(resolve => handle.stdout!.once('error', resolve))
+    await test.started
+    test.finished.reject('connection lost')
+    await expect(handle.done).rejects.toBe('connection lost')
+    expect((await error).message).toBe('connection lost')
+    expect(await handle.waitForExit()).toBe(true)
+  })
+
+  it('reports a live range until termination and rejects new work after disposal', async () => {
+    const test = await setup({ wait: false })
+    const handle = test.runtime.spawn(spec)
+    await test.started
+    expect(await handle.waitForExit()).toBe(false)
+    handle.terminate()
+    expect(await handle.waitForExit()).toBe(true)
+    await test.close()
+    expect(() => test.runtime.spawn(spec)).toThrow('disposed')
+    expect(() => test.runtime.spawn({ ...spec, signal: AbortSignal.abort() })).toThrow()
+  })
+})

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

@@ -0,0 +1,209 @@
+/** Published terminal operations and cleanup failures over the remote provider seam. */
+import { duplexPair } from 'node:stream'
+import { once } from 'node:events'
+import { Context } from '@deepseek-ai/cordis'
+import type { SubprocessTerminalSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { describe, expect, it, onTestFinished, vi } from 'vitest'
+import { z } from 'zod'
+import { SshSubprocessRuntime } from '../src/index.ts'
+
+const id = 'aa910b47-e7d7-467b-8421-3331569dd02b'
+const spec: SubprocessTerminalSpawnSpec = { argv: ['bash'], cwd: '/remote/workspace', rows: 24, cols: 80, graceMs: 100 }
+const completion = { outcome: { exitCode: 0, signal: null }, spills: {}, collected: {} }
+
+async function setup(options: {
+  connectFailure?: Error
+  terminateFailure?: Error | undefined
+  pauseConnect?: boolean
+  missingEndpoint?: boolean
+} = {}) {
+  const ctx = new Context()
+  const cleanupErrors: unknown[] = []
+  ctx.logger.error = ((error: unknown) => { cleanupErrors.push(error) }) as typeof ctx.logger.error
+  const [host, remote] = duplexPair({ allowHalfOpen: true })
+  host.on('error', () => {})
+  remote.on('error', () => {})
+  const entered = Promise.withResolvers<undefined>()
+  const allocationAborted = Promise.withResolvers<undefined>()
+  const release = Promise.withResolvers<undefined>()
+  const finished = Promise.withResolvers<typeof completion>()
+  const calls: { method: string; params: unknown; longRunning?: boolean }[] = []
+  let foreground: { processGroupId: number; inputWaiting: boolean } | null = null
+  const connection = {
+    dispose: vi.fn(() => { host.destroy(); remote.destroy(); return Promise.resolve() }),
+    request: async <T>(method: string, params: unknown, schema: z.ZodType<T>, _signal?: AbortSignal, longRunning?: boolean): Promise<T> => {
+      calls.push({ method, params, ...(longRunning === undefined ? {} : { longRunning }) })
+      let value: unknown
+      if (method === 'process.prepare') {
+        value = { id, streams: options.missingEndpoint ? {} : { terminal: { path: '/tmp/terminal', capability: 'a'.repeat(64) } } }
+      } else if (method === 'process.start') value = { pid: 321 }
+      else if (method === 'process.done') value = await finished.promise
+      else if (method === 'process.terminate') {
+        if (options.terminateFailure !== undefined) throw options.terminateFailure
+        finished.resolve(completion)
+        value = null
+      } else if (method === 'terminal.write') value = null
+      else if (method === 'terminal.inspect') value = foreground
+      else if (method === 'terminal.signal') value = 321
+      else throw new Error(`Unexpected terminal request ${method}`)
+      return schema.parse(value)
+    },
+    connectStream: vi.fn(async (_endpoint: unknown, signal?: AbortSignal) => {
+      entered.resolve(undefined)
+      signal?.addEventListener('abort', () => { allocationAborted.resolve(undefined) }, { once: true })
+      if (options.pauseConnect) await release.promise
+      if (options.connectFailure !== undefined) throw options.connectFailure
+      return host
+    }),
+  }
+  ctx.provide('ssh', connection as never)
+  const fiber = await ctx.plugin(SshSubprocessRuntime)
+  let closing: Promise<void> | undefined
+  const dispose = (): Promise<void> => closing ??= fiber.dispose()
+  const close = (): Promise<void> => {
+    release.resolve(undefined)
+    finished.resolve(completion)
+    host.destroy()
+    remote.destroy()
+    return dispose()
+  }
+  onTestFinished(async () => {
+    host.destroy()
+    remote.destroy()
+    if (closing === undefined) await close()
+  })
+  return {
+    runtime: ctx.subprocess, connection, calls, host, remote, finished, close, dispose, cleanupErrors,
+    entered: entered.promise, release: () => { release.resolve(undefined) },
+    allocationAborted: allocationAborted.promise,
+    setForeground: (value: typeof foreground) => { foreground = value },
+  }
+}
+
+describe('SSH terminal behavior', () => {
+  it('publishes output and forwards terminal operations with remote process observations', async () => {
+    const test = await setup()
+    const handle = await test.runtime.spawnTerminal({ ...spec, env: { KEEP: 'value' } })
+    expect(handle.pid).toBe(321)
+    const data = once(handle.output, 'data')
+    test.remote.write('terminal output')
+    expect(String((await data)[0])).toBe('terminal output')
+    await handle.write('input\n')
+    expect(await handle.inspectForeground()).toBeUndefined()
+    test.setForeground({ processGroupId: 321, inputWaiting: true })
+    expect(await handle.inspectForeground()).toEqual({ processGroupId: 321, inputWaiting: true })
+    expect(await handle.signalForeground('SIGINT')).toBe(321)
+    expect(test.calls.find(call => call.method === 'process.prepare')?.params).toEqual({
+      argv: ['bash'], cwd: spec.cwd, env: { KEEP: 'value' }, graceMs: 100,
+      terminal: { rows: 24, cols: 80 },
+    })
+    expect(test.calls.find(call => call.method === 'terminal.write')?.params).toEqual({ id, value: 'input\n' })
+    expect(test.calls.find(call => call.method === 'terminal.signal')?.params).toEqual({ id, value: 'SIGINT' })
+    test.finished.resolve(completion)
+    expect(await handle.done).toEqual({ exitCode: 0, signal: null })
+    await Promise.all([handle.terminate(), handle.terminate()])
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+    expect(test.calls.find(call => call.method === 'process.terminate')?.longRunning).toBe(true)
+    expect(test.host.destroyed).toBe(true)
+    expect(handle.output.destroyed).toBe(true)
+    await test.close()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+  })
+
+  it('retries termination after a remote cleanup failure without discarding the live handle', async () => {
+    const failure = new Error('range observation failed')
+    const options = { terminateFailure: failure as Error | undefined }
+    const test = await setup(options)
+    const handle = await test.runtime.spawnTerminal(spec)
+    await expect(handle.terminate()).rejects.toBe(failure)
+    expect(handle.output.destroyed).toBe(false)
+    expect(test.connection.dispose).not.toHaveBeenCalled()
+    options.terminateFailure = undefined
+    await handle.terminate()
+    expect(await handle.done).toEqual(completion.outcome)
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(2)
+    expect(handle.output.destroyed).toBe(true)
+  })
+
+  it('keeps termination available after the direct-result observation fails', async () => {
+    const test = await setup()
+    const handle = await test.runtime.spawnTerminal(spec)
+    const failure = new Error('direct result channel closed')
+    const rejected = expect(handle.done).rejects.toBe(failure)
+    test.finished.reject(failure)
+    await rejected
+    await handle.terminate()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+    expect(handle.output.destroyed).toBe(true)
+  })
+
+  it('terminates a published terminal when its caller aborts', async () => {
+    const test = await setup()
+    const controller = new AbortController()
+    const handle = await test.runtime.spawnTerminal({ ...spec, signal: controller.signal })
+    controller.abort(new Error('caller stopped'))
+    await handle.terminate()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+    expect(handle.output.destroyed).toBe(true)
+    expect(await handle.done).toEqual(completion.outcome)
+  })
+
+  it('releases the SSH connection when abort cannot confirm remote termination', async () => {
+    const options = { terminateFailure: new Error('termination lost') as Error | undefined }
+    const test = await setup(options)
+    test.connection.dispose.mockRejectedValueOnce(new Error('transport already lost'))
+    const controller = new AbortController()
+    const handle = await test.runtime.spawnTerminal({ ...spec, signal: controller.signal })
+    controller.abort()
+    await expect.poll(() => test.connection.dispose.mock.calls.length).toBe(1)
+    options.terminateFailure = undefined
+    await handle.terminate()
+    expect(handle.output.destroyed).toBe(true)
+  })
+
+  it('rejects a terminal reservation without its required stream and confirms cleanup', async () => {
+    const test = await setup({ missingEndpoint: true })
+    await expect(test.runtime.spawnTerminal(spec)).rejects.toBeInstanceOf(z.ZodError)
+    expect(test.connection.connectStream).not.toHaveBeenCalled()
+    expect(test.calls.map(call => call.method)).toEqual(['process.prepare', 'process.terminate'])
+  })
+
+  it('reports both unpublished allocation and cleanup failures during provider disposal', async () => {
+    const connectFailure = new Error('stream connection failed')
+    const terminateFailure = new Error('remote cleanup failed')
+    const test = await setup({ connectFailure, terminateFailure, pauseConnect: true })
+    test.connection.dispose.mockRejectedValueOnce(new Error('transport already lost'))
+    const allocation = test.runtime.spawnTerminal(spec)
+    const failed = expect(allocation).rejects.toMatchObject({ errors: [connectFailure, terminateFailure] })
+    await test.entered
+    const disposal = test.dispose()
+    await test.allocationAborted
+    test.release()
+    await Promise.all([failed, disposal])
+    expect(test.cleanupErrors).toMatchObject([{
+      message: 'SSH process cleanup could not be confirmed',
+      errors: [expect.objectContaining({ errors: [connectFailure, terminateFailure] })],
+    }])
+    expect(test.connection.dispose).toHaveBeenCalledOnce()
+    expect(test.calls.some(call => call.method === 'process.start')).toBe(false)
+  })
+
+  it('refuses admission after caller cancellation or provider disposal', async () => {
+    const test = await setup()
+    const stopped = AbortSignal.abort(new Error('cancelled before admission'))
+    await expect(test.runtime.spawnTerminal({ ...spec, signal: stopped })).rejects.toThrow('cancelled before admission')
+    expect(test.calls).toEqual([])
+    await test.close()
+    await expect(test.runtime.spawnTerminal(spec)).rejects.toThrow('disposed')
+    expect(test.calls).toEqual([])
+  })
+
+  it('terminates a published terminal when the provider is disposed', async () => {
+    const test = await setup()
+    const handle = await test.runtime.spawnTerminal(spec)
+    await test.close()
+    expect(test.calls.filter(call => call.method === 'process.terminate')).toHaveLength(1)
+    expect(handle.output.destroyed).toBe(true)
+    expect(await handle.done).toEqual(completion.outcome)
+  })
+})

packages/subprocess/subprocess-local/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/subprocess/subprocess-local/README.md
-README.md: a4d07e4fd287263cacd98b525479421bd961f008
-README.zh.md: 908e267d6de8d1a60a05d52c0b94929961c26919
+README.md: a1a6916e5330737e47eeed29b07b9750b9249b4b
+README.zh.md: f6d04b5d21b53a862e01f5a55b49670e643b0f84

packages/subprocess/subprocess-local/README.md

@@ -44,6 +44,8 @@ Absolute executable paths are verified; bare names resolve against the scrubbed
 
 Collect mode keeps the last `maxBytes` of a stream in memory — errors and final results cluster at the end — and, when a `spill` cap is configured, appends the complete stream to a private file under a per-process directory in the OS temp dir (a `0700` directory, `0600` random-named files). A stream larger than the spill cap discards its incomplete spill and returns only the marked truncated tail. Reads are offset-based and non-consuming, so background and batch readers coexist before and after exit.
 
+The `./output` export shares this collector and retained-spill storage with process adapters. `snapshot()` returns the retained raw bytes and total byte count, allowing remote adapters to preserve offsets without forwarding the complete stream.
+
 ### Control transport
 
 An ordinary spawn can request the [subprocess control pipe](../subprocess/README.md#using-a-control-pipe). A Node target receives fd 7 on every supported host; Windows descriptor numbering requires CRT initialization. POSIX runners preserve that descriptor across `execve`; Windows Job and ACL runners establish it in the child's CRT startup table before Node initializes and close their own carrier copies after spawning. Standard streams and the runner's private management channel remain independent.

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

@@ -44,6 +44,8 @@ kind: "package-reference"
 
 收集模式在内存中保留一条流的最后 `maxBytes`——错误与最终结果通常聚集在末尾——并在配置了 `spill` 上限时把完整流追加到 OS 临时目录下每进程目录中的私有文件（`0700` 目录、`0600` 随机命名文件）。某条流大于 spill 上限时，会丢弃不完整的 spill，只返回带截断标记的尾部。读取基于偏移量且从不消费，因此后台读取与批量读取在退出前后都可以共存。
 
+`./output` 导出向进程适配器共享该收集器与保留 spill 的存储。`snapshot()` 返回保留的原始字节及总字节数，使远程适配器能够保留偏移量，而无需转发完整的流。
+
 ### 控制传输
 
 普通 spawn 可以请求 [subprocess 控制管道](../subprocess/README.zh.md#using-a-control-pipe)。Node 目标在所有受支持的宿主上均收到 fd 7；Windows 描述符编号依赖 CRT 初始化。POSIX runner 在 `execve` 时保留该描述符；Windows Job 和 ACL runner 在 Node 初始化前通过子进程的 CRT 启动表建立它，并在 spawn 后关闭自身的承载副本。标准流与 runner 的私有管理通道保持独立。

packages/subprocess/subprocess-local/package.json

@@ -33,9 +33,9 @@
     "lib/index.js",
     "lib/runner.js",
     "lib/runner-*.js",
+    "lib/output.js",
     "scripts/ensure-spawn-helper.mjs",
-    "lib/types/**/*.d.ts",
-    "lib/output.js"
+    "lib/types/**/*.d.ts"
   ],
   "scripts": {
     "postinstall": "node scripts/ensure-spawn-helper.mjs"

pnpm-lock.yaml

@@ -9176,6 +9176,9 @@ importers:
       '@deepseek-ai/cordis':
         specifier: workspace:^
         version: link:../../../vendor/cordis
+      '@deepseek-ai/dsh-brand':
+        specifier: workspace:^
+        version: link:../../util/brand
       '@deepseek-ai/dsh-code-runtime-node':
         specifier: workspace:^
         version: link:../../code-runtime/code-runtime-node

scripts/check-workspace-constraints.ts

@@ -177,11 +177,18 @@ const packageFileExtras: Readonly<Record<string, readonly string[]>> = {
   '@deepseek-ai/dsh-sandbox-windows-acl': ['lib/runner.js', 'lib/types-*.js'],
   '@deepseek-ai/dsh-skill-badge': ['assets'],
   '@deepseek-ai/dsh-subprocess': ['lib/control.js'],
+  // SSH launches a private helper and shares wire definitions and TLS setup
+  // between that helper and the connection owner.
+  '@deepseek-ai/dsh-ssh': [
+    'lib/helper.js', 'lib/protocol.js', 'lib/schemas.js',
+    'lib/protocol-*.js', 'lib/schemas-*.js', 'lib/stream-security-*.js',
+  ],
   // Ordinary native containment ships a path-loaded runner and its shared
   // runner chunk beside the existing node-pty permission repair.
   '@deepseek-ai/dsh-subprocess-local': [
     'lib/runner.js',
     'lib/runner-*.js',
+    'lib/output.js',
     'scripts/ensure-spawn-helper.mjs',
   ],
   // tsdown shares the repository/pack code between the lib entry and the bin