diff --git a/packages/e2b/e2b/tests/composition.e2e.ts b/packages/e2b/e2b/tests/composition.e2e.ts index 1ece44bb6a..34875d5153 100644 --- a/packages/e2b/e2b/tests/composition.e2e.ts +++ b/packages/e2b/e2b/tests/composition.e2e.ts @@ -185,7 +185,10 @@ describe.skipIf(!process.env.E2B_API_KEY)('E2B live Loader composition', () => { oversizedReply: { error: { kind: 'worker-exit' } }, lingeringCodeRunners: 0, }) - expect((output.terminal as { motd: string }).motd.length).toBeGreaterThan(0) + const terminalMotd = (output.terminal as { motd: string }).motd + expect(terminalMotd.length).toBeGreaterThan(0) + expect(terminalMotd).not.toContain('exec /bin/bash') + expect(terminalMotd).not.toContain('.dsh-e2b/terminals/') expect((output.terminal as { echo: { viewport: string } }).echo.viewport).toContain('PTY-你好') expect((output.terminal as { scrollback: string }).scrollback).toContain('PTY-你好') expect((output.terminal as { signal: { targetPgid: number } }).signal.targetPgid).toBeGreaterThan(0) diff --git a/packages/e2b/subprocess-e2b/README.i18n.yaml b/packages/e2b/subprocess-e2b/README.i18n.yaml index 091a6972cf..2d20e4ad4c 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: b1aa480305fef9ca485340b6bdae50f9fcffd88f -README.zh.md: b4824c90e7fb908d9c12e77c19d2b952055dcb40 +README.md: 03146e02cc76b78fbb93faec7d36d1865418b670 +README.zh.md: 6ed49a355586c6df2a594f83b687e6a565d07cf0 diff --git a/packages/e2b/subprocess-e2b/README.md b/packages/e2b/subprocess-e2b/README.md index b1aa480305..fb3d79433f 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. Before publication, cancellation uses the provisional SDK handle; if publication fails, rollback kills and verifies the provisional group before startup rejects. Service disposal terminates and joins every retained handle before the sandbox owner disposes. +- **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. Setup and teardown own the private state transaction, including failure cleanup. 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 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`. 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. diff --git a/packages/e2b/subprocess-e2b/README.zh.md b/packages/e2b/subprocess-e2b/README.zh.md index b4824c90e7..ec012b6a05 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（资源释放）会在沙箱所有者释放前终止并等待每个保留句柄退出。 +- **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 与清理后的环境，报告前台进程组，发送真实信号，并在结算前清理远程终端会话中的每个进程组。setup 与 teardown 负责私有状态事务，包括失败清理。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 +- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并在结算前清理远程终端会话中的每个进程组。私有随机输出边界会丢弃 E2B 引导 shell 的提示符和回显的 runner 命令，同时保留请求进程的每个字节，包括其第一个提示符。setup 与 teardown 负责私有状态事务，在服务 dispose 期间阻止发布，并包括失败清理。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。 基础 E2B 镜像提供该适配器调用的运行时和 Bash/GNU 工具：`node`、`bash`、`setsid`、`ps`、`awk`、`tr`、`env`、`chmod`、`tee`、`head`、`rm` 和 `kill`。自定义模板必须保留兼容的命令和 E2B PTY 支持。 diff --git a/packages/e2b/subprocess-e2b/src/index.ts b/packages/e2b/subprocess-e2b/src/index.ts index 27efb216da..fcf2c6f0a4 100644 --- a/packages/e2b/subprocess-e2b/src/index.ts +++ b/packages/e2b/subprocess-e2b/src/index.ts @@ -28,6 +28,8 @@ export class E2BSubprocessService extends SubprocessService { private readonly live = new Set() private readonly terminals = new Set() + private readonly terminalSetups = new Set>() + private disposing = false /** @inheritdoc */ readonly cwd: string @@ -41,12 +43,17 @@ export class E2BSubprocessService extends SubprocessService { this.cwd = ctx.e2b.cwd this.runtimeRoot = ctx.e2b.runtimeRoot ctx.effect(() => async () => { + this.disposing = true + await Promise.all([...this.terminalSetups]) const handles = [...this.live] const terminals = [...this.terminals] const pending: Promise[] = [] for (const handle of handles) { handle.terminate() - pending.push(handle.waitForExit().then(() => { this.live.delete(handle) })) + pending.push(handle.waitForExit().then(async () => { + await handle.done.catch(() => undefined) + this.live.delete(handle) + })) } for (const terminal of terminals) { terminal.terminate() @@ -89,6 +96,7 @@ export class E2BSubprocessService extends SubprocessService { /** @inheritdoc */ spawn(spec: SubprocessSpawnSpec): SubprocessHandle { + if (this.isDisposing()) throw new Error('subprocess-e2b: service is disposing') const program = spec.argv[0] if (program === undefined || program.length === 0) { throw new Error('invalid argv: expected a non-empty program name at argv[0]') @@ -112,6 +120,7 @@ export class E2BSubprocessService extends SubprocessService { /** @inheritdoc */ async spawnTerminal(spec: SubprocessTerminalSpawnSpec): Promise { + if (this.isDisposing()) throw new Error('subprocess-e2b: service is disposing') const program = spec.argv[0] if (program === undefined || program.length === 0) { throw new Error('subprocess-e2b: terminal argv must contain a program') @@ -123,14 +132,31 @@ export class E2BSubprocessService extends SubprocessService { } spec.signal?.throwIfAborted() const stateDir = posix.join(this.runtimeRoot, 'terminals', randomUUID()) - const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir) - this.terminals.add(terminal) - const release = async (): Promise => { - await terminal.waitForExit() - this.terminals.delete(terminal) + const setup = Promise.withResolvers() + this.terminalSetups.add(setup.promise) + try { + const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir) + this.terminals.add(terminal) + if (this.isDisposing()) { + terminal.terminate() + await terminal.waitForExit() + this.terminals.delete(terminal) + throw new Error('subprocess-e2b: service disposed during terminal setup') + } + const release = async (): Promise => { + await terminal.waitForExit() + this.terminals.delete(terminal) + } + void terminal.done.then(release, release).catch(() => {}) + return terminal + } finally { + this.terminalSetups.delete(setup.promise) + setup.resolve() } - void terminal.done.then(release, release).catch(() => {}) - return terminal + } + + private isDisposing(): boolean { + return this.disposing } } diff --git a/packages/e2b/subprocess-e2b/src/process.ts b/packages/e2b/subprocess-e2b/src/process.ts index 473c1a0af4..3038f312dc 100644 --- a/packages/e2b/subprocess-e2b/src/process.ts +++ b/packages/e2b/subprocess-e2b/src/process.ts @@ -213,6 +213,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { private invalidHandleQuiescent = false private provisionalHandleQuiescent = false private terminationStarted = false + private terminationSucceeded = false private terminationAttempt: Promise | undefined private terminationFailure: Error | undefined private terminationSignal: NodeJS.Signals | null = null @@ -264,13 +265,14 @@ export class E2BSubprocessHandle implements SubprocessHandle { /** @inheritdoc */ terminate(): void { - if (this.terminationAttempt !== undefined) return + if (this.terminationSucceeded || this.terminationAttempt !== undefined) return this.terminationStarted = true this.terminationFailure = undefined const attempt = this.terminateRemote() this.terminationAttempt = attempt void attempt.then( () => { + this.terminationSucceeded = true this.terminationAttempt = undefined }, (error: unknown) => { diff --git a/packages/e2b/subprocess-e2b/src/terminal.ts b/packages/e2b/subprocess-e2b/src/terminal.ts index fb8dac6f49..78668490d7 100644 --- a/packages/e2b/subprocess-e2b/src/terminal.ts +++ b/packages/e2b/subprocess-e2b/src/terminal.ts @@ -1,6 +1,7 @@ /** E2B PTY allocation and process-session ownership for the subprocess seam. */ import { Buffer } from 'node:buffer' +import { randomUUID } from 'node:crypto' import { PassThrough } from 'node:stream' import { posix } from 'node:path' import { @@ -28,11 +29,13 @@ const TERMINAL_RUNNER_SOURCE = [ 'dsh_state=$1', 'mapfile -d \'\' -t dsh_env < "$dsh_state/environment"', 'mapfile -d \'\' -t dsh_argv < "$dsh_state/argv"', - 'rm -f -- "$dsh_state/environment" "$dsh_state/argv" "$dsh_state/runner.bash"', + 'dsh_output_marker=$(<"$dsh_state/output-marker")', + 'rm -f -- "$dsh_state/environment" "$dsh_state/argv" "$dsh_state/output-marker" "$dsh_state/runner.bash"', 'if (( ${#dsh_argv[@]} == 0 )); then', " printf 'terminal runner received empty argv\\n' >&2", ' exit 125', 'fi', + 'printf \'%s\' "$dsh_output_marker"', "printf 'ready\\n' > \"$dsh_state/ready\"", 'exec env -i "${dsh_env[@]}" "${dsh_argv[@]}"', '', @@ -42,6 +45,7 @@ interface TerminalPaths { runner: string environment: string argv: string + outputMarker: string ready: string } @@ -53,6 +57,73 @@ function delay(ms: number): Promise { return new Promise(resolve => setTimeout(resolve, ms)) } +class BootstrapOutputFilter { + readonly ready: Promise + + private readonly readyState = Promise.withResolvers() + private pending = Buffer.alloc(0) + private published = false + + constructor( + private readonly marker: Buffer, + private readonly output: PassThrough, + ) { + this.ready = this.readyState.promise + } + + push(data: Uint8Array): void { + if (this.published) { + this.write(data) + return + } + const combined = Buffer.concat([this.pending, Buffer.from(data)]) + const markerOffset = combined.indexOf(this.marker) + if (markerOffset < 0) { + const retained = Math.min(combined.length, this.marker.length - 1) + this.pending = Buffer.from(combined.subarray(combined.length - retained)) + return + } + this.published = true + this.pending = Buffer.alloc(0) + this.readyState.resolve() + this.write(combined.subarray(markerOffset + this.marker.length)) + } + + private write(data: Uint8Array): void { + if (data.length > 0 && !this.output.destroyed) this.output.write(data) + } +} + +async function waitForBootstrapOutput( + ready: Promise, + completion: Promise, + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted() + await new Promise((resolve, reject) => { + let settled = false + let removeAbort: (() => void) | undefined + const finish = (complete: () => void): void => { + if (settled) return + settled = true + removeAbort?.() + complete() + } + const onExit = (): void => { + finish(() => { reject(new Error('subprocess-e2b: terminal exited before publishing its output boundary')) }) + } + if (signal !== undefined) { + const onAbort = (): void => { + finish(() => { reject(asError(signal.reason)) }) + } + signal.addEventListener('abort', onAbort, { once: true }) + removeAbort = () => { signal.removeEventListener('abort', onAbort) } + } + void ready.then(() => { finish(resolve) }) + void completion.then(onExit, onExit) + }) +} + function asError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)) } @@ -374,9 +445,12 @@ export async function spawnE2BTerminal( runner: posix.join(stateDir, 'runner.bash'), environment: posix.join(stateDir, 'environment'), argv: posix.join(stateDir, 'argv'), + outputMarker: posix.join(stateDir, 'output-marker'), ready: posix.join(stateDir, 'ready'), } + const outputMarker = Buffer.from(`dsh-e2b-bootstrap:${randomUUID()}`) const output = new PassThrough() + const outputFilter = new BootstrapOutputFilter(outputMarker, output) let handle: CommandHandle | undefined let completion: Promise | undefined let stateDirectoryCreated = false @@ -391,9 +465,10 @@ export async function spawnE2BTerminal( { path: paths.runner, data: TERMINAL_RUNNER_SOURCE }, { path: paths.environment, data: environment }, { path: paths.argv, data: argv }, + { path: paths.outputMarker, data: outputMarker.toString('utf8') }, ], signalOpts(spec.signal)) await sandbox.commands.run( - `chmod 600 -- ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(paths.environment)} ${quoteE2BShellArg(paths.argv)}`, + `chmod 600 -- ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(paths.environment)} ${quoteE2BShellArg(paths.argv)} ${quoteE2BShellArg(paths.outputMarker)}`, signalOpts(spec.signal), ) handle = await sandbox.pty.create({ @@ -403,7 +478,7 @@ export async function spawnE2BTerminal( envs: { TERM: 'dumb' }, timeoutMs: 0, ...signalOpts(spec.signal), - onData: (data) => { if (!output.destroyed) output.write(Buffer.from(data)) }, + onData: (data) => { outputFilter.push(data) }, }) completion = handle.wait() void completion.catch(() => {}) @@ -413,6 +488,7 @@ export async function spawnE2BTerminal( const command = `exec /bin/bash ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(stateDir)}\r` await sandbox.pty.sendInput(handle.pid, Buffer.from(command), signalOpts(spec.signal)) await waitUntilReady(sandbox, paths, completion, spec.signal) + await waitForBootstrapOutput(outputFilter.ready, completion, spec.signal) const sessionId = await terminalSessionId(sandbox, handle.pid, spec.signal) return new E2BTerminalHandle( sandbox, diff --git a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts index 457e68f24d..be8850d905 100644 --- a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts @@ -599,6 +599,12 @@ describe('E2BSubprocessHandle', () => { await expect(handle.waitForExit()).resolves.toBe(true) expect(fake.commandsSeen).toContain('kill -TERM -- -4242') expect(fake.commandsSeen).not.toContain('kill -KILL -- -4242') + 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('escalates a TERM-trapping process group to KILL and uses the SDK kill as fallback', async () => { @@ -1123,6 +1129,24 @@ describe('E2BSubprocessService', () => { expect(fake.alive).toBe(false) }) + it('awaits SDK settlement after the remote process group becomes quiescent', async () => { + const fake = new FakeSandbox() + fake.trapsTerm = true + const { ctx, fiber } = await service(fake) + const handle = ctx.subprocess.spawn(spec()) + await flush() + fake.alive = false + + let disposed = false + const disposing = fiber.dispose().then(() => { disposed = true }) + await flush() + expect(disposed).toBe(false) + + fake.finish() + await disposing + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + }) + it('reports a failed termination transaction from disposal instead of waiting on done', async () => { const fake = new FakeSandbox() fake.signalErrors.push(new Error('TERM transport failed'), new Error('KILL transport failed')) @@ -1178,6 +1202,7 @@ describe('E2BSubprocessService', () => { const { ctx, fiber } = await service(fake) const handle = ctx.subprocess.spawn(spec()) const disposing = fiber.dispose() + expect(() => ctx.subprocess.spawn(spec())).toThrow('service is disposing') fake.releaseStart() await expect(disposing).resolves.toBeUndefined() await expect(handle.done).rejects.toThrow('start failed during disposal') diff --git a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts index bda01a5a61..1eac9a5780 100644 --- a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts @@ -2,7 +2,7 @@ import { Buffer } from 'node:buffer' import { once } from 'node:events' import { PassThrough } from 'node:stream' import { Context } from 'cordis' -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { CommandExitError, FileNotFoundError, @@ -91,6 +91,7 @@ class FakeTerminalSandbox { ambient = 'KEEP=visible\0NPM_TOKEN=secret\0DSH_STALE=old\0BROKEN\0=bad\0' ready: string | Error = 'ready\n' readyMisses = 0 + readyReads = 0 sessionId = '123\n' foreground = '456\n' groups = [123] @@ -108,6 +109,20 @@ class FakeTerminalSandbox { settleOnPtyKill = true ptyKills = 0 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?.() + } readonly sandbox = { files: { @@ -121,6 +136,7 @@ class FakeTerminalSandbox { return files.map(() => ({})) }, read: async (): Promise => { + this.readyReads += 1 if (this.readyMisses > 0) { this.readyMisses -= 1 throw new FileNotFoundError('not ready') @@ -170,6 +186,7 @@ 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() @@ -178,6 +195,17 @@ class FakeTerminalSandbox { options?.signal?.throwIfAborted() this.inputs.push({ pid, data: Buffer.from(data) }) if (this.sendError !== undefined) throw this.sendError + if (this.emitOutputMarker && Buffer.from(data).includes(Buffer.from('runner.bash'))) { + const marker = [...this.writes].find(([path]) => path.endsWith('/output-marker'))?.[1] + const onData = this.createOptions?.onData + if (marker !== undefined && onData !== undefined) { + await onData(Buffer.from(Buffer.from(data).toString().replace(/\r$/, '\r\n'))) + const split = Math.floor(marker.length / 2) + await onData(Buffer.from(marker.slice(0, split))) + await onData(Buffer.from(marker.slice(split))) + await onData(Buffer.from(this.requestedOutput)) + } + } }, kill: async (pid: number): Promise => { this.ptyKills += 1 @@ -211,7 +239,7 @@ function spec(overrides: Partial = {}): SubprocessT } describe('E2B terminal allocation', () => { - it('boots the requested argv through a private runner and preserves buffered bytes', async () => { + it('hides bootstrap-shell bytes and preserves requested-shell bytes across the output boundary', async () => { const fake = new FakeTerminalSandbox() fake.readyMisses = 1 const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/terminal-one') @@ -219,7 +247,9 @@ describe('E2B terminal allocation', () => { terminal.output.on('data', (chunk) => { output += String(chunk) }) await new Promise(resolve => setTimeout(resolve, 0)) - expect(output).toBe('buffered banner\n') + expect(output).toBe('requested-shell$ ') + expect(output).not.toContain('buffered banner') + expect(output).not.toContain('runner.bash') expect(fake.createOptions).toMatchObject({ rows: 24, cols: 80, cwd: '/workspace', timeoutMs: 0, envs: { TERM: 'dumb' } }) expect(fake.inputs[0]?.data.toString()).toContain("exec /bin/bash '/runtime/terminal-one/runner.bash'") expect(fake.writes.get('/runtime/terminal-one/environment')).toContain('KEEP=visible\0') @@ -227,10 +257,17 @@ describe('E2B terminal allocation', () => { expect(fake.writes.get('/runtime/terminal-one/environment')).not.toContain('secret') expect(fake.writes.get('/runtime/terminal-one/environment')).not.toContain('DSH_STALE') expect(fake.writes.get('/runtime/terminal-one/argv')).toBe('/bin/bash\0--noprofile\0--norc\0') + const marker = fake.writes.get('/runtime/terminal-one/output-marker') ?? '' + expect(marker).toMatch(/^dsh-e2b-bootstrap:/) + expect(fake.inputs[0]?.data.toString()).not.toContain(marker) const runner = fake.writes.get('/runtime/terminal-one/runner.bash') ?? '' expect(runner).toContain('if (( ${#dsh_argv[@]} == 0 )); then') + expect(runner).toContain('printf \'%s\' "$dsh_output_marker"') expect(runner).toContain('exec env -i "${dsh_env[@]}" "${dsh_argv[@]}"') expect(runner).not.toContain('\u007f') + terminal.output.destroy() + await fake.createOptions?.onData(Buffer.from('late bootstrap callback')) + expect(output).toBe('requested-shell$ ') await terminal.write(Buffer.from('echo ok\r')) expect(fake.inputs.at(-1)?.data.toString()).toBe('echo ok\r') @@ -395,6 +432,29 @@ describe('E2B terminal allocation', () => { await expect(spawnE2BTerminal(runtime(readFailed), spec(), '/runtime/read')) .rejects.toThrow('ready transport failed') }) + + it('bounds a missing bootstrap-output boundary by process exit or cancellation', async () => { + const exited = new FakeTerminalSandbox() + exited.emitOutputMarker = false + const exiting = spawnE2BTerminal(runtime(exited), spec(), '/runtime/missing-output-boundary') + await vi.waitFor(() => { expect(exited.inputs).toHaveLength(1) }) + exited.handle.succeed(0) + await expect(exiting).rejects.toThrow('terminal exited before publishing its output boundary') + + const cancelled = new FakeTerminalSandbox() + cancelled.emitOutputMarker = false + const controller = new AbortController() + const cancelling = spawnE2BTerminal( + runtime(cancelled), + spec({ signal: controller.signal }), + '/runtime/cancel-output-boundary', + ) + await vi.waitFor(() => { expect(cancelled.inputs).toHaveLength(1) }) + await vi.waitFor(() => { expect(cancelled.readyReads).toBeGreaterThan(0) }) + await new Promise(resolve => setTimeout(resolve, 0)) + controller.abort(new Error('cancel output boundary')) + await expect(cancelling).rejects.toThrow('cancel output boundary') + }) }) describe('E2B terminal lifecycle', () => { @@ -637,6 +697,29 @@ describe('E2B subprocess terminal service', () => { expect(fake.handle.disconnects).toBe(1) }) + it('joins and rejects terminal setup that completes during service disposal', async () => { + const fake = new FakeTerminalSandbox() + fake.deferCreate() + 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.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 expect(subprocess.spawnTerminal(spec())).rejects.toThrow('service is disposing') + fake.releaseCreate() + + await rejected + await disposing + expect(fake.groups).toEqual([]) + expect(fake.handle.disconnects).toBe(1) + expect(fake.removed.some(path => path.includes('/terminals/'))).toBe(true) + }) + it('releases naturally settled terminals and validates terminal requests', async () => { const { ctx, fiber, fake } = await service() for (const request of [