feat(ssh): checkpoint POSIX helper and remote providers

packages/ssh/fs-ssh/package.json

@@ -0,0 +1,47 @@
+{
+  "name": "@deepseek-ai/dsh-fs-ssh",
+  "description": "Filesystem provider over the shared POSIX SSH helper",
+  "version": "0.1.5-rc.2",
+  "publishConfig": {
+    "access": "public"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/deepseek-ai/deepseek-harness.git",
+    "directory": "packages/ssh/fs-ssh"
+  },
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib/*.js",
+    "lib/types/**/*.d.ts"
+  ],
+  "license": "MIT",
+  "peerDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-fs": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-policy": "workspace:^"
+  },
+  "dependencies": {
+    "@deepseek-ai/schemastery": "workspace:^",
+    "zod": "^4.4.3"
+  },
+  "devDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-fs": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-policy": "workspace:^"
+  }
+}

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

@@ -0,0 +1,100 @@
+/** Filesystem provider preserving remote identities and helper-owned atomic mutations. */
+import { posix } from 'node:path'
+import { FileSystem, FsError } from '@deepseek-ai/dsh-fs'
+import type { FsDirEntry, FsEditOutcome, FsEditRequest, FsErrorCode, FsInfo, FsPathInfo, FsTarget, FsVersion, FsWriteIntent, FsWriteOutcome } from '@deepseek-ai/dsh-fs'
+import type { SandboxExecutionPolicy, SandboxMode } from '@deepseek-ai/dsh-sandbox'
+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 { z } from 'zod'
+
+/** Remote filesystem paired with the SSH subprocess and sandbox providers. */
+export class SshFileSystem extends FileSystem {
+  static inject = ['ssh', 'sandboxPolicy']
+
+  override get sandboxMode(): SandboxMode { return this.ctx.sandboxPolicy.defaultMode }
+
+  override async resolve(path: string, opts?: { cwd?: string; signal?: AbortSignal }): Promise<FsTarget> {
+    return await this.call('fs.resolve', { path, cwd: opts?.cwd }, targetSchema, opts?.signal) as FsTarget
+  }
+
+  override processPath(target: FsTarget): string { return String(target.targetKey) }
+
+  override fileUrl(target: FsTarget): string {
+    const url = new URL('file:///')
+    url.pathname = this.processPath(target)
+    return url.href
+  }
+
+  override contains(parent: FsTarget, child: FsTarget): boolean {
+    const path = posix.relative(this.processPath(parent), this.processPath(child))
+    return path === '' || (!path.startsWith('../') && path !== '..' && !posix.isAbsolute(path))
+  }
+
+  override async stat(target: FsTarget, signal?: AbortSignal): Promise<FsInfo | undefined> {
+    return await this.call('fs.stat', { target }, infoSchema.nullable(), signal) as FsInfo | null ?? undefined
+  }
+
+  override async lstat(path: string, opts?: { cwd?: string }, signal?: AbortSignal): Promise<FsPathInfo | undefined> {
+    return await this.call('fs.lstat', { path, cwd: opts?.cwd }, pathInfoSchema.nullable(), signal) as FsPathInfo | null ?? undefined
+  }
+
+  override readText(target: FsTarget, signal?: AbortSignal): Promise<string> {
+    return this.call('fs.readText', { target }, z.string(), signal)
+  }
+
+  override async streamText(target: FsTarget, signal?: AbortSignal): Promise<AsyncIterable<string>> {
+    const id = await this.call('fs.stream', { target }, z.string().uuid(), signal)
+    const call = this.call.bind(this)
+    return (async function* () {
+      let ended = false
+      try {
+        while (!ended) {
+          signal?.throwIfAborted()
+          const next = await call('fs.next', { id }, z.object({ done: z.boolean(), value: z.string() }).strict(), signal)
+          ended = next.done
+          if (next.value.length > 0) yield next.value
+        }
+      } finally {
+        if (!ended) await call('fs.streamClose', { id }, z.null()).catch(() => {})
+      }
+    })()
+  }
+
+  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')
+  }
+
+  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')
+  }
+
+  override async listDir(target: FsTarget, signal?: AbortSignal): Promise<FsDirEntry[]> {
+    return await this.call('fs.list', { target }, entriesSchema, signal) as FsDirEntry[]
+  }
+
+  override async writeText(
+    target: FsTarget, content: string, expected?: FsWriteIntent, signal?: AbortSignal, sandboxPolicy?: SandboxExecutionPolicy,
+  ): Promise<FsWriteOutcome> {
+    const policy = sandboxPolicy ?? this.ctx.sandboxPolicy.resolve()
+    return await this.call('fs.write', { target, content, expected, policy }, writeResultSchema, signal) as FsWriteOutcome
+  }
+
+  override async editText(
+    target: FsTarget, edit: FsEditRequest, expected?: { version: FsVersion },
+    signal?: AbortSignal, sandboxPolicy?: SandboxExecutionPolicy,
+  ): Promise<FsEditOutcome> {
+    const policy = sandboxPolicy ?? this.ctx.sandboxPolicy.resolve()
+    return await this.call('fs.edit', { target, edit, expected, policy }, editResultSchema, signal) as FsEditOutcome
+  }
+
+  private async call<T>(method: string, params: unknown, schema: z.ZodType<T>, signal?: AbortSignal): Promise<T> {
+    try { return await this.ctx.ssh.request(method, params, schema, signal) } catch (error) {
+      if (error instanceof RemoteOperationError && error.code?.startsWith('FS_')) throw new FsError(error.message, error.code as FsErrorCode, { cause: error })
+      throw new FsError(error instanceof Error ? error.message : String(error), signal?.aborted ? 'FS_ABORTED' : 'FS_IO_ERROR', { cause: error })
+    }
+  }
+}
+
+export default SshFileSystem

packages/ssh/fs-ssh/tsconfig.json

@@ -0,0 +1,30 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
+    {
+      "path": "../ssh"
+    },
+    {
+      "path": "../../fs/fs"
+    },
+    {
+      "path": "../../sandbox/sandbox"
+    },
+    {
+      "path": "../../sandbox/sandbox-policy"
+    }
+  ]
+}

packages/ssh/sandbox-ssh/package.json

@@ -0,0 +1,43 @@
+{
+  "name": "@deepseek-ai/dsh-sandbox-ssh",
+  "description": "Remote POSIX sandbox argv provider over the shared SSH helper",
+  "version": "0.1.5-rc.2",
+  "publishConfig": {
+    "access": "public"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/deepseek-ai/deepseek-harness.git",
+    "directory": "packages/ssh/sandbox-ssh"
+  },
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib/*.js",
+    "lib/types/**/*.d.ts"
+  ],
+  "license": "MIT",
+  "peerDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^"
+  },
+  "dependencies": {
+    "@deepseek-ai/schemastery": "workspace:^",
+    "zod": "^4.4.3"
+  },
+  "devDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^"
+  }
+}

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

@@ -0,0 +1,34 @@
+/** Remote argv wrapper that selects and applies the sandbox on the SSH host. */
+import { Service } from '@deepseek-ai/cordis'
+import { SandboxProvider, SandboxUnavailableError } from '@deepseek-ai/dsh-sandbox'
+import type { ConfinedArgv, SandboxPolicy } from '@deepseek-ai/dsh-sandbox'
+import type {} from '@deepseek-ai/dsh-ssh'
+import { z } from 'zod'
+
+const factsSchema = z.object({ argv: z.array(z.string()), enforcement: z.enum(['full', 'partial']), denialSignatures: z.array(z.string()), runnerFailureRules: z.array(z.object({ allowedExitCodes: z.array(z.number().int()).optional(), fatalSignatures: z.array(z.string()), informationalLines: z.array(z.string()).optional() }).strict()) }).strict()
+
+/** Synchronous wrapping uses verified remote backend facts established before the composition becomes ready. */
+export class SshSandboxProvider extends SandboxProvider {
+  static inject = ['ssh']
+  private facts: z.infer<typeof factsSchema> | undefined
+  private node: string | undefined
+
+  /** Verify the remote backend before exposing synchronous argv wrapping. */
+  async [Service.init](): Promise<void> {
+    const hello = await this.ctx.ssh.ready
+    this.node = hello.node
+    this.facts = await this.ctx.ssh.request('sandbox', { argv: ['true'], policy: { mode: 'read-only', workspaceRoot: hello.workspace } }, factsSchema)
+  }
+
+  override confine(argv: readonly string[], policy: SandboxPolicy): ConfinedArgv {
+    if (this.facts === undefined || this.node === undefined) throw new SandboxUnavailableError(policy.mode, 'SSH sandbox has not established its remote backend')
+    return {
+      argv: [this.node, this.ctx.ssh.helperPath, '--confine', JSON.stringify(policy), '--', ...argv],
+      enforcement: this.facts.enforcement,
+      denialSignatures: this.facts.denialSignatures,
+      runnerFailureRules: [...this.facts.runnerFailureRules.map(rule => ({ fatalSignatures: rule.fatalSignatures, ...(rule.allowedExitCodes === undefined ? {} : { allowedExitCodes: rule.allowedExitCodes }), ...(rule.informationalLines === undefined ? {} : { informationalLines: rule.informationalLines }) })), { allowedExitCodes: [127], fatalSignatures: ['dsh-ssh-sandbox: '] }],
+    }
+  }
+}
+
+export default SshSandboxProvider

packages/ssh/sandbox-ssh/tsconfig.json

@@ -0,0 +1,24 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
+    {
+      "path": "../ssh"
+    },
+    {
+      "path": "../../sandbox/sandbox"
+    }
+  ]
+}

packages/ssh/ssh/package.json

@@ -0,0 +1,69 @@
+{
+  "name": "@deepseek-ai/dsh-ssh",
+  "description": "Shared OpenSSH connection and versioned POSIX remote helper",
+  "version": "0.1.5-rc.2",
+  "publishConfig": {
+    "access": "public"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/deepseek-ai/deepseek-harness.git",
+    "directory": "packages/ssh/ssh"
+  },
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json",
+    "./helper": {
+      "types": "./lib/types/helper.d.ts",
+      "default": "./lib/helper.js"
+    },
+    "./protocol": {
+      "types": "./lib/types/protocol.d.ts",
+      "default": "./lib/protocol.js"
+    },
+    "./schemas": {
+      "types": "./lib/types/schemas.d.ts",
+      "default": "./lib/schemas.js"
+    }
+  },
+  "files": [
+    "lib/*.js",
+    "lib/types/**/*.d.ts"
+  ],
+  "license": "MIT",
+  "peerDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-fs": "workspace:^",
+    "@deepseek-ai/dsh-fs-local": "workspace:^",
+    "@deepseek-ai/dsh-fs-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-subprocess-local": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-local": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
+    "@deepseek-ai/dsh-session-projection": "workspace:^"
+  },
+  "dependencies": {
+    "@deepseek-ai/schemastery": "workspace:^",
+    "zod": "^4.4.3"
+  },
+  "devDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-fs": "workspace:^",
+    "@deepseek-ai/dsh-fs-local": "workspace:^",
+    "@deepseek-ai/dsh-fs-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-subprocess-local": "workspace:^",
+    "@deepseek-ai/dsh-sandbox": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-local": "workspace:^",
+    "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
+    "@deepseek-ai/dsh-session-projection": "workspace:^"
+  }
+}

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

