DeepSeek/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness synced 2026-08-15 21:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(e2b): group remote providers

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-28 14:52:37 +08:00
parent 6667102890
commit e64d40837c
81 changed files with 171 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/subprocess/README.i18n.yaml
View File

@@ -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/README.md
README.md: ae18c55205edd6085a0ed8de1bb7f875c411c79f
README.zh.md: e27fd240e4c1e96d8b859b326574e441241e8508
README.md: cdea3803e903b31e895658745bc6dbf5a3c04c71
README.zh.md: 95b02b6491077750f23ff12ffd14c1aa174609be

2
packages/subprocess/README.md
View File

@@ -8,6 +8,6 @@ The shared home for spawning managed child-process trees: fully-specified spawn
|---|---|---|
| [`subprocess`](subprocess/README.md) (`@deepseek-ai/dsh-subprocess`) | `ctx.subprocess` | The seam: abstract `SubprocessService.spawn(spec)`, the fully-explicit `SubprocessSpawnSpec` with per-stream stdio dispositions, `SubprocessHandle` (streams, offset-based readers, terminate/waitForExit/dispose), and the shared scrub + `DSH_*`/`CollectedOutput` vocabulary |
| [`subprocess-local`](subprocess-local/README.md) (`@deepseek-ai/dsh-subprocess-local`) | — | The local implementation: detached process trees, per-disposition stream wiring, tail-keep truncation with bounded private spill files, the `DSH_*` merge order, tree signalling with escalation, the dispose ladder, and terminate-and-join disposal |
| [`subprocess-e2b`](subprocess-e2b/README.md) (`@deepseek-ai/dsh-subprocess-e2b`) | — | Experimental E2B implementation: remote Linux process groups and spill state in the shared `ctx.e2b` sandbox, with asynchronous PID acquisition and SDK buffering limitations |
| [`e2b/subprocess-e2b`](../e2b/subprocess-e2b/README.md) (`@deepseek-ai/dsh-subprocess-e2b`) | — | Experimental E2B implementation: remote Linux process groups and spill state in the shared `ctx.e2b` sandbox, with asynchronous PID acquisition and SDK buffering limitations |
The service owns process lifetime across consumer reloads; consumers own what a process means (a bash command, a future non-shell runner) and every default that shapes one.

2
packages/subprocess/README.zh.md
View File

@@ -8,6 +8,6 @@ spawn 受管子进程树的共用归属位置：完全显式的 spawn spec，其
|---|---|---|
| [`subprocess`](subprocess/README.md)（`@deepseek-ai/dsh-subprocess`） | `ctx.subprocess` | seam 本体：抽象的 `SubprocessService.spawn(spec)`、完全显式且带按流划分 stdio 处置方式的 `SubprocessSpawnSpec`、`SubprocessHandle`（流、基于偏移量的读取器、terminate/waitForExit/dispose），以及共享的凭据清除 + `DSH_*`/`CollectedOutput` 词汇 |
| [`subprocess-local`](subprocess-local/README.md)（`@deepseek-ai/dsh-subprocess-local`） | 无 | 本地实现：detached 进程树、按处置方式接线的流、附带有界私有 spill 文件的尾部保留截断、`DSH_*` 合并次序、带升级的进程树信号发送、dispose 阶梯，以及先终止再等待退出的 dispose |
| [`subprocess-e2b`](subprocess-e2b/README.md)（`@deepseek-ai/dsh-subprocess-e2b`） | 无 | 实验性 E2B 实现：远程 Linux 进程组和共享 `ctx.e2b` 沙箱中的 spill 状态，但 PID 异步获取，且受 SDK 缓冲限制 |
| [`e2b/subprocess-e2b`](../e2b/subprocess-e2b/README.md)（`@deepseek-ai/dsh-subprocess-e2b`） | 无 | 实验性 E2B 实现：远程 Linux 进程组和共享 `ctx.e2b` 沙箱中的 spill 状态，但 PID 异步获取，且受 SDK 缓冲限制 |
服务拥有跨消费方重载的进程存续期；消费方拥有一个进程的含义（一条 bash 命令、未来的非 shell 运行器）以及塑造它的每一项默认值。

6
packages/subprocess/subprocess-e2b/README.i18n.yaml
View File

@@ -1,6 +0,0 @@
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
# 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-e2b/README.md
README.md: 9ec1518103413f33a52023c17722bf5a5fe275ce
README.zh.md: f50ff2f3d85a5ddd9bd352e95d0eb0a23e756404

32
packages/subprocess/subprocess-e2b/README.md
View File

@@ -1,32 +0,0 @@
# @deepseek-ai/dsh-subprocess-e2b
English | [中文](README.zh.md)
E2B implementation of the [`@deepseek-ai/dsh-subprocess`](../subprocess/README.md) seam. It has no config: load [`@deepseek-ai/dsh-e2b`](../../e2b/e2b/README.md) first, then this service in place of `dsh-subprocess-local`. Existing consumers such as [`dsh-bash-local`](../../bash/bash-local/README.md) then execute in the shared remote sandbox without an E2B-specific Bash adapter.
## Behavior
- **Asynchronous remote start** — the synchronous seam returns a handle immediately while `Sandbox.commands.run(..., { background: true })` starts remotely. `pid` is `-1` until the SDK returns the command PID; `done`, stdin, termination, and `waitForExit()` wait for readiness internally.
- **Linux process groups** — a quoted wrapper starts each argv under `setsid --wait` and records its actual process-group id plus private status files beneath `ctx.e2b.runtimeRoot/processes`. The handle waits for that file instead of assuming the SDK command PID is the group id. Termination signals the negative recorded id with `SIGTERM`, waits the caller's `graceMs`, then escalates to `SIGKILL` and the SDK kill fallback. Service disposal terminates and joins every retained handle before the sandbox owner disposes.
- **Environment boundary** — the wrapper starts from the sandbox command environment, removes ambient `DSH_*` and credential-shaped (`*KEY*`, `*SECRET*`, `*TOKEN*`) names, then restores every `spec.env` entry as an explicit caller opt-in. Host ambient variables never enter the sandbox implicitly.
- **Stdio projection** — pipe mode forwards E2B callbacks into host Node streams; inherit mode forwards them to the harness process streams; collect mode retains a bounded host tail with offset reads. Optional complete spill files are written remotely and advertised only while within their cap. Batch and streaming stdin use the SDK handle.
The base E2B image supplies the Bash/GNU utilities this adapter invokes: `bash`, `setsid`, `ps`, `tr`, `env`, `chmod`, `tee`, and `kill`. A custom template must retain compatible commands.
## Model Experience
Indirectly, through consumer seams such as the Bash executor behind `dsh-tool-bash`, which render remote output, exit facts, background deltas, and spill paths.
#### KV Cache effect
No direct invalidation; the named consumers own any request-prefix changes.
## Known Limitations and Deferred Work
- **The SDK still retains complete command output in host memory** — E2B `CommandHandle.stdout` and `.stderr` accumulate even when this adapter exposes bounded tails, so the subprocess seam's normal host-memory bound is not achieved.
- **Pipe output is not byte-faithful** — E2B delivers separately decoded strings rather than raw bytes, so split multibyte sequences and arbitrary binary protocols can be corrupted; LSP and other framed byte-stream consumers are unsupported.
- **Synchronous-PID consumers are unsupported** — `pid` remains `-1` during remote startup; consumers that require a positive PID immediately, including the ACP child backend, cannot use this provider unchanged.
- **Reconnect does not reconstruct handles** — remote PID/status/spill files survive a retained sandbox, but a new harness process does not rebuild live `SubprocessHandle` objects or output cursors from them.
- **Remote state accumulates when retained** — process directories and valid spill files remain under `.dsh-e2b`; this POC supplies no retention sweep.
- **Signal attribution is inferred** — when termination was requested and E2B reports a nonzero exit code, the adapter reports the last requested signal because the SDK result does not identify the terminating signal.
- **Linux utility and E2B transport semantics are assumed** — there is no PTY, Windows, arbitrary-template, or network-partition fidelity layer.

