diff --git a/docs/config-catalog.md b/docs/config-catalog.md index b8a49a7ba2..977921db07 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -1813,6 +1813,20 @@ export interface Config { Source: [`packages/subagent/subagent-spawn/src/index.ts:25`](../packages/subagent/subagent-spawn/src/index.ts) +## `@deepseek-ai/dsh-subprocess-e2b` + +Requires: `e2b` + +```ts config-catalog +/** Configuration for the E2B subprocess adapter. */ +export interface Config { + /** Remote status/liveness poll cadence in milliseconds; each tick is one control-plane request. */ + pollMs?: number +} +``` + +Source: [`packages/e2b/subprocess-e2b/src/index.ts:24`](../packages/e2b/subprocess-e2b/src/index.ts) + ## `@deepseek-ai/dsh-system-prompt` ```ts config-catalog diff --git a/examples/headless-agent/e2b.cordis.yml b/examples/headless-agent/e2b.cordis.yml index a65b0baf81..61bdbc8aa7 100644 --- a/examples/headless-agent/e2b.cordis.yml +++ b/examples/headless-agent/e2b.cordis.yml @@ -1,6 +1,12 @@ # POC overlay: keep the advanced headless agent and model-facing tools, but # place its filesystem and process substrate in one short-lived E2B sandbox; # the generic Bash, PTY, and LSP consumers compose above them. +# +# One-world invariant: e2b.cwd, sandbox-policy.workspaceRoot, and bash-local's +# default workdir (implicit host process.cwd()) must all name the same remote +# directory. Only e2b.cwd is created at sandbox open; dropping its !!js line +# falls back to /home/user/workspace while Bash and PTY keep targeting the +# host path, so every tool call fails with a remote spawn error. - id: base name: '@cordisjs/plugin-include' config: diff --git a/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml b/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml index 3a6deb7e49..e6fd968779 100644 --- a/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml +++ b/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml @@ -1,3 +1,6 @@ +# One-world invariant (same pairing as examples/headless-agent/e2b.cordis.yml): +# e2b.cwd and sandbox-policy.workspaceRoot must name the same remote directory, +# which is also bash-local's implicit default workdir. - id: e2b name: '@deepseek-ai/dsh-e2b' config: diff --git a/packages/e2b/fs-e2b/src/index.ts b/packages/e2b/fs-e2b/src/index.ts index 3b0c835d8c..84c9ac7868 100644 --- a/packages/e2b/fs-e2b/src/index.ts +++ b/packages/e2b/fs-e2b/src/index.ts @@ -232,7 +232,13 @@ export class E2BFileSystem extends FileSystem { await this.requireRegular(target, signal) let stream: ReadableStream try { - stream = await sandbox.files.read(String(target.targetKey), { format: 'stream', ...signalOpts(signal) }) + // The pinned SDK's stream overload lies for empty files: content-length 0 + // returns '' instead of a ReadableStream. + const read = await sandbox.files.read(String(target.targetKey), { format: 'stream', ...signalOpts(signal) }) as + ReadableStream | string + stream = typeof read === 'string' + ? new ReadableStream({ start(controller) { controller.close() } }) + : read } catch (error: unknown) { throw mapError(error, 'read', target.displayPath, signal) } diff --git a/packages/e2b/fs-e2b/tests/filesystem.spec.ts b/packages/e2b/fs-e2b/tests/filesystem.spec.ts index 618472a6b8..9cad00b6eb 100644 --- a/packages/e2b/fs-e2b/tests/filesystem.spec.ts +++ b/packages/e2b/fs-e2b/tests/filesystem.spec.ts @@ -149,7 +149,7 @@ class FakeRemote { } return this.info(path) }, - read: async (path: string, options: { format: 'bytes' | 'stream'; signal?: AbortSignal }): Promise> => { + read: async (path: string, options: { format: 'bytes' | 'stream'; signal?: AbortSignal }): Promise | string> => { this.checkAbort(options) if (this.nextReadError !== undefined) { const error = this.nextReadError @@ -158,6 +158,8 @@ class FakeRemote { } const data = this.followed(path).node.data if (options.format === 'bytes') return data.slice() + // Pinned-SDK fidelity: a content-length-0 response returns '' even in stream format. + if (data.length === 0 && this.streamChunks === undefined) return '' const chunks = this.streamChunks ?? [data.slice()] return new ReadableStream({ start: (controller) => { @@ -375,6 +377,15 @@ describe('E2BFileSystem identity, metadata, and reads', () => { expect(initiallyBuffered).toBe('€') }) + it('streams an empty file even though the pinned SDK returns a non-stream value', async () => { + const remote = new FakeRemote() + remote.file('/workspace/empty.txt', '') + const { fs } = await setup(remote) + let streamed = '' + for await (const chunk of await fs.streamText(await fs.resolve('empty.txt'))) streamed += chunk + expect(streamed).toBe('') + }) + it('cancels a remote stream when its consumer stops early', async () => { const remote = new FakeRemote() remote.file('/workspace/text.txt', 'ab') diff --git a/packages/e2b/subprocess-e2b/README.i18n.yaml b/packages/e2b/subprocess-e2b/README.i18n.yaml index 0a18bf24a1..d0ce148746 100644 --- a/packages/e2b/subprocess-e2b/README.i18n.yaml +++ b/packages/e2b/subprocess-e2b/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/e2b/subprocess-e2b/README.md -README.md: 3bb9802c3afa41f3949fceeb73e76ea5cebce744 -README.zh.md: 520f0ea1cde18ed523ba80015d3d4700730aba7a +README.md: 926d4f22f8e96daa103c236121d35e461290221a +README.zh.md: d1634b6d1a701f7f4815135b4ec6bb86ad1a8c87 diff --git a/packages/e2b/subprocess-e2b/README.md b/packages/e2b/subprocess-e2b/README.md index 3bb9802c3a..926d4f22f8 100644 --- a/packages/e2b/subprocess-e2b/README.md +++ b/packages/e2b/subprocess-e2b/README.md @@ -2,16 +2,22 @@ English | [中文](README.zh.md) -E2B implementation of the [`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam. It has no config: load [`@deepseek-ai/dsh-e2b`](../e2b/README.md) first, then this service in place of `dsh-subprocess-local`. Existing Bash, PTY, and LSP consumers then execute in the shared remote sandbox without E2B-specific capability packages. +E2B implementation of the [`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam. Load [`@deepseek-ai/dsh-e2b`](../e2b/README.md) first, then this service in place of `dsh-subprocess-local`. Existing Bash, PTY, and LSP consumers then execute in the shared remote sandbox without E2B-specific capability packages. + +## Configuration + +| Key | Default | Meaning | +| --- | --- | --- | +| `pollMs` | `20` | Remote status/liveness poll cadence in milliseconds; each tick is one control-plane request, so a larger value trades exit-observation latency for fewer requests. | ## Behavior - **Asynchronous remote start** — the synchronous seam returns a handle immediately while `Sandbox.commands.run(..., { background: true })` starts remotely. `pid` is `-1` until the wrapper publishes and the adapter validates its process-group id; stdin and ordinary observation wait for that publication. An owned startup signal aborts environment and private-state preparation before allocation; once allocation begins, cancellation waits for a provisional SDK handle it can clean. -- **Execution-world coordinates** — `cwd` and private `runtimeRoot` come from the shared owner; executable lookup verifies absolute paths or resolves a bare name against the sandbox PATH plus explicit overrides. +- **Execution-world coordinates** — `cwd` and private `runtimeRoot` come from the shared owner; executable lookup verifies absolute paths or resolves a bare name against the sandbox PATH plus explicit overrides, and rejects relative paths containing separators like every subprocess provider. - **Linux process groups** — a quoted wrapper starts each argv under `exec 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 treating the SDK command PID as its published identity. Termination signals the negative recorded id with `SIGTERM`, waits the caller's `graceMs`, then escalates to `SIGKILL` and the SDK kill fallback; TERM delivery or probe failures also force that escalation. Process-table probes treat groups containing only zombie or dead entries as quiescent. Force cleanup succeeds only after a bounded probe finds the group empty; otherwise `waitForExit()` exposes a retryable failure, while proven quiescence makes later termination a no-op. Publication and monitoring failures apply the same cleanup transaction before rejecting. Service disposal rejects new starts, terminates and joins every retained process group, then awaits SDK settlement and private cleanup before the sandbox owner disposes. - **Environment boundary** — one trusted control-shell probe resolves the sandbox user's login home from its passwd entry and transports the sandbox environment as base64 ASCII for one strict UTF-8 decode; the wrapper then removes ambient `DSH_*` and credential-shaped (`*KEY*`, `*SECRET*`, `*TOKEN*`) names and restores every valid `spec.env` entry as an explicit caller opt-in. Empty names, `=`, and NUL framing violations reject before launch. Subsequent E2B command and PTY login shells receive a fresh randomized root-level `HOME` plus empty overrides for every scrubbed ambient name before user profiles can run; the requested argv receives the serialized environment afterward without changing the sandbox user's umask. Host ambient variables never enter the sandbox implicitly. Private environment files are removed after consumption, and failed command or terminal setup removes its private state before rejecting. - **Stdio projection** — the remote wrapper branches raw bytes into optional bounded spill files, frames each live chunk as newline-delimited base64 ASCII, and the host incrementally restores bytes across arbitrary SDK callback boundaries. Pipe mode writes those bytes to host Node streams; inherit mode writes them to the harness process streams; collect mode retains a bounded host tail with offset reads. The wrapper publishes the direct command status before waiting for inherited writers. For collect or inherit output, the adapter disconnects an incomplete SDK stream after `graceMs`, withholds its partial spill, and returns that status while retaining the remote group for `waitForExit()` and termination. Natural raw-pipe completion instead awaits lossless transport and preserves backpressure; explicit termination destroys the host pipes and releases blocked output before remote cleanup. Batch and streaming stdin use the SDK handle. -- **Terminal sessions** — `spawnTerminal()` uses E2B's byte PTY API, installs the exact argv and scrubbed environment through private mode-`0600` files, reports the foreground process group, sends real signals, and tears down every live group in the remote terminal session through one retryable awaited `terminate()`; termination rejects new handle operations, aborts and joins in-flight writes, inspections, and signals, and treats zombie-only groups as quiescent. A private random output boundary discards the E2B bootstrap shell's prompt and echoed runner command while preserving every requested-process byte, including its first prompt. PTY allocation is awaited through handle publication before cancellation is observed, so owned rollback can clean the published handle. Setup and teardown own the private state transaction, abort pending setup during service disposal, and fence publication; sandbox disposal or timeout bounds a setup rollback that also fails. Prompt detection, scrollback, readiness, and owner policy remain in `dsh-pty-local`. +- **Terminal sessions** — `spawnTerminal()` uses E2B's byte PTY API, installs the exact argv and scrubbed environment through private mode-`0600` files, reports the foreground process group, sends real signals, and tears down every live group in the remote terminal session through one retryable awaited `terminate()`; termination rejects new handle operations, aborts and joins in-flight writes, inspections, and signals, and treats zombie-only groups as quiescent. A private random output boundary discards the E2B bootstrap shell's prompt and echoed runner command while preserving every requested-process byte, including its first prompt. Terminal output is pushed to the handle's stream without awaiting host backpressure: a flowing consumer (the PTY backend attaches one at construction) folds bytes into its own bounded state, while a paused consumer buffers in host memory. PTY allocation is awaited through handle publication before cancellation is observed, so owned rollback can clean the published handle. Setup and teardown own the private state transaction, abort pending setup during service disposal, and fence publication; sandbox disposal or timeout bounds a setup rollback that also fails. Prompt detection, scrollback, readiness, and owner policy remain in `dsh-pty-local`. - **Sandbox disappearance** — `SandboxNotFoundError` during process or terminal liveness, termination, rollback, or disconnect proves the remote execution world cannot retain work, so cleanup treats it as quiescent; unrelated failures remain observable. The default E2B base image supplies the runtime and Bash/GNU utilities this adapter invokes: `node`, `bash`, `setsid`, `ps`, `awk`, `tr`, `env`, `base64`, `chmod`, `tee`, `head`, `rm`, `kill`, `id`, and `getent`. @@ -29,6 +35,7 @@ No direct invalidation; the named consumers own any request-prefix changes. - **The SDK still retains complete command output in host memory** — E2B `CommandHandle.stdout` and `.stderr` accumulate the base64 transport even when this adapter exposes bounded raw-byte tails, so the subprocess seam's normal host-memory bound is not achieved and transport retention is larger than the source stream. - **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. - **Private state lives for the sandbox lifetime** — process directories and valid spill files remain under `.dsh-e2b` until the owner deletes the sandbox; this POC supplies no in-sandbox sweep. +- **Control state shares the sandbox user's UID** — E2B runs every command as the same default user, so `0700`/`0600` modes cannot isolate `.dsh-e2b` control files from concurrently running sandbox processes. A background process could rewrite `pid`/`exit-code` or read a not-yet-consumed `environment` file. The adapter validates published values and refuses group ids whose negative form is unsafe to signal (`<= 1`), but real isolation needs an E2B per-command user or an out-of-band control channel. - **Numeric process identities are not reuse-fenced** — E2B exposes numeric PID/PGID PTY input, signalling, and cleanup operations but no atomic identity-bound alternative. The adapter minimizes host round trips and live coverage exercises the reproducible stale-interrupt overlap; replacement is deferred until E2B adds an identity primitive or a failure demonstrates a narrower protocol. - **The initial environment probe inherits sandbox defaults** — E2B merges command overrides with default environment entries, so the probe cannot blank unknown credential-shaped names before enumerating them. A same-UID untrusted process already in the sandbox could inspect that short-lived control shell; this POC therefore does not support secrets in sandbox-default environment variables and requires an E2B replacement-environment primitive to close the gap. - **E2B exposes no signal fact** — an adapter-requested `SIGTERM` or `SIGKILL` is reported only when no wrapper-published direct exit code wins; every unrequested SDK exit remains an exit code, including values shaped like `128 + signal`. diff --git a/packages/e2b/subprocess-e2b/README.zh.md b/packages/e2b/subprocess-e2b/README.zh.md index 520f0ea1cd..d1634b6d1a 100644 --- a/packages/e2b/subprocess-e2b/README.zh.md +++ b/packages/e2b/subprocess-e2b/README.zh.md @@ -2,16 +2,22 @@ [English](README.md) | 中文 -[`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam 的 E2B 实现。它没有配置：先加载 [`@deepseek-ai/dsh-e2b`](../e2b/README.md)，再用本服务取代 `dsh-subprocess-local`。现有的 Bash、PTY 和 LSP 消费方随后会在共享远程沙箱中执行，无需 E2B 专用的功能包（package）。 +[`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam 的 E2B 实现。先加载 [`@deepseek-ai/dsh-e2b`](../e2b/README.md)，再用本服务取代 `dsh-subprocess-local`。现有的 Bash、PTY 和 LSP 消费方随后会在共享远程沙箱中执行，无需 E2B 专用的功能包（package）。 + +## 配置 + +| 键 | 默认值 | 含义 | +| --- | --- | --- | +| `pollMs` | `20` | 远程状态／存活轮询节奏（毫秒）；每个 tick 是一次控制面请求，调大该值以牺牲退出观察延迟换取更少的请求。 | ## 行为 - **异步远程启动**：同步 seam 会立即返回一个句柄，同时由 `Sandbox.commands.run(..., { background: true })` 在远程启动进程。包装层发布进程组 ID 并由适配器完成验证之前，`pid` 为 `-1`；stdin 和常规观察会等待该发布。自有启动信号会在分配前中止环境和私有状态准备；分配开始后，取消会等待可清理的临时 SDK 句柄。 -- **执行世界坐标**：`cwd` 和私有 `runtimeRoot` 来自共享所有者；可执行文件查找会验证绝对路径，或根据沙箱 PATH 加显式覆盖来解析裸名称。 +- **执行世界坐标**：`cwd` 和私有 `runtimeRoot` 来自共享所有者；可执行文件查找会验证绝对路径，或根据沙箱 PATH 加显式覆盖来解析裸名称，并与所有 subprocess 提供方一致地拒绝含分隔符的相对路径。 - **Linux 进程组**：带引号保护的包装层会在 `exec setsid --wait` 下启动每组 argv，并在 `ctx.e2b.runtimeRoot/processes` 下记录实际进程组 ID 和私有状态文件。句柄会等待该文件，而不会把 SDK 命令 PID 当作已发布的身份。终止操作以记录的负数 ID 发送 `SIGTERM`，等待调用方的 `graceMs`，再升级到 `SIGKILL` 和 SDK kill 回退；TERM 信号发送或探测失败也会强制触发该升级。进程表探测会把仅含僵尸或已死亡条目的进程组视为完全停稳。强制清理只有在有界探测发现进程组为空后才算成功；否则 `waitForExit()` 会公开可重试的失败，而已证明的完全停稳会让后续终止操作不再执行任何动作。发布失败与监控失败都会在拒绝前执行同一清理事务。服务 dispose（资源释放）会拒绝新的启动请求、终止并等待每个保留进程组退出，再等待 SDK 结算和私有清理完成，之后沙箱所有者才会释放。 - **环境边界**：一次受信任的控制 shell 探测会从 passwd 条目解析沙箱用户的登录主目录，以 base64 ASCII 传输沙箱环境，再进行一次严格 UTF-8 解码；随后包装层移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），并把每个有效的 `spec.env` 条目恢复为调用方显式选择。空名称、`=` 和违反 NUL 分帧规则的条目会在启动前被拒绝。在用户 profile 脚本运行前，此后的 E2B 命令 shell 与 PTY 登录 shell 会获得位于根目录下、全新随机生成的 `HOME`，并为每个被清理的环境变量名设置空值覆盖；之后，请求的 argv 会在不改变沙箱用户 umask 的前提下接收序列化环境。宿主环境变量绝不会隐式进入沙箱。私有环境文件在使用后会被删除；命令或终端设置失败时，会先删除其私有状态再拒绝。 - **stdio 投影**：远程包装层先把原始字节分流到可选的有界 spill 文件，再把每个实时分片编码为换行分隔的 base64 ASCII 帧；宿主会跨任意 SDK 回调边界增量恢复字节。pipe 模式把这些字节写入宿主 Node 流；inherit 模式把字节写入 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。包装层会在等待继承管道的写入方之前发布直接命令状态。对于 collect 或 inherit 输出，超过 `graceMs` 后，适配器会断开未完成的 SDK 流，不公开其中不完整的 spill，并返回该状态，同时保留远程进程组供 `waitForExit()` 和终止操作使用。原始 pipe 自然完成时，会等待无损传输完成并保留背压；显式终止则会销毁宿主 pipe，并在远程清理前释放受阻的输出写入。批量 stdin 和流式 stdin 都使用 SDK 句柄。 -- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并通过一项可重试且须等待的 `terminate()` 清理远程终端会话中仍存活的每个进程组；终止会拒绝新的句柄操作，中止并等待在途写入、检查和信号操作结算，并把仅含僵尸进程的进程组视为已经完全停稳。私有随机输出边界会丢弃 E2B 引导 shell 的提示符和回显的 runner 命令，同时保留请求进程的每个字节，包括其第一个提示符。PTY 分配会一直等待到句柄发布后才观察取消，以便由承担清理责任的回滚清理已发布句柄。setup 与 teardown 负责私有状态事务，在服务 dispose 期间中止待处理的 setup 并阻止发布；若 setup 回滚也失败，则由沙箱 dispose 或超时约束其存活时间。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 +- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并通过一项可重试且须等待的 `terminate()` 清理远程终端会话中仍存活的每个进程组；终止会拒绝新的句柄操作，中止并等待在途写入、检查和信号操作结算，并把仅含僵尸进程的进程组视为已经完全停稳。私有随机输出边界会丢弃 E2B 引导 shell 的提示符和回显的 runner 命令，同时保留请求进程的每个字节，包括其第一个提示符。终端输出推入句柄流时不等待宿主背压：流动的消费方（PTY 后端在构造时就挂上一个）把字节折叠进自身的有界状态，而暂停的消费方会在宿主内存中缓冲。PTY 分配会一直等待到句柄发布后才观察取消，以便由承担清理责任的回滚清理已发布句柄。setup 与 teardown 负责私有状态事务，在服务 dispose 期间中止待处理的 setup 并阻止发布；若 setup 回滚也失败，则由沙箱 dispose 或超时约束其存活时间。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 - **沙箱消失**：在进程或终端的存活探测、终止、回滚或断开连接期间出现 `SandboxNotFoundError`，证明远程执行环境无法保留工作，因此清理会将其视为完全停稳；其他故障仍可观察。 E2B 默认基础镜像提供该适配器调用的运行时和 Bash/GNU 工具：`node`、`bash`、`setsid`、`ps`、`awk`、`tr`、`env`、`base64`、`chmod`、`tee`、`head`、`rm`、`kill`、`id` 和 `getent`。 @@ -29,6 +35,7 @@ E2B 默认基础镜像提供该适配器调用的运行时和 Bash/GNU 工具： - **SDK 仍会在宿主内存中保留完整命令输出**：即使本适配器公开的是有界原始字节尾部，E2B `CommandHandle.stdout` 和 `.stderr` 仍会累积 base64 传输内容，因此无法达到进程管理 seam 通常提供的宿主内存边界，而且传输保留量大于源数据流。 - **不支持需要同步 PID 的消费方**：远程启动期间，`pid` 保持为 `-1`；包括 ACP 子进程后端在内，要求立即获得正 PID 的消费方无法原样使用本提供方。 - **私有状态随沙箱生命周期存在**：进程目录和有效的 spill 文件会留在 `.dsh-e2b` 下，直到所有者删除沙箱；本 POC 不提供沙箱内清理。 +- **控制状态与沙箱用户同 UID**：E2B 以同一默认用户运行每条命令，因此 `0700`/`0600` 权限无法把 `.dsh-e2b` 控制文件与并发运行的沙箱进程隔离开。后台进程可以改写 `pid`/`exit-code`，或读取尚未被消费的 `environment` 文件。适配器会验证已发布的值，并拒绝取负后不安全的进程组 ID（`<= 1`），但真正的隔离需要 E2B 提供按命令用户或带外控制通道。 - **数值进程身份没有复用围栏**：E2B 公开基于数值 PID/PGID 的 PTY 输入、信号发送和清理操作，却没有与身份原子绑定的替代方案。适配器会尽量减少宿主往返，真实环境测试会覆盖可复现的陈旧中断重叠；在 E2B 新增身份原语，或实际故障证明需要更窄的协议之前，替代方案会继续延后。 - **初始环境探测会继承沙箱默认值**：E2B 会把命令覆盖与默认环境条目合并，因此探测无法在枚举未知且形似凭据的名称之前将它们置空。一个已在沙箱内运行的同 UID 不可信进程可以检查该短时存在的控制 shell；因此，该 POC 不支持把 secret 放入沙箱默认环境变量，需要 E2B 的替换环境原语才能弥合该缺口。 - **E2B 不公开信号事实**：适配器请求的 `SIGTERM` 或 `SIGKILL` 只有在包装层发布的直接退出码没有胜出时才报告为信号；其他未请求的 SDK 退出始终保留为退出码，包括形似 `128 + signal` 的值。 diff --git a/packages/e2b/subprocess-e2b/package.json b/packages/e2b/subprocess-e2b/package.json index d73607d1f3..fa96e17293 100644 --- a/packages/e2b/subprocess-e2b/package.json +++ b/packages/e2b/subprocess-e2b/package.json @@ -32,6 +32,9 @@ "@deepseek-ai/dsh-subprocess": "^0.0.1", "cordis": "^4.0.0-rc.7" }, + "dependencies": { + "schemastery": "^3.18.0" + }, "devDependencies": { "@deepseek-ai/dsh-e2b": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", diff --git a/packages/e2b/subprocess-e2b/src/index.ts b/packages/e2b/subprocess-e2b/src/index.ts index 15e7feccdd..38b1d32465 100644 --- a/packages/e2b/subprocess-e2b/src/index.ts +++ b/packages/e2b/subprocess-e2b/src/index.ts @@ -7,6 +7,7 @@ import { randomUUID } from 'node:crypto' import { posix } from 'node:path' import { Context } from 'cordis' +import z from 'schemastery' import { SubprocessService } from '@deepseek-ai/dsh-subprocess' import type { SubprocessHandle, @@ -16,30 +17,53 @@ import type { } from '@deepseek-ai/dsh-subprocess' import { e2bControlEnvs, quoteE2BShellArg } from '@deepseek-ai/dsh-e2b' import { E2BSubprocessHandle } from './process.ts' +import { asError, signalOpts } from './remote.ts' import { spawnE2BTerminal } from './terminal.ts' -function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } { - return signal === undefined ? {} : { signal } +/** Configuration for the E2B subprocess adapter. */ +export interface Config { + /** Remote status/liveness poll cadence in milliseconds; each tick is one control-plane request. */ + pollMs?: number +} + +interface SchemaResolvedConfig extends Config { + pollMs: number +} + +interface TerminalSetup { + done: Promise + controller: AbortController } /** E2B command manager registered as `ctx.subprocess`. */ export class E2BSubprocessService extends SubprocessService { static inject = ['e2b'] + static Config: z = z.object({ + pollMs: z.number().default(20), + }) + private readonly live = new Set() private readonly terminals = new Set() - private readonly terminalSetups = new Map, AbortController>() + private readonly terminalSetups = new Set() + private readonly pollMs: number private disposing = false /** Create the E2B subprocess service and bind its disposal policy. */ - constructor(ctx: Context) { + constructor(ctx: Context, config: Config) { super(ctx) + // Schemastery fills pollMs before construction; the type does not encode that step. + const { pollMs } = config as SchemaResolvedConfig + if (!Number.isSafeInteger(pollMs) || pollMs <= 0) { + throw new Error('subprocess-e2b: pollMs must be a positive safe integer') + } + this.pollMs = pollMs ctx.effect(() => async () => { this.disposing = true - for (const controller of this.terminalSetups.values()) { - controller.abort(new Error('subprocess-e2b: service disposed during terminal setup')) + for (const setup of this.terminalSetups) { + setup.controller.abort(new Error('subprocess-e2b: service disposed during terminal setup')) } - await Promise.all([...this.terminalSetups.keys()]) + await Promise.all([...this.terminalSetups].map(setup => setup.done)) const handles = [...this.live] const terminals = [...this.terminals] const pending: Promise[] = [] @@ -54,9 +78,11 @@ export class E2BSubprocessService extends SubprocessService { pending.push(terminal.terminate().then(() => { this.terminals.delete(terminal) })) } const outcomes = await Promise.allSettled(pending) - for (const outcome of outcomes) { - if (outcome.status === 'rejected') throw outcome.reason - } + const failures = outcomes.flatMap(outcome => outcome.status === 'rejected' + ? [outcome.reason as unknown] + : []) + if (failures.length === 1) throw asError(failures[0]) + if (failures.length > 1) throw new AggregateError(failures, 'subprocess-e2b: teardown failed') }, 'e2b subprocess teardown') } @@ -77,6 +103,11 @@ export class E2BSubprocessService extends SubprocessService { signal?.throwIfAborted() return command } + if (command.includes('/')) { + throw new Error( + `subprocess-e2b: command ${JSON.stringify(command)} is a relative path; use an absolute path or a bare PATH name`, + ) + } const path = env?.PATH const prefix = path === undefined ? '' : `PATH=${quoteE2BShellArg(path)} ` const result = await sandbox.commands.run( @@ -88,6 +119,7 @@ export class E2BSubprocessService extends SubprocessService { if (executable.includes('\n') || (!posix.isAbsolute(executable) && !executable.includes('/'))) { throw new Error(`subprocess-e2b: executable ${JSON.stringify(command)} did not resolve to one absolute path`) } + // A relative result comes from a relative PATH entry; the lookup ran with the shared cwd. return posix.resolve(this.ctx.e2b.cwd, executable) } @@ -98,14 +130,11 @@ export class E2BSubprocessService extends SubprocessService { 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)}`) } const stateDir = posix.join(this.ctx.e2b.runtimeRoot, 'processes', randomUUID()) - const handle = new E2BSubprocessHandle(this.ctx.e2b, spec, stateDir) + const handle = new E2BSubprocessHandle(this.ctx.e2b, spec, stateDir, this.pollMs) this.live.add(handle) const release = async (): Promise => { await handle.waitForExit() @@ -124,24 +153,20 @@ export class E2BSubprocessService extends SubprocessService { if (program === undefined || program.length === 0) { throw new Error('subprocess-e2b: terminal argv must contain a program') } - for (const [name, value] of [['rows', spec.rows], ['cols', spec.cols], ['graceMs', spec.graceMs]] as const) { - if (!Number.isSafeInteger(value) || value <= 0) { - throw new Error(`subprocess-e2b: terminal ${name} must be a positive safe integer`) - } - } spec.signal?.throwIfAborted() const stateDir = posix.join(this.ctx.e2b.runtimeRoot, 'terminals', randomUUID()) - const setup = Promise.withResolvers() - const setupController = new AbortController() + const done = Promise.withResolvers() + const setup: TerminalSetup = { done: done.promise, controller: new AbortController() } const setupSignal = spec.signal === undefined - ? setupController.signal - : AbortSignal.any([spec.signal, setupController.signal]) - this.terminalSetups.set(setup.promise, setupController) + ? setup.controller.signal + : AbortSignal.any([spec.signal, setup.controller.signal]) + this.terminalSetups.add(setup) try { const terminal = await spawnE2BTerminal( this.ctx.e2b, { ...spec, signal: setupSignal }, stateDir, + this.pollMs, ) this.terminals.add(terminal) // oxlint-disable-next-line typescript/no-unnecessary-condition -- Remote allocation yields to disposal. @@ -159,8 +184,8 @@ export class E2BSubprocessService extends SubprocessService { }) return terminal } finally { - this.terminalSetups.delete(setup.promise) - setup.resolve() + this.terminalSetups.delete(setup) + done.resolve() } } } diff --git a/packages/e2b/subprocess-e2b/src/process.ts b/packages/e2b/subprocess-e2b/src/process.ts index c0aceda334..feb21a70bc 100644 --- a/packages/e2b/subprocess-e2b/src/process.ts +++ b/packages/e2b/subprocess-e2b/src/process.ts @@ -21,8 +21,8 @@ import type { import type E2BSandboxService from '@deepseek-ai/dsh-e2b' import { bootstrapEnvironment, readRemoteEnvironment, serializeRemoteEnvironment } from './environment.ts' import { E2BBase64Decoder, E2B_OUTPUT_COMPLETE_FRAME, E2BOutputReader } from './output.ts' +import { asError, commandOpts, signalRemoteGroups, waitTick } from './remote.ts' -const GROUP_POLL_MS = 20 const OUTPUT_ENCODER_SOURCE = [ '(async () => {', ' for await (const chunk of process.stdin) {', @@ -48,10 +48,6 @@ function isValidProcessId(value: number): boolean { return Number.isSafeInteger(value) && value > 0 } -function asError(error: unknown): Error { - return error instanceof Error ? error : new Error(String(error)) -} - class DeferredStdin extends Writable { constructor(private readonly ready: Promise) { super({ decodeStrings: false }) @@ -142,28 +138,6 @@ function commandText(spec: SubprocessSpawnSpec, paths: RemotePaths): string { return bootstrap } -function commandOpts( - envs: Record, - signal: AbortSignal | undefined, -): { envs: Record; signal?: AbortSignal } { - return { envs: e2bControlEnvs(envs), ...(signal === undefined ? {} : { signal }) } -} - -function waitTick(signal?: AbortSignal): Promise { - if (signal?.aborted === true) return Promise.resolve(false) - return new Promise((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 }) - }) -} - const WAIT_ABORTED = Symbol('wait aborted') function waitWithSignal(promise: Promise, signal: AbortSignal | undefined): Promise { @@ -194,6 +168,8 @@ export class E2BSubprocessHandle implements SubprocessHandle { private readonly stdoutDecoder = new E2BBase64Decoder() private readonly stderrDecoder = new E2BBase64Decoder() private readonly terminationController = new AbortController() + /** Releases output waits that survive the command outcome, so blocked SDK callbacks settle. */ + private readonly outputReleased = new AbortController() private readonly stdoutReader: E2BOutputReader | undefined private readonly stderrReader: E2BOutputReader | undefined private readonly paths: RemotePaths @@ -212,11 +188,13 @@ export class E2BSubprocessHandle implements SubprocessHandle { * @param runtime - Shared E2B sandbox owner. * @param spec - Fully resolved subprocess request. * @param stateDir - Remote directory retaining process identity, status, and valid spills. + * @param pollMs - Remote status/liveness poll cadence. */ constructor( private readonly runtime: E2BSandboxService, private readonly spec: SubprocessSpawnSpec, readonly stateDir: string, + private readonly pollMs = 20, ) { this.paths = { pid: posix.join(stateDir, 'pid'), @@ -318,7 +296,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { const processGroupId = this.remotePid > 0 ? this.remotePid : handle.pid while (await this.groupAlive(sandbox, processGroupId, signal)) { this.throwTerminationFailure() - if (!await waitTick(signal)) return false + if (!await waitTick(this.pollMs, signal)) return false } this.throwTerminationFailure() if (signal?.aborted === true) return false @@ -483,19 +461,21 @@ export class E2BSubprocessHandle implements SubprocessHandle { await new Promise((resolve, reject) => { const onDrain = (): void => { cleanup(); resolve() } const onClose = (): void => { cleanup(); resolve() } - const onTermination = (): void => { cleanup(); resolve() } + const onRelease = (): void => { cleanup(); resolve() } const onError = (error: Error): void => { cleanup(); reject(error) } const cleanup = (): void => { target.removeListener('drain', onDrain) target.removeListener('close', onClose) target.removeListener('error', onError) - this.terminationController.signal.removeEventListener('abort', onTermination) + this.terminationController.signal.removeEventListener('abort', onRelease) + this.outputReleased.signal.removeEventListener('abort', onRelease) } target.once('drain', onDrain) target.once('close', onClose) target.once('error', onError) - this.terminationController.signal.addEventListener('abort', onTermination, { once: true }) - if (this.terminationController.signal.aborted) onTermination() + this.terminationController.signal.addEventListener('abort', onRelease, { once: true }) + this.outputReleased.signal.addEventListener('abort', onRelease, { once: true }) + if (this.terminationController.signal.aborted || this.outputReleased.signal.aborted) onRelease() }) } @@ -514,9 +494,14 @@ export class E2BSubprocessHandle implements SubprocessHandle { 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)}`) } + // A same-UID sandbox process can rewrite this file; refuse ids whose + // negative form addresses every process (`kill -- -1`) or init's group. + if (pid <= 1) { + throw new Error(`subprocess-e2b: unsafe published process-group id ${pid}`) + } return pid } - const settled = await Promise.race([commandSettled, waitTick().then(() => false)]) + const settled = await Promise.race([commandSettled, waitTick(this.pollMs).then(() => false)]) if (settled) throw new Error('subprocess-e2b: remote command exited before publishing its process-group id') } } @@ -545,13 +530,16 @@ export class E2BSubprocessHandle implements SubprocessHandle { this.outputDrainExpired = true this.stdoutReader?.invalidateSpill() this.stderrReader?.invalidateSpill() + // Release inherited-output waits so a callback blocked on host + // backpressure cannot keep the disconnected SDK settlement pending. + this.outputReleased.abort(new Error('subprocess-e2b: output drain grace expired')) await handle.disconnect() return { exitCode, signal: null } } if (completed !== undefined) return this.commandOutcome(completed) // TODO(e2b-status-watch): Replace collect/inherit control-plane polling // when E2B can observe direct-command exit independently of descendant-held output. - completed = await Promise.race([settlement, waitTick().then(() => undefined)]) + completed = await Promise.race([settlement, waitTick(this.pollMs).then(() => undefined)]) } } @@ -622,7 +610,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { private async terminateGroup(sandbox: Sandbox, handle: CommandHandle, processGroupId: number): Promise { this.terminationSignal = 'SIGTERM' try { - await this.signalGroup(sandbox, processGroupId, 'TERM') + await signalRemoteGroups(sandbox, this.controlEnvs, [processGroupId], 'TERM') if (await this.waitForGroupExit(sandbox, processGroupId)) { this.markQuiescent() return @@ -637,7 +625,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { private async forceKillGroup(sandbox: Sandbox, handle: CommandHandle, processGroupId: number): Promise { try { - await this.signalGroup(sandbox, processGroupId, 'KILL') + await signalRemoteGroups(sandbox, this.controlEnvs, [processGroupId], 'KILL') } catch (_processGroupKillFailure) { // SDK kill and the final liveness probe remain independent cleanup paths. } @@ -654,7 +642,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { const deadline = Date.now() + this.spec.graceMs while (await this.groupAlive(sandbox, processGroupId)) { if (Date.now() >= deadline) return false - await waitTick() + await waitTick(this.pollMs) } return true } @@ -663,21 +651,6 @@ export class E2BSubprocessHandle implements SubprocessHandle { if (this.terminationFailure !== undefined) throw this.terminationFailure } - private async signalGroup(sandbox: Sandbox, pid: number, signal: 'TERM' | 'KILL'): Promise { - // TODO(e2b-pgid-identity): Prefer an atomic identity-bound group signal if E2B adds one; - // a userspace identity precheck cannot close the numeric-PGID reuse race. - try { - await sandbox.commands.run( - `kill -${signal} -- -${pid}`, - commandOpts(this.controlEnvs, undefined), - ) - return true - } catch (error: unknown) { - if (error instanceof CommandExitError || error instanceof SandboxNotFoundError) return false - throw error - } - } - private async groupAlive(sandbox: Sandbox, pid: number, signal?: AbortSignal): Promise { const result = await sandbox.commands.run( `set -o pipefail; ps -eo pgid=,stat= | awk '$1 == ${pid} && $2 !~ /^[ZXx]/ { live=1 } END { if (live) print "live" }'`, diff --git a/packages/e2b/subprocess-e2b/src/remote.ts b/packages/e2b/subprocess-e2b/src/remote.ts new file mode 100644 index 0000000000..1937f54bf0 --- /dev/null +++ b/packages/e2b/subprocess-e2b/src/remote.ts @@ -0,0 +1,97 @@ +/** + * Shared remote-control helpers for the E2B subprocess adapter: SDK option + * shaping, poll ticks, and the one tolerant process-group signal used by both + * the ordinary-process and terminal teardown ladders. + */ + +import { CommandExitError, e2bControlEnvs, SandboxNotFoundError } from '@deepseek-ai/dsh-e2b' +import type { Sandbox } from '@deepseek-ai/dsh-e2b' + +/** + * Normalize an unknown rejection into an Error. + * @param error - Any thrown or rejected value. + * @returns The value itself when already an Error, else a stringified wrapper. + */ +export function asError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)) +} + +/** + * Shape the optional-signal SDK options object. + * @param signal - Optional cancellation for one SDK request. + * @returns An options fragment that omits an undefined signal. + */ +export function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } { + return signal === undefined ? {} : { signal } +} + +/** + * Shape control-shell command options with the isolated HOME override. + * @param envs - Explicit environment entries for the control command. + * @param signal - Optional cancellation for the SDK request. + * @returns Options for `sandbox.commands.run` control invocations. + */ +export function commandOpts( + envs: Record, + signal?: AbortSignal, +): { envs: Record; signal?: AbortSignal } { + return { envs: e2bControlEnvs(envs), ...signalOpts(signal) } +} + +/** + * Resolve after one duration. + * @param ms - Milliseconds to wait. + * @returns Settles after the timeout. + */ +export function delay(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)) +} + +/** + * Wait one poll interval or until the signal aborts. + * @param pollMs - Poll cadence in milliseconds. + * @param signal - Optional abort that ends the wait early. + * @returns `true` after a full tick, `false` when aborted first. + */ +export function waitTick(pollMs: number, signal?: AbortSignal): Promise { + if (signal?.aborted === true) return Promise.resolve(false) + return new Promise((resolve) => { + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort) + resolve(true) + }, pollMs) + const onAbort = (): void => { + clearTimeout(timer) + resolve(false) + } + signal?.addEventListener('abort', onAbort, { once: true }) + }) +} + +/** + * Signal remote process groups, tolerating the shared teardown outcomes: a + * nonzero `kill` (groups already gone) and a disappeared sandbox. Both the + * pgid-keyed process ladder and the sid-keyed terminal ladder deliver signals + * through this single tolerance so they cannot drift apart. + * @param sandbox - Live SDK handle. + * @param envs - Control-shell environment entries. + * @param groups - Positive process-group ids to signal. + * @param signal - `TERM` or `KILL`. + */ +export async function signalRemoteGroups( + sandbox: Sandbox, + envs: Record, + groups: readonly number[], + signal: 'TERM' | 'KILL', +): Promise { + // TODO(e2b-pgid-identity): Prefer an atomic identity-bound group signal if E2B adds one; + // a userspace identity precheck cannot close the numeric-PGID reuse race. + try { + await sandbox.commands.run( + `kill -${signal} -- ${groups.map(group => `-${group}`).join(' ')}`, + commandOpts(envs), + ) + } catch (error: unknown) { + if (!(error instanceof CommandExitError) && !(error instanceof SandboxNotFoundError)) throw error + } +} diff --git a/packages/e2b/subprocess-e2b/src/terminal.ts b/packages/e2b/subprocess-e2b/src/terminal.ts index 3108a4a60a..fd10ecdd8e 100644 --- a/packages/e2b/subprocess-e2b/src/terminal.ts +++ b/packages/e2b/subprocess-e2b/src/terminal.ts @@ -25,8 +25,7 @@ import { readRemoteEnvironment, serializeRemoteEnvironment, } from './environment.ts' - -const POLL_MS = 20 +import { asError, commandOpts, delay, signalOpts, signalRemoteGroups } from './remote.ts' const TERMINAL_RUNNER_SOURCE = [ '#!/bin/bash', @@ -52,21 +51,6 @@ interface TerminalPaths { outputMarker: string } -function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } { - return signal === undefined ? {} : { signal } -} - -function commandOpts( - envs: Record, - signal?: AbortSignal, -): { envs: Record; signal?: AbortSignal } { - return { envs: e2bControlEnvs(envs), ...signalOpts(signal) } -} - -function delay(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)) -} - class BootstrapOutputFilter { readonly ready: Promise @@ -134,10 +118,6 @@ async function waitForBootstrapOutput( }) } -function asError(error: unknown): Error { - return error instanceof Error ? error : new Error(String(error)) -} - function parsePositiveId(value: string, message: string): number { const raw = value.trim() const id = Number(raw) @@ -193,27 +173,12 @@ async function sessionProcessGroups( return [...groups] } -async function signalGroups( - sandbox: Sandbox, - groups: number[], - signal: 'TERM' | 'KILL', - envs: Record, -): Promise { - try { - await sandbox.commands.run( - `kill -${signal} -- ${groups.map(group => `-${group}`).join(' ')}`, - commandOpts(envs), - ) - } catch (error: unknown) { - if (!(error instanceof CommandExitError) && !(error instanceof SandboxNotFoundError)) throw error - } -} - async function awaitSessionEmpty( sandbox: Sandbox, sessionId: number, envs: Record, graceMs: number, + pollMs: number, kill = false, ): Promise { const deadline = Date.now() + graceMs @@ -221,12 +186,12 @@ async function awaitSessionEmpty( const groups = await sessionProcessGroups(sandbox, sessionId, envs) if (groups.length === 0) return groups if (kill) { - await signalGroups(sandbox, groups, 'KILL', envs) + await signalRemoteGroups(sandbox, envs, groups, 'KILL') if (Date.now() >= deadline) return await sessionProcessGroups(sandbox, sessionId, envs) } else if (Date.now() >= deadline) { return groups } - await delay(Math.min(POLL_MS, Math.max(1, deadline - Date.now()))) + await delay(Math.min(pollMs, Math.max(1, deadline - Date.now()))) } } @@ -236,6 +201,7 @@ async function rollbackUnpublishedTerminal( completion: Promise, envs: Record, graceMs: number, + pollMs: number, ): Promise { let topLevelExited = false void completion.then( @@ -256,11 +222,11 @@ async function rollbackUnpublishedTerminal( try { let groups = await sessionProcessGroups(sandbox, sessionId, envs) if (groups.length > 0) { - await signalGroups(sandbox, groups, 'TERM', envs) - groups = await awaitSessionEmpty(sandbox, sessionId, envs, graceMs) + await signalRemoteGroups(sandbox, envs, groups, 'TERM') + groups = await awaitSessionEmpty(sandbox, sessionId, envs, graceMs, pollMs) } if (groups.length > 0) { - await awaitSessionEmpty(sandbox, sessionId, envs, graceMs, true) + await awaitSessionEmpty(sandbox, sessionId, envs, graceMs, pollMs, true) } } catch (error: unknown) { attemptFailures.push(asError(error)) @@ -280,7 +246,7 @@ async function rollbackUnpublishedTerminal( const proofFailures: Error[] = [] if (sessionId !== undefined) { try { - const groups = await awaitSessionEmpty(sandbox, sessionId, envs, graceMs, true) + const groups = await awaitSessionEmpty(sandbox, sessionId, envs, graceMs, pollMs, true) if (groups.length > 0) { proofFailures.push(new Error( `subprocess-e2b: terminal setup rollback failed; surviving process groups: ${groups.join(', ')}`, @@ -328,6 +294,7 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle { private readonly controlEnvs: Record, private readonly stateDir: string, private readonly graceMs: number, + private readonly pollMs: number, ) { this.pid = handle.pid this.done = this.waitForCommand() @@ -441,8 +408,8 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle { let groups = await sessionProcessGroups(this.sandbox, this.sessionId, this.controlEnvs) if (groups.length > 0) { this.terminationSignal = 'SIGTERM' - await signalGroups(this.sandbox, groups, 'TERM', this.controlEnvs) - groups = await awaitSessionEmpty(this.sandbox, this.sessionId, this.controlEnvs, this.graceMs) + await signalRemoteGroups(this.sandbox, this.controlEnvs, groups, 'TERM') + groups = await awaitSessionEmpty(this.sandbox, this.sessionId, this.controlEnvs, this.graceMs, this.pollMs) } if (groups.length === 0 && !this.topLevelExited) { await Promise.race([this.done.catch(() => undefined), delay(this.graceMs)]) @@ -457,7 +424,7 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle { throw error } } - groups = await awaitSessionEmpty(this.sandbox, this.sessionId, this.controlEnvs, this.graceMs, true) + groups = await awaitSessionEmpty(this.sandbox, this.sessionId, this.controlEnvs, this.graceMs, this.pollMs, true) if (!this.topLevelExited) await Promise.race([this.done.catch(() => undefined), delay(this.graceMs)]) } if (groups.length > 0) { @@ -485,12 +452,14 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle { * @param runtime - Shared E2B sandbox owner. * @param spec - Fully specified terminal-process request. * @param stateDir - Private remote directory for one startup transaction. + * @param pollMs - Remote session liveness poll cadence. * @returns The live subprocess terminal handle. */ export async function spawnE2BTerminal( runtime: E2BSandboxService, spec: SubprocessTerminalSpawnSpec, stateDir: string, + pollMs = 20, ): Promise { const sandbox = await runtime.getSandbox() spec.signal?.throwIfAborted() @@ -555,6 +524,7 @@ export async function spawnE2BTerminal( controlEnvs, stateDir, spec.graceMs, + pollMs, ) } catch (error: unknown) { output.destroy() @@ -565,7 +535,7 @@ export async function spawnE2BTerminal( if (!terminalQuiescent && handle !== undefined) { try { if (completion === undefined) await handle.kill() - else await rollbackUnpublishedTerminal(sandbox, handle, completion, controlEnvs, spec.graceMs) + else await rollbackUnpublishedTerminal(sandbox, handle, completion, controlEnvs, spec.graceMs, pollMs) terminalQuiescent = true } catch (cleanupError: unknown) { if (cleanupError instanceof SandboxNotFoundError) terminalQuiescent = true diff --git a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts index 471716ef1a..747dc936c6 100644 --- a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts @@ -519,6 +519,38 @@ describe('E2BSubprocessHandle', () => { await expect(handle.waitForExit()).resolves.toBe(true) }) + it('releases an inherited-output callback blocked on host backpressure at drain expiry', async () => { + const fake = new FakeSandbox() + const written: string[] = [] + const stdoutWrite = vi.spyOn(process.stdout, 'write').mockImplementation(((chunk: Uint8Array) => { + written.push(Buffer.from(chunk).toString()) + return false + }) as typeof process.stdout.write) + try { + const handle = new E2BSubprocessHandle(runtime(fake), spec({ + graceMs: 5, + stdio: { stdin: 'ignore', stdout: 'inherit', stderr: { maxBytes: 4 } }, + }), '/runtime/inherit-backpressure') + await flush() + let callbackSettled = false + const blocked = fake.stdout('blocked bytes').then(() => { callbackSettled = true }) + await flush() + expect(callbackSettled).toBe(false) + fake.exitStatus = '0\n' + + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + await blocked + expect(callbackSettled).toBe(true) + expect(written.join('')).toBe('blocked bytes') + expect(fake.handle.disconnects).toBe(1) + + handle.terminate() + await expect(handle.waitForExit()).resolves.toBe(true) + } finally { + stdoutWrite.mockRestore() + } + }) + it('waits for lossless raw-pipe output after the direct status is published', async () => { const fake = new FakeSandbox() const handle = new E2BSubprocessHandle(runtime(fake), spec({ @@ -1311,6 +1343,17 @@ describe('E2BSubprocessHandle', () => { expect(invalidGroup.commandsSeen).toContain('kill -KILL -- -4242') await expect(invalid.waitForExit()).resolves.toBe(true) + // A rewritten pid file must not aim the kill at every process (`-- -1`). + const unsafeGroup = new FakeSandbox() + unsafeGroup.processGroupId = '1\n' + unsafeGroup.delaysKill = true + unsafeGroup.sdkKillStops = false + unsafeGroup.afterProbe = () => { unsafeGroup.alive = false } + const unsafe = new E2BSubprocessHandle(runtime(unsafeGroup), spec(), '/runtime/unsafe-group') + await expect(unsafe.done).rejects.toThrow(/unsafe published process-group id 1/) + expect(unsafeGroup.commandsSeen).not.toContain('kill -KILL -- -1') + await expect(unsafe.waitForExit()).resolves.toBe(true) + const absentGroup = new FakeSandbox() absentGroup.processGroupId = '' const absent = new E2BSubprocessHandle(runtime(absentGroup), spec(), '/runtime/absent-group') @@ -1588,6 +1631,34 @@ describe('E2BSubprocessService', () => { await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' }) }) + it('aggregates sibling cleanup failures instead of reporting only the first', async () => { + const { ctx, fiber } = await service() + const disposalErrors: unknown[] = [] + ctx.logger.error = ((error: unknown) => { disposalErrors.push(error) }) as typeof ctx.logger.error + const first = { + terminate: vi.fn(), + waitForExit: vi.fn(async () => { throw new Error('first cleanup failed') }), + done: Promise.resolve({ exitCode: 0, signal: null }), + } as unknown as E2BSubprocessHandle + const second = { + terminate: vi.fn(), + waitForExit: vi.fn(async () => { throw new Error('second cleanup failed') }), + done: Promise.resolve({ exitCode: 0, signal: null }), + } as unknown as E2BSubprocessHandle + const live = (ctx.subprocess as unknown as { live: Set }).live + live.add(first) + live.add(second) + + await fiber.dispose() + const failure = disposalErrors[0] + expect(failure).toBeInstanceOf(AggregateError) + if (!(failure instanceof AggregateError)) throw new Error('expected AggregateError') + expect(failure.errors.map(error => (error as Error).message).sort()).toEqual([ + 'first cleanup failed', + 'second cleanup failed', + ]) + }) + it('waits for every owned cleanup before reporting a disposal failure', async () => { const { ctx, fiber } = await service() const failed = { @@ -1667,7 +1738,6 @@ describe('E2BSubprocessService', () => { 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/) }) diff --git a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts index 9973b5151a..826732f057 100644 --- a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts @@ -793,6 +793,8 @@ describe('E2B subprocess terminal service', () => { it('rejects invalid executable lookup inputs and results', async () => { const { ctx, fake } = await service() await expect(ctx.subprocess.resolveExecutable('')).rejects.toThrow('non-empty') + await expect(ctx.subprocess.resolveExecutable('./bin/server')).rejects.toThrow('is a relative path') + await expect(ctx.subprocess.resolveExecutable('node_modules/.bin/server')).rejects.toThrow('is a relative path') await expect(ctx.subprocess.resolveExecutable('node', undefined, AbortSignal.abort(new Error('stop')))) .rejects.toThrow('stop') fake.resolvedExecutable = 'node\n' @@ -801,6 +803,15 @@ describe('E2B subprocess terminal service', () => { await expect(ctx.subprocess.resolveExecutable('node')).rejects.toThrow('did not resolve') }) + it('rejects a non-positive poll cadence at load', async () => { + const ctx = new Context() + ctx.provide('e2b', runtime(new FakeTerminalSandbox())) + await expect(ctx.plugin(E2BSubprocessService, { pollMs: 0 })) + .rejects.toThrow('pollMs must be a positive safe integer') + const explicit = await ctx.plugin(E2BSubprocessService, { pollMs: 5 }) + await explicit.dispose() + }) + it('owns live terminals through service disposal', async () => { const { ctx, fiber, fake } = await service() const terminal = await ctx.subprocess.spawnTerminal(spec({ signal: new AbortController().signal })) @@ -873,9 +884,6 @@ describe('E2B subprocess terminal service', () => { const { ctx, fiber, fake } = await service() for (const request of [ spec({ argv: [] }), - spec({ rows: 0 }), - spec({ cols: 1.5 }), - spec({ graceMs: 0 }), spec({ signal: AbortSignal.abort(new Error('cancelled')) }), ]) { await expect(ctx.subprocess.spawnTerminal(request)).rejects.toThrow()