@@ -0,0 +1,244 @@
+/** Remote process ownership and separately forwarded byte streams. */
+import { randomUUID } from 'node:crypto'
+import { chmod, mkdir, rm } from 'node:fs/promises'
+import { join } from 'node:path'
+import { createServer, type Server, type Socket } from 'node:net'
+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 { OutputCollector } from '@deepseek-ai/dsh-subprocess-local/output'
+import { spawnSchema } from './schemas.ts'
+import { z } from 'zod'
+
+type Request = z.infer<typeof spawnSchema>
+type Channel = 'stdin' | 'stdout' | 'stderr' | 'control' | 'terminal'
+interface Endpoint { path: string; server: Server; connected: Promise<Socket>; socket?: Socket }
+interface ProcessRecord {
+  request: Request
+  directory: string
+  endpoints: Partial<Record<Channel, Endpoint>>
+  ordinary?: SubprocessHandle
+  terminal?: SubprocessTerminalHandle
+  start?: Promise<void>
+  done?: Promise<{ outcome: { exitCode: number | null; signal: NodeJS.Signals | null }; spills: { stdout?: string; stderr?: string } }>
+  expiry: NodeJS.Timeout
+}
+
+/** 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 closing = false
+
+  constructor(
+    private readonly ctx: Context, private readonly root: string,
+    private readonly limit: number, private readonly preparationMs: number,
+  ) {}
+
+  /** Allocate private stream listeners; no target executes until start(). */
+  async prepare(raw: unknown): Promise<{ id: string; streams: Partial<Record<Channel, string>> }> {
+    if (this.closing || this.records.size >= this.limit) throw new Error('SSH process capacity unavailable')
+    const request = spawnSchema.parse(raw)
+    const id = randomUUID()
+    const directory = join(this.root, id)
+    await mkdir(directory, { mode: 0o700 })
+    const record: ProcessRecord = {
+      request, directory, endpoints: {},
+      expiry: setTimeout(() => { void this.release(id).catch(() => {}) }, this.preparationMs),
+    }
+    this.records.set(id, record)
+    try {
+      const names: Channel[] = request.terminal === undefined
+        ? ['stdout', 'stderr', ...(request.stdio?.stdin === 'pipe' ? ['stdin' as const] : []), ...(request.stdio?.control === 'pipe' ? ['control' as const] : [])]
+        : ['terminal']
+      for (const name of names) record.endpoints[name] = await this.endpoint(join(directory, name))
+      return { id, streams: Object.fromEntries(Object.entries(record.endpoints).map(([name, endpoint]) => [name, endpoint.path])) }
+    } catch (error) {
+      await this.release(id)
+      throw error
+    }
+  }
+
+  /** Start once all data channels are connected; duplicate starts refuse rather than replay. */
+  async start(id: string): Promise<{ pid?: number }> {
+    const record = this.record(id)
+    if (record.start !== undefined) throw new Error('SSH process launch was already requested')
+    record.start = this.startOnce(id, record)
+    await record.start
+    return record.terminal === undefined ? {} : { pid: record.terminal.pid }
+  }
+
+  private async startOnce(id: string, 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')
+    const request = record.request
+    const cwd = this.ctx.fs.processPath(await this.ctx.fs.resolve(request.cwd))
+    const env = request.env === undefined ? {} : Object.fromEntries(
+      Object.entries(request.env).map(([key, value]) => [key, value ?? undefined]),
+    )
+    if (request.terminal !== undefined) {
+      const terminal = await this.ctx.subprocess.spawnTerminal({
+        argv: request.argv, cwd,
+        env: Object.fromEntries(Object.entries(env).filter((entry): entry is [string, string] => entry[1] !== undefined)),
+        graceMs: request.graceMs, ...request.terminal,
+      })
+      record.terminal = terminal
+      const socket = await (record.endpoints.terminal as Endpoint).connected
+      const output = pipeline(terminal.output, socket).catch(() => {})
+      record.done = terminal.done.then(outcome => ({ outcome, spills: {} }))
+      void record.done.catch(() => {})
+      void record.done.then(async (result) => {
+        await terminal.terminate()
+        await output
+        this.rememberCompleted(id, record, result)
+      }).catch(() => {})
+      return
+    }
+    const stdio = request.stdio as NonNullable<Request['stdio']>
+    const spec: SubprocessSpawnSpec = { argv: request.argv, cwd, env, graceMs: request.graceMs,
+      stdio: { stdin: stdio.stdin, stdout: 'pipe', stderr: 'pipe', ...(stdio.control === undefined ? {} : { control: stdio.control }) },
+    }
+    const ordinary = this.ctx.subprocess.spawn(spec)
+    record.ordinary = ordinary
+    const collectors: Partial<Record<'stdout' | 'stderr', OutputCollector>> = {}
+    const streams: Promise<void>[] = []
+    for (const name of ['stdout', 'stderr'] as const) {
+      const stream = ordinary[name] as Readable
+      const mode = stdio[name]
+      if (typeof mode === 'object') {
+        const collector = new OutputCollector(mode.maxBytes, mode.spill?.maxBytes, name, record.directory)
+        collectors[name] = collector
+        stream.on('data', (chunk: Buffer) => { collector.push(chunk) })
+      }
+      const socket = await (record.endpoints[name] as Endpoint).connected
+      streams.push(pipeline(stream, socket).catch(() => {}))
+    }
+    if (record.endpoints.stdin !== undefined) {
+      const socket = await record.endpoints.stdin.connected
+      void pipeline(socket, ordinary.stdin as Writable).catch(() => {})
+    }
+    if (record.endpoints.control !== undefined) {
+      const control = (ordinary as SubprocessHandle & { control?: Duplex }).control
+      if (control === undefined) { ordinary.terminate(); throw new Error('Remote subprocess provider did not establish fd 7') }
+      const socket = await record.endpoints.control.connected
+      socket.pipe(control).pipe(socket)
+      socket.on('error', () => { control.destroy() })
+      control.on('error', () => { socket.destroy() })
+      streams.push(finished(socket, { readable: false, cleanup: true }).catch(() => {}))
+    }
+    record.done = ordinary.done.then(async (outcome) => {
+      // A paused output reader may defer EOF, but must not retain the process-range owner.
+      await Promise.race([
+        Promise.all(streams),
+        new Promise<void>((resolve) => { const timer = setTimeout(resolve, request.graceMs); timer.unref() }),
+      ])
+      const spills: { stdout?: string; stderr?: string } = {}
+      for (const name of ['stdout', 'stderr'] as const) {
+        collectors[name]?.seal()
+        const path = collectors[name]?.readFrom(0).spillPath
+        if (path !== undefined) spills[name] = path
+      }
+      return { outcome, spills }
+    })
+    void record.done.catch(() => {})
+    void record.done.then(async (result) => {
+      await ordinary.waitForExit()
+      await Promise.all(streams)
+      this.rememberCompleted(id, record, result)
+    }).catch(() => {})
+  }
+
+  /** Await the direct result without claiming all descendants have exited. */
+  async done(id: string): 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')
+    await record.start
+    return record.done
+  }
+
+  /** Observe the same native managed range used for termination. */
+  async wait(id: string, signal?: AbortSignal): Promise<boolean> {
+    if (this.completed.has(id)) return true
+    const record = this.record(id)
+    await record.start
+    if (record.ordinary !== undefined) return record.ordinary.waitForExit(signal)
+    if (record.terminal !== undefined) { await record.terminal.terminate(); return true }
+    throw new Error('SSH process was not started')
+  }
+
+  /** Start managed termination without serializing it behind output. */
+  async terminate(id: string): Promise<void> {
+    if (this.completed.has(id)) return
+    const record = this.record(id)
+    record.ordinary?.terminate()
+    if (record.terminal !== undefined) await record.terminal.terminate()
+    if (record.start === undefined) await this.release(id)
+  }
+
+  /** Operate on the one terminal owned by the request id. */
+  async terminal(id: string, 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)
+  }
+
+  /** Stop every owned process on lease expiry or disconnect. */
+  async close(): Promise<void> {
+    if (this.closing) return
+    this.closing = true
+    const outcomes = await Promise.allSettled([...this.records.keys()].map(id => this.release(id)))
+    const errors = outcomes.flatMap(result => result.status === 'rejected' ? [result.reason as unknown] : [])
+    if (errors.length > 0) throw new AggregateError(errors, 'SSH remote process cleanup failed')
+  }
+
+  private async release(id: string): Promise<void> {
+    const record = this.records.get(id)
+    if (record === undefined) return
+    clearTimeout(record.expiry)
+    for (const endpoint of Object.values(record.endpoints)) { endpoint.socket?.destroy(); endpoint.server.close() }
+    record.ordinary?.terminate()
+    if (record.ordinary !== undefined) await record.ordinary.waitForExit()
+    if (record.terminal !== undefined) await record.terminal.terminate()
+    this.records.delete(id)
+    await rm(record.directory, { recursive: true, force: true })
+  }
+
+  private record(id: string): ProcessRecord {
+    const record = this.records.get(id)
+    if (record === undefined) throw new Error('Unknown or expired SSH process handle')
+    return record
+  }
+
+  private rememberCompleted(id: string, record: ProcessRecord, result: unknown): 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)
+    // Completed spill files remain in the helper's private root until connection disposal.
+    for (const endpoint of Object.values(record.endpoints)) { endpoint.server.close(); endpoint.socket?.destroy() }
+  }
+
+  private async endpoint(path: string): Promise<Endpoint> {
+    const connected = Promise.withResolvers<Socket>()
+    const server = createServer({ allowHalfOpen: true })
+    const endpoint: Endpoint = { path, server, connected: connected.promise }
+    void connected.promise.catch(() => {})
+    server.on('connection', (socket) => {
+      if (endpoint.socket !== undefined) { socket.destroy(); return }
+      endpoint.socket = socket
+      socket.on('error', () => {})
+      server.close()
+      connected.resolve(socket)
+    })
+    server.on('error', (error) => { connected.reject(error) })
+    server.on('close', () => { if (endpoint.socket === undefined) connected.reject(new Error('SSH stream reservation closed')) })
+    await new Promise<void>((resolve, reject) => { server.once('error', reject); server.listen(path, resolve) })
+    await chmod(path, 0o600)
+    return endpoint
+  }
+}