32
packages/subprocess/subprocess-e2b/README.zh.md
View File

@@ -1,32 +0,0 @@
# @deepseek-ai/dsh-subprocess-e2b
[English](README.md) | 中文
[`@deepseek-ai/dsh-subprocess`](../subprocess/README.md) seam 的 E2B 实现。它没有配置：先加载 [`@deepseek-ai/dsh-e2b`](../../e2b/e2b/README.md)，再用本服务取代 `dsh-subprocess-local`。随后，[`dsh-bash-local`](../../bash/bash-local/README.md) 等现有消费方会在共享远程沙箱中执行，无需 E2B 专用 Bash 适配器。
## 行为
- **异步远程启动**：同步 seam 会立即返回一个句柄，同时由 `Sandbox.commands.run(..., { background: true })` 在远程启动进程。SDK 返回命令 PID 之前，`pid` 为 `-1`；`done`、stdin、终止和 `waitForExit()` 会在内部等待就绪。
- **Linux 进程组**：带引号保护的包装层会在 `setsid --wait` 下启动每组 argv，并在 `ctx.e2b.runtimeRoot/processes` 下记录实际进程组 ID 和私有状态文件。句柄会等待该文件，而不会假设 SDK 命令 PID 就是进程组 ID。终止操作以记录的负数 ID 发送 `SIGTERM`，等待调用方的 `graceMs`，再升级到 `SIGKILL` 和 SDK kill 回退。服务 dispose（资源释放）会在沙箱所有者释放前终止并等待每个保留句柄退出。
- **环境边界**：包装层从沙箱命令环境开始，移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），再把每个 `spec.env` 条目恢复为调用方显式选择。宿主环境变量绝不会隐式进入沙箱。
- **stdio 投影**：pipe 模式把 E2B 回调转发到宿主 Node 流；inherit 模式把回调转发到 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。可选的完整 spill 文件写在远程，并且只有未超过其上限时才会对外公布。批量 stdin 和流式 stdin 都使用 SDK 句柄。
基础 E2B 镜像提供该适配器调用的 Bash/GNU 工具：`bash`、`setsid`、`ps`、`tr`、`env`、`chmod`、`tee` 和 `kill`。自定义模板必须保留兼容的命令。
## 模型体验
通过消费方 seam 间接影响模型，例如 `dsh-tool-bash` 背后的 Bash 执行器；这些消费方会渲染远程输出、退出事实、后台增量和 spill 路径。
#### KV Cache 影响
不会直接失效；请求前缀变更由具名消费方负责。
## 已知限制与延后工作
- **SDK 仍会在宿主内存中保留完整命令输出**：即使本适配器公开的是有界尾部，E2B `CommandHandle.stdout` 和 `.stderr` 仍会持续累积，因此无法达到进程管理 seam 通常提供的宿主内存边界。
- **Pipe 输出并非字节保真**：E2B 交付的是分别解码后的字符串，而不是原始字节，因此拆分的多字节序列和任意二进制协议可能损坏；不支持 LSP 及其他带帧字节流消费方。
- **不支持需要同步 PID 的消费方**：远程启动期间，`pid` 保持为 `-1`；包括 ACP 子进程后端在内，要求立即获得正 PID 的消费方无法原样使用本提供方。
- **重新连接不会重建句柄**：保留沙箱后，远程 PID／状态／spill 文件仍然存在，但新的 harness 进程不会据此重建实时 `SubprocessHandle` 对象或输出游标。
- **保留沙箱时会累积远程状态**：进程目录和有效的 spill 文件会留在 `.dsh-e2b` 下；本 POC 不提供保留清理。
- **信号归因依靠推断**：如果已经请求终止，而 E2B 报告非零退出码，适配器会报告最后请求的信号，因为 SDK 结果不标识终止信号。
- **依赖 Linux 工具与 E2B 传输语义**：没有 PTY、Windows、任意模板或网络分区的保真层。

41
packages/subprocess/subprocess-e2b/package.json
View File

@@ -1,41 +0,0 @@
{
"name": "@deepseek-ai/dsh-subprocess-e2b",
"description": "E2B subprocess implementation for DeepSeek Harness",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./invariant": {
"types": "./lib/types/invariant.d.ts",
"default": "./lib/invariant.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/invariant.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-e2b": "^0.0.1",
"@deepseek-ai/dsh-invariants": "^0.0.1",
"@deepseek-ai/dsh-subprocess": "^0.0.1",
"cordis": "^4.0.0-rc.7"
},
"devDependencies": {
"@deepseek-ai/dsh-e2b": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-subprocess": "workspace:^",
"cordis": "^4.0.0-rc.7"
}
}

58
packages/subprocess/subprocess-e2b/src/index.ts
View File

@@ -1,58 +0,0 @@
/**
* E2B implementation of the subprocess seam. Each handle starts through the
* shared sandbox and retains command output/status paths in that remote world.
* @module @deepseek-ai/dsh-subprocess-e2b
*/
import { randomUUID } from 'node:crypto'
import { posix } from 'node:path'
import { Context } from 'cordis'
import { SubprocessService } from '@deepseek-ai/dsh-subprocess'
import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
import { E2BSubprocessHandle } from './process.ts'
/** E2B command manager registered as `ctx.subprocess`. */
export class E2BSubprocessService extends SubprocessService {
static inject = ['e2b']
private readonly live = new Set<E2BSubprocessHandle>()
/** Create the E2B subprocess service and bind its disposal policy. */
constructor(ctx: Context) {
super(ctx)
ctx.effect(() => async () => {
const handles = [...this.live]
for (const handle of handles) handle.terminate()
await Promise.all(handles.map(async (handle) => {
await handle.done.catch(() => {})
await handle.waitForExit()
}))
this.live.clear()
}, 'e2b subprocess teardown')
}
/** @inheritdoc */
spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
const program = spec.argv[0]
if (program === undefined || program.length === 0) {
throw new Error('invalid argv: expected a non-empty program name at argv[0]')
}
if (!Number.isFinite(spec.graceMs) || spec.graceMs <= 0) {
throw new Error('subprocess-e2b: graceMs must be a positive finite number')
}
if (spec.signal?.aborted === true) {
throw new Error(`aborted before spawn: ${String(spec.signal.reason ?? 'aborted')}`)
}
const stateDir = posix.join(this.ctx.e2b.runtimeRoot, 'processes', randomUUID())
const handle = new E2BSubprocessHandle(this.ctx.e2b, spec, stateDir)
this.live.add(handle)
const release = async (): Promise<void> => {
await handle.waitForExit()
this.live.delete(handle)
}
void handle.done.then(release, release).catch(() => {})
return handle
}
}
export default E2BSubprocessService

27
packages/subprocess/subprocess-e2b/src/invariant.ts
View File

