refactor(e2b): compose portable runtime consumers

2026-07-29 02:51:44 +08:00
parent 97496c3d00
commit 2abc9823e7
64 changed files with 1426 additions and 4229 deletions
--- 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: 066d35a099f560fe40fe629ede632c37535129f7
-README.zh.md: d3af14e8cc58c7d86331156358f78a0abbb24c39
+README.md: e0805345e708c67c6de721d95641e2366a7f990e
+README.zh.md: f83969c794e50f0658aa86218a0f5f3c89d53b56
--- a/packages/e2b/subprocess-e2b/README.md
+++ b/packages/e2b/subprocess-e2b/README.md
@@ -2,16 +2,18 @@

 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 consumers such as [`dsh-bash-local`](../../bash/bash-local/README.md) then execute in the shared remote sandbox without an E2B-specific Bash adapter.
+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, LSP, and subprocess Code Runtime consumers then execute in the shared remote sandbox without E2B-specific capability packages.

 ## 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; `done`, stdin, termination, and `waitForExit()` wait for readiness internally.
+- **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. If publication fails, the SDK PID remains the provisional `exec setsid` group id; rollback kills and verifies that group before startup rejects. Service disposal terminates and joins every retained handle before the sandbox owner disposes.
 - **Environment boundary** — the wrapper starts from the sandbox command environment, removes ambient `DSH_*` and credential-shaped (`*KEY*`, `*SECRET*`, `*TOKEN*`) names, then restores every `spec.env` entry as an explicit caller opt-in. Host ambient variables never enter the sandbox implicitly.
 - **Stdio projection** — pipe mode forwards E2B callbacks into host Node streams; inherit mode forwards them to the harness process streams; collect mode retains a bounded host tail with offset reads. Optional complete spill files are written remotely and advertised only while within their cap. Batch and streaming stdin use the SDK handle.
+- **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. Prompt detection, scrollback, readiness, and owner policy remain in `dsh-pty-local`.

-The base E2B image supplies the Bash/GNU utilities this adapter invokes: `bash`, `setsid`, `ps`, `tr`, `env`, `chmod`, `tee`, `head`, and `kill`. A custom template must retain compatible commands.
+The base E2B image supplies the Bash/GNU utilities this adapter invokes: `bash`, `setsid`, `ps`, `awk`, `tr`, `env`, `chmod`, `tee`, `head`, and `kill`. A custom template must retain compatible commands and E2B PTY support.

 ## Model Experience

@@ -24,9 +26,10 @@ No direct invalidation; the named consumers own any request-prefix changes.
 ## Known Limitations and Deferred Work

 - **The SDK still retains complete command output in host memory** — E2B `CommandHandle.stdout` and `.stderr` accumulate even when this adapter exposes bounded tails, so the subprocess seam's normal host-memory bound is not achieved.
- **Pipe output is not byte-faithful** — E2B delivers separately decoded strings rather than raw bytes, so split multibyte sequences and arbitrary binary protocols can be corrupted; LSP and other framed byte-stream consumers are unsupported.
+- **Command-pipe output is text-decoded by the SDK** — valid UTF-8 protocol traffic, including the exercised LSP composition and Code Runtime's ASCII/base64 frames, is supported; arbitrary binary protocols and invalid UTF-8 are not byte-faithful.
 - **Synchronous-PID consumers are unsupported** — `pid` remains `-1` during remote startup; consumers that require a positive PID immediately, including the ACP child backend, cannot use this provider unchanged.
 - **Reconnect does not reconstruct handles** — remote PID/status/spill files survive a retained sandbox, but a new harness process does not rebuild live `SubprocessHandle` objects or output cursors from them.
 - **Remote state accumulates when retained** — process directories and valid spill files remain under `.dsh-e2b`; this POC supplies no retention sweep.
 - **Signal attribution is inferred** — when termination was requested and E2B reports a nonzero exit code, the adapter reports the last requested signal because the SDK result does not identify the terminating signal.
- **Linux utility and E2B transport semantics are assumed** — there is no PTY, Windows, arbitrary-template, or network-partition fidelity layer.
+- **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.
--- a/packages/e2b/subprocess-e2b/README.zh.md
+++ b/packages/e2b/subprocess-e2b/README.zh.md
@@ -2,16 +2,18 @@

 [English](README.md) | 中文