packages/ssh/ssh/src/helper.ts

@@ -0,0 +1,213 @@
+/** Private POSIX SSH helper; filesystem and process effects use the installed local providers. */
+import { createHash } from 'node:crypto'
+import { readFileSync } from 'node:fs'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { spawnSync } from 'node:child_process'
+import { Context } from '@deepseek-ai/cordis'
+import { FsError, type FsTarget, type FsEditRequest, 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'
+import { SandboxPolicyService } from '@deepseek-ai/dsh-sandbox-policy'
+import { SessionProjectionRegistry } from '@deepseek-ai/dsh-session-projection'
+import type { SandboxExecutionPolicy, SandboxPolicy } from '@deepseek-ai/dsh-sandbox'
+import { z } from 'zod'
+import { SshRpcPeer, SSH_PROTOCOL_VERSION } from './protocol.ts'
+import { RemoteProcesses } from './helper-processes.ts'
+import { editSchema, environmentSchema, intentSchema, policySchema, remotePath, targetSchema } 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()
+
+async function services() {
+  const ctx = new Context()
+  const fibers = [await ctx.plugin(SessionProjectionRegistry)]
+  fibers.push(await ctx.plugin(SandboxPolicyService, { mode: 'read-only', workspaceRoot: process.cwd() }))
+  fibers.push(await ctx.plugin(SandboxedFileSystem, { cwd: process.cwd(), diffBasisMaxBytes: 10 * 1024 * 1024 }))
+  fibers.push(await ctx.plugin(LocalSubprocessRuntime))
+  fibers.push(await ctx.plugin(LocalSandboxProvider, { runnerCommand: [], runnerFailureSignatures: [], probeTimeoutMs: 5000 }))
+  return { ctx, close: async () => { for (const fiber of fibers.reverse()) await fiber.dispose() } }
+}
+
+/** Run the private helper until the SSH channel closes or its client lease expires. */
+export async function runSshHelper(): Promise<void> {
+  if (process.platform !== 'linux' && process.platform !== 'darwin') throw new Error('SSH helper requires a POSIX host')
+  const runtime = await services()
+  const { ctx } = runtime
+  const root = await mkdtemp(join(tmpdir(), 'dsh-ssh-'))
+  const processes = new RemoteProcesses(ctx, root, 128, 30_000)
+  const iterators = new Map<string, AsyncIterator<string>>()
+  let lease: NodeJS.Timeout | undefined
+  let leaseMs = 30_000
+  let initialized = false
+  let cleanup: Promise<void> | undefined
+  const close = (): Promise<void> => {
+    cleanup ??= (async () => {
+      if (lease !== undefined) clearTimeout(lease)
+      await Promise.allSettled([...iterators.values()].map(iterator => iterator.return?.()))
+      await processes.close()
+      await runtime.close()
+      await rm(root, { recursive: true, force: true })
+    })()
+    return cleanup
+  }
+  const touchLease = (): void => {
+    if (lease !== undefined) clearTimeout(lease)
+    lease = setTimeout(() => { peer.close(new Error('SSH helper client lease expired')); void close().catch(() => { process.exitCode = 1 }) }, leaseMs)
+  }
+  const policy = async (raw: unknown): Promise<SandboxExecutionPolicy> => {
+    const parsed = policySchema.parse(raw)
+    const target = await ctx.fs.resolve(parsed.workspaceRoot)
+    return { ...parsed, workspaceRoot: ctx.fs.processPath(target) } as SandboxExecutionPolicy
+  }
+  const asTarget = (raw: unknown): FsTarget => targetSchema.parse(raw) as FsTarget
+  const peer = new SshRpcPeer(process.stdin, process.stdout, MAX_FRAME_BYTES, 128, async (method, raw, signal) => {
+    if (method === 'hello') {
+      if (initialized) throw new Error('SSH helper handshake already completed')
+      const input = z.object({
+        protocol: z.literal(SSH_PROTOCOL_VERSION), workspace: remotePath, leaseMs: z.number().int().min(3000).max(600_000),
+      }).strict().parse(raw)
+      const workspace = ctx.fs.processPath(await ctx.fs.resolve(input.workspace, { signal }))
+      process.chdir(workspace)
+      leaseMs = input.leaseMs
+      initialized = true
+      touchLease()
+      return { protocol: SSH_PROTOCOL_VERSION, hash: createHash('sha256').update(readFileSync(fileURLToPath(import.meta.url))).digest('hex'), platform: process.platform, nodeVersion: process.version, node: process.execPath, root, workspace }
+    }
+    if (!initialized || cleanup !== undefined) throw new Error('SSH helper is not accepting operations')
+    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)
+    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 === 'terminal.write' || method === 'terminal.inspect' || method === 'terminal.signal') {
+      const input = z.object({ id: z.string().uuid(), 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') {
+      const input = z.object({ command: z.string(), env: environmentSchema.optional() }).strict().parse(raw)
+      const env = input.env === undefined ? undefined : Object.fromEntries(
+        Object.entries(input.env).filter((entry): entry is [string, string] => entry[1] !== null),
+      )
+      return ctx.subprocess.resolveExecutable(input.command, env, signal)
+    }
+    if (method === 'sandbox') {
+      const input = z.object({ argv: z.array(z.string()).min(1), policy: policySchema }).strict().parse(raw)
+      const resolved = await policy(input.policy)
+      if (resolved.mode === 'danger-full-access') throw new Error('Unconfined argv does not need a sandbox wrapper')
+      return ctx.sandbox.confine(input.argv, resolved as SandboxPolicy)
+    }
+    if (method === 'fs.resolve' || method === 'fs.lstat') {
+      const input = z.object({ path: z.string(), cwd: remotePath.optional() }).strict().parse(raw)
+      return method === 'fs.resolve' ? ctx.fs.resolve(input.path, { cwd: input.cwd ?? process.cwd(), signal }) : await ctx.fs.lstat(input.path, { cwd: input.cwd ?? process.cwd() }, signal) ?? null
+    }
+    if (method === 'fs.stat' || method === 'fs.list' || method === 'fs.readText' || method === 'fs.stream') {
+      const target = asTarget(z.object({ target: targetSchema }).strict().parse(raw).target)
+      if (method === 'fs.stat') return await ctx.fs.stat(target, signal) ?? null
+      if (method === 'fs.list') return ctx.fs.listDir(target, signal)
+      const stream = await ctx.fs.streamText(target, signal)
+      if (method === 'fs.stream') {
+        const { randomUUID } = await import('node:crypto')
+        const id = randomUUID()
+        if (iterators.size >= 128) throw new Error('SSH text stream limit reached')
+        iterators.set(id, stream[Symbol.asyncIterator]())
+        return id
+      }
+      let text = ''
+      let bytes = 0
+      for await (const chunk of stream) {
+        bytes += Buffer.byteLength(chunk)
+        if (bytes > MAX_TEXT_BYTES) throw new FsError('SSH whole-text transfer exceeds its bounded frame budget; use streaming', 'FS_TOO_LARGE')
+        text += chunk
+      }
+      return text
+    }
+    if (method === 'fs.next' || method === 'fs.streamClose') {
+      const { id } = idRequest.parse(raw)
+      const iterator = iterators.get(id)
+      if (iterator === undefined) throw new Error('Unknown SSH text stream')
+      if (method === 'fs.streamClose') { iterators.delete(id); await iterator.return?.(); return null }
+      const next = await iterator.next()
+      if (next.done) iterators.delete(id)
+      return { done: next.done ?? false, value: next.value ?? '' }
+    }
+    if (method === 'fs.readBytes' || method === 'fs.readRange') {
+      const input = z.object({
+        target: targetSchema, maxBytes: z.number().int().nonnegative().optional(),
+        offset: z.number().int().nonnegative().optional(), length: z.number().int().nonnegative().optional(),
+      }).strict().parse(raw)
+      const target = asTarget(input.target)
+      const bytes = method === 'fs.readBytes'
+        ? await ctx.fs.readBytes(target, signal, Math.min(z.number().int().nonnegative().parse(input.maxBytes), MAX_TEXT_BYTES))
+        : await ctx.fs.readByteRange(target, {
+          offset: z.number().int().nonnegative().parse(input.offset),
+          length: z.number().int().nonnegative().max(MAX_TEXT_BYTES).parse(input.length),
+        }, signal)
+      return Buffer.from(bytes).toString('base64')
+    }
+    if (method === 'fs.write' || method === 'fs.edit') {
+      const input = z.object({
+        target: targetSchema, content: z.string().optional(), edit: editSchema.optional(),
+        expected: z.union([intentSchema, z.object({ version: z.string() }).strict()]).optional(), policy: policySchema,
+      }).strict().parse(raw)
+      const target = asTarget(input.target)
+      const resolved = await policy(input.policy)
+      if (method === 'fs.write') return ctx.fs.writeText(
+        target, z.string().parse(input.content),
+        input.expected === undefined ? undefined : intentSchema.parse(input.expected) as FsWriteIntent, signal, resolved,
+      )
+      return ctx.fs.editText(
+        target, editSchema.parse(input.edit) as FsEditRequest,
+        input.expected === undefined
+          ? undefined : z.object({ version: z.string() }).strict().parse(input.expected) as { version: FsVersion },
+        signal, resolved,
+      )
+    }
+    throw new Error(`Unknown SSH helper operation: ${method}`)
+  })
+  peer.once('closed', () => { void close().catch(() => { process.exitCode = 1 }) })
+  const onSignal = (): void => { peer.close(new Error('SSH helper transport was terminated')) }
+  process.once('SIGTERM', onSignal)
+  process.once('SIGHUP', onSignal)
+  process.once('SIGINT', onSignal)
+  touchLease()
+  try {
+    await new Promise<void>((resolve) => { peer.once('closed', () => { resolve() }) })
+    await close()
+  } finally {
+    process.off('SIGTERM', onSignal)
+    process.off('SIGHUP', onSignal)
+    process.off('SIGINT', onSignal)
+  }
+}
+
+async function confine(): Promise<void> {
+  const runtime = await services()
+  try {
+    const policy = policySchema.parse(JSON.parse(process.argv[3] ?? ''))
+    if (policy.mode === 'danger-full-access' || process.argv[4] !== '--') throw new Error('Invalid remote sandbox invocation')
+    const workspaceRoot = runtime.ctx.fs.processPath(await runtime.ctx.fs.resolve(policy.workspaceRoot))
+    const wrapped = runtime.ctx.sandbox.confine(process.argv.slice(5), { ...policy, workspaceRoot } as SandboxPolicy)
+    const control = process.env.DSH_SUBPROCESS_CONTROL === 'pipe'
+    const result = spawnSync(wrapped.argv[0] as string, wrapped.argv.slice(1), {
+      stdio: control ? ['inherit', 'inherit', 'inherit', 'ignore', 'ignore', 'ignore', 'ignore', 7] : 'inherit',
+      env: process.env,
+    })
+    if (result.error !== undefined) throw result.error
+    process.exitCode = result.status ?? 128
+  } finally { await runtime.close() }
+}
+
+if (import.meta.main) {
+  void (process.argv[2] === '--confine' ? confine() : runSshHelper()).catch((error) => {
+    process.stderr.write(`dsh-ssh-sandbox: ${error instanceof Error ? error.message : String(error)}\n`)
+    process.exitCode = 127
+  })
+}

packages/ssh/ssh/src/index.ts

@@ -0,0 +1,180 @@
+/** OpenSSH connection owner for one version-matched POSIX helper and its independent forwarded streams. */
+
+import { spawn, execFile, type ChildProcessWithoutNullStreams } from 'node:child_process'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { createConnection, type Socket } from 'node:net'
+import { promisify } from 'node:util'
+import { Context, Service } from '@deepseek-ai/cordis'
+import schema from '@deepseek-ai/schemastery'
+import { z } from 'zod'
+import { SshRpcPeer, SSH_PROTOCOL_VERSION } from './protocol.ts'
+import { helloSchema } from './schemas.ts'
+
+const execute = promisify(execFile)
+type Hello = z.infer<typeof helloSchema>
+
+/** Deployment-owned SSH identity and installed helper; no model argument selects these values. */
+export interface Config {
+  /** OpenSSH host alias, including its existing user, key and known-host configuration. */
+  host: string
+  /** Absolute remote Node executable. */
+  node: string
+  /** Absolute path to the installed, bundled helper entry. */
+  helper: string
+  /** SHA-256 of that bundled helper; mismatches refuse the connection. */
+  helperHash: string
+  /** Absolute remote default workspace. */
+  workspace: string
+  /** Bound on connection establishment and individual administrative requests. */
+  requestTimeoutMs?: number
+  /** Maximum JSON payload bytes per helper request or response. */
+  maxFrameBytes?: number
+  /** Maximum outstanding administrative requests. */
+  maxPending?: number
+  /** Remote helper lease; loss of heartbeats starts remote managed cleanup. */
+  leaseMs?: number
+}
+
+declare module '@deepseek-ai/cordis' {
+  interface Context { ssh: SshConnection }
+}
+
+/** One non-reconnecting SSH session; loss invalidates all active operations. */
+export class SshConnection extends Service {
+  static Config: schema<Config> = schema.object({
+    host: schema.string().required(), node: schema.string().required(), helper: schema.string().required(),
+    helperHash: schema.string().required(), workspace: schema.string().required(),
+    requestTimeoutMs: schema.number().default(30_000), maxFrameBytes: schema.number().default(64 * 1024 * 1024),
+    maxPending: schema.number().default(128), leaseMs: schema.number().default(30_000),
+  })
+
+  /** Verified remote helper coordinates; callers must await this before launch. */
+  readonly ready: Promise<Hello>
+  /** Installed helper entry in the remote filesystem. */
+  readonly helperPath: string
+  private rpc: SshRpcPeer | undefined
+  private child: ChildProcessWithoutNullStreams | undefined
+  private childClosed: Promise<void> | undefined
+  private directory: string | undefined
+  private heartbeat: NodeJS.Timeout | undefined
+  private closed = false
+  private failure: Error | undefined
+  private sockets = new Set<Socket>()
+  private nextSocket = 0
+  private readonly config: Required<Config>
+
+  constructor(ctx: Context, config: Config) {
+    super(ctx, 'ssh')
+    if (process.platform !== 'linux' && process.platform !== 'darwin') throw new Error('SSH runtime requires a POSIX client')
+    this.config = z.object({
+      host: z.string().regex(/^[a-zA-Z0-9][a-zA-Z0-9_.@-]*$/),
+      node: z.string().startsWith('/'), helper: z.string().startsWith('/'), helperHash: z.string().regex(/^[0-9a-f]{64}$/),
+      workspace: z.string().startsWith('/'), requestTimeoutMs: z.number().int().positive(),
+      maxFrameBytes: z.number().int().positive().max(256 * 1024 * 1024), maxPending: z.number().int().positive(),
+      leaseMs: z.number().int().min(3000).max(600_000),
+    }).parse(config)
+    this.helperPath = this.config.helper
+    this.ready = this.start()
+    void this.ready.catch((error) => { this.fail(error instanceof Error ? error : new Error(String(error))) })
+    ctx.effect(() => () => this.dispose())
+  }
+
+  /** Hold plugin readiness until the remote identity and helper digest are verified. */
+  async [Service.init](): Promise<void> { await this.ready }
+
+  /** Send a helper operation; cancellation never automatically replays an ambiguous mutation. */
+  async request<T>(method: string, params: unknown, result: z.ZodType<T>, signal?: AbortSignal, wait = false): Promise<T> {
+    await this.ready
+    if (this.failure !== undefined) throw this.failure
+    const bounded = wait ? signal : signal === undefined
+      ? AbortSignal.timeout(this.config.requestTimeoutMs)
+      : AbortSignal.any([signal, AbortSignal.timeout(this.config.requestTimeoutMs)])
+    return (this.rpc as SshRpcPeer).request(method, params, result, bounded)
+  }
+
+  /** Forward one private remote socket through an independent SSH channel. */
+  async connectStream(remote: string, signal?: AbortSignal): Promise<Socket> {
+    const hello = await this.ready
+    if (!remote.startsWith(`${hello.root}/`) || /[:\r\n\0]/u.test(remote)) throw new Error('SSH helper returned an invalid stream path')
+    signal?.throwIfAborted()
+    const local = join(this.directory as string, `s${this.nextSocket++}`)
+    const forward = `${local}:${remote}`
+    await execute('ssh', ['-S', this.controlPath(), '-O', 'forward', '-o', 'ExitOnForwardFailure=yes', '-L', forward, this.config.host], {
+      timeout: this.config.requestTimeoutMs, ...(signal === undefined ? {} : { signal }), maxBuffer: 64 * 1024,
+    })
+    if (this.closed) throw new Error('SSH connection closed before stream establishment')
+    const socket = createConnection({ path: local, ...(signal === undefined ? {} : { signal }) })
+    this.sockets.add(socket)
+    socket.once('close', () => {
+      this.sockets.delete(socket)
+      if (!this.closed) void execute('ssh', ['-S', this.controlPath(), '-O', 'cancel', '-L', forward, this.config.host], { timeout: this.config.requestTimeoutMs }).catch(() => {})
+    })
+    await new Promise<void>((resolve, reject) => {
+      socket.once('connect', resolve)
+      socket.once('error', reject)
+    })
+    return socket
+  }
+
+  /** Tear down the helper's remote managed ranges before releasing the SSH master when reachable. */
+  async dispose(): Promise<void> {
+    if (this.closed) return
+    this.closed = true
+    if (this.heartbeat !== undefined) clearInterval(this.heartbeat)
+    try {
+      await this.ready
+      if (this.failure === undefined) await this.rpc?.request('close', {}, z.null(), AbortSignal.timeout(this.config.requestTimeoutMs))
+    } finally {
+      this.rpc?.close()
+      for (const socket of this.sockets) socket.destroy()
+      this.child?.kill('SIGTERM')
+      const force = setTimeout(() => { this.child?.kill('SIGKILL') }, this.config.requestTimeoutMs)
+      try { await this.childClosed } finally { clearTimeout(force) }
+      if (this.directory !== undefined) await rm(this.directory, { recursive: true, force: true })
+    }
+  }
+
+  private controlPath(): string { return join(this.directory as string, 'master') }
+
+  private fail(error: Error): void {
+    if (this.failure !== undefined) return
+    this.failure = error
+    if (this.heartbeat !== undefined) clearInterval(this.heartbeat)
+    this.rpc?.close(error)
+    for (const socket of this.sockets) socket.destroy(error)
+    this.child?.kill('SIGTERM')
+  }
+
+  private async start(): Promise<Hello> {
+    this.directory = await mkdtemp(join(tmpdir(), 'dsh-ssh-'))
+    if (this.closed) throw new Error('SSH connection closed before startup')
+    const quote = (value: string): string => `'${value.replaceAll("'", "'\\''")}'`
+    const command = [this.config.node, this.config.helper].map(quote).join(' ')
+    const child = spawn('ssh', [
+      '-T', '-M', '-S', this.controlPath(), '-o', 'ControlPersist=no', '-o', 'BatchMode=yes',
+      '-o', 'StrictHostKeyChecking=yes', '-o', 'ForwardAgent=no', '-o', 'ClearAllForwardings=yes',
+      '-o', 'ServerAliveInterval=10', '-o', 'ServerAliveCountMax=3', this.config.host, command,
+    ], { stdio: ['pipe', 'pipe', 'pipe'] })
+    this.child = child
+    this.childClosed = new Promise((resolve) => { child.once('close', () => { resolve() }) })
+    child.stderr.resume() // SSH diagnostics can contain configured paths; operation errors remain structured.
+    child.once('error', (error) => { this.fail(error) })
+    child.once('close', () => { this.fail(new Error('SSH helper disconnected; remote outcomes and cleanup are unknown')) })
+    const rpc = new SshRpcPeer(child.stdout, child.stdin, this.config.maxFrameBytes, this.config.maxPending)
+    this.rpc = rpc
+    rpc.once('closed', (error) => { this.fail(error as Error) })
+    const hello = await rpc.request('hello', {
+      protocol: SSH_PROTOCOL_VERSION, workspace: this.config.workspace, leaseMs: this.config.leaseMs,
+    }, helloSchema, AbortSignal.timeout(this.config.requestTimeoutMs))
+    if (hello.hash !== this.config.helperHash) throw new Error('SSH helper digest differs from the configured artifact')
+    this.heartbeat = setInterval(() => {
+      void rpc.request('heartbeat', {}, z.null(), AbortSignal.timeout(this.config.leaseMs / 2)).catch((error) => { this.fail(error instanceof Error ? error : new Error(String(error))) })
+    }, Math.floor(this.config.leaseMs / 3))
+    this.heartbeat.unref()
+    return hello
+  }
+}
+
+export default SshConnection

packages/ssh/ssh/src/protocol.ts

@@ -0,0 +1,170 @@
+/** Bounded, versioned requests between an SSH client and its private remote helper. */
+
+import { randomUUID } from 'node:crypto'
+import { EventEmitter, once } from 'node:events'
+import type { Readable, Writable } from 'node:stream'
+import { z } from 'zod'
+
+/** Wire version shared by the installed helper and client package. */
+export const SSH_PROTOCOL_VERSION = 1
+
+const errorSchema = z.object({ name: z.string(), message: z.string(), code: z.string().optional() }).strict()
+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(),
+])
+type Frame = z.infer<typeof frameSchema>
+type RequestHandler = (method: string, params: unknown, signal: AbortSignal) => Promise<unknown>
+
+/** A remote error retains its typed filesystem or sandbox code. */
+export class RemoteOperationError extends Error {
+  constructor(message: string, readonly code?: string) {
+    super(message)
+    this.name = 'RemoteOperationError'
+  }
+}
+
+/** 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 }>()
+  private readonly active = new Map<string, AbortController>()
+  private writeTail = Promise.resolve()
+  private queuedBytes = 0
+  private failure: Error | undefined
+
+  constructor(
+    private readonly input: Readable,
+    private readonly output: Writable,
+    private readonly maxFrameBytes: number,
+    private readonly maxPending: number,
+    private readonly handler?: RequestHandler,
+  ) {
+    super()
+    input.on('error', (error) => { this.close(error) })
+    output.on('error', (error) => { this.close(error) })
+    void this.readFrames().catch((error) => { this.close(error instanceof Error ? error : new Error(String(error))) })
+  }
+
+  /** Send one request and validate its response before exposing it to the caller. */
+  async request<T>(method: string, params: unknown, schema: z.ZodType<T>, signal?: AbortSignal): Promise<T> {
+    signal?.throwIfAborted()
+    if (this.failure !== undefined) throw this.failure
+    if (this.pending.size >= this.maxPending) throw new Error('SSH helper pending request limit reached')
+    const id = randomUUID()
+    const result = Promise.withResolvers<unknown>()
+    void result.promise.catch(() => {})
+    this.pending.set(id, result)
+    const abort = (): void => {
+      this.pending.delete(id)
+      result.reject(new Error('SSH operation cancelled; a completed remote mutation is not rolled back'))
+      void this.send({ type: 'cancel', id }).catch(() => {})
+    }
+    signal?.addEventListener('abort', abort, { once: true })
+    try {
+      await this.send({ type: 'request', id, method, params })
+      return schema.parse(await result.promise)
+    } finally {
+      this.pending.delete(id)
+      signal?.removeEventListener('abort', abort)
+    }
+  }
+
+  /** Fail pending operations and abort remote handlers without claiming their effects were undone. */
+  close(error = new Error('SSH connection lost; remote operation outcome and cleanup are unknown')): void {
+    if (this.failure !== undefined) return
+    this.failure = error
+    for (const pending of this.pending.values()) pending.reject(error)
+    this.pending.clear()
+    for (const controller of this.active.values()) controller.abort(error)
+    this.active.clear()
+    this.input.destroy()
+    this.output.destroy()
+    this.emit('closed', error)
+  }
+
+  private send(frame: Frame): Promise<void> {
+    if (this.failure !== undefined) return Promise.reject(this.failure)
+    const body = Buffer.from(JSON.stringify(frame))
+    if (body.length > this.maxFrameBytes || this.queuedBytes + body.length + 4 > this.maxFrameBytes * 2) {
+      return Promise.reject(new Error('SSH helper frame or write queue limit exceeded'))
+    }
+    const header = Buffer.alloc(4)
+    header.writeUInt32BE(body.length)
+    const bytes = Buffer.concat([header, body])
+    this.queuedBytes += bytes.length
+    const write = this.writeTail.then(async () => {
+      if (this.failure !== undefined) throw this.failure
+      if (!this.output.write(bytes)) await once(this.output, 'drain')
+    })
+    this.writeTail = write.catch((error) => { this.close(error instanceof Error ? error : new Error(String(error))) })
+    return write.finally(() => { this.queuedBytes -= bytes.length })
+  }
+
+  private async readFrames(): Promise<void> {
+    const header = Buffer.alloc(4)
+    let headerBytes = 0
+    let payload: Buffer | undefined
+    let payloadBytes = 0
+    for await (const raw of this.input) {
+      const chunk: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(raw as Uint8Array)
+      let offset = 0
+      while (offset < chunk.length) {
+        if (payload === undefined) {
+          const count = Math.min(4 - headerBytes, chunk.length - offset)
+          chunk.copy(header, headerBytes, offset, offset + count)
+          headerBytes += count
+          offset += count
+          if (headerBytes < 4) continue
+          const size = header.readUInt32BE(0)
+          if (size === 0 || size > this.maxFrameBytes) throw new Error('SSH helper sent an invalid frame length')
+          payload = Buffer.alloc(size)
+          payloadBytes = 0
+        }
+        const count = Math.min(payload.length - payloadBytes, chunk.length - offset)
+        chunk.copy(payload, payloadBytes, offset, offset + count)
+        payloadBytes += count
+        offset += count
+        if (payloadBytes === payload.length) {
+          const frame = frameSchema.parse(JSON.parse(payload.toString('utf8')))
+          payload = undefined
+          headerBytes = 0
+          this.receive(frame)
+        }
+      }
+    }
+    throw new Error(headerBytes > 0 || payload !== undefined ? 'SSH helper disconnected during a frame; outcome is unknown' : 'SSH helper disconnected; outcome is unknown')
+  }
+
+  private receive(frame: Frame): void {
+    if (frame.type === 'result' || frame.type === 'error') {
+      const pending = this.pending.get(frame.id)
+      if (pending === undefined) return // A cancelled request can still complete remotely.
+      this.pending.delete(frame.id)
+      if (frame.type === 'result') pending.resolve(frame.value)
+      else pending.reject(new RemoteOperationError(frame.error.message, frame.error.code))
+      return
+    }
+    if (frame.type === 'cancel') {
+      this.active.get(frame.id)?.abort(new Error('SSH caller cancelled the operation'))
+      return
+    }
+    if (this.handler === undefined || this.active.has(frame.id) || this.active.size >= this.maxPending) {
+      throw new Error('SSH helper received an unexpected or excessive request')
+    }
+    const controller = new AbortController()
+    this.active.set(frame.id, controller)
+    void this.handler(frame.method, frame.params, controller.signal).then(
+      value => this.send({ type: 'result', id: frame.id, value: value ?? null }),
+      (error: unknown) => {
+        const detail = error instanceof Error ? error : new Error(String(error))
+        const code = 'code' in detail && typeof detail.code === 'string' ? detail.code : undefined
+        return this.send({ type: 'error', id: frame.id, error: {
+          name: detail.name, message: detail.message, ...(code === undefined ? {} : { code }),
+        } })
+      },
+    ).catch((error) => { this.close(error instanceof Error ? error : new Error(String(error))) })
+      .finally(() => { this.active.delete(frame.id) })
+  }
+}