@@ -1,27 +0,0 @@
/** Package-owned invariant companion for `@deepseek-ai/dsh-subprocess-e2b`. */
/* jscpd:ignore-start */
import type { Context } from 'cordis'
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
const PACKAGE_NAME = '@deepseek-ai/dsh-subprocess-e2b'
/** Cordis companion plugin name. */
export const name = 'subprocess-e2b-invariant'
/** Service required before reserving package ownership. */
export const inject = ['invariants']
/**
* No runtime invariant: live remote handles are private teardown ownership,
* and the E2B command event stream is the sole outcome authority.
*/
const install: InvariantInstaller = () => {}
/**
* Register this package's invariant companion.
* @param ctx - Cordis context carrying the invariant service.
* @returns the installed registration's disposer after setup succeeds.
*/
export const apply = (ctx: Context): Promise<() => void> =>
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
/* jscpd:ignore-end */

70
packages/subprocess/subprocess-e2b/src/output.ts
View File

@@ -1,70 +0,0 @@
/** Bounded host-side projection of a complete output file retained in E2B. */
import { Buffer } from 'node:buffer'
import type { SubprocessOutputRead, SubprocessOutputReader } from '@deepseek-ai/dsh-subprocess'
/** Offset reader used for one collect-mode E2B stream. */
export class E2BOutputReader implements SubprocessOutputReader {
private chunks: Buffer[] = []
private retainedBytes = 0
private totalBytes = 0
/**
* Create a bounded reader over one remote spill path.
* @param maxBytes - In-memory tail cap.
* @param maxSpillBytes - Maximum complete remote file size the caller accepts.
* @param spillPath - Remote full-output path.
*/
constructor(
private readonly maxBytes: number,
private readonly maxSpillBytes: number | undefined,
private readonly spillPath: string,
) {}
/** Total bytes observed from the SDK stream. */
get size(): number {
return this.totalBytes
}
/**
* Append one decoded SDK output event.
* @param text - Event text delivered by E2B.
*/
push(text: string): void {
if (text.length === 0) return
const chunk = Buffer.from(text)
this.totalBytes += chunk.length
this.chunks.push(chunk)
this.retainedBytes += chunk.length
while (this.retainedBytes > this.maxBytes) {
const head = this.chunks[0] as Buffer
const excess = this.retainedBytes - this.maxBytes
if (head.length <= excess) {
this.chunks.shift()
this.retainedBytes -= head.length
} else {
this.chunks[0] = head.subarray(excess)
this.retainedBytes -= excess
}
}
}
/** @inheritdoc */
readFrom(fromByte: number): SubprocessOutputRead {
if (!Number.isSafeInteger(fromByte) || fromByte < 0) {
throw new Error('subprocess output offset must be a non-negative safe integer')
}
const retained = Buffer.concat(this.chunks, this.retainedBytes)
const firstRetained = this.totalBytes - this.retainedBytes
const lossy = fromByte < firstRetained
const start = lossy ? 0 : Math.min(retained.length, Math.max(0, fromByte - firstRetained))
return {
text: retained.subarray(start).toString('utf8'),
nextOffset: this.totalBytes,
lossy,
...(lossy && this.maxSpillBytes !== undefined && this.totalBytes <= this.maxSpillBytes
? { spillPath: this.spillPath }
: {}),
}
}
}

417
packages/subprocess/subprocess-e2b/src/process.ts
View File

