From 219999cd2f3e0abf3878d354de3bf42cfc2e6134 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:19:46 +0800 Subject: [PATCH] fix(e2b): close adapter lifecycle review gaps --- packages/e2b/fs-e2b/README.i18n.yaml | 4 +- packages/e2b/fs-e2b/README.md | 2 +- packages/e2b/fs-e2b/README.zh.md | 2 +- packages/e2b/fs-e2b/src/index.ts | 24 ++- packages/e2b/fs-e2b/tests/filesystem.spec.ts | 33 +++- packages/e2b/subprocess-e2b/README.i18n.yaml | 4 +- packages/e2b/subprocess-e2b/README.md | 10 +- packages/e2b/subprocess-e2b/README.zh.md | 10 +- .../e2b/subprocess-e2b/src/environment.ts | 20 +++ packages/e2b/subprocess-e2b/src/index.ts | 25 ++- packages/e2b/subprocess-e2b/src/process.ts | 67 +++++--- packages/e2b/subprocess-e2b/src/terminal.ts | 61 ++++---- .../subprocess-e2b/tests/subprocess.spec.ts | 146 +++++++++++++++++- .../e2b/subprocess-e2b/tests/terminal.spec.ts | 70 ++++++--- 14 files changed, 375 insertions(+), 103 deletions(-) diff --git a/packages/e2b/fs-e2b/README.i18n.yaml b/packages/e2b/fs-e2b/README.i18n.yaml index 6e45ccaaf6..3c2b30ad15 100644 --- a/packages/e2b/fs-e2b/README.i18n.yaml +++ b/packages/e2b/fs-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/fs-e2b/README.md -README.md: 86ad8720d3e4c7ce70ee0ac8c41713af03ed297a -README.zh.md: 90dff57a5b9fe786aea3315633125536277ecc20 +README.md: 8abf130ce16e2b79cda5fc858182159442f9e3c3 +README.zh.md: 943d9603046ad9b2a65f41d5801cac882040a0ff diff --git a/packages/e2b/fs-e2b/README.md b/packages/e2b/fs-e2b/README.md index 86ad8720d3..8abf130ce1 100644 --- a/packages/e2b/fs-e2b/README.md +++ b/packages/e2b/fs-e2b/README.md @@ -10,7 +10,7 @@ E2B implementation of the [`@deepseek-ai/dsh-fs`](../../fs/fs/README.md) provide - **Execution-world paths** — canonical targets expose absolute POSIX process paths, percent-encoded `file:` URIs, and provider-owned containment checks, so generic subprocess consumers never parse E2B target ids or apply host path rules. - **UTF-8 reads** — whole reads and streamed reads preserve cross-chunk decoding, reject invalid UTF-8, and use the seam's 8192-byte NUL sample for binary detection. The model-facing tool still owns size selection and line windowing. - **Stable bounded reads** — a dependency-free Node helper walks directory descriptors with no-follow opens and reads one held regular-file descriptor through the byte cap. Generic LSP queries therefore reject parent swaps, non-files, invalid UTF-8, and growth past the configured document limit before server startup. -- **Atomic mutations** — writes upload a mode-`0600` temporary sibling, preserve an existing file's POSIX mode, and publish through E2B's same-directory atomic rename. The rename response supplies the committed version, so no fallible metadata request follows the commit point. E2B creates missing parent directories. Literal edits LF-normalize for matching, restore dominant CRLF storage, and serialize mutations per canonical target within the host process. Optional create/version guards keep the base seam's observed-state semantics. +- **Atomic mutations** — writes create a random sibling staging directory, change it to mode `0700` before uploading content, preserve an existing file's POSIX mode, and publish the staged file through E2B's same-filesystem atomic rename. The rename response supplies the committed version, so no fallible metadata request follows the commit point. E2B creates missing parent directories. Literal edits LF-normalize for matching, restore dominant CRLF storage, and serialize mutations per canonical target within the host process. Optional create/version guards keep the base seam's observed-state semantics. - **Failures and cancellation** — E2B not-found, permission, abort, and other controller failures map to the existing `FsError` vocabulary. Cancellation is best-effort at SDK request boundaries; a successful rename is the commit point. The provider does not copy, mount, or reconcile the host workspace. Giving it a host path as `cwd` creates a remote directory with the same spelling only. diff --git a/packages/e2b/fs-e2b/README.zh.md b/packages/e2b/fs-e2b/README.zh.md index 90dff57a5b..943d960304 100644 --- a/packages/e2b/fs-e2b/README.zh.md +++ b/packages/e2b/fs-e2b/README.zh.md @@ -10,7 +10,7 @@ - **执行世界路径**：规范化目标公开绝对 POSIX 进程路径、百分号编码的 `file:` URI，以及由提供方负责的包含关系检查，因此通用进程管理消费方无需解析 E2B 目标 ID，也不会套用宿主路径规则。 - **UTF-8 读取**：完整读取和流式读取会保留跨分片解码、拒绝无效 UTF-8，并使用 seam 的 8192 字节 NUL 样本检测二进制内容。面向模型的工具仍负责选择大小和行窗口。 - **稳定的有界读取**：一个零依赖 Node 辅助程序会以不跟随链接的方式逐级打开目录描述符，并通过一个持续持有的常规文件描述符读取至字节上限。因此，通用 LSP 查询会在服务器启动前拒绝父目录交换、非文件、无效 UTF-8，以及增长后超出所配置文档上限的文件。 -- **原子变更**：写入会上传 mode 为 `0600` 的同级临时文件，保留现有文件的 POSIX mode，并通过 E2B 的同目录原子重命名发布。重命名响应会提供已提交的版本，因此提交点之后不会再进行可能失败的元数据请求。E2B 会创建缺失的父目录。字面量编辑匹配时会规范化为 LF，存储时恢复占主导的 CRLF，并在宿主进程内按规范化目标串行执行变更。可选的创建／版本防护会保留基础 seam 的已观察状态语义。 +- **原子变更**：写入会创建随机的同级暂存目录，在上传内容前将其 mode 改为 `0700`，保留现有文件的 POSIX mode，并通过 E2B 的同一文件系统原子重命名发布暂存文件。重命名响应会提供已提交的版本，因此提交点之后不会再进行可能失败的元数据请求。E2B 会创建缺失的父目录。字面量编辑匹配时会规范化为 LF，存储时恢复占主导的 CRLF，并在宿主进程内按规范化目标串行执行变更。可选的创建／版本防护会保留基础 seam 的已观察状态语义。 - **失败与取消**：E2B 的未找到、权限、中止及其他控制器故障会映射到现有 `FsError` 词汇。取消在 SDK 请求边界上采用尽力而为语义；成功 rename 是提交点。 该提供方不会复制、挂载或协调宿主工作区。把宿主路径用作 `cwd`，只会在远程创建一个拼写相同的目录。 diff --git a/packages/e2b/fs-e2b/src/index.ts b/packages/e2b/fs-e2b/src/index.ts index 2452e856ca..0824c3f87c 100644 --- a/packages/e2b/fs-e2b/src/index.ts +++ b/packages/e2b/fs-e2b/src/index.ts @@ -498,8 +498,15 @@ export class E2BFileSystem extends FileSystem { const sandbox = await this.ctx.e2b.getSandbox() const targetPath = String(target.targetKey) const versionId = randomUUID() - const temporary = posix.join(posix.dirname(targetPath), `.${posix.basename(targetPath)}.dsh-${randomUUID()}.tmp`) + const stagingDirectory = posix.join(posix.dirname(targetPath), `.dsh-${randomUUID()}.tmp`) + const temporary = posix.join(stagingDirectory, 'content') + let stagingDirectoryCreated = false try { + const created = await sandbox.files.makeDir(stagingDirectory, signalOpts(signal)) + if (!created) throw new Error('private staging directory already exists') + stagingDirectoryCreated = true + await sandbox.commands.run(`chmod 700 -- ${quoteE2BShellArg(stagingDirectory)}`, signalOpts(signal)) + assertNotAborted(signal, 'write') await sandbox.files.write(temporary, content, { metadata: { [VERSION_METADATA_KEY]: versionId }, ...signalOpts(signal), @@ -512,12 +519,19 @@ export class E2BFileSystem extends FileSystem { ) assertNotAborted(signal, 'write') const committed = await sandbox.files.rename(temporary, targetPath, signalOpts(signal)) + try { + await sandbox.files.remove(stagingDirectory) + } catch (_committedStagingCleanupFailure) { + // The target is already committed; an empty private directory cannot turn that write into a failure. + } return entryVersion(committed) } catch (error: unknown) { - try { - await sandbox.files.remove(temporary) - } catch (_temporaryAlreadyAbsent) { - // Only the private staging path is swallowed; the original failure owns the operation. + if (stagingDirectoryCreated) { + try { + await sandbox.files.remove(stagingDirectory) + } catch (_stagingDirectoryAlreadyAbsentOrCleanupFailed) { + // Only the private staging directory is swallowed; the original failure owns the operation. + } } throw mapError(error, 'write', target.displayPath, signal) } diff --git a/packages/e2b/fs-e2b/tests/filesystem.spec.ts b/packages/e2b/fs-e2b/tests/filesystem.spec.ts index 648631c7ee..303f99c1e8 100644 --- a/packages/e2b/fs-e2b/tests/filesystem.spec.ts +++ b/packages/e2b/fs-e2b/tests/filesystem.spec.ts @@ -35,6 +35,7 @@ function commandError(exitCode: number, stderr = ''): CommandExitError { class FakeRemote { readonly nodes = new Map() readonly writes: Array<{ path: string; data: string; metadata?: Record }> = [] + readonly writeParentModes: number[] = [] readonly renames: Array<{ from: string; to: string }> = [] readonly removals: string[] = [] readonly commands: string[] = [] @@ -42,6 +43,7 @@ class FakeRemote { streamKeepOpen = false readonly streamCancel = vi.fn() nextCommandError: unknown + nextMakeDirResult: boolean | undefined nextInfoError: unknown nextListError: unknown nextReadError: unknown @@ -130,7 +132,13 @@ class FakeRemote { readonly sandbox = { sandboxId: 'fake', files: { - makeDir: async (path: string): Promise => { + makeDir: async (path: string, options?: { signal?: AbortSignal }): Promise => { + this.checkAbort(options) + if (this.nextMakeDirResult !== undefined) { + const result = this.nextMakeDirResult + this.nextMakeDirResult = undefined + return result + } if (this.nodes.has(path)) return false this.dir(path) return true @@ -178,6 +186,7 @@ class FakeRemote { this.checkAbort(options) const parent = dirname(path) if (!this.nodes.has(parent)) this.dir(parent) + this.writeParentModes.push(this.required(parent).mode) this.nodes.set(path, { type: FileType.FILE, data: bytes(data), @@ -209,7 +218,9 @@ class FakeRemote { this.nextRemoveError = undefined throw error } - this.nodes.delete(path) + for (const candidate of this.nodes.keys()) { + if (candidate === path || candidate.startsWith(`${path}/`)) this.nodes.delete(candidate) + } }, }, commands: { @@ -499,6 +510,10 @@ describe('E2BFileSystem atomic writes and edits', () => { expect(outcome).toMatchObject({ operation: 'create', before: null, after: 'one\ntwo\rthree' }) expect(remote.nodes.get('/workspace/new.txt')?.mode).toBe(0o600) expect(remote.nodes.get('/workspace/new.txt')?.metadata?.['dsh-version']).toBeDefined() + expect(remote.writeParentModes).toEqual([0o700]) + const stagingDirectory = posix.dirname(remote.writes[0]!.path) + expect(posix.dirname(stagingDirectory)).toBe('/workspace') + expect(remote.removals).toContain(stagingDirectory) await expect(fs.stat(target)).resolves.toMatchObject({ version: outcome.version, size: 14 }) }) @@ -558,6 +573,15 @@ describe('E2BFileSystem atomic writes and edits', () => { expect(controller.signal.aborted).toBe(true) }) + it('does not turn post-commit staging cleanup failure into a failed write', async () => { + const remote = new FakeRemote() + remote.nextRemoveError = new Error('empty staging cleanup failed') + const { fs } = await setup(remote) + await expect(fs.writeText(await fs.resolve('committed'), 'yes')) + .resolves.toMatchObject({ operation: 'create' }) + expect(new TextDecoder().decode(remote.nodes.get('/workspace/committed')?.data)).toBe('yes') + }) + it('returns committed rename metadata without a fallible post-commit lookup', async () => { const remote = new FakeRemote() const getInfo = vi.spyOn(remote.sandbox.files, 'getInfo') @@ -582,6 +606,11 @@ describe('E2BFileSystem atomic writes and edits', () => { remote.nextRemoveError = new Error('cleanup also failed') remote.nextRenameError = new DOMException('aborted', 'AbortError') await expectCode(fs.writeText(await fs.resolve('abort'), 'x'), 'FS_ABORTED') + + const removalsBeforeCollision = remote.removals.length + remote.nextMakeDirResult = false + await expectCode(fs.writeText(await fs.resolve('collision'), 'x'), 'FS_IO_ERROR') + expect(remote.removals).toHaveLength(removalsBeforeCollision) }) it('applies literal edits atomically and restores the detected CRLF style', async () => { diff --git a/packages/e2b/subprocess-e2b/README.i18n.yaml b/packages/e2b/subprocess-e2b/README.i18n.yaml index 3a577227f8..41d964a58a 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: fb3d79433fba8bb86097cffb09a257ad5756cc2b -README.zh.md: ec012b6a05c1a5f911d5ede9730f0f7207d5a645 +README.md: 42cd4951edb101dc75980e114e704c4aab50ae25 +README.zh.md: c4b318184cb7616a378e3336f4f4a7c366a98807 diff --git a/packages/e2b/subprocess-e2b/README.md b/packages/e2b/subprocess-e2b/README.md index fb3d79433f..42cd4951ed 100644 --- a/packages/e2b/subprocess-e2b/README.md +++ b/packages/e2b/subprocess-e2b/README.md @@ -8,10 +8,10 @@ E2B implementation of the [`@deepseek-ai/dsh-subprocess`](../../subprocess/subpr - **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, while cancellation can stop the provisional SDK handle first. - **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. -- **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. A failed transaction is observable through `waitForExit()` and may be retried, while successful termination remains a permanent no-op. Before publication, cancellation uses the provisional SDK handle; if publication fails, rollback kills and verifies the provisional group before startup rejects. 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** — 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. 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; after `graceMs`, the adapter disconnects an incomplete SDK stream, withholds its partial spill, and returns that status while retaining the remote group for `waitForExit()` and termination. 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 group in the remote terminal session before settlement. 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. Setup and teardown own the private state transaction, fence publication during service disposal, and include failure cleanup. Prompt detection, scrollback, readiness, and owner policy remain in `dsh-pty-local`. +- **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. A failed transaction is observable through `waitForExit()` and may be retried, while any proven quiescence permanently fences later termination against PID reuse. Before publication, cancellation uses the provisional SDK handle; if publication fails, rollback kills and verifies the provisional group before startup rejects. After publication, a monitoring failure also rolls back the group 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** — the wrapper starts from the sandbox command environment, removes ambient `DSH_*` and credential-shaped (`*KEY*`, `*SECRET*`, `*TOKEN*`) names, then restores every valid `spec.env` entry as an explicit caller opt-in; empty names, `=`, and NUL framing violations reject before launch. 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; raw pipes instead await lossless transport completion and preserve backpressure. 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 group in the remote terminal session before settlement. 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. Setup and teardown own the private state transaction, abort pending setup during service disposal, fence publication, and retain an unproven setup cleanup for disposal retry. Prompt detection, scrollback, readiness, and owner policy remain in `dsh-pty-local`. The base E2B image supplies the runtime and Bash/GNU utilities this adapter invokes: `node`, `bash`, `setsid`, `ps`, `awk`, `tr`, `env`, `chmod`, `tee`, `head`, `rm`, and `kill`. A custom template must retain compatible commands and E2B PTY support. @@ -29,6 +29,6 @@ No direct invalidation; the named consumers own any request-prefix changes. - **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. -- **E2B exposes no signal fact** — only an adapter-requested `SIGTERM` or `SIGKILL` is reported as a signal; every unrequested SDK exit remains an exit code, including values shaped like `128 + signal`. +- **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`. - **Exact terminal stdin-wait inspection is unavailable** — E2B exposes the foreground process group but not the syscall evidence needed to prove it is waiting on fd 0, so the generic PTY backend falls back to controlled prompt markers and bounded silence. - **Linux utility and E2B transport semantics are assumed** — there is no Windows, arbitrary-template, escaped-session recovery, or network-partition fidelity layer. diff --git a/packages/e2b/subprocess-e2b/README.zh.md b/packages/e2b/subprocess-e2b/README.zh.md index ec012b6a05..c4b318184c 100644 --- a/packages/e2b/subprocess-e2b/README.zh.md +++ b/packages/e2b/subprocess-e2b/README.zh.md @@ -8,10 +8,10 @@ - **异步远程启动**：同步 seam 会立即返回一个句柄，同时由 `Sandbox.commands.run(..., { background: true })` 在远程启动进程。包装层发布进程组 ID 并由适配器完成验证之前，`pid` 为 `-1`；stdin 和常规观察会等待该发布，而取消操作可以先停止临时 SDK 句柄。 - **执行世界坐标**：`cwd` 和私有 `runtimeRoot` 来自共享所有者；可执行文件查找会验证绝对路径，或根据沙箱 PATH 加显式覆盖来解析裸名称。 -- **Linux 进程组**：带引号保护的包装层会在 `exec setsid --wait` 下启动每组 argv，并在 `ctx.e2b.runtimeRoot/processes` 下记录实际进程组 ID 和私有状态文件。句柄会等待该文件，而不会把 SDK 命令 PID 当作已发布的身份。终止操作以记录的负数 ID 发送 `SIGTERM`，等待调用方的 `graceMs`，再升级到 `SIGKILL` 和 SDK kill 回退；TERM 信号发送或探测失败也会强制触发该升级。失败的事务可通过 `waitForExit()` 观察，并可重试；成功终止后，重复终止将永久为空操作。发布前，取消操作使用临时 SDK 句柄；如果发布失败，回滚会终止并验证临时进程组，随后启动操作才会拒绝。服务 dispose（资源释放）会拒绝新的启动请求、终止并等待每个保留进程组退出，再等待 SDK 结算和私有清理完成，之后沙箱所有者才会释放。 -- **环境边界**：包装层从沙箱命令环境开始，移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），再把每个 `spec.env` 条目恢复为调用方显式选择。宿主环境变量绝不会隐式进入沙箱。私有环境文件在使用后会被删除；命令或终端设置失败时，会先删除其私有状态再拒绝。 -- **stdio 投影**：远程包装层先把原始字节分流到可选的有界 spill 文件，再把每个实时分片编码为换行分隔的 base64 ASCII 帧；宿主会跨任意 SDK 回调边界增量恢复字节。pipe 模式把这些字节写入宿主 Node 流；inherit 模式把字节写入 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。包装层会在等待继承管道的写入方之前发布直接命令状态；超过 `graceMs` 后，适配器会断开未完成的 SDK 流，不公开其中不完整的 spill，并返回该状态，同时保留远程进程组供 `waitForExit()` 和终止操作使用。批量 stdin 和流式 stdin 都使用 SDK 句柄。 -- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并在结算前清理远程终端会话中的每个进程组。私有随机输出边界会丢弃 E2B 引导 shell 的提示符和回显的 runner 命令，同时保留请求进程的每个字节，包括其第一个提示符。setup 与 teardown 负责私有状态事务，在服务 dispose 期间阻止发布，并包括失败清理。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 +- **Linux 进程组**：带引号保护的包装层会在 `exec setsid --wait` 下启动每组 argv，并在 `ctx.e2b.runtimeRoot/processes` 下记录实际进程组 ID 和私有状态文件。句柄会等待该文件，而不会把 SDK 命令 PID 当作已发布的身份。终止操作以记录的负数 ID 发送 `SIGTERM`，等待调用方的 `graceMs`，再升级到 `SIGKILL` 和 SDK kill 回退；TERM 信号发送或探测失败也会强制触发该升级。失败的事务可通过 `waitForExit()` 观察，并可重试；任何已证明的完全停稳都会永久防止后续终止操作命中复用的 PID。发布前，取消操作使用临时 SDK 句柄；如果发布失败，回滚会终止并验证临时进程组，随后启动操作才会拒绝。发布后，监控失败也会在拒绝前回滚进程组。服务 dispose（资源释放）会拒绝新的启动请求、终止并等待每个保留进程组退出，再等待 SDK 结算和私有清理完成，之后沙箱所有者才会释放。 +- **环境边界**：包装层从沙箱命令环境开始，移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），再把每个有效的 `spec.env` 条目恢复为调用方显式选择；空名称、`=` 和违反 NUL 分帧规则的条目会在启动前被拒绝。宿主环境变量绝不会隐式进入沙箱。私有环境文件在使用后会被删除；命令或终端设置失败时，会先删除其私有状态再拒绝。 +- **stdio 投影**：远程包装层先把原始字节分流到可选的有界 spill 文件，再把每个实时分片编码为换行分隔的 base64 ASCII 帧；宿主会跨任意 SDK 回调边界增量恢复字节。pipe 模式把这些字节写入宿主 Node 流；inherit 模式把字节写入 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。包装层会在等待继承管道的写入方之前发布直接命令状态。对于 collect 或 inherit 输出，超过 `graceMs` 后，适配器会断开未完成的 SDK 流，不公开其中不完整的 spill，并返回该状态，同时保留远程进程组供 `waitForExit()` 和终止操作使用；原始 pipe 则会等待无损传输完成并保留背压。批量 stdin 和流式 stdin 都使用 SDK 句柄。 +- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并在结算前清理远程终端会话中的每个进程组。私有随机输出边界会丢弃 E2B 引导 shell 的提示符和回显的 runner 命令，同时保留请求进程的每个字节，包括其第一个提示符。setup 与 teardown 负责私有状态事务，在服务 dispose 期间中止待处理的 setup、阻止发布，并保留未证明已完成的 setup 清理事务，供 dispose 重试。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 基础 E2B 镜像提供该适配器调用的运行时和 Bash/GNU 工具：`node`、`bash`、`setsid`、`ps`、`awk`、`tr`、`env`、`chmod`、`tee`、`head`、`rm` 和 `kill`。自定义模板必须保留兼容的命令和 E2B PTY 支持。 @@ -29,6 +29,6 @@ - **不支持需要同步 PID 的消费方**：远程启动期间，`pid` 保持为 `-1`；包括 ACP 子进程后端在内，要求立即获得正 PID 的消费方无法原样使用本提供方。 - **重新连接不会重建句柄**：保留沙箱后，远程 PID／状态／spill 文件仍然存在，但新的 harness 进程不会据此重建实时 `SubprocessHandle` 对象或输出游标。 - **保留沙箱时会累积远程状态**：进程目录和有效的 spill 文件会留在 `.dsh-e2b` 下；本 POC 不提供保留清理。 -- **E2B 不公开信号事实**：只有适配器请求的 `SIGTERM` 或 `SIGKILL` 会报告为信号；其他未请求的 SDK 退出始终保留为退出码，包括形似 `128 + signal` 的值。 +- **E2B 不公开信号事实**：适配器请求的 `SIGTERM` 或 `SIGKILL` 只有在包装层发布的直接退出码没有胜出时才报告为信号；其他未请求的 SDK 退出始终保留为退出码，包括形似 `128 + signal` 的值。 - **无法精确检查终端 stdin 等待状态**：E2B 会公开前台进程组，但不提供证明其正在等待 fd 0 所需的 syscall 证据，因此通用 PTY 后端会回退到受控提示符标记与有界静默机制。 - **依赖 Linux 工具与 E2B 传输语义**：没有 Windows、任意模板、逃逸会话恢复或网络分区的保真层。 diff --git a/packages/e2b/subprocess-e2b/src/environment.ts b/packages/e2b/subprocess-e2b/src/environment.ts index cc9b41296a..f9c30d378f 100644 --- a/packages/e2b/subprocess-e2b/src/environment.ts +++ b/packages/e2b/subprocess-e2b/src/environment.ts @@ -19,3 +19,23 @@ export function scrubRemoteEnvironment(raw: string): Map { } return environment } + +/** + * Overlay explicit entries and serialize one validated E2B environment. + * @param raw - The complete NUL-delimited remote environment. + * @param explicit - Deliberate caller overrides applied after ambient scrubbing. + * @returns NUL-delimited `name=value` entries accepted by `env -i`. + */ +export function serializeRemoteEnvironment( + raw: string, + explicit: Readonly> | undefined, +): string { + const environment = scrubRemoteEnvironment(raw) + for (const [name, value] of Object.entries(explicit ?? {})) { + if (name.length === 0 || name.includes('=') || name.includes('\0') || value.includes('\0')) { + throw new Error('subprocess-e2b: environment entries require non-empty NUL-free names without = and NUL-free values') + } + environment.set(name, value) + } + return [...environment].map(([name, value]) => `${name}=${value}\0`).join('') +} diff --git a/packages/e2b/subprocess-e2b/src/index.ts b/packages/e2b/subprocess-e2b/src/index.ts index fcf2c6f0a4..bf7edc41a1 100644 --- a/packages/e2b/subprocess-e2b/src/index.ts +++ b/packages/e2b/subprocess-e2b/src/index.ts @@ -28,7 +28,8 @@ export class E2BSubprocessService extends SubprocessService { private readonly live = new Set() private readonly terminals = new Set() - private readonly terminalSetups = new Set>() + private readonly terminalSetups = new Map, AbortController>() + private readonly failedTerminalSetupCleanups = new Set<() => Promise>() private disposing = false /** @inheritdoc */ @@ -44,9 +45,13 @@ export class E2BSubprocessService extends SubprocessService { this.runtimeRoot = ctx.e2b.runtimeRoot ctx.effect(() => async () => { this.disposing = true - await Promise.all([...this.terminalSetups]) + for (const controller of this.terminalSetups.values()) { + controller.abort(new Error('subprocess-e2b: service disposed during terminal setup')) + } + await Promise.all([...this.terminalSetups.keys()]) const handles = [...this.live] const terminals = [...this.terminals] + const failedTerminalSetupCleanups = [...this.failedTerminalSetupCleanups] const pending: Promise[] = [] for (const handle of handles) { handle.terminate() @@ -59,6 +64,9 @@ export class E2BSubprocessService extends SubprocessService { terminal.terminate() pending.push(terminal.waitForExit().then(() => { this.terminals.delete(terminal) })) } + for (const cleanup of failedTerminalSetupCleanups) { + pending.push(cleanup().then(() => { this.failedTerminalSetupCleanups.delete(cleanup) })) + } await Promise.all(pending) }, 'e2b subprocess teardown') } @@ -133,9 +141,18 @@ export class E2BSubprocessService extends SubprocessService { spec.signal?.throwIfAborted() const stateDir = posix.join(this.runtimeRoot, 'terminals', randomUUID()) const setup = Promise.withResolvers() - this.terminalSetups.add(setup.promise) + const setupController = new AbortController() + const setupSignal = spec.signal === undefined + ? setupController.signal + : AbortSignal.any([spec.signal, setupController.signal]) + this.terminalSetups.set(setup.promise, setupController) try { - const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir) + const terminal = await spawnE2BTerminal( + this.ctx.e2b, + { ...spec, signal: setupSignal }, + stateDir, + (cleanup) => { this.failedTerminalSetupCleanups.add(cleanup) }, + ) this.terminals.add(terminal) if (this.isDisposing()) { terminal.terminate() diff --git a/packages/e2b/subprocess-e2b/src/process.ts b/packages/e2b/subprocess-e2b/src/process.ts index 3038f312dc..4855b27b81 100644 --- a/packages/e2b/subprocess-e2b/src/process.ts +++ b/packages/e2b/subprocess-e2b/src/process.ts @@ -17,7 +17,7 @@ import type { SubprocessSpawnSpec, } from '@deepseek-ai/dsh-subprocess' import type E2BSandboxService from '@deepseek-ai/dsh-e2b' -import { scrubRemoteEnvironment } from './environment.ts' +import { serializeRemoteEnvironment } from './environment.ts' import { E2BBase64Decoder, E2B_OUTPUT_COMPLETE_FRAME, E2BOutputReader } from './output.ts' const GROUP_POLL_MS = 20 @@ -92,12 +92,6 @@ function withinMs(promise: Promise, timeoutMs: number): Promise> | undefined): string { - const environment = scrubRemoteEnvironment(raw) - for (const [name, value] of Object.entries(explicit ?? {})) environment.set(name, value) - return [...environment].map(([name, value]) => `${name}=${value}\0`).join('') -} - function commandText(spec: SubprocessSpawnSpec, paths: RemotePaths): string { const encoder = `"$dsh_e2b_env_bin" -i "$dsh_e2b_node" -e ${quoteE2BShellArg(OUTPUT_ENCODER_SOURCE)}` const stdoutRedirect = hasSpill(spec.stdio.stdout) @@ -213,7 +207,8 @@ export class E2BSubprocessHandle implements SubprocessHandle { private invalidHandleQuiescent = false private provisionalHandleQuiescent = false private terminationStarted = false - private terminationSucceeded = false + private terminationFenced = false + private quiescenceProven = false private terminationAttempt: Promise | undefined private terminationFailure: Error | undefined private terminationSignal: NodeJS.Signals | null = null @@ -265,18 +260,18 @@ export class E2BSubprocessHandle implements SubprocessHandle { /** @inheritdoc */ terminate(): void { - if (this.terminationSucceeded || this.terminationAttempt !== undefined) return + if (this.terminationFenced || this.quiescenceProven || this.terminationAttempt !== undefined) return this.terminationStarted = true this.terminationFailure = undefined const attempt = this.terminateRemote() this.terminationAttempt = attempt void attempt.then( () => { - this.terminationSucceeded = true + this.terminationFenced = true this.terminationAttempt = undefined }, (error: unknown) => { - this.terminationFailure = asError(error) + if (!this.quiescenceProven) this.terminationFailure = asError(error) this.terminationAttempt = undefined }, ) @@ -284,17 +279,24 @@ export class E2BSubprocessHandle implements SubprocessHandle { /** @inheritdoc */ async waitForExit(signal?: AbortSignal): Promise { + if (this.quiescenceProven) return true let handle: CommandHandle | undefined if (this.terminationStarted) { const observed = await waitWithSignal(this.commandState.promise, signal) if (observed === WAIT_ABORTED) return false handle = observed - if (handle === undefined) return true + if (handle === undefined) { + this.markQuiescent() + return true + } if (this.remotePid <= 0) { const attempt = this.terminationAttempt if (attempt !== undefined && await waitWithSignal(attempt, signal) === WAIT_ABORTED) return false this.throwTerminationFailure() - if (this.invalidHandleQuiescent || this.provisionalHandleQuiescent) return true + if (this.invalidHandleQuiescent || this.provisionalHandleQuiescent) { + this.markQuiescent() + return true + } } } else { try { @@ -303,7 +305,10 @@ export class E2BSubprocessHandle implements SubprocessHandle { handle = observed } catch { handle = this.commandHandle - if (handle === undefined) return true + if (handle === undefined) { + this.markQuiescent() + return true + } } } this.throwTerminationFailure() @@ -320,11 +325,18 @@ export class E2BSubprocessHandle implements SubprocessHandle { if (!await waitTick(signal)) return false } this.throwTerminationFailure() - return !isAborted(signal) + if (isAborted(signal)) return false + this.markQuiescent() + return true } private readonly onAbort = (): void => { this.terminate() } + private markQuiescent(): void { + this.quiescenceProven = true + this.terminationFailure = undefined + } + private async run(): Promise { let sandbox: Sandbox | undefined try { @@ -385,7 +397,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { return outcome } catch (error: unknown) { this.commandState.resolve(undefined) - let failure = error + let failure = await this.rollbackPublishedFailure(error) if (sandbox !== undefined && this.stateDirectoryCreated) { try { await this.removeFailedState(sandbox) @@ -413,7 +425,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { const files = [ { path: this.paths.pid, data: '' }, { path: this.paths.status, data: '' }, - { path: this.paths.environment, data: remoteEnvironment(ambient.stdout, this.spec.env) }, + { path: this.paths.environment, data: serializeRemoteEnvironment(ambient.stdout, this.spec.env) }, ...(hasSpill(this.spec.stdio.stdout) ? [{ path: this.paths.stdout, data: '' }] : []), ...(hasSpill(this.spec.stdio.stderr) ? [{ path: this.paths.stderr, data: '' }] : []), ] @@ -508,15 +520,16 @@ export class E2BSubprocessHandle implements SubprocessHandle { if (!/^(?:0|[1-9][0-9]*)$/.test(rawStatus) || !Number.isSafeInteger(exitCode) || exitCode > 255) { throw new Error(`subprocess-e2b: remote wrapper published invalid exit code ${JSON.stringify(rawStatus)}`) } + if (this.spec.stdio.stdout === 'pipe' || this.spec.stdio.stderr === 'pipe') { + return this.commandOutcome(await settlement) + } const completed = await withinMs(settlement, this.spec.graceMs) if (completed !== undefined) return this.commandOutcome(completed) this.outputDrainExpired = true this.stdoutReader?.invalidateSpill() this.stderrReader?.invalidateSpill() await handle.disconnect() - return this.terminationSignal === null - ? { exitCode, signal: null } - : { exitCode: null, signal: this.terminationSignal } + return { exitCode, signal: null } } const completed = await Promise.race([settlement, waitTick().then(() => undefined)]) if (completed !== undefined) return this.commandOutcome(completed) @@ -533,6 +546,20 @@ export class E2BSubprocessHandle implements SubprocessHandle { throw settlement.error } + private async rollbackPublishedFailure(error: unknown): Promise { + if (this.remotePid <= 0 || this.commandHandle === undefined || this.quiescenceProven) return error + this.terminate() + try { + await this.waitForExit() + return error + } catch (cleanupError: unknown) { + return new AggregateError( + [asError(error), asError(cleanupError)], + 'subprocess-e2b: command monitoring failed and process-group rollback did not reach quiescence', + ) + } + } + private async rollbackUnpublishedGroup(sandbox: Sandbox, handle: CommandHandle): Promise { // The bootstrap ends in an exec chain through the scrubbed environment and // `setsid`, so E2B's command PID is the provisional group id even before the diff --git a/packages/e2b/subprocess-e2b/src/terminal.ts b/packages/e2b/subprocess-e2b/src/terminal.ts index 78668490d7..0f5809fa44 100644 --- a/packages/e2b/subprocess-e2b/src/terminal.ts +++ b/packages/e2b/subprocess-e2b/src/terminal.ts @@ -19,7 +19,7 @@ import type { SubprocessTerminalSpawnSpec, } from '@deepseek-ai/dsh-subprocess' import type E2BSandboxService from '@deepseek-ai/dsh-e2b' -import { scrubRemoteEnvironment } from './environment.ts' +import { serializeRemoteEnvironment } from './environment.ts' const POLL_MS = 20 @@ -142,17 +142,6 @@ function serializeValues(values: readonly string[], kind: string): string { return values.map(value => `${value}\0`).join('') } -function remoteEnvironment(raw: string, explicit: Readonly> | undefined): string { - const environment = scrubRemoteEnvironment(raw) - for (const [name, value] of Object.entries(explicit ?? {})) { - if (name.length === 0 || name.includes('=') || name.includes('\0') || value.includes('\0')) { - throw new Error('subprocess-e2b: terminal environment entries require non-empty NUL-free names without = and NUL-free values') - } - environment.set(name, value) - } - return serializeValues([...environment].map(([name, value]) => `${name}=${value}`), 'environment') -} - async function terminalSessionId(sandbox: Sandbox, pid: number, signal?: AbortSignal): Promise { const result = await sandbox.commands.run(`ps -o sid= -p ${pid}`, signalOpts(signal)) signal?.throwIfAborted() @@ -432,12 +421,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 retainFailedCleanup - Optional owner for retrying a cleanup transaction that could not prove quiescence. * @returns The live subprocess terminal handle. */ export async function spawnE2BTerminal( runtime: E2BSandboxService, spec: SubprocessTerminalSpawnSpec, stateDir: string, + retainFailedCleanup?: (cleanup: () => Promise) => void, ): Promise { const sandbox = await runtime.getSandbox() spec.signal?.throwIfAborted() @@ -456,7 +447,7 @@ export async function spawnE2BTerminal( let stateDirectoryCreated = false try { const ambient = await sandbox.commands.run('env -0', signalOpts(spec.signal)) - const environment = remoteEnvironment(ambient.stdout, spec.env) + const environment = serializeRemoteEnvironment(ambient.stdout, spec.env) const argv = serializeValues(spec.argv, 'argv') await sandbox.files.makeDir(stateDir) stateDirectoryCreated = true @@ -502,25 +493,37 @@ export async function spawnE2BTerminal( ) } catch (error: unknown) { output.destroy() - const cleanupErrors: Error[] = [] - if (handle !== undefined && completion !== undefined) { - try { - await rollbackUnpublishedTerminal(sandbox, handle, completion, spec.graceMs) - } catch (rollbackError: unknown) { - cleanupErrors.push(asError(rollbackError)) + let terminalQuiescent = handle === undefined + let stateRemoved = !stateDirectoryCreated + const retryCleanup = async (): Promise => { + const failures: Error[] = [] + if (!terminalQuiescent && handle !== undefined) { + try { + if (completion === undefined) await handle.kill() + else await rollbackUnpublishedTerminal(sandbox, handle, completion, spec.graceMs) + terminalQuiescent = true + } catch (cleanupError: unknown) { + failures.push(asError(cleanupError)) + } } - } else if (handle !== undefined) { - await handle.kill().catch(() => false) - } - if (stateDirectoryCreated) { - try { - await sandbox.files.remove(stateDir) - } catch (stateError: unknown) { - if (!(stateError instanceof FileNotFoundError)) cleanupErrors.push(asError(stateError)) + if (!stateRemoved) { + try { + await sandbox.files.remove(stateDir) + stateRemoved = true + } catch (stateError: unknown) { + if (stateError instanceof FileNotFoundError) stateRemoved = true + else failures.push(asError(stateError)) + } + } + if (failures.length > 0) { + throw new AggregateError(failures, 'subprocess-e2b: terminal setup cleanup did not complete') } } - if (cleanupErrors.length > 0) { - throw new AggregateError([asError(error), ...cleanupErrors], asError(error).message) + try { + await retryCleanup() + } catch (cleanupError: unknown) { + retainFailedCleanup?.(retryCleanup) + throw new AggregateError([asError(error), asError(cleanupError)], asError(error).message) } throw error } diff --git a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts index 8e6e420bcc..462e943a1b 100644 --- a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts @@ -97,6 +97,7 @@ class FakeSandbox { startOptions: StartOptions | undefined backgroundError: unknown envError: unknown + statusError: unknown nextRemoveError: unknown probeError: unknown signalError: unknown @@ -117,6 +118,8 @@ class FakeSandbox { private openStart: (() => void) | undefined private processGroupReadGate: Promise | undefined private openProcessGroupRead: (() => void) | undefined + private signalGate: Promise | undefined + private openSignal: (() => void) | undefined constructor() { this.handle = new FakeCommandHandle(() => { @@ -147,6 +150,16 @@ class FakeSandbox { this.openProcessGroupRead?.() } + deferSignals(): void { + const gate = Promise.withResolvers() + this.signalGate = gate.promise + this.openSignal = () => { gate.resolve(undefined) } + } + + releaseSignals(): void { + this.openSignal?.() + } + finish(exitCode = 0): void { this.alive = false void this.completeOutput().then( @@ -198,6 +211,11 @@ class FakeSandbox { await this.processGroupReadGate return this.processGroupReads.shift() ?? this.processGroupId } + if (this.statusError !== undefined) { + const error = this.statusError + this.statusError = undefined + throw error + } this.afterStatusRead?.() return this.exitStatus }, @@ -230,6 +248,7 @@ class FakeSandbox { return { exitCode: 0, stdout: '', stderr: '' } } if (command.startsWith('kill -TERM ')) { + await this.signalGate const error = this.signalErrors.shift() ?? this.signalError if (error !== undefined) { if (this.signalErrors.length === 0) this.signalError = undefined @@ -242,6 +261,7 @@ class FakeSandbox { return { exitCode: 0, stdout: '', stderr: '' } } if (command.startsWith('kill -KILL ')) { + await this.signalGate const error = this.signalErrors.shift() ?? this.signalError if (error !== undefined) { if (this.signalErrors.length === 0) this.signalError = undefined @@ -399,6 +419,16 @@ describe('E2BSubprocessHandle', () => { await expect(handle.waitForExit()).resolves.toBe(true) }) + it('rejects malformed environment entries before command start', async () => { + for (const env of [{ 'BAD=NAME': 'x' }, { BAD: 'x\0INJECTED=1' }]) { + const fake = new FakeSandbox() + const handle = new E2BSubprocessHandle(runtime(fake), spec({ env }), '/runtime/invalid-environment') + await expect(handle.done).rejects.toThrow('environment entries') + expect(fake.startOptions).toBeUndefined() + expect(fake.removed).toContain('/runtime/invalid-environment') + } + }) + it('preserves UTF-8 bytes when the ASCII transport is split across callbacks', async () => { const fake = new FakeSandbox() const handle = new E2BSubprocessHandle(runtime(fake), spec({ @@ -460,6 +490,29 @@ describe('E2BSubprocessHandle', () => { await expect(handle.waitForExit()).resolves.toBe(true) }) + 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({ + graceMs: 1, + stdio: { stdin: 'ignore', stdout: 'pipe', stderr: { maxBytes: 4 } }, + }), '/runtime/pipe-drain') + let output = '' + handle.stdout!.on('data', (chunk) => { output += String(chunk) }) + await flush() + fake.exitStatus = '0\n' + + let settled = false + void handle.done.then(() => { settled = true }) + await new Promise(resolve => setTimeout(resolve, 50)) + expect(settled).toBe(false) + expect(fake.handle.disconnects).toBe(0) + + await fake.stdout('complete protocol frame') + fake.finish() + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + expect(output).toBe('complete protocol frame') + }) + it('accepts clean encoder completion inside the output-drain grace', async () => { const fake = new FakeSandbox() const handle = new E2BSubprocessHandle(runtime(fake), spec({ graceMs: 100 }), '/runtime/drain-complete') @@ -474,7 +527,7 @@ describe('E2BSubprocessHandle', () => { expect(fake.handle.disconnects).toBe(0) }) - it('preserves a requested signal when output draining expires', async () => { + it('preserves a published exit code when requested termination outlives output draining', async () => { const fake = new FakeSandbox() fake.trapsTerm = true fake.delaysKill = true @@ -485,9 +538,9 @@ describe('E2BSubprocessHandle', () => { handle.terminate() await vi.waitFor(() => { expect(fake.commandsSeen).toContain('kill -KILL -- -4242') }) - fake.exitStatus = '143\n' + fake.exitStatus = '0\n' - await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' }) + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) expect(fake.handle.disconnects).toBe(1) fake.alive = false await expect(handle.waitForExit()).resolves.toBe(true) @@ -503,6 +556,31 @@ describe('E2BSubprocessHandle', () => { await expect(handle.waitForExit()).resolves.toBe(true) }) + it('rolls back a published process group before rejecting a monitoring failure', async () => { + const fake = new FakeSandbox() + fake.statusError = new Error('status transport failed') + const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/status-failure') + + await expect(handle.done).rejects.toThrow('status transport failed') + expect(fake.commandsSeen).toContain('kill -TERM -- -4242') + expect(fake.alive).toBe(false) + await expect(handle.waitForExit()).resolves.toBe(true) + + const failed = new FakeSandbox() + failed.statusError = new Error('status transport failed') + failed.signalErrors.push(new Error('TERM transport failed'), new Error('KILL transport failed')) + failed.handle.killError = new Error('SDK kill failed') + const retained = new E2BSubprocessHandle(runtime(failed), spec({ graceMs: 1 }), '/runtime/status-cleanup-failure') + + await expect(retained.done).rejects.toThrow( + 'command monitoring failed and process-group rollback did not reach quiescence', + ) + expect(failed.alive).toBe(true) + failed.handle.killError = undefined + retained.terminate() + await expect(retained.waitForExit()).resolves.toBe(true) + }) + it('surfaces deferred piped-stdin write and close failures as stream errors', async () => { const writeFake = new FakeSandbox() writeFake.deferStart() @@ -607,6 +685,46 @@ describe('E2BSubprocessHandle', () => { expect(fake.commandsSeen.filter(command => command.startsWith('kill -'))).toHaveLength(signals) }) + it('makes termination a permanent no-op after natural quiescence is observed', async () => { + const fake = new FakeSandbox() + const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/natural-quiescence') + await flush() + fake.finish() + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + await expect(handle.waitForExit()).resolves.toBe(true) + + const signals = fake.commandsSeen.filter(command => command.startsWith('kill -')).length + fake.alive = true + handle.terminate() + await flush() + expect(fake.alive).toBe(true) + expect(fake.commandsSeen.filter(command => command.startsWith('kill -'))).toHaveLength(signals) + }) + + it('keeps proven quiescence after a concurrent termination transport fails', async () => { + const fake = new FakeSandbox() + fake.signalErrors.push(new Error('TERM transport failed'), new Error('KILL transport failed')) + fake.handle.killError = new Error('SDK kill failed') + fake.deferSignals() + const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/quiescent-race') + await flush() + + handle.terminate() + await vi.waitFor(() => { expect(fake.commandsSeen).toContain('kill -TERM -- -4242') }) + fake.alive = false + await expect(handle.waitForExit()).resolves.toBe(true) + + fake.probeError = new Error('post-quiescence probe failed') + fake.releaseSignals() + await vi.waitFor(() => { expect(fake.handle.kills).toBe(1) }) + fake.finish() + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + const signals = fake.commandsSeen.filter(command => command.startsWith('kill -')).length + handle.terminate() + await expect(handle.waitForExit()).resolves.toBe(true) + expect(fake.commandsSeen.filter(command => command.startsWith('kill -'))).toHaveLength(signals) + }) + it('escalates a TERM-trapping process group to KILL and uses the SDK kill as fallback', async () => { const fake = new FakeSandbox() fake.trapsTerm = true @@ -620,6 +738,28 @@ describe('E2BSubprocessHandle', () => { expect(fake.handle.kills).toBe(1) }) + it('does not treat successful termination transport as observed quiescence', async () => { + const fake = new FakeSandbox() + fake.trapsTerm = true + fake.delaysKill = true + fake.delaysKillCompletion = true + fake.sdkKillStops = false + const handle = new E2BSubprocessHandle(runtime(fake), spec({ graceMs: 1 }), '/runtime/termination-fence') + await flush() + handle.terminate() + await vi.waitFor(() => { expect(fake.handle.kills).toBe(1) }) + + let quiescent = false + const waiting = handle.waitForExit().then((value) => { quiescent = value }) + await new Promise(resolve => setTimeout(resolve, 10)) + expect(quiescent).toBe(false) + + fake.alive = false + fake.finish() + await waiting + expect(quiescent).toBe(true) + }) + it('honors termination requested before asynchronous startup finishes', async () => { const fake = new FakeSandbox() fake.deferStart() diff --git a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts index 1eac9a5780..e77646e586 100644 --- a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts @@ -111,18 +111,7 @@ class FakeTerminalSandbox { resolvedExecutable = '/usr/bin/node\n' requestedOutput = 'requested-shell$ ' emitOutputMarker = true - private createGate: Promise | undefined - private releaseCreateGate: (() => void) | undefined - - deferCreate(): void { - const gate = Promise.withResolvers() - this.createGate = gate.promise - this.releaseCreateGate = () => { gate.resolve(undefined) } - } - - releaseCreate(): void { - this.releaseCreateGate?.() - } + afterSessionLookup: (() => void) | undefined readonly sandbox = { files: { @@ -163,7 +152,10 @@ class FakeTerminalSandbox { if (command.includes('command -v -- ')) { return { exitCode: 0, stdout: this.resolvedExecutable, stderr: '' } } - if (command.startsWith('ps -o sid=')) return { exitCode: 0, stdout: this.sessionId, stderr: '' } + if (command.startsWith('ps -o sid=')) { + this.afterSessionLookup?.() + return { exitCode: 0, stdout: this.sessionId, stderr: '' } + } if (command.startsWith('ps -o tpgid=')) { if (this.foregroundFailure !== undefined) throw this.foregroundFailure return { exitCode: 0, stdout: this.foreground, stderr: '' } @@ -186,7 +178,6 @@ class FakeTerminalSandbox { pty: { create: async (options: Parameters[0]): Promise => { this.createOptions = options - await this.createGate if (this.createError !== undefined) throw this.createError await options.onData(Buffer.from('buffered banner\n')) return this.handle.asHandle() @@ -691,7 +682,7 @@ describe('E2B subprocess terminal service', () => { it('owns live terminals through service disposal', async () => { const { ctx, fiber, fake } = await service() - const terminal = await ctx.subprocess.spawnTerminal(spec()) + const terminal = await ctx.subprocess.spawnTerminal(spec({ signal: new AbortController().signal })) await fiber.dispose() await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' }) expect(fake.handle.disconnects).toBe(1) @@ -699,19 +690,19 @@ describe('E2B subprocess terminal service', () => { it('joins and rejects terminal setup that completes during service disposal', async () => { const fake = new FakeTerminalSandbox() - fake.deferCreate() const { ctx, fiber } = await service(fake) + let disposing: Promise | undefined + fake.afterSessionLookup = () => { + fake.afterSessionLookup = undefined + queueMicrotask(() => { + queueMicrotask(() => { disposing = fiber.dispose() }) + }) + } + const subprocess = ctx.subprocess const spawning = ctx.subprocess.spawnTerminal(spec()) const rejected = expect(spawning).rejects.toThrow('service disposed during terminal setup') - await vi.waitFor(() => { expect(fake.createOptions).toBeDefined() }) - - let disposed = false - const subprocess = ctx.subprocess - const disposing = fiber.dispose().then(() => { disposed = true }) - await new Promise(resolve => setTimeout(resolve, 0)) - expect(disposed).toBe(false) + await vi.waitFor(() => { expect(disposing).toBeDefined() }) await expect(subprocess.spawnTerminal(spec())).rejects.toThrow('service is disposing') - fake.releaseCreate() await rejected await disposing @@ -720,6 +711,37 @@ describe('E2B subprocess terminal service', () => { expect(fake.removed.some(path => path.includes('/terminals/'))).toBe(true) }) + it('aborts and rolls back terminal setup that cannot publish readiness during disposal', async () => { + const fake = new FakeTerminalSandbox() + fake.ready = new FileNotFoundError('not ready') + const { ctx, fiber } = await service(fake) + const spawning = ctx.subprocess.spawnTerminal(spec()) + const rejected = expect(spawning).rejects.toThrow('service disposed during terminal setup') + await vi.waitFor(() => { expect(fake.readyReads).toBeGreaterThan(0) }) + + await fiber.dispose() + await rejected + expect(fake.groups).toEqual([]) + expect(fake.handle.disconnects).toBe(1) + }) + + it('retains failed terminal setup cleanup for disposal retry', async () => { + const fake = new FakeTerminalSandbox() + fake.sendError = new Error('bootstrap failed') + fake.clearOnTerm = false + fake.clearOnKill = false + const { ctx, fiber } = await service(fake) + + await expect(ctx.subprocess.spawnTerminal(spec({ graceMs: 1 }))).rejects.toThrow('bootstrap failed') + expect(fake.groups).toEqual([123]) + expect(fake.handle.disconnects).toBe(0) + + fake.clearOnKill = true + await fiber.dispose() + expect(fake.groups).toEqual([]) + expect(fake.handle.disconnects).toBe(1) + }) + it('releases naturally settled terminals and validates terminal requests', async () => { const { ctx, fiber, fake } = await service() for (const request of [