packages/ssh/ssh/src/schemas.ts

@@ -0,0 +1,55 @@
+/** Strict JSON validation for SSH helper requests and remote observations. */
+import { z } from 'zod'
+
+/** 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')
+/** Filesystem identity returned by the remote filesystem provider. */
+export const targetSchema = z.object({ targetKey: remotePath, displayPath: z.string() }).strict()
+/** Remote metadata observation. */
+export const infoSchema = z.object({ version: z.string(), type: z.enum(['file', 'directory', 'other']), size: z.number().nonnegative().optional() }).strict()
+/** Metadata that preserves a final symlink. */
+export const pathInfoSchema = infoSchema.extend({ type: z.enum(['file', 'directory', 'symlink', 'other']) })
+/** A resolved file-effect policy; the remote helper owns path canonicalization. */
+export const policySchema = z.object({ mode: z.enum(['read-only', 'workspace-write', 'danger-full-access']), workspaceRoot: remotePath, sessionId: z.string().optional() }).strict()
+/** Complete directory entries. */
+export const entriesSchema = z.array(z.object({ name: z.string(), type: z.enum(['file', 'directory', 'other']), target: targetSchema, version: z.string().optional(), size: z.number().nonnegative().optional() }).strict())
+/** Guarded write intent. */
+export const intentSchema = z.discriminatedUnion('kind', [z.object({ kind: z.literal('createIfAbsent') }).strict(), z.object({ kind: z.literal('replaceIfVersion'), version: z.string() }).strict()])
+/** Literal text edit. */
+export const editSchema = z.object({ oldString: z.string(), newString: z.string(), replaceAll: z.boolean() }).strict()
+/** Atomic write observation. */
+export const writeResultSchema = z.object({ operation: z.enum(['create', 'update']), version: z.string(), before: z.string().nullable(), after: z.string() }).strict()
+/** Atomic edit observation. */
+export const editResultSchema = z.object({ version: z.string(), before: z.string(), after: z.string() }).strict()
+/** Explicit child environment; null encodes an environment tombstone. */
+export const environmentSchema = z.record(z.string(), z.string().nullable())
+const collection = z.object({
+  maxBytes: z.number().int().positive(),
+  spill: z.object({ maxBytes: z.number().int().positive() }).strict().optional(),
+}).strict()
+/** Ordinary or terminal process requested by a trusted SSH client. */
+export const spawnSchema = z.object({
+  argv: z.array(z.string().refine(value => !value.includes('\0'))).min(1),
+  cwd: remotePath,
+  env: environmentSchema.optional(), graceMs: z.number().int().positive().max(2_147_483_647),
+  stdio: z.object({
+    stdin: z.union([z.literal('ignore'), z.literal('pipe'), z.object({ data: z.string() }).strict()]),
+    stdout: z.union([z.literal('pipe'), z.literal('inherit'), collection]),
+    stderr: z.union([z.literal('pipe'), z.literal('inherit'), collection]),
+    control: z.literal('pipe').optional(),
+  }).strict().optional(),
+  terminal: z.object({ rows: z.number().int().positive(), cols: z.number().int().positive() }).strict().optional(),
+}).strict().refine(value => (value.stdio === undefined) !== (value.terminal === undefined), 'select ordinary or terminal execution')
+/** Connection handshake binds sockets and workspace to one helper process. */
+export const helloSchema = z.object({ protocol: z.literal(1), hash: z.string().regex(/^[0-9a-f]{64}$/), platform: z.enum(['linux', 'darwin']), nodeVersion: z.string(), node: remotePath, root: remotePath, workspace: remotePath }).strict()
+/** A prepared process publishes its sockets before target code may execute. */
+export const preparedSchema = z.object({ id: z.string().uuid(), streams: z.partialRecord(z.enum(['stdin', 'stdout', 'stderr', 'control', 'terminal']), remotePath) }).strict()
+/** Direct process exit facts. */
+export const outcomeSchema = z.object({ exitCode: z.number().int().nullable(), signal: z.string().nullable() }).strict()
+/** Final output locations are remote paths, never host copies. */
+export const doneSchema = z.object({
+  outcome: outcomeSchema,
+  spills: z.object({ stdout: remotePath.optional(), stderr: remotePath.optional() }).strict(),
+}).strict()
+/** Remote terminal foreground observation. */
+export const foregroundSchema = z.object({ processGroupId: z.number().int().positive(), inputWaiting: z.boolean() }).strict().nullable()

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