@@ -1,417 +0,0 @@
/** One asynchronously-started E2B command projected onto the subprocess seam. */
import { Buffer } from 'node:buffer'
import { PassThrough, Writable } from 'node:stream'
import { posix } from 'node:path'
import {
CommandExitError,
quoteE2BShellArg,
} from '@deepseek-ai/dsh-e2b'
import type { CommandHandle, CommandResult, Sandbox } from '@deepseek-ai/dsh-e2b'
import type {
SubprocessCollect,
SubprocessHandle,
SubprocessOutcome,
SubprocessOutputMode,
SubprocessSpawnSpec,
} from '@deepseek-ai/dsh-subprocess'
import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
import { E2BOutputReader } from './output.ts'
const GROUP_POLL_MS = 20
function isCollect(mode: SubprocessOutputMode): mode is SubprocessCollect {
return mode !== 'pipe' && mode !== 'inherit'
}
function hasSpill(mode: SubprocessOutputMode): mode is SubprocessCollect & { spill: { maxBytes: number } } {
return isCollect(mode) && mode.spill !== undefined
}
function asError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error))
}
class DeferredStdin extends Writable {
constructor(private readonly ready: Promise<CommandHandle>) {
super({ decodeStrings: false })
}
override _write(chunk: string | Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
void this.ready.then(handle => handle.sendStdin(chunk)).then(
() => { callback() },
(error: unknown) => { callback(asError(error)) },
)
}
override _final(callback: (error?: Error | null) => void): void {
void this.ready.then(handle => handle.closeStdin()).then(
() => { callback() },
(error: unknown) => { callback(asError(error)) },
)
}
}
interface RemotePaths {
pid: string
status: string
stdout: string
stderr: string
}
function explicitEnvironmentNames(env: Readonly<Record<string, string>> | undefined): string {
return Object.keys(env ?? {})
.map(quoteE2BShellArg)
.join(' ')
}
function commandText(spec: SubprocessSpawnSpec, paths: RemotePaths): string {
const stdoutRedirect = hasSpill(spec.stdio.stdout)
? `> >(tee -a -- ${quoteE2BShellArg(paths.stdout)})`
: ''
const stderrRedirect = hasSpill(spec.stdio.stderr)
? `2> >(tee -a -- ${quoteE2BShellArg(paths.stderr)} >&2)`
: ''
const environmentNames = explicitEnvironmentNames(spec.env)
const inner = [
'set +e',
'umask 077',
'dsh_e2b_pgid="$(ps -o pgid= -p "$$" | tr -d " ")"',
`printf '%s\\n' "$dsh_e2b_pgid" > ${quoteE2BShellArg(paths.pid)}`,
'dsh_e2b_env=()',
`dsh_e2b_explicit=(${environmentNames})`,
'while IFS= read -r dsh_e2b_name; do',
' case "${dsh_e2b_name^^}" in DSH_*|*KEY*|*SECRET*|*TOKEN*) continue ;; esac',
' dsh_e2b_env+=("$dsh_e2b_name=${!dsh_e2b_name}")',
'done < <(compgen -e)',
'for dsh_e2b_name in "${dsh_e2b_explicit[@]}"; do dsh_e2b_env+=("$dsh_e2b_name=${!dsh_e2b_name}"); done',
`env -i "\${dsh_e2b_env[@]}" "$@" ${stdoutRedirect} ${stderrRedirect}`.trimEnd(),
'dsh_e2b_status=$?',
'wait',
`printf '%s\\n' "$dsh_e2b_status" > ${quoteE2BShellArg(paths.status)}`,
'exit "$dsh_e2b_status"',
].join('\n')
const argv = spec.argv.map(quoteE2BShellArg).join(' ')
return `exec setsid --wait -- bash -c ${quoteE2BShellArg(inner)} dsh-e2b ${argv}`
}
function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } {
return signal === undefined ? {} : { signal }
}
function isAborted(signal: AbortSignal | undefined): boolean {
return signal?.aborted === true
}
function waitTick(signal?: AbortSignal): Promise<boolean> {
if (signal?.aborted === true) return Promise.resolve(false)
return new Promise<boolean>((resolve) => {
const timer = setTimeout(() => {
signal?.removeEventListener('abort', onAbort)
resolve(true)
}, GROUP_POLL_MS)
const onAbort = (): void => {
clearTimeout(timer)
resolve(false)
}
signal?.addEventListener('abort', onAbort, { once: true })
})
}
/** E2B-backed subprocess handle with deferred remote PID acquisition. */
export class E2BSubprocessHandle implements SubprocessHandle {
readonly stdin: Writable | undefined
readonly stdout: PassThrough | undefined
readonly stderr: PassThrough | undefined
readonly collected: SubprocessHandle['collected']
readonly done: Promise<SubprocessOutcome>
private readonly readyState = Promise.withResolvers<CommandHandle>()
private readonly stdoutReader: E2BOutputReader | undefined
private readonly stderrReader: E2BOutputReader | undefined
private readonly paths: RemotePaths
private remotePid = -1
private settled = false
private terminationRequested = false
private terminationSignal: NodeJS.Signals | null = null
private termination: Promise<void> | undefined
/**
* Begin an E2B command without blocking the synchronous subprocess spawn seam.
* @param runtime - Shared E2B sandbox owner.
* @param spec - Fully resolved subprocess request.
* @param stateDir - Remote directory retaining process identity, status, and valid spills.
*/
constructor(
private readonly runtime: E2BSandboxService,
private readonly spec: SubprocessSpawnSpec,
readonly stateDir: string,
) {
this.paths = {
pid: posix.join(stateDir, 'pid'),
status: posix.join(stateDir, 'exit-code'),
stdout: posix.join(stateDir, 'stdout.log'),
stderr: posix.join(stateDir, 'stderr.log'),
}
const outMode = spec.stdio.stdout
const errMode = spec.stdio.stderr
this.stdout = outMode === 'pipe' ? new PassThrough() : undefined
this.stderr = errMode === 'pipe' ? new PassThrough() : undefined
this.stdoutReader = isCollect(outMode)
? new E2BOutputReader(outMode.maxBytes, outMode.spill?.maxBytes, this.paths.stdout)
: undefined
this.stderrReader = isCollect(errMode)
? new E2BOutputReader(errMode.maxBytes, errMode.spill?.maxBytes, this.paths.stderr)
: undefined
this.collected = {
...(this.stdoutReader !== undefined ? { stdout: this.stdoutReader } : {}),
...(this.stderrReader !== undefined ? { stderr: this.stderrReader } : {}),
}
this.stdin = spec.stdio.stdin === 'pipe' ? new DeferredStdin(this.readyState.promise) : undefined
void this.readyState.promise.catch(() => {})
spec.signal?.addEventListener('abort', this.onAbort, { once: true })
this.done = this.run()
void this.done.catch(() => {})
if (spec.signal?.aborted === true) this.terminate()
}
/** Remote process id after start; `-1` while E2B startup is pending or after it fails. */
get pid(): number {
return this.remotePid
}
/** @inheritdoc */
terminate(): void {
if (this.terminationRequested || this.settled) return
this.terminationRequested = true
this.termination = this.terminateRemote()
void this.termination.catch(() => {})
}
/** @inheritdoc */
async waitForExit(signal?: AbortSignal): Promise<boolean> {
let handle: CommandHandle | undefined
try {
handle = await this.readyForWait(signal)
} catch {
return true
}
if (handle === undefined) return false
let sandbox: Sandbox
try {
sandbox = await this.runtime.getSandbox()
} catch (error: unknown) {
if (isAborted(signal)) return false
throw error
}
while (await this.groupAlive(sandbox, this.remotePid, signal)) {
if (!await waitTick(signal)) return false
}
return !isAborted(signal)
}
private readyForWait(signal: AbortSignal | undefined): Promise<CommandHandle | undefined> {
if (signal === undefined) return this.readyState.promise
return new Promise<CommandHandle | undefined>((resolve, reject) => {
const onAbort = (): void => { cleanup(); resolve(undefined) }
const cleanup = (): void => { signal.removeEventListener('abort', onAbort) }
signal.addEventListener('abort', onAbort, { once: true })
if (signal.aborted) {
onAbort()
return
}
void this.readyState.promise.then(
(handle) => { cleanup(); resolve(handle) },
(error: unknown) => { cleanup(); reject(asError(error)) },
)
})
}
private readonly onAbort = (): void => { this.terminate() }
private async run(): Promise<SubprocessOutcome> {
try {
const sandbox = await this.runtime.getSandbox()
await this.prepareState(sandbox)
const handle = await sandbox.commands.run(
commandText(this.spec, this.paths),
{
background: true,
cwd: this.spec.cwd,
stdin: this.spec.stdio.stdin !== 'ignore',
timeoutMs: 0,
...(this.spec.env !== undefined ? { envs: this.spec.env } : {}),
onStdout: async (data) => { await this.dispatchOutput('stdout', data) },
onStderr: async (data) => { await this.dispatchOutput('stderr', data) },
},
)
if (!Number.isSafeInteger(handle.pid) || handle.pid <= 0) {
throw new Error(`subprocess-e2b: E2B returned invalid command pid ${handle.pid}`)
}
const completion = handle.wait()
void completion.catch(() => {})
this.remotePid = await this.waitForProcessGroupId(sandbox, completion)
this.readyState.resolve(handle)
await this.writeBatchStdin(handle)
const outcome = await this.waitForCommand(completion)
await this.finalizeSpills(sandbox)
return outcome
} catch (error: unknown) {
this.readyState.reject(error)
throw error
} finally {
this.settled = true
this.spec.signal?.removeEventListener('abort', this.onAbort)
this.stdout?.end()
this.stderr?.end()
}
}
private async prepareState(sandbox: Sandbox): Promise<void> {
await sandbox.files.makeDir(this.stateDir)
const files = [
{ path: this.paths.pid, data: '' },
{ path: this.paths.status, data: '' },
...(hasSpill(this.spec.stdio.stdout) ? [{ path: this.paths.stdout, data: '' }] : []),
...(hasSpill(this.spec.stdio.stderr) ? [{ path: this.paths.stderr, data: '' }] : []),
]
await sandbox.files.write(files)
await sandbox.commands.run([
`chmod 700 -- ${quoteE2BShellArg(this.stateDir)}`,
`chmod 600 -- ${files.map(file => quoteE2BShellArg(file.path)).join(' ')}`,
].join('\n'))
}
private async writeBatchStdin(handle: CommandHandle): Promise<void> {
if (typeof this.spec.stdio.stdin !== 'object') return
try {
await handle.sendStdin(this.spec.stdio.stdin.data)
await handle.closeStdin()
} catch (_processClosedItsInput) {
// Like the local adapter, batch stdin is best-effort; exit and output remain authoritative.
}
}
private async dispatchOutput(stream: 'stdout' | 'stderr', data: string): Promise<void> {
try {
if (stream === 'stdout') {
this.stdoutReader?.push(data)
await this.writeOutput(this.stdout, this.spec.stdio.stdout === 'inherit' ? process.stdout : undefined, data)
return
}
this.stderrReader?.push(data)
await this.writeOutput(this.stderr, this.spec.stdio.stderr === 'inherit' ? process.stderr : undefined, data)
} catch (error: unknown) {
const target = stream === 'stdout' ? this.stdout : this.stderr
target?.destroy(asError(error))
}
}
private async writeOutput(pipe: PassThrough | undefined, inherited: NodeJS.WriteStream | undefined, data: string): Promise<void> {
const target = pipe ?? inherited
if (target === undefined || data.length === 0) return
if (target.destroyed) throw new Error('subprocess output stream is closed')
if (target.write(Buffer.from(data))) return
await new Promise<void>((resolve, reject) => {
const onDrain = (): void => { cleanup(); resolve() }
const onError = (error: Error): void => { cleanup(); reject(error) }
const cleanup = (): void => {
target.removeListener('drain', onDrain)
target.removeListener('error', onError)
}
target.once('drain', onDrain)
target.once('error', onError)
})
}
private async waitForProcessGroupId(sandbox: Sandbox, completion: Promise<CommandResult>): Promise<number> {
const commandSettled = completion.then(
() => true,
() => true,
)
while (true) {
const raw = await sandbox.files.read(this.paths.pid)
const value = raw.trim()
if (value.length > 0) {
const pid = Number(value)
if (!/^[1-9][0-9]*$/.test(value) || !Number.isSafeInteger(pid)) {
throw new Error(`subprocess-e2b: remote wrapper published invalid process-group id ${JSON.stringify(value)}`)
}
return pid
}
const settled = await Promise.race([commandSettled, waitTick().then(() => false)])
if (settled) throw new Error('subprocess-e2b: remote command exited before publishing its process-group id')
}
}
private async waitForCommand(completion: Promise<CommandResult>): Promise<SubprocessOutcome> {
try {
const result = await completion
return { exitCode: result.exitCode, signal: null }
} catch (error: unknown) {
if (error instanceof CommandExitError) {
return this.terminationSignal === null
? { exitCode: error.exitCode, signal: null }
: { exitCode: null, signal: this.terminationSignal }
}
throw error
}
}
private async terminateRemote(): Promise<void> {
let handle: CommandHandle
try {
handle = await this.readyState.promise
} catch {
return
}
const sandbox = await this.runtime.getSandbox()
this.terminationSignal = 'SIGTERM'
await this.signalGroup(sandbox, this.remotePid, 'TERM')
const deadline = Date.now() + this.spec.graceMs
while (Date.now() < deadline && await this.groupAlive(sandbox, this.remotePid)) {
await waitTick()
}
if (!await this.groupAlive(sandbox, this.remotePid)) return
this.terminationSignal = 'SIGKILL'
try {
await this.signalGroup(sandbox, this.remotePid, 'KILL')
} finally {
await handle.kill().catch(() => false)
}
}
private async signalGroup(sandbox: Sandbox, pid: number, signal: 'TERM' | 'KILL'): Promise<void> {
try {
await sandbox.commands.run(`kill -${signal} -- -${pid}`)
} catch (error: unknown) {
if (!(error instanceof CommandExitError)) throw error
}
}
private async groupAlive(sandbox: Sandbox, pid: number, signal?: AbortSignal): Promise<boolean> {
try {
await sandbox.commands.run(`kill -0 -- -${pid}`, signalOpts(signal))
return true
} catch (error: unknown) {
if (signal?.aborted === true) return false
if (error instanceof CommandExitError) return false
throw error
}
}
private async finalizeSpills(sandbox: Sandbox): Promise<void> {
const removals: Promise<void>[] = []
const collect = (mode: SubprocessOutputMode, reader: E2BOutputReader | undefined, path: string): void => {
if (!hasSpill(mode)) return
// A spill mode is a collect mode, so construction always created its reader.
const size = (reader as E2BOutputReader).size
if (size <= mode.maxBytes || size > mode.spill.maxBytes) {
removals.push(sandbox.files.remove(path).catch(() => {}))
}
}
collect(this.spec.stdio.stdout, this.stdoutReader, this.paths.stdout)
collect(this.spec.stdio.stderr, this.stderrReader, this.paths.stderr)
await Promise.all(removals)
}
}