-[`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam 的 E2B 实现。它没有配置：先加载 [`@deepseek-ai/dsh-e2b`](../e2b/README.md)，再用本服务取代 `dsh-subprocess-local`。随后，[`dsh-bash-local`](../../bash/bash-local/README.md) 等现有消费方会在共享远程沙箱中执行，无需 E2B 专用 Bash 适配器。
+[`@deepseek-ai/dsh-subprocess`](../../subprocess/subprocess/README.md) seam 的 E2B 实现。它没有配置：先加载 [`@deepseek-ai/dsh-e2b`](../e2b/README.md)，再用本服务取代 `dsh-subprocess-local`。现有的 Bash、PTY、LSP 以及使用 subprocess 的 Code Runtime 消费方随后会在共享远程沙箱中执行，无需 E2B 专用的功能包（package）。

 ## 行为

 - **异步远程启动**：同步 seam 会立即返回一个句柄，同时由 `Sandbox.commands.run(..., { background: true })` 在远程启动进程。包装层发布进程组 ID 并由适配器完成验证之前，`pid` 为 `-1`；`done`、stdin、终止和 `waitForExit()` 会在内部等待就绪。
+- **执行世界坐标**：`cwd` 和私有 `runtimeRoot` 来自共享所有者；可执行文件查找会验证绝对路径，或根据沙箱 PATH 加显式覆盖来解析裸名称。
 - **Linux 进程组**：带引号保护的包装层会在 `exec setsid --wait` 下启动每组 argv，并在 `ctx.e2b.runtimeRoot/processes` 下记录实际进程组 ID 和私有状态文件。句柄会等待该文件，而不会把 SDK 命令 PID 当作已发布的身份。终止操作以记录的负数 ID 发送 `SIGTERM`，等待调用方的 `graceMs`，再升级到 `SIGKILL` 和 SDK kill 回退。如果发布失败，SDK PID 仍为临时的 `exec setsid` 进程组 ID；回滚会终止并验证该进程组，随后启动操作才会以拒绝结束。服务 dispose（资源释放）会在沙箱所有者释放前终止并等待每个保留句柄退出。
 - **环境边界**：包装层从沙箱命令环境开始，移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），再把每个 `spec.env` 条目恢复为调用方显式选择。宿主环境变量绝不会隐式进入沙箱。
 - **stdio 投影**：pipe 模式把 E2B 回调转发到宿主 Node 流；inherit 模式把回调转发到 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。可选的完整 spill 文件写在远程，并且只有未超过其上限时才会对外公布。批量 stdin 和流式 stdin 都使用 SDK 句柄。
+- **终端会话**：`spawnTerminal()` 使用 E2B 的字节 PTY API，以 mode 为 `0600` 的私有文件传入原样 argv 与清理后的环境，报告前台进程组，发送真实信号，并在结算前清理远程终端会话中的每个进程组。提示符检测、scrollback、就绪状态与所有者策略仍归 `dsh-pty-local` 所有。

-基础 E2B 镜像提供该适配器调用的 Bash/GNU 工具：`bash`、`setsid`、`ps`、`tr`、`env`、`chmod`、`tee`、`head` 和 `kill`。自定义模板必须保留兼容的命令。
+基础 E2B 镜像提供该适配器调用的 Bash/GNU 工具：`bash`、`setsid`、`ps`、`awk`、`tr`、`env`、`chmod`、`tee`、`head` 和 `kill`。自定义模板必须保留兼容的命令和 E2B PTY 支持。

 ## 模型体验

@@ -24,9 +26,10 @@
 ## 已知限制与延后工作

 - **SDK 仍会在宿主内存中保留完整命令输出**：即使本适配器公开的是有界尾部，E2B `CommandHandle.stdout` 和 `.stderr` 仍会持续累积，因此无法达到进程管理 seam 通常提供的宿主内存边界。
- **Pipe 输出并非字节保真**：E2B 交付的是分别解码后的字符串，而不是原始字节，因此拆分的多字节序列和任意二进制协议可能损坏；不支持 LSP 及其他带帧字节流消费方。
+- **命令管道输出由 SDK 解码为文本**：支持有效的 UTF-8 协议流量，包括已经过测试的 LSP 组合与 Code Runtime 的 ASCII/base64 帧；任意二进制协议和无效 UTF-8 不具备字节保真。
 - **不支持需要同步 PID 的消费方**：远程启动期间，`pid` 保持为 `-1`；包括 ACP 子进程后端在内，要求立即获得正 PID 的消费方无法原样使用本提供方。
 - **重新连接不会重建句柄**：保留沙箱后，远程 PID／状态／spill 文件仍然存在，但新的 harness 进程不会据此重建实时 `SubprocessHandle` 对象或输出游标。
 - **保留沙箱时会累积远程状态**：进程目录和有效的 spill 文件会留在 `.dsh-e2b` 下；本 POC 不提供保留清理。
 - **信号归因依靠推断**：如果已经请求终止，而 E2B 报告非零退出码，适配器会报告最后请求的信号，因为 SDK 结果不标识终止信号。
- **依赖 Linux 工具与 E2B 传输语义**：没有 PTY、Windows、任意模板或网络分区的保真层。
+- **无法精确检查终端 stdin 等待状态**：E2B 会公开前台进程组，但不提供证明其正在等待 fd 0 所需的 syscall 证据，因此通用 PTY 后端会回退到受控提示符标记与有界静默机制。
+- **依赖 Linux 工具与 E2B 传输语义**：没有 Windows、任意模板、逃逸会话恢复或网络分区的保真层。
--- a/packages/e2b/subprocess-e2b/src/index.ts
+++ b/packages/e2b/subprocess-e2b/src/index.ts
@@ -8,29 +8,87 @@ import { randomUUID } from 'node:crypto'
 import { posix } from 'node:path'
 import { Context } from 'cordis'
 import { SubprocessService } from '@deepseek-ai/dsh-subprocess'
-import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import type {
+  SubprocessHandle,
+  SubprocessSpawnSpec,
+  SubprocessTerminalHandle,
+  SubprocessTerminalSpawnSpec,
+} from '@deepseek-ai/dsh-subprocess'
+import { quoteE2BShellArg } from '@deepseek-ai/dsh-e2b'
 import { E2BSubprocessHandle } from './process.ts'
+import { spawnE2BTerminal } from './terminal.ts'
+
+function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } {
+  return signal === undefined ? {} : { signal }
+}

 /** E2B command manager registered as `ctx.subprocess`. */
 export class E2BSubprocessService extends SubprocessService {
  static inject = ['e2b']

  private readonly live = new Set<E2BSubprocessHandle>()
+  private readonly terminals = new Set<SubprocessTerminalHandle>()
+
+  /** @inheritdoc */
+  readonly cwd: string
+
+  /** @inheritdoc */
+  readonly runtimeRoot: string

  /** Create the E2B subprocess service and bind its disposal policy. */
  constructor(ctx: Context) {
    super(ctx)
+    this.cwd = ctx.e2b.cwd
+    this.runtimeRoot = ctx.e2b.runtimeRoot
    ctx.effect(() => async () => {
      const handles = [...this.live]
-      for (const handle of handles) handle.terminate()
-      await Promise.all(handles.map(async (handle) => {
-        await handle.done.catch(() => {})
-        await handle.waitForExit()
-      }))
+      const terminals = [...this.terminals]
+      const pending: Promise<unknown>[] = []
+      for (const handle of handles) {
+        handle.terminate()
+        pending.push(handle.done.catch(() => {}).then(() => handle.waitForExit()))
+      }
+      for (const terminal of terminals) {
+        terminal.terminate()
+        pending.push(terminal.waitForExit())
+      }
      this.live.clear()
+      this.terminals.clear()
+      await Promise.all(pending)
    }, 'e2b subprocess teardown')
  }

+  /** @inheritdoc */
+  async resolveExecutable(
+    command: string,
+    env?: Readonly<Record<string, string>>,
+    signal?: AbortSignal,
+  ): Promise<string> {
+    if (command.length === 0) throw new Error('subprocess-e2b: executable name must be non-empty')
+    signal?.throwIfAborted()
+    const sandbox = await this.ctx.e2b.getSandbox()
+    if (posix.isAbsolute(command)) {
+      await sandbox.commands.run(
+        `test -f ${quoteE2BShellArg(command)} -a -x ${quoteE2BShellArg(command)}`,
+        signalOpts(signal),
+      )
+      signal?.throwIfAborted()
+      return command
+    }
+    const path = env?.PATH
+    const prefix = path === undefined ? '' : `PATH=${quoteE2BShellArg(path)} `
+    const result = await sandbox.commands.run(
+      `${prefix}command -v -- ${quoteE2BShellArg(command)}`,
+      signalOpts(signal),
+    )
+    signal?.throwIfAborted()
+    const executable = result.stdout.trim()
+    if (!posix.isAbsolute(executable) || executable.includes('\n')) {
+      throw new Error(`subprocess-e2b: executable ${JSON.stringify(command)} did not resolve to one absolute path`)
+    }
+    return executable
+  }
+
  /** @inheritdoc */
  spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
    const program = spec.argv[0]
@@ -53,6 +111,29 @@ export class E2BSubprocessService extends SubprocessService {
    void handle.done.then(release, release).catch(() => {})
    return handle
  }
+
+  /** @inheritdoc */
+  async spawnTerminal(spec: SubprocessTerminalSpawnSpec): Promise<SubprocessTerminalHandle> {
+    const program = spec.argv[0]
+    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.runtimeRoot, 'terminals', randomUUID())
+    const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir)
+    this.terminals.add(terminal)
+    const release = async (): Promise<void> => {
+      await terminal.waitForExit()
+      this.terminals.delete(terminal)
+    }
+    void terminal.done.then(release, release).catch(() => {})
+    return terminal
+  }
 }

 export default E2BSubprocessService
--- a/packages/e2b/subprocess-e2b/src/terminal.ts
+++ b/packages/e2b/subprocess-e2b/src/terminal.ts
@@ -0,0 +1,386 @@
+/** E2B PTY allocation and process-session ownership for the subprocess seam. */
+
+import { Buffer } from 'node:buffer'
+import { constants } from 'node:os'
+import { PassThrough } from 'node:stream'
+import { posix } from 'node:path'
+import {
+  CommandExitError,
+  FileNotFoundError,
+  quoteE2BShellArg,
+} from '@deepseek-ai/dsh-e2b'
+import type { CommandHandle, CommandResult, Sandbox } from '@deepseek-ai/dsh-e2b'
+import { SENSITIVE_ENV_PATTERN } from '@deepseek-ai/dsh-subprocess'
+import type {
+  SubprocessOutcome,
+  SubprocessTerminalForeground,
+  SubprocessTerminalHandle,
+  SubprocessTerminalSignal,
+  SubprocessTerminalSpawnSpec,
+} from '@deepseek-ai/dsh-subprocess'
+import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
+
+const POLL_MS = 20
+
+const TERMINAL_RUNNER_SOURCE = [
+  '#!/bin/bash',
+  'set -euo pipefail',
+  '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"',
+  'if (( ${#dsh_argv[@]} == 0 )); then',
+  "  printf 'terminal runner received empty argv\\n' >&2",
+  '  exit 125',
+  'fi',
+  "printf 'ready\\n' > \"$dsh_state/ready\"",
+  'exec env -i "${dsh_env[@]}" "${dsh_argv[@]}"',
+  '',
+].join('\n')
+
+interface TerminalPaths {
+  runner: string
+  environment: string
+  argv: string
+  ready: string
+}
+
+function signalOpts(signal: AbortSignal | undefined): { signal?: AbortSignal } {
+  return signal === undefined ? {} : { signal }
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise(resolve => setTimeout(resolve, ms))
+}
+
+function commandSignal(exitCode: number): NodeJS.Signals | null {
+  const number = exitCode - 128
+  if (number <= 0) return null
+  for (const [name, value] of Object.entries(constants.signals)) {
+    if (value === number) return name as NodeJS.Signals
+  }
+  return null
+}
+
+function parsePositiveId(value: string, message: string): number {
+  const raw = value.trim()
+  const id = Number(raw)
+  if (!/^[1-9][0-9]*$/.test(raw) || !Number.isSafeInteger(id)) throw new Error(message)
+  return id
+}
+
+function serializeValues(values: readonly string[], kind: string): string {
+  for (const value of values) {
+    if (value.includes('\0')) throw new Error(`subprocess-e2b: terminal ${kind} must not contain NUL bytes`)
+  }
+  return values.map(value => `${value}\0`).join('')
+}
+
+function remoteEnvironment(raw: string, explicit: Readonly<Record<string, string>> | undefined): string {
+  const environment = new Map<string, string>()
+  for (const entry of raw.split('\0')) {
+    if (entry.length === 0) continue
+    const separator = entry.indexOf('=')
+    if (separator <= 0) continue
+    const name = entry.slice(0, separator)
+    if (name.startsWith('DSH_') || SENSITIVE_ENV_PATTERN.test(name)) continue
+    environment.set(name, entry.slice(separator + 1))
+  }
+  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<number> {
+  const result = await sandbox.commands.run(`ps -o sid= -p ${pid}`, signalOpts(signal))
+  signal?.throwIfAborted()
+  return parsePositiveId(result.stdout, `subprocess-e2b: cannot resolve process session for terminal ${pid}`)
+}
+
+async function waitUntilReady(
+  sandbox: Sandbox,
+  paths: TerminalPaths,
+  completion: Promise<CommandResult>,
+  signal?: AbortSignal,
+): Promise<void> {
+  const settled = completion.then(() => true, () => true)
+  for (;;) {
+    signal?.throwIfAborted()
+    try {
+      if ((await sandbox.files.read(paths.ready, signalOpts(signal))).trim() === 'ready') return
+    } catch (error: unknown) {
+      if (!(error instanceof FileNotFoundError)) throw error
+    }
+    if (await Promise.race([settled, delay(POLL_MS).then(() => false)])) {
+      throw new Error('subprocess-e2b: terminal exited before publishing readiness')
+    }
+  }
+}
+
+/** One E2B PTY and all process groups in its remote process session. */
+export class E2BTerminalHandle implements SubprocessTerminalHandle {
+  readonly pid: number
+  readonly done: Promise<SubprocessOutcome>
+
+  private topLevelExited = false
+  private termination: Promise<void> | undefined
+  private terminationSignal: NodeJS.Signals | null = null
+  private removeAbort: (() => void) | undefined
+
+  constructor(
+    private readonly sandbox: Sandbox,
+    private readonly handle: CommandHandle,
+    readonly output: PassThrough,
+    private readonly completion: Promise<CommandResult>,
+    private readonly sessionId: number,
+    private readonly stateDir: string,
+    private readonly graceMs: number,
+    signal?: AbortSignal,
+  ) {
+    this.pid = handle.pid
+    this.done = this.waitForCommand()
+    void this.done.then(() => { this.terminate() }, () => { this.terminate() })
+    if (signal !== undefined) {
+      const onAbort = (): void => { this.terminate() }
+      signal.addEventListener('abort', onAbort, { once: true })
+      this.removeAbort = () => { signal.removeEventListener('abort', onAbort) }
+      if (signal.aborted) this.terminate()
+    }
+  }
+
+  /** @inheritdoc */
+  async write(data: Uint8Array): Promise<void> {
+    if (this.topLevelExited) throw new Error('terminal process has exited')
+    await this.sandbox.pty.sendInput(this.pid, data)
+  }
+
+  /** @inheritdoc */
+  async inspectForeground(): Promise<SubprocessTerminalForeground | undefined> {
+    try {
+      const result = await this.sandbox.commands.run(`ps -o tpgid= -p ${this.pid}`)
+      return {
+        processGroupId: parsePositiveId(
+          result.stdout,
+          `subprocess-e2b: cannot resolve foreground process group for terminal ${this.pid}`,
+        ),
+        // E2B exposes process-table commands but not the /proc memory access
+        // needed to prove a specific syscall is waiting on fd 0.
+        inputWaiting: false,
+      }
+    } catch (error: unknown) {
+      if (error instanceof CommandExitError && this.topLevelExited) return undefined
+      throw error
+    }
+  }
+
+  /** @inheritdoc */
+  async signalForeground(signal: SubprocessTerminalSignal): Promise<number> {
+    const foreground = await this.inspectForeground()
+    if (foreground === undefined) {
+      throw new Error(`subprocess-e2b: cannot resolve foreground process group for terminal ${this.pid}`)
+    }
+    if (signal === 'SIGKILL' && foreground.processGroupId === this.pid) {
+      throw new Error('refusing to SIGKILL the terminal shell; terminate the terminal session instead')
+    }
+    await this.sandbox.commands.run(`kill -${signal.slice(3)} -- -${foreground.processGroupId}`)
+    return foreground.processGroupId
+  }
+
+  /** @inheritdoc */
+  terminate(): void {
+    this.termination ??= this.closeOnce().catch((error: unknown) => {
+      this.termination = undefined
+      throw error
+    })
+    void this.termination.catch(() => {})
+  }
+
+  /** @inheritdoc */
+  async waitForExit(signal?: AbortSignal): Promise<boolean> {
+    const quiescence = this.termination ?? this.done.then(
+      () => { this.terminate(); return this.termination },
+      () => { this.terminate(); return this.termination },
+    )
+    if (signal === undefined) {
+      await quiescence
+      return true
+    }
+    if (signal.aborted) return false
+    return await new Promise<boolean>((resolve, reject) => {
+      const onAbort = (): void => { cleanup(); resolve(false) }
+      const cleanup = (): void => { signal.removeEventListener('abort', onAbort) }
+      signal.addEventListener('abort', onAbort, { once: true })
+      void quiescence.then(
+        () => { cleanup(); resolve(true) },
+        (error: unknown) => {
+          cleanup()
+          reject(error instanceof Error ? error : new Error(String(error)))
+        },
+      )
+    })
+  }
+
+  private async waitForCommand(): Promise<SubprocessOutcome> {
+    try {
+      const result = await this.completion
+      return { exitCode: result.exitCode, signal: null }
+    } catch (error: unknown) {
+      if (error instanceof CommandExitError) {
+        const signal = this.terminationSignal ?? commandSignal(error.exitCode)
+        return signal === null ? { exitCode: error.exitCode, signal: null } : { exitCode: null, signal }
+      }
+      this.output.destroy(error instanceof Error ? error : new Error(String(error)))
+      throw error
+    } finally {
+      this.topLevelExited = true
+      if (!this.output.destroyed) this.output.end()
+    }
+  }
+
+  private async sessionProcessGroups(): Promise<number[]> {
+    const result = await this.sandbox.commands.run(
+      `ps -eo sid=,pgid= | awk '$1 == ${this.sessionId} { print $2 }'`,
+    )
+    const groups = new Set<number>()
+    for (const raw of result.stdout.trim().split(/\s+/)) {
+      if (raw.length === 0) continue
+      const group = parsePositiveId(
+        raw,
+        `subprocess-e2b: invalid process group ${JSON.stringify(raw)} in terminal session ${this.sessionId}`,
+      )
+      if (group <= 1) {
+        throw new Error(`subprocess-e2b: unsafe process group ${group} in terminal session ${this.sessionId}`)
+      }
+      groups.add(group)
+    }
+    return [...groups]
+  }
+
+  private async signalGroups(groups: number[], signal: 'TERM' | 'KILL'): Promise<void> {
+    if (groups.length === 0) return
+    try {
+      await this.sandbox.commands.run(`kill -${signal} -- ${groups.map(group => `-${group}`).join(' ')}`)
+    } catch (error: unknown) {
+      if (!(error instanceof CommandExitError)) throw error
+    }
+  }
+
+  private async awaitSessionEmpty(kill = false): Promise<number[]> {
+    const deadline = Date.now() + this.graceMs
+    for (;;) {
+      const groups = await this.sessionProcessGroups()
+      if (groups.length === 0 || Date.now() >= deadline) return groups
+      if (kill) await this.signalGroups(groups, 'KILL')
+      await delay(Math.min(POLL_MS, Math.max(1, deadline - Date.now())))
+    }
+  }
+
+  private async closeOnce(): Promise<void> {
+    let groups = await this.sessionProcessGroups()
+    if (groups.length > 0) {
+      this.terminationSignal = 'SIGTERM'
+      await this.signalGroups(groups, 'TERM')
+      groups = await this.awaitSessionEmpty()
+    }
+    if (groups.length === 0 && !this.topLevelExited) {
+      await Promise.race([this.done.catch(() => undefined), delay(this.graceMs)])
+    }
+    if (groups.length > 0 || !this.topLevelExited) {
+      this.terminationSignal = 'SIGKILL'
+      if (!this.topLevelExited) await this.sandbox.pty.kill(this.pid)
+      groups = await this.awaitSessionEmpty(true)
+      if (!this.topLevelExited) await Promise.race([this.done.catch(() => undefined), delay(this.graceMs)])
+    }
+    if (groups.length > 0) {
+      throw new Error(`subprocess-e2b: terminal cleanup failed; surviving process groups: ${groups.join(', ')}`)
+    }
+    if (!this.topLevelExited) {
+      throw new Error(`subprocess-e2b: terminal cleanup failed; surviving pid: ${this.pid}`)
+    }
+    this.removeAbort?.()
+    this.removeAbort = undefined
+    await this.handle.disconnect()
+    await this.sandbox.files.remove(this.stateDir).catch(() => {})
+  }
+}
+
+/**
+ * Allocate an E2B PTY, replace its bootstrap shell with the requested argv,
+ * and return only after the private runner has published readiness.
+ * @param runtime - Shared E2B sandbox owner.
+ * @param spec - Fully specified terminal-process request.
+ * @param stateDir - Private remote directory for one startup transaction.
+ * @returns The live subprocess terminal handle.
+ */
+export async function spawnE2BTerminal(
+  runtime: E2BSandboxService,
+  spec: SubprocessTerminalSpawnSpec,
+  stateDir: string,
+): Promise<E2BTerminalHandle> {
+  const sandbox = await runtime.getSandbox()
+  spec.signal?.throwIfAborted()
+  const paths: TerminalPaths = {
+    runner: posix.join(stateDir, 'runner.bash'),
+    environment: posix.join(stateDir, 'environment'),
+    argv: posix.join(stateDir, 'argv'),
+    ready: posix.join(stateDir, 'ready'),
+  }
+  const ambient = await sandbox.commands.run('env -0', signalOpts(spec.signal))
+  const environment = remoteEnvironment(ambient.stdout, spec.env)
+  const argv = serializeValues(spec.argv, 'argv')
+  await sandbox.files.makeDir(stateDir)
+  await sandbox.commands.run(`chmod 700 -- ${quoteE2BShellArg(stateDir)}`, signalOpts(spec.signal))
+  await sandbox.files.write([
+    { path: paths.runner, data: TERMINAL_RUNNER_SOURCE },
+    { path: paths.environment, data: environment },
+    { path: paths.argv, data: argv },
+  ], signalOpts(spec.signal))
+  await sandbox.commands.run(
+    `chmod 600 -- ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(paths.environment)} ${quoteE2BShellArg(paths.argv)}`,
+    signalOpts(spec.signal),
+  )
+
+  const output = new PassThrough()
+  let handle: CommandHandle | undefined
+  let completion: Promise<CommandResult> | undefined
+  try {
+    handle = await sandbox.pty.create({
+      rows: spec.rows,
+      cols: spec.cols,
+      cwd: spec.cwd,
+      envs: { TERM: 'dumb' },
+      timeoutMs: 0,
+      ...signalOpts(spec.signal),
+      onData: (data) => { if (!output.destroyed) output.write(Buffer.from(data)) },
+    })
+    completion = handle.wait()
+    void completion.catch(() => {})
+    if (!Number.isSafeInteger(handle.pid) || handle.pid <= 0) {
+      throw new Error(`subprocess-e2b: E2B returned invalid terminal pid ${handle.pid}`)
+    }
+    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)
+    const sessionId = await terminalSessionId(sandbox, handle.pid, spec.signal)
+    return new E2BTerminalHandle(
+      sandbox,
+      handle,
+      output,
+      completion,
+      sessionId,
+      stateDir,
+      spec.graceMs,
+      spec.signal,
+    )
+  } catch (error: unknown) {
+    output.destroy()
+    if (handle !== undefined) await handle.kill().catch(() => false)
+    if (completion !== undefined) await completion.catch(() => {})
+    await sandbox.files.remove(stateDir).catch(() => {})
+    throw error
+  }
+}
--- a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts
+++ b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts
@@ -0,0 +1,541 @@
+import { Buffer } from 'node:buffer'
+import { once } from 'node:events'
+import { Context } from 'cordis'
+import { describe, expect, it } from 'vitest'
+import {
+  CommandExitError,
+  FileNotFoundError,
+  type CommandHandle,
+  type CommandResult,
+  type Sandbox,
+} from '@deepseek-ai/dsh-e2b'
+import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
+import type { SubprocessTerminalSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b'
+import { spawnE2BTerminal } from '../src/terminal.ts'
+
+function commandError(exitCode: number): CommandExitError {
+  return new CommandExitError({ exitCode, stdout: '', stderr: '', error: `exit ${exitCode}` })
+}
+
+class FakeTerminalCommandHandle {
+  pid = 123
+  disconnects = 0
+  sdkKills = 0
+  disconnectError: unknown
+  sdkKillError: unknown
+  private readonly result = Promise.withResolvers<CommandResult>()
+  private settled = false
+
+  wait(): Promise<CommandResult> {
+    return this.result.promise
+  }
+
+  async disconnect(): Promise<void> {
+    this.disconnects += 1
+    if (this.disconnectError !== undefined) throw this.disconnectError
+  }
+
+  async kill(): Promise<boolean> {
+    this.sdkKills += 1
+    if (this.sdkKillError !== undefined) {
+      const error = this.sdkKillError
+      this.fail(137)
+      throw error
+    }
+    this.fail(137)
+    return true
+  }
+
+  succeed(exitCode = 0): void {
+    if (this.settled) return
+    this.settled = true
+    this.result.resolve({ exitCode, stdout: '', stderr: '' })
+  }
+
+  fail(exitCode: number): void {
+    if (this.settled) return
+    this.settled = true
+    this.result.reject(commandError(exitCode))
+  }
+
+  crash(error: unknown): void {
+    if (this.settled) return
+    this.settled = true
+    this.result.reject(error)
+  }
+
+  asHandle(): CommandHandle {
+    return this as unknown as CommandHandle
+  }
+}
+
+class FakeTerminalSandbox {
+  readonly handle = new FakeTerminalCommandHandle()
+  readonly commands: string[] = []
+  readonly inputs: Array<{ pid: number; data: Buffer }> = []
+  readonly removed: string[] = []
+  readonly directories: string[] = []
+  readonly writes = new Map<string, string>()
+  createOptions: Parameters<Sandbox['pty']['create']>[0] | undefined
+  ambient = 'KEEP=visible\0NPM_TOKEN=secret\0DSH_STALE=old\0BROKEN\0=bad\0'
+  ready: string | Error = 'ready\n'
+  sessionId = '123\n'
+  foreground = '456\n'
+  groups = [123]
+  createError: unknown
+  sendError: unknown
+  commandFailure: unknown
+  foregroundFailure: unknown
+  termFailure: unknown
+  removeError: unknown
+  clearOnTerm = true
+  clearOnKill = true
+  settleOnPtyKill = true
+  ptyKills = 0
+  resolvedExecutable = '/usr/bin/node\n'
+
+  readonly sandbox = {
+    files: {
+      makeDir: async (path: string): Promise<boolean> => {
+        this.directories.push(path)
+        return true
+      },
+      write: async (files: Array<{ path: string; data: string }>): Promise<object[]> => {
+        for (const file of files) this.writes.set(file.path, file.data)
+        return files.map(() => ({}))
+      },
+      read: async (): Promise<string> => {
+        if (this.ready instanceof Error) throw this.ready
+        return this.ready
+      },
+      remove: async (path: string): Promise<void> => {
+        this.removed.push(path)
+        if (this.removeError !== undefined) throw this.removeError
+      },
+    },
+    commands: {
+      run: async (command: string, options?: { signal?: AbortSignal }): Promise<CommandResult> => {
+        this.commands.push(command)
+        options?.signal?.throwIfAborted()
+        if (this.commandFailure !== undefined) {
+          const error = this.commandFailure
+          this.commandFailure = undefined
+          throw error
+        }
+        if (command === 'env -0') return { exitCode: 0, stdout: this.ambient, stderr: '' }
+        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 tpgid=')) {
+          if (this.foregroundFailure !== undefined) throw this.foregroundFailure
+          return { exitCode: 0, stdout: this.foreground, stderr: '' }
+        }
+        if (command.startsWith('ps -eo sid=')) {
+          return { exitCode: 0, stdout: this.groups.map(group => `${group}\n`).join(''), stderr: '' }
+        }
+        if (command.startsWith('kill -TERM -- ')) {
+          if (this.termFailure !== undefined) throw this.termFailure
+          if (this.clearOnTerm) {
+            this.groups = []
+            this.handle.fail(143)
+          }
+        }
+        if (command.startsWith('kill -KILL -- ') && this.clearOnKill) this.groups = []
+        return { exitCode: 0, stdout: '', stderr: '' }
+      },
+    },
+    pty: {
+      create: async (options: Parameters<Sandbox['pty']['create']>[0]): Promise<CommandHandle> => {
+        this.createOptions = options
+        if (this.createError !== undefined) throw this.createError
+        await options.onData(Buffer.from('buffered banner\n'))
+        return this.handle.asHandle()
+      },
+      sendInput: async (pid: number, data: Uint8Array, options?: { signal?: AbortSignal }): Promise<void> => {
+        options?.signal?.throwIfAborted()
+        this.inputs.push({ pid, data: Buffer.from(data) })
+        if (this.sendError !== undefined) throw this.sendError
+      },
+      kill: async (pid: number): Promise<boolean> => {
+        this.ptyKills += 1
+        if (this.settleOnPtyKill) this.handle.fail(137)
+        return pid === this.handle.pid
+      },
+    },
+  } as unknown as Sandbox
+}
+
+function runtime(fake: FakeTerminalSandbox): E2BSandboxService {
+  return {
+    cwd: '/workspace',
+    runtimeRoot: '/workspace/.dsh-e2b',
+    disposeMode: 'kill',
+    getSandbox: async () => fake.sandbox,
+  } as unknown as E2BSandboxService
+}
+
+function spec(overrides: Partial<SubprocessTerminalSpawnSpec> = {}): SubprocessTerminalSpawnSpec {
+  return {
+    argv: ['/bin/bash', '--noprofile', '--norc'],
+    cwd: '/workspace',
+    rows: 24,
+    cols: 80,
+    graceMs: 5,
+    env: { TERM: 'dumb', DSH_SESSION_ID: 'owner', TOKEN_EXPLICIT: 'kept' },
+    ...overrides,
+  }
+}
+
+describe('E2B terminal allocation', () => {
+  it('boots the requested argv through a private runner and preserves buffered bytes', async () => {
+    const fake = new FakeTerminalSandbox()
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/terminal-one')
+    let output = ''
+    terminal.output.on('data', (chunk) => { output += String(chunk) })
+    await new Promise(resolve => setTimeout(resolve, 0))
+
+    expect(output).toBe('buffered banner\n')
+    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')
+    expect(fake.writes.get('/runtime/terminal-one/environment')).toContain('TOKEN_EXPLICIT=kept\0')
+    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 runner = fake.writes.get('/runtime/terminal-one/runner.bash') ?? ''
+    expect(runner).toContain('if (( ${#dsh_argv[@]} == 0 )); then')
+    expect(runner).toContain('exec env -i "${dsh_env[@]}" "${dsh_argv[@]}"')
+    expect(runner).not.toContain('\u007f')
+
+    await terminal.write(Buffer.from('echo ok\r'))
+    expect(fake.inputs.at(-1)?.data.toString()).toBe('echo ok\r')
+    await expect(terminal.inspectForeground()).resolves.toEqual({ processGroupId: 456, inputWaiting: false })
+    await expect(terminal.signalForeground('SIGINT')).resolves.toBe(456)
+    expect(fake.commands).toContain('kill -INT -- -456')
+
+    terminal.terminate()
+    await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+    expect(fake.handle.disconnects).toBe(1)
+    expect(fake.removed).toContain('/runtime/terminal-one')
+  })
+
+  it('inherits only safe ambient values and binds live abort to terminal cleanup', async () => {
+    const fake = new FakeTerminalSandbox()
+    const controller = new AbortController()
+    const terminal = await spawnE2BTerminal(
+      runtime(fake),
+      spec({ env: undefined, signal: controller.signal }),
+      '/runtime/abort-live',
+    )
+    const environment = fake.writes.get('/runtime/abort-live/environment') ?? ''
+    expect(environment).toContain('KEEP=visible\0')
+    expect(environment).not.toContain('secret')
+    expect(environment).not.toContain('DSH_STALE')
+
+    controller.abort(new Error('stop'))
+    await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
+    await expect(terminal.waitForExit(controller.signal)).resolves.toBe(false)
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+
+  it('rejects malformed environment and argv values before PTY allocation', async () => {
+    const invalidName = new FakeTerminalSandbox()
+    await expect(spawnE2BTerminal(runtime(invalidName), spec({ env: { 'BAD=NAME': 'x' } }), '/runtime/name'))
+      .rejects.toThrow('environment entries')
+    expect(invalidName.createOptions).toBeUndefined()
+
+    const invalidValue = new FakeTerminalSandbox()
+    await expect(spawnE2BTerminal(runtime(invalidValue), spec({ env: { BAD: 'x\0y' } }), '/runtime/value'))
+      .rejects.toThrow('environment entries')
+
+    const invalidArg = new FakeTerminalSandbox()
+    await expect(spawnE2BTerminal(runtime(invalidArg), spec({ argv: ['/bin/bash', 'x\0y'] }), '/runtime/argv'))
+      .rejects.toThrow('argv must not contain NUL')
+  })
+
+  it('cleans malformed handles, bootstrap failures, and readiness failures', async () => {
+    const invalidPid = new FakeTerminalSandbox()
+    invalidPid.handle.pid = 0
+    await expect(spawnE2BTerminal(runtime(invalidPid), spec(), '/runtime/invalid-pid'))
+      .rejects.toThrow('invalid terminal pid 0')
+    expect(invalidPid.handle.sdkKills).toBe(1)
+    expect(invalidPid.removed).toContain('/runtime/invalid-pid')
+
+    const failedInput = new FakeTerminalSandbox()
+    failedInput.sendError = new Error('bootstrap failed')
+    await expect(spawnE2BTerminal(runtime(failedInput), spec(), '/runtime/input'))
+      .rejects.toThrow('bootstrap failed')
+    expect(failedInput.handle.sdkKills).toBe(1)
+
+    const exited = new FakeTerminalSandbox()
+    exited.ready = new FileNotFoundError('not ready')
+    queueMicrotask(() => { exited.handle.succeed(0) })
+    await expect(spawnE2BTerminal(runtime(exited), spec(), '/runtime/exited'))
+      .rejects.toThrow('exited before publishing readiness')
+
+    const invalidSession = new FakeTerminalSandbox()
+    invalidSession.sessionId = 'not-a-session\n'
+    await expect(spawnE2BTerminal(runtime(invalidSession), spec(), '/runtime/session'))
+      .rejects.toThrow('cannot resolve process session')
+    expect(invalidSession.handle.sdkKills).toBe(1)
+
+    const cleanupFailed = new FakeTerminalSandbox()
+    cleanupFailed.handle.pid = 0
+    cleanupFailed.handle.sdkKillError = new Error('kill transport failed')
+    cleanupFailed.removeError = new Error('remove transport failed')
+    await expect(spawnE2BTerminal(runtime(cleanupFailed), spec(), '/runtime/cleanup-failed'))
+      .rejects.toThrow('invalid terminal pid 0')
+  })
+
+  it('propagates setup cancellation and provider failures', async () => {
+    const aborted = new FakeTerminalSandbox()
+    await expect(spawnE2BTerminal(runtime(aborted), spec({ signal: AbortSignal.abort(new Error('stop')) }), '/runtime/abort'))
+      .rejects.toThrow('stop')
+
+    const createFailed = new FakeTerminalSandbox()
+    createFailed.createError = new Error('create failed')
+    await expect(spawnE2BTerminal(runtime(createFailed), spec(), '/runtime/create'))
+      .rejects.toThrow('create failed')
+
+    const readFailed = new FakeTerminalSandbox()
+    readFailed.ready = new Error('ready transport failed')
+    await expect(spawnE2BTerminal(runtime(readFailed), spec(), '/runtime/read'))
+      .rejects.toThrow('ready transport failed')
+  })
+})
+
+describe('E2B terminal lifecycle', () => {
+  it('maps ordinary exits, closes output, and reports an absent foreground after exit', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = []
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/natural')
+    terminal.output.resume()
+    const ended = once(terminal.output, 'end')
+    fake.handle.succeed(7)
+    await expect(terminal.done).resolves.toEqual({ exitCode: 7, signal: null })
+    await ended
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+    await expect(terminal.write(Buffer.from('late'))).rejects.toThrow('exited')
+    fake.foregroundFailure = commandError(1)
+    await expect(terminal.inspectForeground()).resolves.toBeUndefined()
+  })
+
+  it.each([
+    [7, { exitCode: 7, signal: null }],
+    [143, { exitCode: null, signal: 'SIGTERM' }],
+    [255, { exitCode: 255, signal: null }],
+  ] as const)('classifies an unrequested command exit %i', async (exitCode, expected) => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = []
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), `/runtime/exit-${exitCode}`)
+    fake.handle.fail(exitCode)
+    await expect(terminal.done).resolves.toEqual(expected)
+    await expect(terminal.waitForExit(new AbortController().signal)).resolves.toBe(true)
+  })
+
+  it('lets an early quiescence observer follow a transport rejection', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = []
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/early-observer')
+    terminal.output.on('error', () => {})
+    const quiescence = terminal.waitForExit()
+    fake.handle.crash(new Error('transport failed'))
+    await expect(terminal.done).rejects.toThrow('transport failed')
+    await expect(quiescence).resolves.toBe(true)
+  })
+
+  it('rejects killing the terminal shell and propagates live foreground failures', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.foreground = '123\n'
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/signal')
+    await expect(terminal.signalForeground('SIGKILL')).rejects.toThrow('refusing to SIGKILL')
+    fake.foreground = 'invalid\n'
+    await expect(terminal.inspectForeground()).rejects.toThrow('cannot resolve foreground')
+    fake.foregroundFailure = commandError(1)
+    await expect(terminal.inspectForeground()).rejects.toBeInstanceOf(CommandExitError)
+    fake.clearOnTerm = true
+    terminal.terminate()
+    await terminal.waitForExit()
+  })
+
+  it('escalates surviving process groups and bounds an observing wait', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = [123, 456]
+    fake.clearOnTerm = false
+    const terminal = await spawnE2BTerminal(runtime(fake), spec({ graceMs: 1 }), '/runtime/escalate')
+    const controller = new AbortController()
+    const observing = terminal.waitForExit(controller.signal)
+    controller.abort()
+    await expect(observing).resolves.toBe(false)
+
+    terminal.terminate()
+    await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+    expect(fake.commands).toContain('kill -TERM -- -123 -456')
+    expect(fake.commands).toContain('kill -KILL -- -123 -456')
+  })
+
+  it('surfaces cleanup failures and allows a later retry', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = [1]
+    const terminal = await spawnE2BTerminal(runtime(fake), spec({ graceMs: 1 }), '/runtime/retry')
+    terminal.terminate()
+    await expect(terminal.waitForExit(new AbortController().signal)).rejects.toThrow('unsafe process group 1')
+
+    fake.groups = []
+    fake.handle.succeed(0)
+    await terminal.done
+    terminal.terminate()
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+
+  it('propagates a process-group signalling transport failure before retry', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.termFailure = new Error('signal transport failed')
+    const terminal = await spawnE2BTerminal(runtime(fake), spec({ graceMs: 1 }), '/runtime/signal-failure')
+    terminal.terminate()
+    await expect(terminal.waitForExit()).rejects.toThrow('signal transport failed')
+
+    fake.groups = []
+    fake.handle.succeed(0)
+    await terminal.done
+    terminal.terminate()
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+
+  it('keeps command rejection authoritative while cleanup is already waiting', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = []
+    fake.removeError = new Error('private state already gone')
+    const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/reject-during-cleanup')
+    terminal.output.on('error', () => {})
+    terminal.terminate()
+    await Promise.resolve()
+    fake.handle.crash(new Error('command transport failed'))
+    await expect(terminal.done).rejects.toThrow('command transport failed')
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+
+  it('keeps a late command rejection authoritative after PTY kill', async () => {
+    const fake = new FakeTerminalSandbox()
+    fake.groups = []
+    fake.settleOnPtyKill = false
+    const terminal = await spawnE2BTerminal(runtime(fake), spec({ graceMs: 1 }), '/runtime/reject-after-kill')
+    terminal.output.on('error', () => {})
+    terminal.terminate()
+    while (fake.ptyKills === 0) await new Promise(resolve => setTimeout(resolve, 0))
+    await Promise.resolve()
+    fake.handle.crash(new Error('late command transport failed'))
+    await expect(terminal.done).rejects.toThrow('late command transport failed')
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+
+  it('reports surviving groups, a surviving top-level pid, and transport failure', async () => {
+    const survivor = new FakeTerminalSandbox()
+    survivor.clearOnTerm = false
+    survivor.clearOnKill = false
+    const terminal = await spawnE2BTerminal(runtime(survivor), spec({ graceMs: 1 }), '/runtime/survivor')
+    terminal.terminate()
+    await expect(terminal.waitForExit()).rejects.toThrow('surviving process groups: 123')
+
+    const livePid = new FakeTerminalSandbox()
+    livePid.groups = []
+    livePid.settleOnPtyKill = false
+    const live = await spawnE2BTerminal(runtime(livePid), spec({ graceMs: 1 }), '/runtime/live-pid')
+    live.terminate()
+    await expect(live.waitForExit()).rejects.toThrow('surviving pid: 123')
+    livePid.handle.succeed(0)
+    await live.done
+
+    const crashed = new FakeTerminalSandbox()
+    crashed.groups = []
+    const failed = await spawnE2BTerminal(runtime(crashed), spec(), '/runtime/crashed')
+    const outputError = once(failed.output, 'error')
+    crashed.handle.crash('transport gone')
+    await expect(failed.done).rejects.toEqual('transport gone')
+    await expect(outputError).resolves.toMatchObject([{ message: 'transport gone' }])
+    await expect(failed.waitForExit()).resolves.toBe(true)
+  })
+})
+
+describe('E2B subprocess terminal service', () => {
+  async function service(fake = new FakeTerminalSandbox()): Promise<{
+    ctx: Context
+    fiber: Awaited<ReturnType<Context['plugin']>>
+    fake: FakeTerminalSandbox
+  }> {
+    const ctx = new Context()
+    ctx.provide('e2b', runtime(fake))
+    const fiber = await ctx.plugin(E2BSubprocessService)
+    return { ctx, fiber, fake }
+  }
+
+  it('publishes execution-world coordinates and resolves remote executables', async () => {
+    const { ctx } = await service()
+    expect(ctx.subprocess.cwd).toBe('/workspace')
+    expect(ctx.subprocess.runtimeRoot).toBe('/workspace/.dsh-e2b')
+    await expect(ctx.subprocess.resolveExecutable('/bin/bash')).resolves.toBe('/bin/bash')
+    await expect(ctx.subprocess.resolveExecutable('node', { PATH: '/custom/bin' }, new AbortController().signal))
+      .resolves.toBe('/usr/bin/node')
+    expect((ctx.e2b)).toBeDefined()
+  })
+
+  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('node', undefined, AbortSignal.abort(new Error('stop'))))
+      .rejects.toThrow('stop')
+    fake.resolvedExecutable = 'relative/node\n'
+    await expect(ctx.subprocess.resolveExecutable('node')).rejects.toThrow('did not resolve')
+    fake.resolvedExecutable = '/one\n/two\n'
+    await expect(ctx.subprocess.resolveExecutable('node')).rejects.toThrow('did not resolve')
+  })
+
+  it('owns live terminals through service disposal', async () => {
+    const { ctx, fiber, fake } = await service()
+    const terminal = await ctx.subprocess.spawnTerminal(spec())
+    await fiber.dispose()
+    await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
+    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 [
+      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()
+    }
+
+    fake.groups = []
+    const terminal = await ctx.subprocess.spawnTerminal(spec())
+    fake.handle.succeed(0)
+    await terminal.done
+    await terminal.waitForExit()
+    const signals = fake.commands.filter(command => command.startsWith('kill -')).length
+    await fiber.dispose()
+    expect(fake.commands.filter(command => command.startsWith('kill -'))).toHaveLength(signals)
+  })
+
+  it('contains a failed automatic terminal release until service disposal retries it', async () => {
+    const { fiber, fake } = await service()
+    fake.clearOnTerm = false
+    fake.clearOnKill = false
+    const terminal = await (fiber.ctx).subprocess.spawnTerminal(spec({ graceMs: 1 }))
+    fake.handle.succeed(0)
+    await terminal.done
+    await new Promise(resolve => setTimeout(resolve, 10))
+    expect(fake.commands).toContain('kill -KILL -- -123')
+
+    fake.groups = []
+    await fiber.dispose()
+    await expect(terminal.waitForExit()).resolves.toBe(true)
+  })
+})