@@ -0,0 +1,78 @@
+import { PassThrough } from 'node:stream'
+import { once } from 'node:events'
+import { describe, expect, it } from 'vitest'
+import { z } from 'zod'
+import { SshRpcPeer, RemoteOperationError } from '../src/protocol.ts'
+
+describe('SSH helper protocol', () => {
+  it('validates responses and retains typed operation errors', async () => {
+    const inbound = new PassThrough()
+    const outbound = new PassThrough()
+    const server = new SshRpcPeer(outbound, inbound, 4096, 8, async (method, raw) => {
+      if (method === 'write') throw Object.assign(new Error('stale'), { code: 'FS_STALE_VERSION' })
+      return z.object({ value: z.string() }).parse(raw).value
+    })
+    const client = new SshRpcPeer(inbound, outbound, 4096, 8)
+    try {
+      await expect(client.request('echo', { value: 'a\0b\n中文' }, z.string())).resolves.toBe('a\0b\n中文')
+      await expect(client.request('write', {}, z.null())).rejects.toMatchObject({ code: 'FS_STALE_VERSION', message: 'stale' })
+      await expect(client.request('echo', { value: 'not a number' }, z.number())).rejects.toThrow()
+    } finally { client.close(); server.close() }
+  })
+
+  it('rejects excessive lengths before waiting for the payload', async () => {
+    const inbound = new PassThrough()
+    const outbound = new PassThrough()
+    const client = new SshRpcPeer(inbound, outbound, 64, 2)
+    const closed = once(client, 'closed')
+    const header = Buffer.alloc(4)
+    header.writeUInt32BE(65)
+    inbound.write(header)
+    await expect(closed).resolves.toEqual([expect.objectContaining({ message: 'SSH helper sent an invalid frame length' })])
+  })
+
+  it('cancels an admitted operation without replaying it', async () => {
+    const inbound = new PassThrough()
+    const outbound = new PassThrough()
+    const entered = Promise.withResolvers<undefined>()
+    const aborted = Promise.withResolvers<undefined>()
+    let calls = 0
+    const server = new SshRpcPeer(outbound, inbound, 4096, 8, async (_method, _raw, signal) => {
+      calls++
+      signal.addEventListener('abort', () => { aborted.resolve(undefined) }, { once: true })
+      entered.resolve(undefined)
+      await aborted.promise
+      return null
+    })
+    const client = new SshRpcPeer(inbound, outbound, 4096, 8)
+    const controller = new AbortController()
+    try {
+      const response = client.request('mutate', {}, z.null(), controller.signal)
+      const rejection = expect(response).rejects.toThrow('not rolled back')
+      await entered.promise
+      controller.abort()
+      await rejection
+      await aborted.promise
+      expect(calls).toBe(1)
+    } finally { client.close(); server.close() }
+  })
+
+  it('reports an unknown outcome when a connection disappears after admission', async () => {
+    const inbound = new PassThrough()
+    const outbound = new PassThrough()
+    const entered = Promise.withResolvers<undefined>()
+    const server = new SshRpcPeer(outbound, inbound, 4096, 8, async (_method, _raw, signal) => {
+      entered.resolve(undefined)
+      await new Promise<void>((resolve) => { signal.addEventListener('abort', () => { resolve() }, { once: true }) })
+      return null
+    })
+    const client = new SshRpcPeer(inbound, outbound, 4096, 8)
+    const pending = client.request('write', {}, z.null())
+    const rejection = expect(pending).rejects.toThrow('unknown')
+    await entered.promise
+    client.close()
+    server.close()
+    await rejection
+    expect(new RemoteOperationError('remote failure', 'FS_IO_ERROR').code).toBe('FS_IO_ERROR')
+  })
+})