725
packages/subprocess/subprocess-e2b/tests/subprocess.spec.ts
View File

@@ -1,725 +0,0 @@
import { once } from 'node:events'
import { Context } from 'cordis'
import {
CommandExitError,
type CommandHandle,
type CommandResult,
type Sandbox,
} from '@deepseek-ai/dsh-e2b'
import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b'
import * as E2BSubprocessInvariant from '../src/invariant.ts'
import { E2BOutputReader } from '../src/output.ts'
import { E2BSubprocessHandle } from '../src/process.ts'
import InvariantService from '@deepseek-ai/dsh-invariants'
import { describe, expect, it, vi } from 'vitest'
function commandError(exitCode: number): CommandExitError {
return new CommandExitError({ exitCode, stdout: '', stderr: '', error: `exit ${exitCode}` })
}
interface StartOptions {
background: true
cwd: string
stdin: boolean
timeoutMs: number
signal?: AbortSignal
envs?: Record<string, string>
onStdout?: (data: string) => void | Promise<void>
onStderr?: (data: string) => void | Promise<void>
}
class FakeCommandHandle {
pid = 4242
readonly sent: Array<string | Uint8Array> = []
closes = 0
kills = 0
killError: unknown
private readonly result = Promise.withResolvers<CommandResult>()
private settled = false
wait(): Promise<CommandResult> {
return this.result.promise
}
async sendStdin(data: string | Uint8Array): Promise<void> {
this.sent.push(data)
}
async closeStdin(): Promise<void> {
this.closes += 1
}
async kill(): Promise<boolean> {
this.kills += 1
if (this.killError !== undefined) throw this.killError
return true
}
succeed(exitCode = 0): void {
if (this.settled) return
this.settled = true
this.result.resolve({ exitCode, stdout: '', stderr: '' })
}
fail(exitCode: number): void {
if (this.settled) return
this.settled = true
this.result.reject(commandError(exitCode))
}
crash(error: unknown): void {
if (this.settled) return
this.settled = true
this.result.reject(error)
}
}
class FakeSandbox {
readonly handle = new FakeCommandHandle()
readonly commandsSeen: string[] = []
readonly writtenFiles: string[][] = []
readonly removed: string[] = []
readonly directories: string[] = []
startOptions: StartOptions | undefined
backgroundError: unknown
nextRemoveError: unknown
probeError: unknown
signalError: unknown
trapsTerm = false
alive = true
processGroupId = '4242\n'
readonly processGroupReads: string[] = []
beforeProbe: (() => void) | undefined
afterProbe: (() => void) | undefined
private startGate: Promise<void> | undefined
private openStart: (() => void) | undefined
deferStart(): void {
const gate = Promise.withResolvers<undefined>()
this.startGate = gate.promise
this.openStart = () => { gate.resolve(undefined) }
}
releaseStart(): void {
this.openStart?.()
}
finish(exitCode = 0): void {
this.alive = false
if (exitCode === 0) this.handle.succeed(0)
else this.handle.fail(exitCode)
}
async stdout(data: string): Promise<void> {
await this.startOptions?.onStdout?.(data)
}
async stderr(data: string): Promise<void> {
await this.startOptions?.onStderr?.(data)
}
readonly sandbox = {
sandboxId: 'fake',
files: {
makeDir: async (path: string): Promise<boolean> => {
this.directories.push(path)
return true
},
write: async (files: Array<{ path: string; data: string }>): Promise<object[]> => {
this.writtenFiles.push(files.map(file => file.path))
return files.map(() => ({}))
},
read: async (): Promise<string> => this.processGroupReads.shift() ?? this.processGroupId,
remove: async (path: string): Promise<void> => {
this.removed.push(path)
if (this.nextRemoveError !== undefined) {
const error = this.nextRemoveError
this.nextRemoveError = undefined
throw error
}
},
},
commands: {
run: async (command: string, options?: StartOptions | { signal?: AbortSignal }): Promise<CommandHandle | CommandResult> => {
this.commandsSeen.push(command)
if (command.startsWith('kill -0 ')) {
this.beforeProbe?.()
if (options?.signal?.aborted === true) throw new DOMException('aborted', 'AbortError')
if (this.probeError !== undefined) {
const error = this.probeError
this.probeError = undefined
throw error
}
if (!this.alive) throw commandError(1)
this.afterProbe?.()
return { exitCode: 0, stdout: '', stderr: '' }
}
if (command.startsWith('kill -TERM ')) {
if (this.signalError !== undefined) {
const error = this.signalError
this.signalError = undefined
throw error
}
if (!this.trapsTerm) {
this.alive = false
this.handle.fail(143)
}
return { exitCode: 0, stdout: '', stderr: '' }
}
if (command.startsWith('kill -KILL ')) {
if (this.signalError !== undefined) {
const error = this.signalError
this.signalError = undefined
throw error
}
this.alive = false
this.handle.fail(137)
return { exitCode: 0, stdout: '', stderr: '' }
}
if ((options as StartOptions | undefined)?.background === true) {
this.startOptions = options as StartOptions
await this.startGate
if (this.backgroundError !== undefined) throw this.backgroundError
return this.handle as unknown as CommandHandle
}
return { exitCode: 0, stdout: '', stderr: '' }
},
},
} as unknown as Sandbox
}
function spec(overrides: Partial<SubprocessSpawnSpec> = {}): SubprocessSpawnSpec {
return {
argv: ['bash', '-c', 'printf ok'],
cwd: '/workspace',
stdio: {
stdin: 'ignore',
stdout: { maxBytes: 4, spill: { maxBytes: 16 } },
stderr: { maxBytes: 4 },
},
graceMs: 5,
...overrides,
}
}
function runtime(fake: FakeSandbox, getSandbox: () => Promise<Sandbox> = async () => fake.sandbox): E2BSandboxService {
return {
cwd: '/workspace',
runtimeRoot: '/workspace/.dsh-e2b',
disposeMode: 'kill',
getSandbox,
} as unknown as E2BSandboxService
}
async function flush(): Promise<void> {
await new Promise(resolve => setTimeout(resolve, 0))
}
describe('E2BOutputReader', () => {
it('keeps a byte-exact tail with independent whole-stream cursors', () => {
const reader = new E2BOutputReader(4, 10, '/remote/spill')
reader.push('')
reader.push('ab')
reader.push('cdef')
expect(reader.size).toBe(6)
expect(reader.readFrom(0)).toEqual({ text: 'cdef', nextOffset: 6, lossy: true, spillPath: '/remote/spill' })
expect(reader.readFrom(2)).toEqual({ text: 'cdef', nextOffset: 6, lossy: false })
expect(reader.readFrom(5)).toEqual({ text: 'f', nextOffset: 6, lossy: false })
expect(reader.readFrom(99)).toEqual({ text: '', nextOffset: 6, lossy: false })
})
it('drops whole head chunks and withholds absent or over-cap spills', () => {
const withoutSpill = new E2BOutputReader(2, undefined, '/unused')
withoutSpill.push('ab')
withoutSpill.push('cd')
expect(withoutSpill.readFrom(0)).toEqual({ text: 'cd', nextOffset: 4, lossy: true })
const overCap = new E2BOutputReader(2, 3, '/too-small')
overCap.push('abcd')
expect(overCap.readFrom(0)).toEqual({ text: 'cd', nextOffset: 4, lossy: true })
expect(() => overCap.readFrom(-1)).toThrow(/non-negative safe integer/)
expect(() => overCap.readFrom(1.5)).toThrow(/non-negative safe integer/)
})
})
describe('E2BSubprocessHandle', () => {
it('starts asynchronously, keeps secrets out of the command, and supports deferred piped stdin/output', async () => {
const fake = new FakeSandbox()
fake.processGroupId = '4343\n'
fake.deferStart()
const handle = new E2BSubprocessHandle(runtime(fake), spec({
argv: ['tool', 'argument with spaces'],
stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: 8, spill: { maxBytes: 32 } } },
env: { PATH: '/bin', DEEPSEEK_API_KEY: 'explicit-secret', DSH_MODE: 'test' },
}), '/workspace/.dsh-e2b/processes/one')
expect(handle.pid).toBe(-1)
handle.stdin!.write('hello')
handle.stdin!.end()
fake.releaseStart()
await flush()
expect(handle.pid).toBe(4343)
expect(fake.handle.sent.map(value => String(value))).toEqual(['hello'])
expect(fake.handle.closes).toBe(1)
expect(fake.startOptions?.envs).toEqual({ PATH: '/bin', DEEPSEEK_API_KEY: 'explicit-secret', DSH_MODE: 'test' })
const command = fake.commandsSeen.find(value => value.startsWith('exec setsid'))!
expect(command).toContain('exec setsid --wait -- bash -c')
expect(command).toContain('DEEPSEEK_API_KEY')
expect(command).toContain('DSH_MODE')
expect(command).not.toContain('explicit-secret')
expect(fake.writtenFiles[0]).toEqual([
'/workspace/.dsh-e2b/processes/one/pid',
'/workspace/.dsh-e2b/processes/one/exit-code',
'/workspace/.dsh-e2b/processes/one/stderr.log',
])
let piped = ''
handle.stdout!.on('data', (chunk) => { piped += String(chunk) })
await fake.stdout('pipe-data')
await fake.stderr('err')
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
expect(piped).toBe('pipe-data')
expect(handle.collected.stderr!.readFrom(0)).toMatchObject({ text: 'err', lossy: false })
expect(fake.removed).toContain('/workspace/.dsh-e2b/processes/one/stderr.log')
await expect(handle.waitForExit()).resolves.toBe(true)
})
it('surfaces deferred piped-stdin write and close failures as stream errors', async () => {
const writeFake = new FakeSandbox()
writeFake.deferStart()
vi.spyOn(writeFake.handle, 'sendStdin').mockRejectedValueOnce('stdin rejected')
const writeHandle = new E2BSubprocessHandle(runtime(writeFake), spec({
stdio: { stdin: 'pipe', stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } },
}), '/runtime/stdin-write-error')
const writeError = once(writeHandle.stdin!, 'error')
writeHandle.stdin!.write('input')
writeFake.releaseStart()
await expect(writeError).resolves.toMatchObject([{ message: 'stdin rejected' }])
writeFake.finish()
await writeHandle.done
const closeFake = new FakeSandbox()
vi.spyOn(closeFake.handle, 'closeStdin').mockRejectedValueOnce(new Error('close rejected'))
const closeHandle = new E2BSubprocessHandle(runtime(closeFake), spec({
stdio: { stdin: 'pipe', stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } },
}), '/runtime/stdin-close-error')
await flush()
const closeError = once(closeHandle.stdin!, 'error')
closeHandle.stdin!.end()
await expect(closeError).resolves.toMatchObject([{ message: 'close rejected' }])
closeFake.finish()
await closeHandle.done
})
it('collects bounded tails, retains valid spills, and maps natural nonzero exits', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: {
stdin: { data: 'batch' },
stdout: { maxBytes: 4, spill: { maxBytes: 16 } },
stderr: { maxBytes: 3 },
},
}), '/runtime/two')
await flush()
await fake.stdout('abcdef')
await fake.stderr('12345')
fake.finish(7)
await expect(handle.done).resolves.toEqual({ exitCode: 7, signal: null })
expect(fake.handle.sent).toEqual(['batch'])
expect(fake.handle.closes).toBe(1)
expect(handle.collected.stdout!.readFrom(0)).toEqual({
text: 'cdef',
nextOffset: 6,
lossy: true,
spillPath: '/runtime/two/stdout.log',
})
expect(handle.collected.stderr!.readFrom(0)).toEqual({ text: '345', nextOffset: 5, lossy: true })
expect(fake.removed).not.toContain('/runtime/two/stdout.log')
})
it('removes a spill once the complete stream exceeds its cap', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: { stdin: 'ignore', stdout: { maxBytes: 2, spill: { maxBytes: 3 } }, stderr: 'inherit' },
}), '/runtime/oversize')
await flush()
await fake.stdout('abcd')
await fake.stderr('')
fake.finish()
await handle.done
expect(handle.collected.stdout!.readFrom(0)).toEqual({ text: 'cd', nextOffset: 4, lossy: true })
expect(fake.removed).toContain('/runtime/oversize/stdout.log')
})
it('contains remote spill-removal failures and routes empty inherited output', async () => {
const fake = new FakeSandbox()
fake.nextRemoveError = new Error('already removed')
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: { stdin: 'ignore', stdout: 'inherit', stderr: { maxBytes: 4, spill: { maxBytes: 8 } } },
}), '/runtime/remove-error')
await flush()
await fake.stdout('')
await fake.stderr('')
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
expect(fake.removed).toContain('/runtime/remove-error/stderr.log')
})
it('terminates a process group with TERM and reports the signal outcome', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/term')
await flush()
handle.terminate()
handle.terminate()
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
await expect(handle.waitForExit()).resolves.toBe(true)
expect(fake.commandsSeen).toContain('kill -TERM -- -4242')
expect(fake.commandsSeen).not.toContain('kill -KILL -- -4242')
})
it('escalates a TERM-trapping process group to KILL and uses the SDK kill as fallback', async () => {
const fake = new FakeSandbox()
fake.trapsTerm = true
fake.handle.killError = new Error('already gone')
const handle = new E2BSubprocessHandle(runtime(fake), spec({ graceMs: 1 }), '/runtime/kill')
await flush()
handle.terminate()
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
await expect(handle.waitForExit()).resolves.toBe(true)
expect(fake.commandsSeen).toContain('kill -KILL -- -4242')
expect(fake.handle.kills).toBe(1)
})
it('honors termination requested before asynchronous startup finishes', async () => {
const fake = new FakeSandbox()
fake.deferStart()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/deferred-kill')
handle.terminate()
fake.releaseStart()
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
})
it('honors an already-aborted signal when constructing the asynchronous handle directly', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec({ signal: AbortSignal.abort('stop') }), '/runtime/pre-aborted')
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
})
it('reacts to a signal that aborts after the remote command has started', async () => {
const fake = new FakeSandbox()
const controller = new AbortController()
const handle = new E2BSubprocessHandle(runtime(fake), spec({ signal: controller.signal }), '/runtime/live-abort')
await flush()
controller.abort('stop')
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
})
it('bounds waitForExit while startup or a live group is pending', async () => {
const fake = new FakeSandbox()
fake.deferStart()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/wait')
const beforeStart = new AbortController()
const pending = handle.waitForExit(beforeStart.signal)
beforeStart.abort()
await expect(pending).resolves.toBe(false)
await expect(handle.waitForExit(AbortSignal.abort())).resolves.toBe(false)
fake.releaseStart()
await flush()
const live = new AbortController()
const liveWait = handle.waitForExit(live.signal)
live.abort()
await expect(liveWait).resolves.toBe(false)
fake.finish()
await handle.done
})
it('bounds both sides of the liveness-poll abort race', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/poll-abort')
await flush()
const beforeTick = new AbortController()
fake.afterProbe = () => { beforeTick.abort(); fake.afterProbe = undefined }
await expect(handle.waitForExit(beforeTick.signal)).resolves.toBe(false)
const duringTick = new AbortController()
fake.afterProbe = () => {
fake.afterProbe = undefined
setTimeout(() => { duringTick.abort() }, 0)
}
await expect(handle.waitForExit(duringTick.signal)).resolves.toBe(false)
const duringProbe = new AbortController()
fake.beforeProbe = () => { duringProbe.abort(); fake.beforeProbe = undefined }
await expect(handle.waitForExit(duringProbe.signal)).resolves.toBe(false)
fake.finish()
await handle.done
})
it('observes a live group across one successful bounded poll', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/poll-success')
await flush()
setTimeout(() => { fake.finish() }, 1)
await expect(handle.waitForExit(new AbortController().signal)).resolves.toBe(true)
await handle.done
})
it('treats startup failure as no live tree and contains readiness rejection', async () => {
const fake = new FakeSandbox()
fake.backgroundError = new Error('start failed')
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/fail')
await expect(handle.done).rejects.toThrow('start failed')
expect(handle.pid).toBe(-1)
await expect(handle.waitForExit()).resolves.toBe(true)
handle.terminate()
})
it('bounds a readiness rejection with a still-live caller signal', async () => {
const fake = new FakeSandbox()
fake.deferStart()
fake.backgroundError = new Error('start failed')
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/fail-with-signal')
const waiting = handle.waitForExit(new AbortController().signal)
fake.releaseStart()
await expect(handle.done).rejects.toThrow('start failed')
await expect(waiting).resolves.toBe(true)
})
it('propagates an unavailable sandbox unless the caller aborts the wait', async () => {
const fake = new FakeSandbox()
let calls = 0
const unavailable = runtime(fake, async () => {
calls += 1
if (calls === 1) return fake.sandbox
throw new Error('connection unavailable')
})
const handle = new E2BSubprocessHandle(unavailable, spec(), '/runtime/unavailable')
await flush()
await expect(handle.waitForExit()).rejects.toThrow('connection unavailable')
fake.finish()
await handle.done
})
it('returns false when the caller aborts while reconnecting for liveness', async () => {
const fake = new FakeSandbox()
const reconnect = Promise.withResolvers<Sandbox>()
let calls = 0
const unavailable = runtime(fake, async () => {
calls += 1
return calls === 1 ? fake.sandbox : await reconnect.promise
})
const handle = new E2BSubprocessHandle(unavailable, spec(), '/runtime/reconnect-abort')
await flush()
const controller = new AbortController()
const waiting = handle.waitForExit(controller.signal)
await flush()
controller.abort()
reconnect.reject(new Error('connection unavailable'))
await expect(waiting).resolves.toBe(false)
fake.finish()
await handle.done
})
it('returns false when a liveness request itself is aborted and surfaces other probe failures', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/probe')
await flush()
const controller = new AbortController()
controller.abort()
await expect(handle.waitForExit(controller.signal)).resolves.toBe(false)
fake.probeError = new Error('probe failed')
await expect(handle.waitForExit()).rejects.toThrow('probe failed')
fake.finish()
await handle.done
})
it('makes batch stdin close failures best-effort', async () => {
const fake = new FakeSandbox()
vi.spyOn(fake.handle, 'sendStdin').mockRejectedValueOnce(new Error('closed'))
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: { stdin: { data: 'ignored' }, stdout: { maxBytes: 4 }, stderr: { maxBytes: 4 } },
}), '/runtime/stdin-closed')
await flush()
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
})
it('rejects malformed SDK process ids and non-command settlement failures', async () => {
const invalidPid = new FakeSandbox()
invalidPid.handle.pid = 0
const invalid = new E2BSubprocessHandle(runtime(invalidPid), spec(), '/runtime/invalid-pid')
await expect(invalid.done).rejects.toThrow(/invalid command pid 0/)
await expect(invalid.waitForExit()).resolves.toBe(true)
const crashedFake = new FakeSandbox()
const crashed = new E2BSubprocessHandle(runtime(crashedFake), spec(), '/runtime/crashed')
await flush()
crashedFake.alive = false
crashedFake.handle.crash(new Error('command transport failed'))
await expect(crashed.done).rejects.toThrow('command transport failed')
})
it('rejects invalid or absent process-group publication', async () => {
const invalidGroup = new FakeSandbox()
invalidGroup.processGroupId = 'not-a-pid\n'
const invalid = new E2BSubprocessHandle(runtime(invalidGroup), spec(), '/runtime/invalid-group')
await expect(invalid.done).rejects.toThrow(/invalid process-group id/)
const absentGroup = new FakeSandbox()
absentGroup.processGroupId = ''
const absent = new E2BSubprocessHandle(runtime(absentGroup), spec(), '/runtime/absent-group')
await flush()
absentGroup.finish()
await expect(absent.done).rejects.toThrow(/exited before publishing/)
})
it('waits for delayed process-group publication', async () => {
const fake = new FakeSandbox()
fake.processGroupReads.push('', '4242\n')
const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/delayed-group')
await vi.waitFor(() => { expect(handle.pid).toBe(4242) })
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
})
it('handles output backpressure and contains a stderr sink failure', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' },
}), '/runtime/backpressure')
await flush()
handle.stdout!.on('error', () => {})
const stdoutWrite = vi.spyOn(handle.stdout!, 'write').mockReturnValueOnce(false)
const stdoutPending = fake.stdout('blocked')
queueMicrotask(() => { handle.stdout!.emit('drain') })
await stdoutPending
stdoutWrite.mockRestore()
handle.stderr!.on('error', () => {})
const stderrWrite = vi.spyOn(handle.stderr!, 'write').mockReturnValueOnce(false)
const stderrPending = fake.stderr('broken')
queueMicrotask(() => { handle.stderr!.emit('error', new Error('sink failed')) })
await stderrPending
stderrWrite.mockRestore()
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
})
it('contains a pipe callback failure instead of rejecting command settlement', async () => {
const fake = new FakeSandbox()
const handle = new E2BSubprocessHandle(runtime(fake), spec({
stdio: { stdin: 'ignore', stdout: 'pipe', stderr: { maxBytes: 4 } },
}), '/runtime/pipe-error')
await flush()
const emitted = once(handle.stdout!, 'error')
handle.stdout!.destroy(new Error('consumer failed'))
await emitted
await fake.stdout('late output')
fake.finish()
await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null })
})
it('contains an already-gone group signal and observes non-command signal failures', async () => {
const gone = new FakeSandbox()
gone.trapsTerm = true
gone.signalError = commandError(1)
const goneHandle = new E2BSubprocessHandle(runtime(gone), spec({ graceMs: 1 }), '/runtime/gone-signal')
await flush()
goneHandle.terminate()
await expect(goneHandle.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
const failed = new FakeSandbox()
failed.signalError = new Error('signal transport failed')
const failedHandle = new E2BSubprocessHandle(runtime(failed), spec(), '/runtime/failed-signal')
await flush()
failedHandle.terminate()
await flush()
failed.finish()
await expect(failedHandle.done).resolves.toEqual({ exitCode: 0, signal: null })
})
})
describe('E2BSubprocessService', () => {
async function service(
fake = new FakeSandbox(),
providedRuntime: E2BSandboxService = runtime(fake),
): Promise<{ ctx: Context; fiber: Awaited<ReturnType<Context['plugin']>> }> {
const ctx = new Context()
ctx.provide('e2b', providedRuntime)
const fiber = await ctx.plugin(E2BSubprocessService)
return { ctx, fiber }
}
it('registers handles and disposal terminates and joins live remote groups regardless of sandbox policy', async () => {
const fake = new FakeSandbox()
fake.trapsTerm = true
const { ctx, fiber } = await service(fake)
const handle = ctx.subprocess.spawn(spec({ graceMs: 1 }))
await flush()
await fiber.dispose()
await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
expect(fake.alive).toBe(false)
})
it('releases naturally settled handles before later service disposal', async () => {
const fake = new FakeSandbox()
const { ctx, fiber } = await service(fake)
const handle = ctx.subprocess.spawn(spec())
await flush()
fake.finish()
await handle.done
await flush()
const signalsBefore = fake.commandsSeen.filter(command => command.startsWith('kill -')).length
await fiber.dispose()
expect(fake.commandsSeen.filter(command => command.startsWith('kill -')).length).toBe(signalsBefore)
})
it('contains a release liveness failure and retries quiescence during disposal', async () => {
const fake = new FakeSandbox()
let calls = 0
const reconnecting = runtime(fake, async () => {
calls += 1
if (calls === 2) throw new Error('transient liveness failure')
return fake.sandbox
})
const { ctx, fiber } = await service(fake, reconnecting)
const handle = ctx.subprocess.spawn(spec())
await flush()
fake.finish()
await handle.done
await flush()
await fiber.dispose()
expect(calls).toBeGreaterThanOrEqual(3)
})
it('contains spawn rejection while disposal is joining the pending handle', async () => {
const fake = new FakeSandbox()
fake.deferStart()
fake.backgroundError = new Error('start failed during disposal')
const { ctx, fiber } = await service(fake)
const handle = ctx.subprocess.spawn(spec())
const disposing = fiber.dispose()
fake.releaseStart()
await expect(disposing).resolves.toBeUndefined()
await expect(handle.done).rejects.toThrow('start failed during disposal')
})
it('validates synchronous spawn preconditions', async () => {
const { ctx } = await service()
expect(() => ctx.subprocess.spawn(spec({ argv: [] }))).toThrow(/non-empty program/)
expect(() => ctx.subprocess.spawn(spec({ graceMs: 0 }))).toThrow(/positive finite/)
expect(() => ctx.subprocess.spawn(spec({ signal: AbortSignal.abort('stop') }))).toThrow(/aborted before spawn/)
expect(() => ctx.subprocess.spawn(spec({ signal: { aborted: true, reason: undefined } as AbortSignal }))).toThrow(/aborted$/)
})
it('registers the package-owned empty invariant installer', async () => {
const ctx = new Context()
await ctx.plugin(InvariantService, { enabled: true })
const fiber = await ctx.plugin(E2BSubprocessInvariant).await()
await fiber.dispose()
})
})

25
packages/subprocess/subprocess-e2b/tsconfig.json
View File

@@ -1,25 +0,0 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": ["src"],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../e2b/e2b"
},
{
"path": "../subprocess"
},
{
"path": "../../support/invariants"
}
]
}