packages/ssh/ssh/tsconfig.json

@@ -0,0 +1,45 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
+    {
+      "path": "../../fs/fs"
+    },
+    {
+      "path": "../../fs/fs-local"
+    },
+    {
+      "path": "../../fs/fs-sandbox"
+    },
+    {
+      "path": "../../subprocess/subprocess"
+    },
+    {
+      "path": "../../subprocess/subprocess-local"
+    },
+    {
+      "path": "../../sandbox/sandbox"
+    },
+    {
+      "path": "../../sandbox/sandbox-local"
+    },
+    {
+      "path": "../../sandbox/sandbox-policy"
+    },
+    {
+      "path": "../../session/session-projection"
+    }
+  ]
+}

packages/ssh/ssh/tsdown.config.ts

@@ -0,0 +1,6 @@
+import { defineConfig } from 'tsdown'
+
+export default defineConfig({
+  entry: { index: 'lib/types/index.js', protocol: 'lib/types/protocol.js', schemas: 'lib/types/schemas.js', helper: 'lib/types/helper.js' },
+  outDir: 'lib', format: ['esm'], platform: 'node', target: 'es2024', fixedExtension: false, dts: false, clean: false,
+})

packages/ssh/subprocess-ssh/package.json

@@ -0,0 +1,45 @@
+{
+  "name": "@deepseek-ai/dsh-subprocess-ssh",
+  "description": "Subprocess and terminal provider over the shared POSIX SSH helper",
+  "version": "0.1.5-rc.2",
+  "publishConfig": {
+    "access": "public"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/deepseek-ai/deepseek-harness.git",
+    "directory": "packages/ssh/subprocess-ssh"
+  },
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib/*.js",
+    "lib/types/**/*.d.ts"
+  ],
+  "license": "MIT",
+  "peerDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-subprocess-local": "workspace:^"
+  },
+  "dependencies": {
+    "@deepseek-ai/schemastery": "workspace:^",
+    "zod": "^4.4.3"
+  },
+  "devDependencies": {
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-ssh": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-subprocess-local": "workspace:^"
+  }
+}

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

@@ -0,0 +1,187 @@
+/** Remote subprocess and PTY handles with independent SSH streams and helper-owned process lifetimes. */
+import { Duplex, PassThrough, type Readable, type Writable } from 'node:stream'
+import type { Socket } from 'node:net'
+import { Context } from '@deepseek-ai/cordis'
+import { SubprocessRuntime } from '@deepseek-ai/dsh-subprocess'
+import type { SubprocessCollectedOutputs, SubprocessHandle, SubprocessOutcome, SubprocessOutputMode, SubprocessSpawnSpec, SubprocessTerminalHandle, SubprocessTerminalSignal, SubprocessTerminalSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { OutputCollector } from '@deepseek-ai/dsh-subprocess-local/output'
+import type { SshConnection } from '@deepseek-ai/dsh-ssh'
+import { doneSchema, foregroundSchema, preparedSchema, remotePath } from '@deepseek-ai/dsh-ssh/schemas'
+import { z } from 'zod'
+
+function environment(env?: NodeJS.ProcessEnv): Record<string, string | null> | undefined {
+  return env === undefined ? undefined : Object.fromEntries(Object.entries(env).map(([key, value]) => [key, value ?? null]))
+}
+
+/** One remote ordinary process; stdin and control remain usable during asynchronous SSH allocation. */
+class RemoteProcess implements SubprocessHandle {
+  readonly stdin: Writable | undefined
+  readonly stdout: Readable | undefined
+  readonly stderr: Readable | undefined
+  readonly control: Duplex | undefined
+  readonly collected: SubprocessCollectedOutputs
+  readonly done: Promise<SubprocessOutcome>
+  private readonly inbound = new PassThrough()
+  private readonly out = new PassThrough()
+  private readonly err = new PassThrough()
+  private readonly toControl = new PassThrough()
+  private readonly fromControl = new PassThrough()
+  private readonly controller = new AbortController()
+  private readonly started: Promise<void>
+  private id: string | undefined
+  private sockets: Socket[] = []
+  private failure: unknown
+  private quiescent = false
+  private readonly detachAbort: () => void
+  private spills: { stdout?: string | undefined; stderr?: string | undefined } = {}
+
+  constructor(private readonly ssh: SshConnection, private readonly spec: SubprocessSpawnSpec) {
+    this.stdin = spec.stdio.stdin === 'pipe' ? this.inbound : undefined
+    this.stdout = spec.stdio.stdout === 'pipe' ? this.out : undefined
+    this.stderr = spec.stdio.stderr === 'pipe' ? this.err : undefined
+    const requestedControl = (spec.stdio as typeof spec.stdio & { control?: 'pipe' }).control
+    this.control = requestedControl === undefined ? undefined : Duplex.from({ writable: this.toControl, readable: this.fromControl })
+    for (const stream of [this.inbound, this.out, this.err, this.toControl, this.fromControl, this.control]) stream?.on('error', () => {})
+    const collect = (name: 'stdout' | 'stderr', stream: PassThrough, mode: SubprocessOutputMode) => {
+      if (mode === 'pipe') return undefined
+      if (mode === 'inherit') { stream.pipe(name === 'stdout' ? process.stdout : process.stderr, { end: false }); return undefined }
+      const collector = new OutputCollector(mode.maxBytes, undefined, name, '')
+      stream.on('data', (chunk: Buffer) => { collector.push(chunk) })
+      return {
+        readFrom: (offset: number) => ({
+          ...collector.readFrom(offset), ...(this.spills[name] === undefined ? {} : { spillPath: this.spills[name] }),
+        }),
+      }
+    }
+    const stdout = collect('stdout', this.out, spec.stdio.stdout)
+    const stderr = collect('stderr', this.err, spec.stdio.stderr)
+    this.collected = { ...(stdout === undefined ? {} : { stdout }), ...(stderr === undefined ? {} : { stderr }) }
+    const onAbort = (): void => { this.terminate() }
+    spec.signal?.addEventListener('abort', onAbort, { once: true })
+    this.detachAbort = () => { spec.signal?.removeEventListener('abort', onAbort) }
+    this.started = this.start()
+    this.done = this.started.then(async () => {
+      const result = await ssh.request('process.done', { id: this.id }, doneSchema, undefined, true)
+      this.spills = result.spills
+      return { exitCode: result.outcome.exitCode, signal: result.outcome.signal as NodeJS.Signals | null }
+    }).catch((error: unknown) => {
+      this.failure = error
+      this.terminate()
+      for (const stream of [this.inbound, this.out, this.err, this.toControl, this.fromControl]) {
+        stream.destroy(error instanceof Error ? error : new Error(String(error)))
+      }
+      throw error
+    })
+    void this.done.catch(() => {})
+  }
+
+  private async start(): Promise<void> {
+    this.spec.signal?.throwIfAborted()
+    const prepared = await this.ssh.request(
+      'process.prepare', { ...this.spec, signal: undefined, env: environment(this.spec.env) }, preparedSchema, this.controller.signal,
+    )
+    this.id = prepared.id
+    try {
+      const sockets = await Promise.all(Object.entries(prepared.streams).map(async ([name, path]) =>
+        [name, await this.ssh.connectStream(path, this.controller.signal)] as const))
+      this.sockets = sockets.map(([, socket]) => socket)
+      for (const [name, socket] of sockets) {
+        if (name === 'stdout') socket.pipe(this.out)
+        if (name === 'stderr') socket.pipe(this.err)
+        if (name === 'stdin') this.inbound.pipe(socket)
+        if (name === 'control') { this.toControl.pipe(socket); socket.pipe(this.fromControl) }
+      }
+      this.controller.signal.throwIfAborted()
+      await this.ssh.request('process.start', { id: this.id }, z.object({}).strict(), this.controller.signal)
+    } catch (error) {
+      await this.ssh.request('process.terminate', { id: this.id }, z.null()).catch(() => {})
+      for (const socket of this.sockets) socket.destroy()
+      throw error
+    }
+  }
+
+  terminate(): void {
+    this.controller.abort(new Error('SSH process terminated'))
+    if (this.id !== undefined) void this.ssh.request('process.terminate', { id: this.id }, z.null()).catch((error: unknown) => { this.failure = error })
+  }
+
+  async waitForExit(signal?: AbortSignal): Promise<boolean> {
+    if (this.quiescent) return true
+    await this.started
+    if (this.failure !== undefined) throw this.failure
+    const result = await this.ssh.request('process.wait', { id: this.id }, z.boolean(), signal, true)
+    if (result) { this.quiescent = true; this.detachAbort() }
+    return result
+  }
+}
+
+/** SSH provider paired with the SSH filesystem; the remote helper selects POSIX process ownership. */
+export class SshSubprocessRuntime extends SubprocessRuntime {
+  static inject = ['ssh']
+  private readonly live = new Set<SubprocessHandle>()
+  private readonly terminals = new Set<SubprocessTerminalHandle>()
+
+  constructor(ctx: Context) {
+    super(ctx)
+    ctx.effect(() => async () => {
+      for (const handle of this.live) handle.terminate()
+      const results = await Promise.allSettled([
+        ...[...this.live].map(handle => handle.waitForExit()),
+        ...[...this.terminals].map(handle => handle.terminate()),
+      ])
+      const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason as unknown] : [])
+      if (errors.length > 0) throw new AggregateError(errors, 'SSH process cleanup could not be confirmed')
+    })
+  }
+
+  override resolveExecutable(command: string, env?: Readonly<Record<string, string>>, signal?: AbortSignal): Promise<string> {
+    return this.ctx.ssh.request('executable', { command, env }, remotePath, signal)
+  }
+
+  override spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
+    spec.signal?.throwIfAborted()
+    const handle = new RemoteProcess(this.ctx.ssh, spec)
+    this.live.add(handle)
+    void handle.done.then(() => handle.waitForExit()).then(() => { this.live.delete(handle) }).catch(() => {})
+    return handle
+  }
+
+  override async spawnTerminal(spec: SubprocessTerminalSpawnSpec): Promise<SubprocessTerminalHandle> {
+    const ssh = this.ctx.ssh
+    const prepared = await ssh.request('process.prepare', {
+      argv: spec.argv, cwd: spec.cwd, env: environment(spec.env), graceMs: spec.graceMs,
+      terminal: { rows: spec.rows, cols: spec.cols },
+    }, preparedSchema, spec.signal)
+    const id = prepared.id
+    let socket: Socket | undefined
+    try {
+      socket = await ssh.connectStream(z.string().parse(prepared.streams.terminal), spec.signal)
+      const started = await ssh.request('process.start', { id }, z.object({ pid: z.number().int().positive() }).strict(), spec.signal)
+      const output = socket
+      const done = ssh.request('process.done', { id }, doneSchema, undefined, true).then(result => ({ exitCode: result.outcome.exitCode, signal: result.outcome.signal as NodeJS.Signals | null }))
+      void done.catch(() => {})
+      let closing: Promise<void> | undefined
+      const handle: SubprocessTerminalHandle = {
+        pid: started.pid, output, done,
+        write: async (data) => { await ssh.request('terminal.write', { id, value: data }, z.null()) },
+        inspectForeground: async () => await ssh.request('terminal.inspect', { id }, foregroundSchema) ?? undefined,
+        signalForeground: (signal: SubprocessTerminalSignal) => ssh.request('terminal.signal', { id, value: signal }, z.number().int().positive()),
+        terminate: () => {
+          closing ??= ssh.request('process.terminate', { id }, z.null(), undefined, true).then(() => {
+            output.destroy()
+            this.terminals.delete(handle)
+          }).catch((error: unknown) => { closing = undefined; throw error })
+          return closing
+        },
+      }
+      this.terminals.add(handle)
+      return handle
+    } catch (error) {
+      socket?.destroy()
+      await ssh.request('process.terminate', { id }, z.null()).catch(() => {})
+      throw error
+    }
+  }
+}
+
+export default SshSubprocessRuntime

packages/ssh/subprocess-ssh/tsconfig.json

@@ -0,0 +1,27 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
+    {
+      "path": "../ssh"
+    },
+    {
+      "path": "../../subprocess/subprocess"
+    },
+    {
+      "path": "../../subprocess/subprocess-local"
+    }
+  ]
+}

packages/subprocess/subprocess-local/package.json

@@ -23,14 +23,19 @@
       "default": "./lib/runner.js"
     },
     "./src/*": "./src/*",
-    "./package.json": "./package.json"
+    "./package.json": "./package.json",
+    "./output": {
+      "types": "./lib/types/output.d.ts",
+      "default": "./lib/output.js"
+    }
   },
   "files": [
     "lib/index.js",
     "lib/runner.js",
     "lib/runner-*.js",
     "scripts/ensure-spawn-helper.mjs",
-    "lib/types/**/*.d.ts"
+    "lib/types/**/*.d.ts",
+    "lib/output.js"
   ],
   "scripts": {
     "postinstall": "node scripts/ensure-spawn-helper.mjs"

packages/subprocess/subprocess-local/src/output.ts

@@ -0,0 +1,2 @@
+/** Bounded stream collection shared by local processes and SSH stream adapters. */
+export { OutputCollector } from './spawn.ts'

packages/subprocess/subprocess-local/tsdown.config.ts

@@ -4,6 +4,7 @@ export default defineConfig({
   entry: {
     index: 'lib/types/index.js',
     runner: 'lib/types/bin.js',
+    output: 'lib/types/output.js',
   },
   outDir: 'lib',
   format: ['esm'],

pnpm-lock.yaml

@@ -9120,6 +9120,112 @@ importers:
         specifier: workspace:^
         version: link:../../core/tools
 
+  packages/ssh/fs-ssh:
+    dependencies:
+      '@deepseek-ai/schemastery':
+        specifier: link:../../../vendor/schemastery
+        version: link:../../../vendor/schemastery
+      zod:
+        specifier: ^4.4.3
+        version: 4.4.3
+    devDependencies:
+      '@deepseek-ai/cordis':
+        specifier: workspace:^
+        version: link:../../../vendor/cordis
+      '@deepseek-ai/dsh-fs':
+        specifier: workspace:^
+        version: link:../../fs/fs
+      '@deepseek-ai/dsh-sandbox':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox
+      '@deepseek-ai/dsh-sandbox-policy':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox-policy
+      '@deepseek-ai/dsh-ssh':
+        specifier: workspace:^
+        version: link:../ssh
+
+  packages/ssh/sandbox-ssh:
+    dependencies:
+      '@deepseek-ai/schemastery':
+        specifier: link:../../../vendor/schemastery
+        version: link:../../../vendor/schemastery
+      zod:
+        specifier: ^4.4.3
+        version: 4.4.3
+    devDependencies:
+      '@deepseek-ai/cordis':
+        specifier: workspace:^
+        version: link:../../../vendor/cordis
+      '@deepseek-ai/dsh-sandbox':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox
+      '@deepseek-ai/dsh-ssh':
+        specifier: workspace:^
+        version: link:../ssh
+
+  packages/ssh/ssh:
+    dependencies:
+      '@deepseek-ai/schemastery':
+        specifier: link:../../../vendor/schemastery
+        version: link:../../../vendor/schemastery
+      zod:
+        specifier: ^4.4.3
+        version: 4.4.3
+    devDependencies:
+      '@deepseek-ai/cordis':
+        specifier: workspace:^
+        version: link:../../../vendor/cordis
+      '@deepseek-ai/dsh-fs':
+        specifier: workspace:^
+        version: link:../../fs/fs
+      '@deepseek-ai/dsh-fs-local':
+        specifier: workspace:^
+        version: link:../../fs/fs-local
+      '@deepseek-ai/dsh-fs-sandbox':
+        specifier: workspace:^
+        version: link:../../fs/fs-sandbox
+      '@deepseek-ai/dsh-sandbox':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox
+      '@deepseek-ai/dsh-sandbox-local':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox-local
+      '@deepseek-ai/dsh-sandbox-policy':
+        specifier: workspace:^
+        version: link:../../sandbox/sandbox-policy
+      '@deepseek-ai/dsh-session-projection':
+        specifier: workspace:^
+        version: link:../../session/session-projection
+      '@deepseek-ai/dsh-subprocess':
+        specifier: workspace:^
+        version: link:../../subprocess/subprocess
+      '@deepseek-ai/dsh-subprocess-local':
+        specifier: workspace:^
+        version: link:../../subprocess/subprocess-local
+
+  packages/ssh/subprocess-ssh:
+    dependencies:
+      '@deepseek-ai/schemastery':
+        specifier: link:../../../vendor/schemastery
+        version: link:../../../vendor/schemastery
+      zod:
+        specifier: ^4.4.3
+        version: 4.4.3
+    devDependencies:
+      '@deepseek-ai/cordis':
+        specifier: workspace:^
+        version: link:../../../vendor/cordis
+      '@deepseek-ai/dsh-ssh':
+        specifier: workspace:^
+        version: link:../ssh
+      '@deepseek-ai/dsh-subprocess':
+        specifier: workspace:^
+        version: link:../../subprocess/subprocess
+      '@deepseek-ai/dsh-subprocess-local':
+        specifier: workspace:^
+        version: link:../../subprocess/subprocess-local
+
   packages/storage/storage:
     devDependencies:
       '@deepseek-ai/cordis':
@@ -12946,81 +13052,6 @@ packages:
     resolution: {integrity: sha512-Mmjg4anFBD5OzbPnGJOA0jPPN8645ERhQk38HQLpSenx1ox9bfdPkmAzUnNjeQtqQGFLtKe13J20RtLBmUKMZA==}
     hasBin: true
 
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-5+nkh8i3gt6lqS/d2jTZ1xAn6tdgtB4Lf1mW6T0Qm5/rXNwBuV1sAEyLEWan5o9gJPU/GuvHR3rvSeZ+FaGrbw==}
-    engines: {node: '>= 10'}
-    cpu: [arm64]
-    os: [android]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-7jQ47gi+fZ7KJmfc/5rNyy1CYw/cu4kZ0KPIYbo9UUgSdW0bKQJpt+WihEor6s4Lyp7+xc3a+3HeyXmAEbbnPg==}
-    engines: {node: '>= 10'}
-    cpu: [arm64]
-    os: [darwin]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-rRjDMZs9pIRKGxgijwezplKc1RnJsqUokrA9h88bbTkqQ+7ePj0ZN4ZnZDy8Vu0tXs7KRlI2tQLaK4mx9QlxHg==}
-    engines: {node: '>= 10'}
-    cpu: [x64]
-    os: [darwin]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-jGcCd+8ra6Q61xKqZeiItujTpp9a9eRLcQ0jW6qYNku+WpupqOPFPY0SrsuSnXFviJwkpKYT9p7QrB4lsf3LNQ==}
-    engines: {node: '>= 10'}
-    cpu: [arm]
-    os: [linux]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-od6I2Y7kU7i1SwZYG2EKW8rWz6JiedtPpko4WEe1DDsiikrfaotVBCRaUTM5/yeZKaZ92EatoAS+5xG+6uJlYA==}
-    engines: {node: '>= 10'}
-    cpu: [arm64]
-    os: [linux]
-    libc: [glibc]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-yYkPbJDJiWj6N0gASA3CAvRypZmVpJnxU0DQg3aBhneLDQde9TPLKADsQkobNoJUtTT/lj46aWpzT48PDb3Qcg==}
-    engines: {node: '>= 10'}
-    cpu: [arm64]
-    os: [linux]
-    libc: [musl]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-PB00MSKAp4VwK/xwe6duKxRKmH8UH4GIl1pqHSbxng0jnU9Dr7FwaDypDiqwNFZ774N+8G7mJLGuLtg9NTcQsg==}
-    engines: {node: '>= 10'}
-    cpu: [riscv64]
-    os: [linux]
-    libc: [glibc]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-TWM2XWJoitLiIPCvgJh7SriC+L/T9qkYCVzC66AidsZy0QP1hkKzBzVwshCdcA3q6fIn3yE0ISbq4lMJSy8jFw==}
-    engines: {node: '>= 10'}
-    cpu: [x64]
-    os: [linux]
-    libc: [glibc]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-hb20MxKXXb5IB7AAwN8UHz9WRsa2HmdZfjsDCzjElwJoeV1aotVEwFU4FrFQcYQVzsJQLeaCc/2Qdt/0Q72mMg==}
-    engines: {node: '>= 10'}
-    cpu: [x64]
-    os: [linux]
-    libc: [musl]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-WwPN08IXE4SkL+FhJyPz/iFnycMAUkbphFIT4cmKLlvbSU0Zfn1R7BGJ3Hqky1S89QUYc0Q4IOScXb/42Re9wQ==}
-    engines: {node: '>= 10'}
-    cpu: [arm64]
-    os: [win32]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-XkrVqKb+pxyba7kjy2LJvABFVBTE0DNpEl7MrG4OYUmaWarrXH+t54z/Czj2YxCKtizYTV4mg6phNm3x24qjhQ==}
-    engines: {node: '>= 10'}
-    cpu: [x64]
-    os: [win32]
-
-  '@napi-rs/[email protected]':
-    resolution: {integrity: sha512-/SaLcvlqGWdm0HSCWMgHu7cjJiQXfP8/mOY+6dUyV9flQz7sPBBZ+ed2zYtoukojPmxOaL7bm+d/G4GeWWoN7g==}
-    engines: {node: '>= 10'}
-
   '@napi-rs/[email protected]':
     resolution: {integrity: sha512-AWPoBRJ9tsnVhor4sjO7rkni+7p+2IAEFj6cx06UgP10jkQHqay/36uRV/bFkgrh18D9vb4cr8Q0Pthskgzy+Q==}
     peerDependencies:
@@ -19669,54 +19700,6 @@ snapshots:
       - supports-color
       - zod
 
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optional: true
-
-  '@napi-rs/[email protected]':
-    optionalDependencies:
-      '@napi-rs/canvas-android-arm64': 1.0.8
-      '@napi-rs/canvas-darwin-arm64': 1.0.8
-      '@napi-rs/canvas-darwin-x64': 1.0.8
-      '@napi-rs/canvas-linux-arm-gnueabihf': 1.0.8
-      '@napi-rs/canvas-linux-arm64-gnu': 1.0.8
-      '@napi-rs/canvas-linux-arm64-musl': 1.0.8
-      '@napi-rs/canvas-linux-riscv64-gnu': 1.0.8
-      '@napi-rs/canvas-linux-x64-gnu': 1.0.8
-      '@napi-rs/canvas-linux-x64-musl': 1.0.8
-      '@napi-rs/canvas-win32-arm64-msvc': 1.0.8
-      '@napi-rs/canvas-win32-x64-msvc': 1.0.8
-    optional: true
-
   '@napi-rs/[email protected](@emnapi/[email protected])(@emnapi/[email protected])':
     dependencies:
       '@emnapi/core': 1.10.0
@@ -23833,9 +23816,7 @@ snapshots:
 
   [email protected]: {}
 
-  [email protected]:
-    optionalDependencies:
-      '@napi-rs/canvas': 1.0.8
+  [email protected]: {}
 
   [email protected]: {}
 

tsconfig.base.json

@@ -28,6 +28,9 @@
     // not declaration path aliases, keep each package/vendor source compiled
     // under its own tsconfig boundary.
     "paths": {
+      "@deepseek-ai/dsh-ssh/protocol": ["./packages/ssh/ssh/src/protocol.ts"],
+      "@deepseek-ai/dsh-ssh/schemas": ["./packages/ssh/ssh/src/schemas.ts"],
+      "@deepseek-ai/dsh-subprocess-local/output": ["./packages/subprocess/subprocess-local/src/output.ts"],
       "@deepseek-ai/cordis": ["./vendor/cordis/src"],
       "@deepseek-ai/cosmokit": ["./vendor/cosmokit/src"],
       "@deepseek-ai/schemastery": ["./vendor/schemastery/src"],
@@ -300,6 +303,7 @@
       "@deepseek-ai/dsh-fs-local": ["./packages/fs/fs-local/src"],
       "@deepseek-ai/dsh-fs-observation-policy": ["./packages/fs/fs-observation-policy/src"],
       "@deepseek-ai/dsh-fs-sandbox": ["./packages/fs/fs-sandbox/src"],
+      "@deepseek-ai/dsh-fs-ssh": ["./packages/ssh/fs-ssh/src"],
       "@deepseek-ai/dsh-goal": ["./packages/goal/goal/src"],
       "@deepseek-ai/dsh-goal/invariant": ["./packages/goal/goal/src/invariant.ts"],
       "@deepseek-ai/dsh-goal-round-driver": ["./packages/goal/goal-round-driver/src"],
@@ -344,6 +348,7 @@
       "@deepseek-ai/dsh-sandbox-local": ["./packages/sandbox/sandbox-local/src"],
       "@deepseek-ai/dsh-sandbox-policy": ["./packages/sandbox/sandbox-policy/src"],
       "@deepseek-ai/dsh-sandbox-policy/invariant": ["./packages/sandbox/sandbox-policy/src/invariant.ts"],
+      "@deepseek-ai/dsh-sandbox-ssh": ["./packages/ssh/sandbox-ssh/src"],
       "@deepseek-ai/dsh-sandbox-windows-acl": ["./packages/sandbox/sandbox-windows-acl/src"],
       "@deepseek-ai/dsh-schedule": ["./packages/schedule/schedule/src"],
       "@deepseek-ai/dsh-schedule/invariant": ["./packages/schedule/schedule/src/invariant.ts"],
@@ -387,6 +392,7 @@
       "@deepseek-ai/dsh-spill": ["./packages/spill/spill/src"],
       "@deepseek-ai/dsh-spill-local": ["./packages/spill/spill-local/src"],
       "@deepseek-ai/dsh-spill-policy": ["./packages/spill/spill-policy/src"],
+      "@deepseek-ai/dsh-ssh": ["./packages/ssh/ssh/src"],
       "@deepseek-ai/dsh-storage": ["./packages/storage/storage/src"],
       "@deepseek-ai/dsh-storage-domain": ["./packages/storage/storage-domain/src"],
       "@deepseek-ai/dsh-storage-domain/invariant": ["./packages/storage/storage-domain/src/invariant.ts"],
@@ -403,6 +409,7 @@
       "@deepseek-ai/dsh-subagent-spawn-in-process": ["./packages/subagent/subagent-spawn-in-process/src"],
       "@deepseek-ai/dsh-subprocess": ["./packages/subprocess/subprocess/src"],
       "@deepseek-ai/dsh-subprocess-local": ["./packages/subprocess/subprocess-local/src"],
+      "@deepseek-ai/dsh-subprocess-ssh": ["./packages/ssh/subprocess-ssh/src"],
       "@deepseek-ai/dsh-system-prompt": ["./packages/core/system-prompt/src"],
       "@deepseek-ai/dsh-system-prompt/invariant": ["./packages/core/system-prompt/src/invariant.ts"],
       "@deepseek-ai/dsh-terminal": ["./packages/terminal/terminal/src"],

tsconfig.host.json

@@ -244,6 +244,10 @@
     { "path": "./packages/llm/llm-retry" },
     { "path": "./packages/subprocess/subprocess" },
     { "path": "./packages/subprocess/subprocess-local" },
+    { "path": "./packages/ssh/ssh" },
+    { "path": "./packages/ssh/fs-ssh" },
+    { "path": "./packages/ssh/subprocess-ssh" },
+    { "path": "./packages/ssh/sandbox-ssh" },
     { "path": "./packages/subprocess/win32-process" },
     { "path": "./packages/shell/shell" },
     { "path": "./packages/terminal/terminal" },