diff --git a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.i18n.yaml b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.i18n.yaml index beb49fba76..04ebe420ac 100644 --- a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.i18n.yaml +++ b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.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 .agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md -2026-07-27-e2b-remote-runtime-poc.md: efb91fc166f36dc8fb047d686e8cd82d93b9af6a -2026-07-27-e2b-remote-runtime-poc.zh.md: eb14cccc284d38cdc5cc704512b6899621a9a840 +2026-07-27-e2b-remote-runtime-poc.md: 4e6414b7d919d765712ac9314ce73a69c5c6344e +2026-07-27-e2b-remote-runtime-poc.zh.md: 4d39ed7eb07cdd17c5bfa9155bc7efabaa798117 diff --git a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md index efb91fc166..4e6414b7d9 100644 --- a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md +++ b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md @@ -32,7 +32,7 @@ E2B owns the mutable filesystem, command and Bash processes, PTY shell and termi The host owns Cordis and plugin objects, the agent loop, agent/session/goal state, session logs and persistence, LLM calls, prompts and tools, authority decisions, skills, subagent orchestration, PTY buffers and readiness state, LSP JSON-RPC ids/queues/protocol state, Code Runtime type stripping/output accounting/binding dispatch, and E2B SDK/network orchestration. The overlay does not upload, mount, or synchronize the host workspace; identical cwd strings name independent host and remote directories. -Byte-sensitive protocols use the narrowest adapter required by E2B's callback shapes. PTY consumes the SDK's byte callback directly. LSP installs a bounded remote source reader, while Code Runtime keeps framed stdout in a launcher process isolated from the controller and worker descriptors. Their dependency-free helpers encode protocol payloads as validated newline-delimited base64 JSON, keeping E2B's decoded command callbacks on an ASCII transport. +Byte-sensitive protocols use the narrowest adapter required by E2B's callback shapes. PTY consumes the SDK's byte callback directly and carries send identity across asynchronous foreground-group lookup. LSP installs a bounded remote source reader that walks no-follow directory descriptors beneath the canonical workspace. Code Runtime keeps framed stdout in a launcher process isolated from the controller and worker descriptors, and gives each controller a process group that is killed before its inherited pipes drain. Their dependency-free helpers encode protocol payloads as validated newline-delimited base64 JSON, keeping E2B's decoded command callbacks on an ASCII transport. Retaining a sandbox preserves remote files and unmanaged remote state only. Reconnect does not reconstruct host PTY sessions, buffers, process handles, LSP connections or requests, code workers, binding calls, timers, output cursors, or locks. Managed groups terminate and join when their provider disposes before the shared owner pauses, leaves, or kills the sandbox. @@ -40,9 +40,9 @@ The POC has no session-persistence backend, template builder, volume, snapshot, ## Verification -Focused package suites pin owner lifecycle cleanup, filesystem semantics and commit metadata, subprocess process groups, configuration and verified publication rollback, byte framing and multibyte boundaries, PTY readiness/signals/default-environment scrubbing/terminal-session cleanup, stable bounded LSP source reads, Code Runtime binding and descriptor isolation, worker-pipe draining, hostile traffic, output limits, timeout/abort ordering, disposal to quiescence, and package-owned invariant registrations. Adjacent local-backend suites pin the shared PTY utilities and the LSP cross-namespace `processId` behavior. +Focused package suites pin owner lifecycle cleanup, filesystem semantics and commit metadata, subprocess process groups, configuration and verified publication rollback, byte framing and multibyte boundaries, PTY readiness/signal identity/default-environment scrubbing/terminal-session cleanup, descriptor-walked bounded LSP source reads, Code Runtime binding and descriptor isolation, worker and descendant-held pipe draining, hostile traffic, output limits, timeout/abort ordering, disposal to quiescence, and package-owned invariant registrations. Adjacent local-backend suites pin the shared PTY utilities and the LSP cross-namespace `processId` behavior. -A credential-gated Loader composition creates real E2B sandboxes and exercises FS-to-Bash and Bash-to-FS visibility, process-publication rollback, bounded spill output, PTY default-secret scrubbing and process-tree cleanup, stable bounded LSP source reads, Code Runtime host bindings and descriptor-isolated output accounting, wall timeout, abort, runner cleanup, host-workspace isolation, and final sandbox deletion. The same composition runs through source imports and built package exports. +A credential-gated Loader composition creates real E2B sandboxes and exercises FS-to-Bash and Bash-to-FS visibility, process-publication rollback, bounded spill output, PTY default-secret scrubbing, stale-interrupt identity, and process-tree cleanup, parent-swap-safe bounded LSP source reads, Code Runtime host bindings, descriptor-isolated output accounting, descendant-held pipe cleanup, wall timeout, abort, runner cleanup, host-workspace isolation, and final sandbox deletion. The same composition runs through source imports and built package exports. ## Alternatives considered diff --git a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.zh.md b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.zh.md index eb14cccc28..4d39ed7eb0 100644 --- a/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.zh.md +++ b/.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.zh.md @@ -32,7 +32,7 @@ E2B 拥有可变文件系统、命令和 Bash 进程、PTY shell 与终端会话 宿主拥有 Cordis 与插件对象、agent loop、agent／会话／goal 状态、会话日志及持久化、LLM（大语言模型）调用、提示词与工具、权限决策、skill（技能）、subagent 编排、PTY 缓冲与就绪状态、LSP JSON-RPC id／队列／协议状态、Code Runtime 类型剥离／输出计量／绑定分发，以及 E2B SDK／网络编排。该 overlay 不会上传、挂载或同步宿主工作区；拼写相同的 cwd 字符串分别指向彼此独立的宿主与远程目录。 -对字节敏感的协议只使用适配 E2B 回调形状所需的最窄适配器。PTY 直接消费 SDK 的字节回调。LSP 会安装一个有界的远程源码读取器；Code Runtime 则把分帧 stdout 保留在与 controller 和 worker 描述符隔离的 launcher 进程内。它们的无依赖辅助程序会把协议载荷编码为经过验证、以换行分隔的 base64 JSON，并通过 ASCII 传输承载 E2B 已解码的命令回调。 +对字节敏感的协议只使用适配 E2B 回调形状所需的最窄适配器。PTY 直接消费 SDK 的字节回调，并在异步查找前台进程组的过程中保留发送操作身份。LSP 会安装一个有界的远程源码读取器，通过不跟随符号链接打开的目录描述符在规范化工作区下逐级遍历。Code Runtime 则把分帧 stdout 保留在与 controller 和 worker 描述符隔离的 launcher 进程内，并为每个 controller 分配一个进程组，在 controller 继承的管道排空前终止该组。它们的无依赖辅助程序会把协议载荷编码为经过验证、以换行分隔的 base64 JSON，并通过 ASCII 传输承载 E2B 已解码的命令回调。 保留沙箱只会保存远程文件与未受管的远程状态。重新连接不会重建宿主 PTY 会话、缓冲、进程句柄、LSP 连接或请求、代码 worker、绑定调用、定时器、输出游标或锁。受管进程组会在所属提供方 dispose（资源释放）时终止并等待退出，之后共享所有者才会暂停、脱离或终止沙箱。 @@ -40,9 +40,9 @@ E2B 拥有可变文件系统、命令和 Bash 进程、PTY shell 与终端会话 ## 验证 -聚焦包测试套件固定所有者生命周期清理、文件系统语义与提交元数据、进程管理的进程组、配置与经过验证的发布回滚、字节分帧与多字节边界、PTY 就绪状态／信号／默认环境清理／终端会话清理、稳定且有界的 LSP 源码读取、Code Runtime 绑定与描述符隔离、worker 管道排空、恶意通信、输出上限、超时／中止顺序、等待完全停稳的资源释放，以及包自有不变式注册。相邻本地后端测试套件固定共享 PTY 工具函数，以及 LSP 跨命名空间 `processId` 行为。 +聚焦包测试套件固定所有者生命周期清理、文件系统语义与提交元数据、进程管理的进程组、配置与经过验证的发布回滚、字节分帧与多字节边界、PTY 就绪状态／信号身份／默认环境清理／终端会话清理、基于描述符逐级遍历的有界 LSP 源码读取、Code Runtime 绑定与描述符隔离、worker 管道及后代进程所持管道的排空、恶意通信、输出上限、超时／中止顺序、等待完全停稳的资源释放，以及包自有不变式注册。相邻本地后端测试套件固定共享 PTY 工具函数，以及 LSP 跨命名空间 `processId` 行为。 -凭据门控的 Loader 组合会创建真实 E2B 沙箱，并演练 FS-to-Bash 与 Bash-to-FS 可见性、进程发布回滚、有界 spill 输出、PTY 默认秘密清理与进程树清理、稳定且有界的 LSP 源码读取、Code Runtime 宿主绑定与描述符隔离的输出记账、墙钟超时、中止、runner 清理、宿主工作区隔离，以及最终删除沙箱。同一组合分别通过源代码导入与已构建包导出运行。 +凭据门控的 Loader 组合会创建真实 E2B 沙箱，并演练 FS-to-Bash 与 Bash-to-FS 可见性、进程发布回滚、有界 spill 输出、PTY 默认秘密清理、陈旧中断身份与进程树清理、可抵御父目录替换的有界 LSP 源码读取、Code Runtime 宿主绑定、描述符隔离的输出记账、后代进程所持管道的清理、墙钟超时、中止、runner 清理、宿主工作区隔离，以及最终删除沙箱。同一组合分别通过源代码导入与已构建包导出运行。 ## 曾考虑的替代方案 diff --git a/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts b/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts index 4f2401d2c7..a997752711 100644 --- a/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts +++ b/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts @@ -147,18 +147,23 @@ try { workspaceRoot: process.cwd(), }) - const swappedSource = await ctx.fs.resolve('swapped-source.ts') - const swappedSourcePath = posix.join(process.cwd(), 'swapped-source.ts') - await ctx.fs.writeText(swappedSource, 'const safe = true\n', { kind: 'createIfAbsent' }) + const swappedParentPath = posix.join(process.cwd(), 'swapped-parent') + const swappedSourcePath = posix.join(swappedParentPath, 'source.ts') + const swappedOutsidePath = '/tmp/dsh-e2b-lsp-outside' + await sandbox.commands.run( + `mkdir -p -- ${quoteE2BShellArg(swappedParentPath)} ${quoteE2BShellArg(swappedOutsidePath)} && printf 'const safe = true\\n' > ${quoteE2BShellArg(swappedSourcePath)} && printf 'const outside = true\\n' > ${quoteE2BShellArg(posix.join(swappedOutsidePath, 'source.ts'))}`, + ) const remoteCommands = sandbox.commands as unknown as { run(command: string, options?: unknown): Promise<{ exitCode: number; stdout: string; stderr: string }> } const runRemoteCommand = remoteCommands.run.bind(sandbox.commands) let containmentFaultInjected = false remoteCommands.run = async (command, options) => { - if (!containmentFaultInjected && command.includes('dsh-e2b-source-reader') && command.includes('swapped-source.ts')) { + if (!containmentFaultInjected && command.includes('dsh-e2b-source-reader') && command.includes('swapped-parent/source.ts')) { containmentFaultInjected = true - await runRemoteCommand(`rm -f -- ${quoteE2BShellArg(swappedSourcePath)} && ln -s -- /etc/hosts ${quoteE2BShellArg(swappedSourcePath)}`) + await runRemoteCommand( + `rm -rf -- ${quoteE2BShellArg(swappedParentPath)} && ln -s -- ${quoteE2BShellArg(swappedOutsidePath)} ${quoteE2BShellArg(swappedParentPath)}`, + ) } return await runRemoteCommand(command, options) } @@ -166,7 +171,7 @@ try { try { await ctx.lsp.query({ operation: 'hover', - filePath: 'swapped-source.ts', + filePath: 'swapped-parent/source.ts', position: { line: 0, character: 1 }, workspaceRoot: process.cwd(), }) @@ -200,8 +205,28 @@ try { text: "printf 'PTY-你好\\n'", submit: true, }).done + const foregroundLookup = Promise.withResolvers() + let delayedForegroundLookup = false + remoteCommands.run = async (command, options) => { + if (!delayedForegroundLookup && command.startsWith('ps -o tpgid=')) { + delayedForegroundLookup = true + await foregroundLookup.promise + } + return await runRemoteCommand(command, options) + } + const staleInterrupt = ctx.pty.startSend(owner, terminal.sessionId, { text: 'sleep 0.2', submit: true }) + if (!staleInterrupt.cancel()) throw new Error('E2B PTY refused the stale-interrupt probe cancellation') + await staleInterrupt.done const sleeping = ctx.pty.startSend(owner, terminal.sessionId, { text: 'sleep 30', submit: true }) - await new Promise(resolveDelay => setTimeout(resolveDelay, 150)) + foregroundLookup.resolve(undefined) + const interruptIdentitySafe = await Promise.race([ + sleeping.done.then(() => false), + new Promise(resolveDelay => setTimeout(() => { resolveDelay(true) }, 300)), + ]) + remoteCommands.run = runRemoteCommand + if (!delayedForegroundLookup || !interruptIdentitySafe) { + throw new Error('E2B PTY stale interrupt affected its successor send') + } const terminalSignal = await ctx.pty.signal(owner, terminal.sessionId, 'SIGINT') const interrupted = await sleeping.done const stubborn = await ctx.pty.startSend(owner, terminal.sessionId, { @@ -296,6 +321,23 @@ try { `, bindings: [], }) + const descendantPipe = await ctx.codeRuntime.run({ + program: ` + const childProcess = await import('node:child_process') + const child = childProcess.spawn( + process.execPath, + ['-e', 'setInterval(() => {}, 1000)', 'dsh-code-runtime-descendant'], + { stdio: ['ignore', 'inherit', 'inherit'] }, + ) + return child.pid > 0 + `, + bindings: [], + }) + const descendantProcesses = await sandbox.commands.list() + const descendantCleanup = !descendantProcesses.some(processInfo => + JSON.stringify([processInfo.cmd, processInfo.args]).includes('dsh-code-runtime-descendant'), + ) + if (!descendantCleanup) throw new Error('E2B Code Runtime left a pipe-holding descendant alive') const timedOut = await ctx.codeRuntime.run({ program: 'await new Promise(() => {})', bindings: [], @@ -340,6 +382,7 @@ try { echo: terminalEcho, signal: terminalSignal, interrupted, + interruptIdentitySafe, treeCleanup: terminalTreeCleanup, scrollback: terminalScrollback.text, }, @@ -348,6 +391,8 @@ try { nativeOutput, descriptorOutput, inheritedOutput, + descendantPipe, + descendantCleanup, timedOut, aborted, oversizedBoot, diff --git a/packages/e2b/code-runtime-e2b/README.i18n.yaml b/packages/e2b/code-runtime-e2b/README.i18n.yaml index ba43d787b8..eb4e75f297 100644 --- a/packages/e2b/code-runtime-e2b/README.i18n.yaml +++ b/packages/e2b/code-runtime-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/code-runtime-e2b/README.md -README.md: 171e63a8c1ca70deb63e9860f5088b83401a37d0 -README.zh.md: b70984e765d32985436c15dd543263cc06a25715 +README.md: a8623f95d16b54b29e53bb9cf2c528b36f121283 +README.zh.md: 2b4a37864e3c6755a74f0d2ef6ce38aa39aae24b diff --git a/packages/e2b/code-runtime-e2b/README.md b/packages/e2b/code-runtime-e2b/README.md index 171e63a8c1..a8623f95d1 100644 --- a/packages/e2b/code-runtime-e2b/README.md +++ b/packages/e2b/code-runtime-e2b/README.md @@ -19,7 +19,7 @@ Every value is a positive safe integer. `maxOutputBytes` is at least four bytes, ## Execution and bridge contract -Setup uploads one dependency-free runner under `ctx.e2b.runtimeRoot` and resolves remote Node. For each run, the host wraps and type-strips erasable TypeScript with Node's `stripTypeScriptTypes`, then starts the runner in `ctx.e2b.cwd`. The runner keeps the framed host protocol in a launcher process, forks a controller whose stdout and stderr are bounded data pipes, and creates a fresh worker thread with an empty environment and heap limit. Model writes to native descriptors and inherited child output therefore cannot enter the frame stream; worker and controller pipes drain before the terminal frame. The worker measures active event-loop time and is destroyed after one completion. The enclosing E2B subprocess group is terminated and awaited after every result, timeout, abort, or disposal, so ordinary child processes in that group stop with the run. +Setup uploads one dependency-free runner under `ctx.e2b.runtimeRoot` and resolves remote Node. For each run, the host wraps and type-strips erasable TypeScript with Node's `stripTypeScriptTypes`, then starts the runner in `ctx.e2b.cwd`. The runner keeps the framed host protocol in a launcher process, forks a controller process group whose stdout and stderr are bounded data pipes, and creates a fresh worker thread with an empty environment and heap limit. Model writes to native descriptors and inherited child output therefore cannot enter the frame stream; completion kills the controller group before draining its pipes and emitting the terminal frame. The worker measures active event-loop time and is destroyed after one completion. The enclosing E2B subprocess group is terminated and awaited after every result, timeout, abort, or disposal, so ordinary child processes in either managed group stop with the run. The bridge uses validated newline-delimited base64 JSON frames because E2B subprocess callbacks expose decoded text. Binding arguments and resolutions use the worker runtime's iterative lossless-JSON wire shape; binding functions execute on the host and typed rejection classes are materialized inside the remote worker. The worker captures the JavaScript intrinsics that its adapter boundary invokes before model code runs, hardening binding transport, output accounting, and completion validation against mutation of those references. The host repeats message validation, call-id deduplication, lossless-JSON checks, and the outer-output ledger. diff --git a/packages/e2b/code-runtime-e2b/README.zh.md b/packages/e2b/code-runtime-e2b/README.zh.md index b70984e765..2b4a37864e 100644 --- a/packages/e2b/code-runtime-e2b/README.zh.md +++ b/packages/e2b/code-runtime-e2b/README.zh.md @@ -19,7 +19,7 @@ ## 执行与桥接契约 -设置阶段会在 `ctx.e2b.runtimeRoot` 下上传一个无依赖的 runner，并解析远程 Node。每次运行时，宿主会包装仅使用可擦除语法的 TypeScript，再用 Node 的 `stripTypeScriptTypes` 剥离类型，然后在 `ctx.e2b.cwd` 中启动 runner。runner 会把面向宿主的分帧协议保留在 launcher 进程内，派生一个以 stdout 和 stderr 作为有界数据管道的 controller，再创建一个具有空环境与堆上限的全新 worker 线程。因此，模型对原生描述符的写入和继承的子进程输出无法进入分帧流；worker 与 controller 管道会在发出终结帧前排空。worker 会测量事件循环活跃时间，并在一次运行结算后销毁。每当运行返回结果、超时、中止或因资源释放终止时，系统都会终止外围的 E2B 进程组并等待其退出，因此组内的普通子进程会随本次运行一同停止。 +设置阶段会在 `ctx.e2b.runtimeRoot` 下上传一个无依赖的 runner，并解析远程 Node。每次运行时，宿主会包装仅使用可擦除语法的 TypeScript，再用 Node 的 `stripTypeScriptTypes` 剥离类型，然后在 `ctx.e2b.cwd` 中启动 runner。runner 会把面向宿主的分帧协议保留在 launcher 进程内，派生一个以 stdout 和 stderr 作为有界数据管道的 controller 进程组，再创建一个具有空环境与堆上限的全新 worker 线程。因此，模型对原生描述符的写入和继承的子进程输出无法进入分帧流；运行结算会先终止 controller 进程组，再排空其管道并发出终结帧。worker 会测量事件循环活跃时间，并在一次运行结算后销毁。每当运行返回结果、超时、中止或因资源释放终止时，系统都会终止外围的 E2B 进程组并等待其退出，因此任一受管组内的普通子进程会随本次运行一同停止。 由于 E2B 进程管理回调公开的是已解码文本，桥接层使用经过验证、以换行分隔的 base64 JSON 帧。绑定参数与 resolve 值使用 worker 运行时的迭代式无损 JSON wire 形状；绑定函数在宿主执行，类型化的 reject 类则在远程 worker 内物化。worker 会在模型代码运行前捕获其适配器边界调用的 JavaScript intrinsic，从而增强绑定传输、输出记账与完成值验证对这些引用修改的抵御能力。宿主会再次执行消息验证、调用 id 去重和无损 JSON 检查，并用外层输出账本再次计量。 diff --git a/packages/e2b/code-runtime-e2b/src/runner-source.ts b/packages/e2b/code-runtime-e2b/src/runner-source.ts index e5cf6ebefa..527f21114e 100644 --- a/packages/e2b/code-runtime-e2b/src/runner-source.ts +++ b/packages/e2b/code-runtime-e2b/src/runner-source.ts @@ -35,6 +35,18 @@ const waitForChildExit = child => { return new Promise(resolve => { child.once('exit', resolve) }) } +const killControllerGroup = child => { + if (process.platform !== 'win32' && Number.isSafeInteger(child.pid)) { + try { + process.kill(-child.pid, 'SIGKILL') + } catch (error) { + if (!error || typeof error !== 'object' || error.code !== 'ESRCH') throw error + } + return + } + child.kill('SIGKILL') +} + const jsonStringBytes = text => Buffer.byteLength(JSON.stringify(text)) const truncateLog = (text, available) => { @@ -74,7 +86,7 @@ const runLauncher = () => { const stdoutDrained = waitForPipeDrain(current.stdout) const stderrDrained = waitForPipeDrain(current.stderr) const exited = waitForChildExit(current) - current.kill('SIGKILL') + killControllerGroup(current) await Promise.all([exited, stdoutDrained, stderrDrained]) }) : Promise.resolve() @@ -112,6 +124,7 @@ const runLauncher = () => { maxOutputBytes = message.maxOutputBytes controller = fork(fileURLToPath(import.meta.url), [], { env: { DSH_CODE_RUNTIME_CONTROLLER: '1' }, + detached: process.platform !== 'win32', execArgv: [], stdio: ['ignore', 'pipe', 'pipe', 'ipc'], }) @@ -173,7 +186,7 @@ const runLauncher = () => { }) } }) - input.on('close', () => { if (controller && !settling) controller.kill('SIGKILL') }) + input.on('close', () => { if (controller && !settling) killControllerGroup(controller) }) } const runController = () => { diff --git a/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts b/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts index 9743c361c5..f99485a929 100644 --- a/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts +++ b/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts @@ -1,8 +1,9 @@ import { spawn } from 'node:child_process' -import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { access, mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { PassThrough, Writable } from 'node:stream' +import { setTimeout as delay } from 'node:timers/promises' import { Context } from 'cordis' import { describe, expect, it, vi } from 'vitest' import type { Sandbox } from '@deepseek-ai/dsh-e2b' @@ -271,6 +272,55 @@ describe('E2BCodeRuntime', () => { expect(Buffer.byteLength(nativeOutput)).toBe(expectedBytes) }) + it.skipIf(process.platform === 'win32')('reaps descendant-held controller pipes before completion', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dsh-e2b-code-descendant-')) + const marker = join(directory, 'started') + const release = join(directory, 'release') + const childSource = ` + const fs = require('node:fs') + fs.writeFileSync(${JSON.stringify(marker)}, 'started') + const timer = setInterval(() => { + if (fs.existsSync(${JSON.stringify(release)})) clearInterval(timer) + }, 10) + ` + let running: ReturnType | undefined + try { + running = runInstalledRunner(` + const fs = await import('node:fs') + const childProcess = await import('node:child_process') + childProcess.spawn(process.execPath, ['-e', ${JSON.stringify(childSource)}], { + stdio: ['ignore', 'inherit', 'inherit'], + }) + while (!fs.existsSync(${JSON.stringify(marker)})) await new Promise(resolve => setTimeout(resolve, 5)) + return true + `) + const deadline = Date.now() + 2_000 + for (;;) { + try { + await access(marker) + break + } catch (error: unknown) { + if (Date.now() >= deadline) throw error + await delay(10) + } + } + const completed = await Promise.race([ + running.then(() => true), + delay(500).then(() => false), + ]) + await writeFile(release, '') + const { messages, stderr } = await running + + expect(completed).toBe(true) + expect(stderr).toBe('') + expect(messages.at(-1)).toEqual({ type: 'done', value: [true] }) + } finally { + await writeFile(release, '').catch(() => undefined) + await running?.catch(() => undefined) + await rm(directory, { recursive: true, force: true }) + } + }) + it('prepares the remote runner and returns logs and a lossless completion', async () => { const handle = new FakeHandle((message, current) => { if ((message as { type?: string }).type !== 'boot') return diff --git a/packages/e2b/e2b/tests/composition.e2e.ts b/packages/e2b/e2b/tests/composition.e2e.ts index 4f4ea19eb9..1bfbab1e8e 100644 --- a/packages/e2b/e2b/tests/composition.e2e.ts +++ b/packages/e2b/e2b/tests/composition.e2e.ts @@ -109,12 +109,15 @@ describe.skipIf(!process.env.E2B_API_KEY)('E2B live Loader composition', () => { echo: { waitReason: 'stdin_read', sessionStatus: { kind: 'running' } }, signal: { delivered: true }, interrupted: { sessionStatus: { kind: 'running' } }, + interruptIdentitySafe: true, treeCleanup: true, }, hostileOutput: { error: { kind: 'output-limit' } }, nativeOutput: { error: { kind: 'output-limit' } }, descriptorOutput: { error: { kind: 'output-limit' } }, inheritedOutput: { error: { kind: 'output-limit' } }, + descendantPipe: { value: true, logs: [] }, + descendantCleanup: true, timedOut: { error: { kind: 'timeout' } }, aborted: { error: { kind: 'abort', message: 'live abort' } }, oversizedBoot: { error: { kind: 'worker-exit' } }, diff --git a/packages/e2b/lsp-e2b/README.i18n.yaml b/packages/e2b/lsp-e2b/README.i18n.yaml deleted file mode 100644 index 066820d9fd..0000000000 --- a/packages/e2b/lsp-e2b/README.i18n.yaml +++ /dev/null @@ -1,6 +0,0 @@ -# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each -# side as of the last confirmed-consistent state. Both languages carry equal authority; -# after editing either side, bring the other along and re-record with: -# pnpm run verify-translation-pairing --write packages/e2b/lsp-e2b/README.md -README.md: 6cc118767adc5d2df855b009f48faab596928fb7 -README.zh.md: b384e9efd0e5c62b0ee6e69f73755e1bbe0192df diff --git a/packages/e2b/lsp-e2b/README.md b/packages/e2b/lsp-e2b/README.md deleted file mode 100644 index 6cc118767a..0000000000 --- a/packages/e2b/lsp-e2b/README.md +++ /dev/null @@ -1,47 +0,0 @@ -# @deepseek-ai/dsh-lsp-e2b - -English | [中文](README.zh.md) - -Generic E2B language-server backend for [`ctx.lsp`](../../lsp/lsp/README.md). It runs configured stdio servers and reads their source documents inside the shared `ctx.e2b` sandbox; the provider registry, normalized query results, queues, and protocol connection state remain on the host. - -## Plugin and configuration - -The `lsp-e2b` plugin injects `e2b`, `lsp`, and the concrete `dsh-subprocess-e2b` service. `servers` is a non-empty provider-id table: - -| Server key | Default | Meaning | -|---|---|---| -| `command` | required | Remote executable, absolute or resolved on the sandbox PATH at load. | -| `args` | `[]` | Remote server arguments. | -| `env` | `{}` | Explicit environment entries passed through the subprocess adapter. | -| `extensionToLanguage` | required | Lowercase leading-dot extension to LSP language id. | -| `initializationOptions` / `configuration` | `null` / `null` | Static initialize options and `workspace/configuration` answer. | -| `maxMessageBytes` | `16000000` | Largest LSP message accepted from the server. | -| `maxStderrBytes` | `1000000` | Retained raw server stderr tail. | -| `maxDocumentBytes` | `4000000` | Largest remote source opened for one query. | -| `shutdownTimeoutMs` | `5000` | Graceful protocol-shutdown budget. | -| `killGraceMs` | `2000` | Request-cancel and TERM-to-KILL grace. | - -Provider ids and commands are non-empty; numeric bounds are positive safe integers, and timer values cannot exceed Node's maximum timer delay. Setup uploads one owner-private proxy under `ctx.e2b.runtimeRoot`, resolves Node and every configured server executable remotely, then registers all providers atomically. - -## Remote protocol and filesystem - -E2B command callbacks are text, while LSP is byte-framed. The installed proxy therefore base64-frames raw server stdout, stderr, and stdin as newline-delimited ASCII JSON; the host validates and decodes every frame before handing bytes to the shared `LspInstance` protocol engine. `initialize.processId` is `null` because host and server do not share a process namespace. - -One language-server process is pooled per provider and canonical remote workspace. Queries serialize per workspace but different workspaces run concurrently. Each query canonicalizes the remote workspace and source with `realpath`, rejects paths outside that workspace, then uses a remote helper to open the canonical source without following the final symlink and to verify and read one stable descriptor. The helper requires a regular file and reads at most `maxDocumentBytes + 1` bytes before strict UTF-8 decoding. Queries use the ordinary transient `didOpen` / request / `didClose` lifecycle. A transport failure disposes the instance and retries the read-only query once on a fresh remote process. - -The subprocess adapter owns process groups and escalation, so cancellation and disposal await remote server quiescence. The host owns LSP request ids, pending requests, provider queues, and normalized results. - -## Model Experience - -Indirectly, through `@deepseek-ai/dsh-tool-lsp`, which exposes normalized semantic navigation and hover results without changing its model-facing schema. - -#### KV Cache effect - -No direct invalidation; `dsh-tool-lsp` owns request-prefix changes. - -## Known Limitations and Deferred Work - -- **Configured servers only** — this package does not install language servers, select presets, or synchronize a host workspace into E2B. -- **Host protocol state is not reconnectable** — retaining a sandbox does not restore provider queues, JSON-RPC requests, subprocess handles, or document lifecycle state. -- **SDK output retention remains** — ASCII framing preserves protocol bytes, but E2B and the subprocess adapter still retain callback output in host memory. -- **Sandbox policy is template-owned** — this provider adds no volume, snapshot, credential, or network-policy layer. diff --git a/packages/e2b/lsp-e2b/README.zh.md b/packages/e2b/lsp-e2b/README.zh.md deleted file mode 100644 index b384e9efd0..0000000000 --- a/packages/e2b/lsp-e2b/README.zh.md +++ /dev/null @@ -1,47 +0,0 @@ -# @deepseek-ai/dsh-lsp-e2b - -[English](README.md) | 中文 - -用于 [`ctx.lsp`](../../lsp/lsp/README.md) 的通用 E2B 语言服务器后端。它在共享的 `ctx.e2b` 沙箱内运行已配置的 stdio 服务器并读取其源文档；提供方注册表、规范化查询结果、队列和协议连接状态仍保留在宿主侧。 - -## 插件与配置 - -`lsp-e2b` 插件注入 `e2b`、`lsp` 和具体的 `dsh-subprocess-e2b` 服务。`servers` 是一张非空的提供方 id 表： - -| 服务器键 | 默认值 | 含义 | -|---|---|---| -| `command` | 必填 | 远程可执行文件：绝对路径，或在加载时通过沙箱 PATH 解析。 | -| `args` | `[]` | 远程服务器参数。 | -| `env` | `{}` | 经由进程管理适配器传入的显式环境条目。 | -| `extensionToLanguage` | 必填 | 小写、以点开头的扩展名到 LSP language id 的映射。 | -| `initializationOptions` / `configuration` | `null` / `null` | 静态初始化选项和 `workspace/configuration` 应答。 | -| `maxMessageBytes` | `16000000` | 从服务器接受的 LSP 消息大小上限。 | -| `maxStderrBytes` | `1000000` | 保留的服务器原始 stderr 尾部上限。 | -| `maxDocumentBytes` | `4000000` | 单次查询可打开的最大远程源文件。 | -| `shutdownTimeoutMs` | `5000` | 协议优雅关闭预算。 | -| `killGraceMs` | `2000` | 请求取消与 TERM 到 KILL 升级的宽限期。 | - -提供方 id 与命令必须非空；数值上限必须是正的安全整数，定时器取值不得超过 Node 的最大定时器延迟。设置阶段会在 `ctx.e2b.runtimeRoot` 下上传一个仅供所有者使用的私有代理，在远程解析 Node 和每个已配置服务器的可执行文件，再以原子方式注册所有提供方。 - -## 远程协议与文件系统 - -E2B 命令回调交付的是文本，而 LSP 按字节分帧。因此，已安装的代理会把服务器 stdout、stderr 和 stdin 的原始字节进行 base64 编码，封装为以换行分隔的 ASCII JSON 帧；宿主会验证并解码每一帧，再把字节交给共享的 `LspInstance` 协议引擎。`initialize.processId` 为 `null`，因为宿主与服务器不共享进程命名空间。 - -每个提供方与规范化远程工作区的组合共享一个池化语言服务器进程。同一工作区的查询串行执行，不同工作区的查询并发运行。每项查询都会使用 `realpath` 规范化远程工作区与源文件，拒绝工作区外的路径，再由远程 helper 以不跟随最终符号链接的方式打开规范化源文件，并在同一个稳定描述符上完成验证与读取。该 helper 要求目标为普通文件，最多读取 `maxDocumentBytes + 1` 字节，随后执行严格的 UTF-8 解码。查询采用常规的临时 `didOpen`／请求／`didClose` 生命周期。传输失败会 dispose（资源释放）该实例，并在全新的远程进程上重试一次只读查询。 - -进程管理适配器负责进程组和终止升级，因此取消与资源释放都会等待远程服务器完全停稳。宿主负责 LSP 请求 id、待完成请求、提供方队列和规范化结果。 - -## 模型体验 - -通过 `@deepseek-ai/dsh-tool-lsp` 间接影响模型；该包会公开规范化的语义导航与悬停结果，而不改变其面向模型的 schema。 - -#### KV Cache 影响 - -不会直接失效；请求前缀变更由 `dsh-tool-lsp` 负责。 - -## 已知限制与暂缓工作 - -- **仅支持已配置的服务器**：本包不会安装语言服务器、选择 preset，或把宿主工作区同步到 E2B。 -- **宿主协议状态不可重连**：保留沙箱并不会恢复提供方队列、JSON-RPC 请求、进程管理句柄或文档生命周期状态。 -- **SDK 仍会保留输出**：ASCII 分帧能保留协议字节，但 E2B 和进程管理适配器仍会在宿主内存中保留回调输出。 -- **沙箱策略归模板负责**：本提供方不会额外增加卷、快照、凭据或网络策略层。 diff --git a/packages/e2b/lsp-e2b/package.json b/packages/e2b/lsp-e2b/package.json deleted file mode 100644 index 0671341f3c..0000000000 --- a/packages/e2b/lsp-e2b/package.json +++ /dev/null @@ -1,52 +0,0 @@ -{ - "name": "@deepseek-ai/dsh-lsp-e2b", - "description": "E2B language-server provider for DeepSeek Harness", - "version": "0.0.1", - "private": true, - "type": "module", - "main": "lib/index.js", - "types": "lib/types/index.d.ts", - "exports": { - ".": { - "types": "./lib/types/index.d.ts", - "default": "./lib/index.js" - }, - "./invariant": { - "types": "./lib/types/invariant.d.ts", - "default": "./lib/invariant.js" - }, - "./src/*": "./src/*", - "./package.json": "./package.json" - }, - "files": [ - "lib/index.js", - "lib/invariant.js", - "lib/types/**/*.d.ts", - "lib/types/**/*.d.ts.map", - "src" - ], - "license": "BSD-3-Clause", - "peerDependencies": { - "@deepseek-ai/dsh-e2b": "^0.0.1", - "@deepseek-ai/dsh-invariants": "^0.0.1", - "@deepseek-ai/dsh-lsp": "^0.0.1", - "@deepseek-ai/dsh-lsp-local": "^0.0.1", - "@deepseek-ai/dsh-subprocess": "^0.0.1", - "@deepseek-ai/dsh-subprocess-e2b": "^0.0.1", - "@deepseek-ai/dsh-timeout": "^0.0.1", - "cordis": "^4.0.0-rc.7" - }, - "dependencies": { - "schemastery": "^3.18.0" - }, - "devDependencies": { - "@deepseek-ai/dsh-e2b": "workspace:^", - "@deepseek-ai/dsh-invariants": "workspace:^", - "@deepseek-ai/dsh-lsp": "workspace:^", - "@deepseek-ai/dsh-lsp-local": "workspace:^", - "@deepseek-ai/dsh-subprocess": "workspace:^", - "@deepseek-ai/dsh-subprocess-e2b": "workspace:^", - "@deepseek-ai/dsh-timeout": "workspace:^", - "cordis": "^4.0.0-rc.7" - } -} diff --git a/packages/e2b/lsp-e2b/src/index.ts b/packages/e2b/lsp-e2b/src/index.ts deleted file mode 100644 index fa6b3729da..0000000000 --- a/packages/e2b/lsp-e2b/src/index.ts +++ /dev/null @@ -1,441 +0,0 @@ -/** E2B filesystem and process backend for the harness LSP capability seam. */ - -import { Buffer } from 'node:buffer' -import { posix } from 'node:path' -import type { Context } from 'cordis' -import z from 'schemastery' -import { - FileType, - quoteE2BShellArg, - resolveE2BExecutable, -} from '@deepseek-ai/dsh-e2b' -import type { Sandbox } from '@deepseek-ai/dsh-e2b' -import { LspError, LspProviderId } from '@deepseek-ai/dsh-lsp' -import type { LspProvider, LspProviderQuery, LspQueryResult } from '@deepseek-ai/dsh-lsp' -import { LspInstance } from '@deepseek-ai/dsh-lsp-local' -import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess' -import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b' -import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' -import { LSP_PROXY_SOURCE } from './proxy-source.ts' -import { E2BLspTransport } from './transport.ts' - -export { E2BLspTransport } from './transport.ts' - -/** Cordis plugin name. */ -export const name = 'lsp-e2b' -/** Services required by the remote provider. */ -export const inject = ['e2b', 'lsp', 'subprocess'] - -const DEFAULT_MAX_MESSAGE_BYTES = 16_000_000 -const DEFAULT_MAX_STDERR_BYTES = 1_000_000 -const DEFAULT_MAX_DOCUMENT_BYTES = 4_000_000 -const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000 -const DEFAULT_KILL_GRACE_MS = 2_000 - -/* jscpd:ignore-start -- Loader requires each backend to expose its own statically walkable server schema. */ -/** One configured language server inside the shared E2B sandbox. */ -export interface LspE2BServerConfig { - /** Remote executable, absolute or resolved on the sandbox PATH. */ - command: string - /** Lowercase leading-dot extension to LSP language id. */ - extensionToLanguage: Record - /** Remote executable arguments. */ - args?: string[] - /** Explicit remote environment overrides. */ - env?: Record - /** Static `initialize` options. */ - initializationOptions?: unknown - /** Static answer to every `workspace/configuration` item. */ - configuration?: unknown - /** Largest LSP message accepted from the server. */ - maxMessageBytes?: number - /** Largest remote stderr tail retained for diagnostics. */ - maxStderrBytes?: number - /** Largest remote source opened for one query. */ - maxDocumentBytes?: number - /** Graceful LSP shutdown budget. */ - shutdownTimeoutMs?: number - /** Request-cancel and TERM-to-KILL grace. */ - killGraceMs?: number -} - -/** Plugin configuration. */ -export interface Config { - /** Non-empty provider-id to remote-server table. */ - servers: Record -} - -type ResolvedServerConfig = Required - -const ServerConfig: z = z.object({ - command: z.string().required(), - args: z.array(String).default([]), - env: z.dict(String).default({}), - extensionToLanguage: z.dict(String).required(), - initializationOptions: z.any().default(null), - configuration: z.any().default(null), - maxMessageBytes: z.number().default(DEFAULT_MAX_MESSAGE_BYTES), - maxStderrBytes: z.number().default(DEFAULT_MAX_STDERR_BYTES), - maxDocumentBytes: z.number().default(DEFAULT_MAX_DOCUMENT_BYTES), - shutdownTimeoutMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_SHUTDOWN_TIMEOUT_MS), - killGraceMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_KILL_GRACE_MS), -}) - -/** Schemastery plugin configuration. */ -export const Config: z = z.object({ - servers: z.dict(ServerConfig).required(), -}) -/* jscpd:ignore-end */ - -interface RemoteSource { - canonicalPath: string - text: string -} - -interface RemoteSourceReadResponse { - kind: 'ok' | 'not-file' | 'oversize' | 'grew' | 'open-error' - data?: string - size?: number - message?: string -} - -const SOURCE_READER_SOURCE = String.raw` -/* dsh-e2b-source-reader */ -const fs = require('node:fs') -const path = process.argv[1] -const maxBytes = Number(process.argv[2]) -let descriptor -let response -try { - descriptor = fs.openSync(path, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK) - const info = fs.fstatSync(descriptor) - if (!info.isFile()) response = { kind: 'not-file' } - else if (info.size > maxBytes) response = { kind: 'oversize', size: info.size } - else { - const chunks = [] - let total = 0 - while (total <= maxBytes) { - const chunk = Buffer.allocUnsafe(Math.min(65536, maxBytes - total + 1)) - const bytesRead = fs.readSync(descriptor, chunk, 0, chunk.length, null) - if (bytesRead === 0) break - chunks.push(chunk.subarray(0, bytesRead)) - total += bytesRead - } - response = total > maxBytes - ? { kind: 'grew' } - : { kind: 'ok', data: Buffer.concat(chunks, total).toString('base64') } - } -} catch (error) { - response = { kind: 'open-error', message: error instanceof Error ? error.message : String(error) } -} -if (descriptor !== undefined) fs.closeSync(descriptor) -process.stdout.write(JSON.stringify(response)) -` - -function abortReason(signal: AbortSignal): unknown { - try { - signal.throwIfAborted() - } catch (error: unknown) { - return error - } - return new DOMException('The operation was aborted', 'AbortError') -} - -function abortable(promise: Promise, signal?: AbortSignal): Promise { - if (signal === undefined) return promise - // AbortSignal permits opaque reasons, and callers observe the exact reason. - // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors - if (signal.aborted) return Promise.reject(abortReason(signal)) - return new Promise((resolve, reject) => { - const onAbort = (): void => { - cleanup() - // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- Preserve the signal's exact opaque reason. - reject(abortReason(signal)) - } - const cleanup = (): void => { signal.removeEventListener('abort', onAbort) } - signal.addEventListener('abort', onAbort, { once: true }) - void promise.then( - (value) => { cleanup(); resolve(value) }, - (error: unknown) => { cleanup(); reject(error instanceof Error ? error : new Error(String(error))) }, - ) - }) -} - -function validateServerConfig(providerId: string, config: ResolvedServerConfig): void { - if (config.command.length === 0) throw new Error(`lsp-e2b: servers.${providerId}.command must be non-empty`) - for (const name of ['maxMessageBytes', 'maxStderrBytes', 'maxDocumentBytes', 'shutdownTimeoutMs', 'killGraceMs'] as const) { - const value = config[name] - if (!Number.isSafeInteger(value) || value <= 0 || (name.endsWith('Ms') && value > MAX_TIMER_DELAY_MS)) { - throw new Error(`lsp-e2b: servers.${providerId}.${name} must be a positive safe integer${name.endsWith('Ms') ? ` no greater than ${MAX_TIMER_DELAY_MS}` : ''}`) - } - } -} - -async function canonicalRemotePath(sandbox: Sandbox, path: string, signal?: AbortSignal): Promise { - signal?.throwIfAborted() - const result = await sandbox.commands.run(`realpath -e -- ${quoteE2BShellArg(path)}`, signal === undefined ? {} : { signal }) - signal?.throwIfAborted() - const canonical = result.stdout.trim() - if (!posix.isAbsolute(canonical) || canonical.includes('\n')) throw new Error(`remote path ${JSON.stringify(path)} did not resolve canonically`) - return canonical -} - -/** - * Canonicalize and validate one workspace inside E2B. - * @param sandbox - Shared sandbox that owns the workspace. - * @param workspaceRoot - Remote workspace path supplied by the query. - * @param signal - Optional query cancellation signal. - * @returns The canonical remote directory path. - */ -export async function canonicalizeE2BWorkspace( - sandbox: Sandbox, - workspaceRoot: string, - signal?: AbortSignal, -): Promise { - const canonical = await canonicalRemotePath(sandbox, workspaceRoot, signal) - const info = await sandbox.files.getInfo(canonical, signal === undefined ? {} : { signal }) - signal?.throwIfAborted() - if (info.type !== FileType.DIR) throw new Error(`workspace root ${JSON.stringify(workspaceRoot)} is not a directory`) - return canonical -} - -/** - * Resolve, contain, and read one UTF-8 query source inside E2B. - * @param sandbox - Shared sandbox that owns the source. - * @param filePath - Absolute path or path relative to the canonical workspace. - * @param workspace - Canonical remote workspace directory. - * @param maxDocumentBytes - Maximum bytes read through the stable remote handle. - * @param nodeExecutable - Resolved remote Node executable used by the bounded reader. - * @param signal - Optional query cancellation signal. - * @returns The canonical source path and decoded text. - */ -export async function readE2BSource( - sandbox: Sandbox, - filePath: string, - workspace: string, - maxDocumentBytes: number, - nodeExecutable: string, - signal?: AbortSignal, -): Promise { - const requested = posix.isAbsolute(filePath) ? filePath : posix.resolve(workspace, filePath) - const canonicalPath = await canonicalRemotePath(sandbox, requested, signal) - const relative = posix.relative(workspace, canonicalPath) - if (relative === '..' || relative.startsWith('../') || posix.isAbsolute(relative)) { - throw new Error(`source ${JSON.stringify(filePath)} resolves outside the workspace`) - } - const command = [ - quoteE2BShellArg(nodeExecutable), - '--input-type=commonjs', - '-e', - quoteE2BShellArg(SOURCE_READER_SOURCE), - quoteE2BShellArg(canonicalPath), - String(maxDocumentBytes), - ].join(' ') - const result = await sandbox.commands.run(command, signal === undefined ? {} : { signal }) - signal?.throwIfAborted() - let response: RemoteSourceReadResponse - try { - response = JSON.parse(result.stdout) as RemoteSourceReadResponse - } catch (error: unknown) { - throw new Error(`source ${JSON.stringify(filePath)} reader returned an invalid response`, { cause: error }) - } - if (response.kind === 'not-file') throw new Error(`source ${JSON.stringify(filePath)} is not a regular file`) - if (response.kind === 'oversize' && Number.isSafeInteger(response.size)) { - throw new Error(`source ${JSON.stringify(filePath)} is ${response.size} bytes, over the ${maxDocumentBytes}-byte limit`) - } - if (response.kind === 'grew') { - throw new Error(`source ${JSON.stringify(filePath)} grew past the ${maxDocumentBytes}-byte limit while reading`) - } - if (response.kind === 'open-error' && typeof response.message === 'string') { - throw new Error(`source ${JSON.stringify(filePath)} could not be opened safely: ${response.message}`) - } - if (response.kind !== 'ok' || typeof response.data !== 'string') { - throw new Error(`source ${JSON.stringify(filePath)} reader returned an invalid response`) - } - const bytes = Buffer.from(response.data, 'base64') - if (bytes.toString('base64') !== response.data || bytes.length > maxDocumentBytes) { - throw new Error(`source ${JSON.stringify(filePath)} reader returned invalid bounded bytes`) - } - let text: string - try { - text = new TextDecoder('utf-8', { fatal: true }).decode(bytes) - } catch (error: unknown) { - throw new Error(`source ${JSON.stringify(filePath)} is not valid UTF-8 text`, { cause: error }) - } - return { canonicalPath, text } -} - -/** - * Encode one absolute remote Linux path as a host-independent file URI. - * @param path - Canonical POSIX path inside E2B. - * @returns The equivalent percent-encoded file URI. - */ -export function e2bFileUri(path: string): string { - if (!posix.isAbsolute(path)) throw new Error(`lsp-e2b: expected an absolute remote path, received ${JSON.stringify(path)}`) - return `file://${path.split('/').map(segment => encodeURIComponent(segment)).join('/')}` -} - -/* jscpd:ignore-start -- Provider identity mirrors the seam while remote source and process ownership stay local. */ -/** One pooled remote provider with an isolated server per canonical workspace. */ -export class E2BLspProvider implements LspProvider { - readonly id: LspProviderId - readonly extensionToLanguage: Readonly> - private readonly instances = new Map() - private readonly queues = new Map>() - private disposed = false - - constructor( - providerId: string, - private readonly sandbox: Sandbox, - private readonly subprocess: E2BSubprocessService, - private readonly config: ResolvedServerConfig, - private readonly executable: string, - private readonly nodeExecutable: string, - private readonly proxyPath: string, - ) { - this.id = LspProviderId(providerId) - this.extensionToLanguage = config.extensionToLanguage - } - /* jscpd:ignore-end */ - - async query(request: LspProviderQuery, signal?: AbortSignal): Promise { - this.assertActive(signal) - const workspace = await canonicalizeE2BWorkspace(this.sandbox, request.workspaceRoot, signal) - this.assertActive(signal) - return this.enqueue(workspace, signal, async () => { - this.assertActive(signal) - const source = await readE2BSource( - this.sandbox, - request.filePath, - workspace, - this.config.maxDocumentBytes, - this.nodeExecutable, - signal, - ) - this.assertActive(signal) - let instance = this.instanceFor(workspace) - try { - return await instance.query(request, source, signal) - } catch (error: unknown) { - if (!instance.isTransportFailure(error)) throw error - await instance.dispose() - this.evict(workspace, instance) - this.assertActive(signal) - instance = this.instanceFor(workspace) - return await instance.query(request, source, signal) - } finally { - if (instance.dead) { - await instance.dispose() - this.evict(workspace, instance) - } - } - }) - } - - /* jscpd:ignore-start -- Queue and pooling semantics are shared; transport failure and disposal identities differ. */ - /** Stop accepting work and await every remote server and queued query. */ - async disposeAll(): Promise { - this.disposed = true - const instances = [...this.instances.values()] - const queues = [...this.queues.values()] - this.instances.clear() - await Promise.all([...instances.map(instance => instance.dispose()), ...queues]) - this.queues.clear() - } - - private assertActive(signal?: AbortSignal): void { - if (this.disposed) throw new LspError('lsp-e2b provider is disposed', 'LSP_DISPOSED') - signal?.throwIfAborted() - } - - private enqueue(workspace: string, signal: AbortSignal | undefined, run: () => Promise): Promise { - const previous = this.queues.get(workspace) ?? Promise.resolve() - const result = abortable(previous, signal).then(run) - const tail = previous.then(() => result).then(() => undefined, () => undefined) - this.queues.set(workspace, tail) - void tail.then(() => { if (this.queues.get(workspace) === tail) this.queues.delete(workspace) }) - return result - } - - private instanceFor(workspace: string): LspInstance { - this.assertActive() - const existing = this.instances.get(workspace) - if (existing !== undefined) return existing - const created = this.createInstance(workspace) - this.instances.set(workspace, created) - return created - } - /* jscpd:ignore-end */ - - private createInstance(workspace: string): LspInstance { - return new LspInstance({ - command: this.executable, - args: this.config.args, - cwd: workspace, - env: this.config.env, - configuration: this.config.configuration, - initializationOptions: this.config.initializationOptions, - maxMessageBytes: this.config.maxMessageBytes, - maxStderrBytes: this.config.maxStderrBytes, - shutdownTimeoutMs: this.config.shutdownTimeoutMs, - killGraceMs: this.config.killGraceMs, - clientProcessId: null, - pathToFileUri: e2bFileUri, - }, (spec: SubprocessSpawnSpec) => { - const originalArgv = Buffer.from(JSON.stringify(spec.argv)).toString('base64') - const inner = this.subprocess.spawn({ - ...spec, - argv: [this.nodeExecutable, this.proxyPath, originalArgv], - stdio: { - stdin: 'pipe', - stdout: 'pipe', - stderr: { maxBytes: this.config.maxStderrBytes }, - }, - }) - const rawBound = Math.max(this.config.maxMessageBytes, this.config.maxStderrBytes) - return new E2BLspTransport(inner, rawBound * 2 + 1024, this.config.maxStderrBytes) - }) - } - - private evict(workspace: string, instance: LspInstance): void { - if (this.instances.get(workspace) === instance) this.instances.delete(workspace) - } -} - -/** Install the proxy, resolve remote commands, and atomically register providers. */ -export async function apply(ctx: Context, config: Config): Promise { - if (!(ctx.subprocess instanceof E2BSubprocessService)) { - throw new Error('lsp-e2b requires @deepseek-ai/dsh-subprocess-e2b as ctx.subprocess') - } - const subprocess = ctx.subprocess - const entries = Object.entries(config.servers) - if (entries.length === 0) throw new Error('lsp-e2b: servers must contain at least one server') - const sandbox = await ctx.e2b.getSandbox() - const proxyPath = posix.join(ctx.e2b.runtimeRoot, 'lsp-stdio-proxy.mjs') - await sandbox.files.write([{ path: proxyPath, data: LSP_PROXY_SOURCE }]) - await sandbox.commands.run(`chmod 600 -- ${quoteE2BShellArg(proxyPath)}`) - const nodeExecutable = await resolveE2BExecutable(sandbox, 'node') - const providers = await Promise.all(entries.map(async ([providerId, raw]) => { - if (providerId.trim() === '') throw new Error('lsp-e2b: server ids must be non-empty strings') - const resolved = raw as ResolvedServerConfig - validateServerConfig(providerId, resolved) - const executable = await resolveE2BExecutable(sandbox, resolved.command) - return new E2BLspProvider(providerId, sandbox, subprocess, resolved, executable, nodeExecutable, proxyPath) - })) - - /* jscpd:ignore-start -- Every provider table publishes atomically through the same registry contract. */ - ctx.effect(() => { - const disposers: Array<() => void> = [] - try { - for (const provider of providers) disposers.push(ctx.lsp.registerProvider(provider)) - } catch (error: unknown) { - for (const dispose of disposers.reverse()) dispose() - throw error - } - return async () => { - for (const dispose of disposers.reverse()) dispose() - await Promise.all(providers.map(provider => provider.disposeAll())) - } - }, 'lsp-e2b.registerProviders') - /* jscpd:ignore-end */ -} diff --git a/packages/e2b/lsp-e2b/src/invariant.ts b/packages/e2b/lsp-e2b/src/invariant.ts deleted file mode 100644 index 129cf0a6d4..0000000000 --- a/packages/e2b/lsp-e2b/src/invariant.ts +++ /dev/null @@ -1,20 +0,0 @@ -/** Package-owned invariant companion for `@deepseek-ai/dsh-lsp-e2b`. */ - -/* jscpd:ignore-start */ -import type { Context } from 'cordis' -import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants' - -const PACKAGE_NAME = '@deepseek-ai/dsh-lsp-e2b' - -/** Cordis companion plugin name. */ -export const name = 'lsp-e2b-invariant' -/** Service required before the companion can reserve package ownership. */ -export const inject = ['invariants'] - -/** No runtime invariant: the LSP registry owns provider publication. */ -const install: InvariantInstaller = () => {} - -/** Register this package's invariant companion. */ -export const apply = (ctx: Context): Promise<() => void> => - Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install)) -/* jscpd:ignore-end */ diff --git a/packages/e2b/lsp-e2b/src/proxy-source.ts b/packages/e2b/lsp-e2b/src/proxy-source.ts deleted file mode 100644 index 5b6fe42d07..0000000000 --- a/packages/e2b/lsp-e2b/src/proxy-source.ts +++ /dev/null @@ -1,55 +0,0 @@ -/** Dependency-free remote stdio proxy installed inside the E2B sandbox. */ - -/** - * Node program that base64-frames raw child stdio so E2B's text callbacks - * never decode the language server's byte stream. - */ -export const LSP_PROXY_SOURCE = String.raw`import { Buffer } from 'node:buffer' -import { spawn } from 'node:child_process' -import { createInterface } from 'node:readline' - -const emit = (message) => { - process.stdout.write(Buffer.from(JSON.stringify(message)).toString('base64') + '\n') -} - -let argv -try { - argv = JSON.parse(Buffer.from(process.argv[2], 'base64').toString('utf8')) - if (!Array.isArray(argv) || argv.length === 0 || argv.some(value => typeof value !== 'string')) throw new Error('invalid argv') -} catch (error) { - process.stderr.write('lsp-e2b proxy argv error: ' + String(error) + '\n') - process.exitCode = 125 - process.stdin.destroy() -} - -if (argv) { - const child = spawn(argv[0], argv.slice(1), { stdio: ['pipe', 'pipe', 'pipe'], env: process.env }) - const input = createInterface({ input: process.stdin, crlfDelay: Infinity }) - child.stdout.on('data', data => { emit({ type: 'stdout', data: data.toString('base64') }) }) - child.stderr.on('data', data => { emit({ type: 'stderr', data: data.toString('base64') }) }) - child.on('error', error => { - emit({ type: 'stderr', data: Buffer.from('language server spawn failed: ' + error.message).toString('base64') }) - }) - child.on('close', (code, signal) => { - emit({ type: 'exit', code, signal }) - input.close() - process.stdin.destroy() - process.exitCode = code === null ? 1 : code - }) - input.on('line', line => { - input.pause() - try { - const message = JSON.parse(Buffer.from(line, 'base64').toString('utf8')) - if (!message || message.type !== 'stdin' || typeof message.data !== 'string') throw new Error('invalid stdin frame') - const data = Buffer.from(message.data, 'base64') - if (data.toString('base64') !== message.data) throw new Error('invalid stdin base64') - if (child.stdin.write(data)) input.resume() - else child.stdin.once('drain', () => { input.resume() }) - } catch (error) { - process.stderr.write('lsp-e2b proxy stdin error: ' + String(error) + '\n') - child.kill('SIGTERM') - } - }) - input.on('close', () => { child.stdin.end() }) -} -` diff --git a/packages/e2b/lsp-e2b/src/transport.ts b/packages/e2b/lsp-e2b/src/transport.ts deleted file mode 100644 index ddd3445db8..0000000000 --- a/packages/e2b/lsp-e2b/src/transport.ts +++ /dev/null @@ -1,183 +0,0 @@ -/** Byte-faithful stdio transport over an E2B subprocess and ASCII/base64 frames. */ - -import { Buffer } from 'node:buffer' -import { PassThrough, Writable } from 'node:stream' -import { E2BFrameDecoder, encodeE2BFrame } from '@deepseek-ai/dsh-e2b' -import type { - SubprocessHandle, - SubprocessOutcome, - SubprocessOutputRead, - SubprocessOutputReader, -} from '@deepseek-ai/dsh-subprocess' - -function asError(error: unknown): Error { - return error instanceof Error ? error : new Error(String(error)) -} - -class ByteTailReader implements SubprocessOutputReader { - private chunks: Buffer[] = [] - private totalBytes = 0 - private retainedBytes = 0 - private dropped = false - - constructor(private readonly maxBytes: number) {} - - append(data: Buffer): void { - if (data.length === 0) return - this.chunks.push(data) - this.totalBytes += data.length - this.retainedBytes += data.length - while (this.retainedBytes > this.maxBytes && this.chunks.length > 0) { - const first = this.chunks[0] as Buffer - const excess = this.retainedBytes - this.maxBytes - if (first.length <= excess) { - this.chunks.shift() - this.retainedBytes -= first.length - } else { - this.chunks[0] = first.subarray(excess) - this.retainedBytes -= excess - } - this.dropped = true - } - } - - readFrom(fromByte: number): SubprocessOutputRead { - if (!Number.isSafeInteger(fromByte) || fromByte < 0) { - throw new Error('subprocess output offset must be a non-negative safe integer') - } - const retainedStart = this.totalBytes - this.retainedBytes - const lossy = fromByte < retainedStart - const start = lossy ? 0 : Math.min(this.retainedBytes, fromByte - retainedStart) - const bytes = Buffer.concat(this.chunks).subarray(start) - return { text: bytes.toString('utf8'), nextOffset: this.totalBytes, lossy: lossy || this.dropped && fromByte === 0 } - } -} - -class FramedInput extends Writable { - constructor(private readonly target: Writable) { - super() - target.on('error', (error: Error) => { this.destroy(error) }) - } - - override _write(chunk: Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void { - this.target.write(encodeE2BFrame({ type: 'stdin', data: chunk.toString('base64') }), callback) - } - - override _final(callback: (error?: Error | null) => void): void { - this.target.end(callback) - } -} - -/** Subprocess handle that decodes a remote proxy's stdout/stderr byte frames. */ -export class E2BLspTransport implements SubprocessHandle { - readonly stdin: Writable - readonly stdout = new PassThrough() - readonly stderr = undefined - readonly collected: SubprocessHandle['collected'] - readonly done: Promise - private readonly stderrTail: ByteTailReader - private readonly decoder: E2BFrameDecoder - private failed: Error | undefined - - /** - * @param inner - E2B subprocess running the installed proxy. - * @param maxFrameBytes - Maximum decoded proxy frame size. - * @param maxStderrBytes - Retained raw language-server stderr tail. - */ - constructor( - private readonly inner: SubprocessHandle, - maxFrameBytes: number, - maxStderrBytes: number, - ) { - if (inner.stdin === undefined || inner.stdout === undefined) { - inner.terminate() - throw new Error('lsp-e2b: proxy subprocess dropped a piped stream') - } - this.stdin = new FramedInput(inner.stdin) - this.stderrTail = new ByteTailReader(maxStderrBytes) - this.collected = { stderr: this.stderrTail } - this.decoder = new E2BFrameDecoder(maxFrameBytes) - inner.stdout.on('data', (chunk: Buffer) => { this.onProxyData(chunk) }) - inner.stdout.on('error', (error: Error) => { this.fail(error) }) - this.done = inner.done.then( - (outcome) => { - this.finishFrames() - this.captureProxyStderr() - this.stdout.end() - if (this.failed !== undefined) throw this.failed - return outcome - }, - (error: unknown) => { - this.captureProxyStderr() - this.stdout.end() - throw error - }, - ) - void this.done.catch(() => {}) - } - - get pid(): number { - return this.inner.pid - } - - terminate(): void { - this.inner.terminate() - } - - async waitForExit(signal?: AbortSignal): Promise { - return await this.inner.waitForExit(signal) - } - - private onProxyData(chunk: Buffer): void { - if (this.failed !== undefined) return - let frames: unknown[] - try { - frames = this.decoder.push(chunk.toString('utf8')) - } catch (error: unknown) { - this.fail(asError(error)) - return - } - for (const frame of frames) this.dispatch(frame) - } - - private dispatch(frame: unknown): void { - if (typeof frame !== 'object' || frame === null) { - this.fail(new Error('lsp-e2b: proxy emitted a malformed frame')) - return - } - const record = frame as Record - if (record.type === 'exit' && (record.code === null || typeof record.code === 'number') && (record.signal === null || typeof record.signal === 'string')) return - if ((record.type !== 'stdout' && record.type !== 'stderr') || typeof record.data !== 'string') { - this.fail(new Error('lsp-e2b: proxy emitted a malformed frame')) - return - } - const data = Buffer.from(record.data, 'base64') - if (data.toString('base64') !== record.data) { - this.fail(new Error('lsp-e2b: proxy emitted invalid base64')) - return - } - if (record.type === 'stdout') this.stdout.write(data) - else this.stderrTail.append(data) - } - - private finishFrames(): void { - if (this.failed !== undefined) return - try { - this.decoder.finish() - } catch (error: unknown) { - this.fail(asError(error)) - } - } - - private captureProxyStderr(): void { - const diagnostic = this.inner.collected.stderr?.readFrom(0).text - if (diagnostic !== undefined && diagnostic.length > 0) this.stderrTail.append(Buffer.from(diagnostic)) - } - - private fail(error: Error): void { - if (this.failed !== undefined) return - this.failed = error - this.inner.terminate() - this.stdout.end() - } -} diff --git a/packages/e2b/lsp-e2b/tests/provider.spec.ts b/packages/e2b/lsp-e2b/tests/provider.spec.ts deleted file mode 100644 index dc92053e4a..0000000000 --- a/packages/e2b/lsp-e2b/tests/provider.spec.ts +++ /dev/null @@ -1,418 +0,0 @@ -import { PassThrough } from 'node:stream' -import { Context } from 'cordis' -import { beforeEach, describe, expect, it, vi } from 'vitest' -import { - FileType, - type Sandbox, -} from '@deepseek-ai/dsh-e2b' -import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess' -import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b' - -const mockedLsp = vi.hoisted(() => { - interface Plan { - query?: (...args: unknown[]) => unknown - transportFailure?: unknown - dead?: boolean - deadAfterQuery?: boolean - disposeError?: unknown - } - - class FakeLspInstance { - static readonly instances: FakeLspInstance[] = [] - static readonly plans: Plan[] = [] - readonly plan: Plan - readonly transport: unknown - readonly queries: unknown[][] = [] - dead: boolean - disposals = 0 - - constructor( - readonly spec: Record, - spawner: (spec: SubprocessSpawnSpec) => unknown, - ) { - this.plan = FakeLspInstance.plans.shift() ?? {} - this.dead = this.plan.dead === true - this.transport = spawner({ - argv: [String(spec.command), ...(spec.args as string[])], - cwd: String(spec.cwd), - env: spec.env as Record, - graceMs: Number(spec.killGraceMs), - stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: Number(spec.maxStderrBytes) } }, - }) - FakeLspInstance.instances.push(this) - } - - async query(...args: unknown[]): Promise { - this.queries.push(args) - const result = await Promise.resolve(this.plan.query?.(...args) ?? { kind: 'hover', hover: null }) - if (this.plan.deadAfterQuery === true) this.dead = true - return result - } - - isTransportFailure(error: unknown): boolean { - return error === this.plan.transportFailure - } - - async dispose(): Promise { - this.disposals += 1 - this.dead = true - if (this.plan.disposeError !== undefined) throw this.plan.disposeError - } - } - - return { FakeLspInstance } -}) - -vi.mock('@deepseek-ai/dsh-lsp-local', () => ({ LspInstance: mockedLsp.FakeLspInstance })) - -import { - E2BLspProvider, - apply, - canonicalizeE2BWorkspace, - e2bFileUri, - readE2BSource, -} from '@deepseek-ai/dsh-lsp-e2b' -import type { LspE2BServerConfig } from '@deepseek-ai/dsh-lsp-e2b' -import * as E2BLspInvariant from '../src/invariant.ts' -import InvariantService from '@deepseek-ai/dsh-invariants' - -class FakeInnerHandle implements SubprocessHandle { - readonly pid = 777 - readonly stdin = new PassThrough() - readonly stdout = new PassThrough() - readonly stderr = undefined - readonly collected = { stderr: { readFrom: () => ({ text: '', nextOffset: 0, lossy: false }) } } - readonly done = Promise.resolve({ exitCode: 0, signal: null }) - terminate(): void {} - async waitForExit(): Promise { return true } -} - -class FakeRemote { - readonly writes: Array> = [] - readonly commands: string[] = [] - readonly infos = new Map() - readonly contents = new Map() - readonly realpaths = new Map() - forcedRealpath: string | undefined - readerResponse: unknown - readerOutput: string | undefined - - constructor() { - this.infos.set('/workspace', { type: FileType.DIR, size: 0 }) - this.infos.set('/workspace/file.ts', { type: FileType.FILE, size: 12 }) - this.contents.set('/workspace/file.ts', Buffer.from('const x = 1')) - this.readerResponse = { kind: 'ok', data: Buffer.from('const x = 1').toString('base64') } - } - - readonly sandbox = { - commands: { - run: async (command: string) => { - this.commands.push(command) - if (command.startsWith('realpath ')) { - const match = /'([^']*)'$/.exec(command) - const requested = match?.[1] ?? '' - return { exitCode: 0, stdout: `${this.forcedRealpath ?? this.realpaths.get(requested) ?? requested}\n`, stderr: '' } - } - if (command.includes('dsh-e2b-source-reader')) { - return { exitCode: 0, stdout: this.readerOutput ?? JSON.stringify(this.readerResponse), stderr: '' } - } - if (command.startsWith('command -v')) return { exitCode: 0, stdout: '/usr/bin/node\n', stderr: '' } - return { exitCode: 0, stdout: '', stderr: '' } - }, - }, - files: { - write: async (files: Array<{ path: string; data: string }>) => { - this.writes.push(files) - return files.map(() => ({})) - }, - getInfo: async (path: string) => { - const info = this.infos.get(path) - if (info === undefined) throw new Error(`missing info for ${path}`) - return info - }, - read: async (path: string) => this.contents.get(path) ?? new Uint8Array(), - }, - } as unknown as Sandbox -} - -function subprocess(spawn = vi.fn((_spec: SubprocessSpawnSpec) => new FakeInnerHandle())): E2BSubprocessService { - const service = Object.create(E2BSubprocessService.prototype) as E2BSubprocessService - Object.defineProperty(service, 'spawn', { value: spawn }) - return service -} - -function server(overrides: Partial = {}): Required { - return { - command: '/usr/bin/server', args: ['--stdio'], env: {}, - extensionToLanguage: { '.ts': 'typescript' }, - initializationOptions: null, configuration: null, - maxMessageBytes: 1_024, maxStderrBytes: 128, maxDocumentBytes: 1_024, - shutdownTimeoutMs: 100, killGraceMs: 50, - ...overrides, - } -} - -function provider(remote = new FakeRemote(), service = subprocess()): E2BLspProvider { - return new E2BLspProvider( - 'fixture', remote.sandbox, service, server(), - '/usr/bin/server', '/usr/bin/node', '/workspace/.dsh-e2b/lsp-proxy.mjs', - ) -} - -function query(workspaceRoot = '/workspace') { - return { - operation: 'hover' as const, - filePath: 'file.ts', - position: { line: 0, character: 1 }, - workspaceRoot, - languageId: 'typescript', - } -} - -beforeEach(() => { - mockedLsp.FakeLspInstance.instances.length = 0 - mockedLsp.FakeLspInstance.plans.length = 0 -}) - -describe('E2B LSP filesystem boundary', () => { - it('canonicalizes a directory and reads a contained UTF-8 source', async () => { - const remote = new FakeRemote() - await expect(canonicalizeE2BWorkspace(remote.sandbox, '/workspace')).resolves.toBe('/workspace') - await expect(readE2BSource(remote.sandbox, 'file.ts', '/workspace', 1_024, '/usr/bin/node')).resolves.toEqual({ - canonicalPath: '/workspace/file.ts', - text: 'const x = 1', - }) - await expect(readE2BSource(remote.sandbox, '/workspace/file.ts', '/workspace', 1_024, '/usr/bin/node')).resolves.toMatchObject({ - canonicalPath: '/workspace/file.ts', - }) - const signal = new AbortController().signal - await expect(canonicalizeE2BWorkspace(remote.sandbox, '/workspace', signal)).resolves.toBe('/workspace') - await expect(readE2BSource(remote.sandbox, 'file.ts', '/workspace', 1_024, '/usr/bin/node', signal)).resolves.toMatchObject({ - canonicalPath: '/workspace/file.ts', - }) - }) - - it('rejects malformed workspaces and source containment/type/size/encoding failures', async () => { - const malformed = new FakeRemote() - malformed.forcedRealpath = 'relative' - await expect(canonicalizeE2BWorkspace(malformed.sandbox, '/workspace')).rejects.toThrow('did not resolve canonically') - malformed.forcedRealpath = '/workspace\nother' - await expect(canonicalizeE2BWorkspace(malformed.sandbox, '/workspace')).rejects.toThrow('did not resolve canonically') - - const notDirectory = new FakeRemote() - notDirectory.infos.set('/workspace', { type: FileType.FILE, size: 0 }) - await expect(canonicalizeE2BWorkspace(notDirectory.sandbox, '/workspace')).rejects.toThrow('not a directory') - - const outside = new FakeRemote() - outside.realpaths.set('/workspace/file.ts', '/outside/file.ts') - await expect(readE2BSource(outside.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('outside the workspace') - - const notFile = new FakeRemote() - notFile.readerResponse = { kind: 'not-file' } - await expect(readE2BSource(notFile.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('not a regular file') - - const tooLarge = new FakeRemote() - tooLarge.readerResponse = { kind: 'oversize', size: 21 } - await expect(readE2BSource(tooLarge.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('over the 20-byte limit') - - const grew = new FakeRemote() - grew.readerResponse = { kind: 'grew' } - await expect(readE2BSource(grew.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('grew past') - - const invalid = new FakeRemote() - invalid.readerResponse = { kind: 'ok', data: Buffer.from([0xff]).toString('base64') } - await expect(readE2BSource(invalid.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('not valid UTF-8') - - const swapped = new FakeRemote() - swapped.readerResponse = { kind: 'open-error', message: 'ELOOP: symbolic link encountered' } - await expect(readE2BSource(swapped.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('opened safely') - - const malformedReader = new FakeRemote() - malformedReader.readerResponse = { kind: 'ok', data: '*' } - await expect(readE2BSource(malformedReader.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('invalid bounded bytes') - malformedReader.readerResponse = { kind: 'oversize', size: 'large' } - await expect(readE2BSource(malformedReader.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('invalid response') - malformedReader.readerOutput = '{' - await expect(readE2BSource(malformedReader.sandbox, 'file.ts', '/workspace', 20, '/usr/bin/node')).rejects.toThrow('invalid response') - - await expect(canonicalizeE2BWorkspace(new FakeRemote().sandbox, '/workspace', AbortSignal.abort('stop'))) - .rejects.toBe('stop') - }) -}) - -describe('E2BLspProvider pooling and lifecycle', () => { - it('reuses one canonical-workspace instance and constructs the remote proxy transport', async () => { - const spawn = vi.fn((_spec: SubprocessSpawnSpec) => new FakeInnerHandle()) - const remote = new FakeRemote() - mockedLsp.FakeLspInstance.plans.push({ query: async () => ({ kind: 'hover', hover: { contents: 'ok' } }) }) - const current = provider(remote, subprocess(spawn)) - - await expect(current.query(query())).resolves.toEqual({ kind: 'hover', hover: { contents: 'ok' } }) - await expect(current.query(query())).resolves.toEqual({ kind: 'hover', hover: { contents: 'ok' } }) - expect(mockedLsp.FakeLspInstance.instances).toHaveLength(1) - expect(mockedLsp.FakeLspInstance.instances[0]?.spec).toMatchObject({ clientProcessId: null, cwd: '/workspace' }) - expect(e2bFileUri('/workspace/a b#c.ts')).toBe('file:///workspace/a%20b%23c.ts') - expect(() => e2bFileUri('relative.ts')).toThrow('absolute remote path') - const pathToFileUri = mockedLsp.FakeLspInstance.instances[0]?.spec.pathToFileUri as (path: string) => string - expect(pathToFileUri('/workspace/a b#c.ts')).toBe('file:///workspace/a%20b%23c.ts') - expect(spawn).toHaveBeenCalledWith(expect.objectContaining({ - argv: ['/usr/bin/node', '/workspace/.dsh-e2b/lsp-proxy.mjs', expect.any(String)], - stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: 128 } }, - })) - expect(current.id).toBe('fixture') - expect(current.extensionToLanguage).toEqual({ '.ts': 'typescript' }) - await current.disposeAll() - expect(mockedLsp.FakeLspInstance.instances[0]?.disposals).toBe(1) - }) - - it('replaces one transport failure, but preserves ordinary query errors', async () => { - const transportFailure = new Error('transport failed') - mockedLsp.FakeLspInstance.plans.push( - { transportFailure, query: async () => { throw transportFailure } }, - { query: async () => ({ kind: 'hover', hover: { contents: 'retried' } }) }, - ) - const retried = provider() - await expect(retried.query(query())).resolves.toMatchObject({ hover: { contents: 'retried' } }) - expect(mockedLsp.FakeLspInstance.instances[0]?.disposals).toBe(1) - expect(mockedLsp.FakeLspInstance.instances).toHaveLength(2) - - const ordinary = new Error('ordinary failure') - mockedLsp.FakeLspInstance.plans.push( - { query: async () => { throw ordinary }, dead: true }, - { query: async () => ({ kind: 'hover', hover: null }) }, - ) - const failed = provider() - await expect(failed.query(query())).rejects.toBe(ordinary) - await expect(failed.query(query())).resolves.toMatchObject({ kind: 'hover' }) - }) - - it('evicts a server that dies after a successful query', async () => { - mockedLsp.FakeLspInstance.plans.push( - { deadAfterQuery: true, query: async () => ({ kind: 'hover', hover: null }) }, - { query: async () => ({ kind: 'hover', hover: null }) }, - ) - const current = provider() - await current.query(query()) - await current.query(query()) - expect(mockedLsp.FakeLspInstance.instances).toHaveLength(2) - expect(mockedLsp.FakeLspInstance.instances[0]?.disposals).toBe(1) - }) - - it('serializes a workspace queue, observes queued abort, and awaits work on disposal', async () => { - const first = Promise.withResolvers() - mockedLsp.FakeLspInstance.plans.push({ query: () => first.promise }) - const current = provider() - const running = current.query(query()) - const controller = new AbortController() - const queued = current.query(query(), controller.signal) - await new Promise(resolve => setImmediate(resolve)) - controller.abort('queued stop') - await expect(queued).rejects.toBe('queued stop') - - const disposing = current.disposeAll() - first.resolve({ kind: 'hover', hover: null }) - await expect(running).resolves.toMatchObject({ kind: 'hover' }) - await disposing - await expect(current.query(query())).rejects.toMatchObject({ code: 'LSP_DISPOSED' }) - }) - - it('covers pre-abort, synthetic abort, resolve, and rejection in the queue race', async () => { - const current = provider() - const internal = current as unknown as { - queues: Map> - enqueue(workspace: string, signal: AbortSignal | undefined, run: () => Promise): Promise - } - await expect(internal.enqueue('pre', AbortSignal.abort('pre-stop'), async () => 'unused')).rejects.toBe('pre-stop') - - const signal = new AbortController().signal - await expect(internal.enqueue('resolve', signal, async () => 'ok')).resolves.toBe('ok') - - const failure = new Error('queue failed') - const rejected = Promise.reject(failure) - void rejected.catch(() => {}) - internal.queues.set('reject', rejected) - await expect(internal.enqueue('reject', signal, async () => 'unused')).rejects.toBe(failure) - - // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- Exercise normalization at the promise boundary. - const opaque = Promise.reject('opaque queue failure') - void opaque.catch(() => {}) - internal.queues.set('opaque', opaque) - await expect(internal.enqueue('opaque', signal, async () => 'unused')).rejects.toEqual(new Error('opaque queue failure')) - - const synthetic = { - aborted: false, - reason: undefined, - throwIfAborted() {}, - addEventListener(_type: string, listener: () => void) { listener() }, - removeEventListener() {}, - } as unknown as AbortSignal - await expect(internal.enqueue('synthetic', synthetic, async () => 'unused')).rejects.toMatchObject({ name: 'AbortError' }) - await current.disposeAll() - }) -}) - -describe('lsp-e2b plugin composition', () => { - function pluginContext( - remote: FakeRemote, - service: E2BSubprocessService, - registerProvider = vi.fn(() => vi.fn()), - ) { - const effects: Array<() => void | Promise> = [] - const ctx = { - subprocess: service, - e2b: { - runtimeRoot: '/workspace/.dsh-e2b', - getSandbox: async () => remote.sandbox, - }, - lsp: { registerProvider }, - effect: (callback: () => (() => void | Promise)) => { effects.push(callback()) }, - } as unknown as Context - return { ctx, effects, registerProvider } - } - - it('installs one proxy, resolves commands, registers providers, and disposes them', async () => { - const remote = new FakeRemote() - const fixture = pluginContext(remote, subprocess()) - await apply(fixture.ctx, { servers: { one: server(), two: server({ command: 'server-two' }) } }) - expect(remote.writes).toHaveLength(1) - expect(remote.writes[0]?.[0]?.path).toBe('/workspace/.dsh-e2b/lsp-stdio-proxy.mjs') - expect(remote.commands).toContain("chmod 600 -- '/workspace/.dsh-e2b/lsp-stdio-proxy.mjs'") - expect(fixture.registerProvider).toHaveBeenCalledTimes(2) - await fixture.effects[0]?.() - }) - - it('rolls back partial registration and rejects invalid composition/configuration', async () => { - const remote = new FakeRemote() - const firstDispose = vi.fn() - const register = vi.fn() - .mockReturnValueOnce(firstDispose) - .mockImplementationOnce(() => { throw new Error('duplicate provider') }) - const rollback = pluginContext(remote, subprocess(), register) - await expect(apply(rollback.ctx, { servers: { one: server(), two: server() } })).rejects.toThrow('duplicate provider') - expect(firstDispose).toHaveBeenCalledOnce() - - const wrong = pluginContext(remote, {} as E2BSubprocessService) - await expect(apply(wrong.ctx, { servers: { one: server() } })).rejects.toThrow('dsh-subprocess-e2b') - - const empty = pluginContext(remote, subprocess()) - await expect(apply(empty.ctx, { servers: {} })).rejects.toThrow('at least one server') - - for (const [id, config] of [ - ['', server()], - ['one', server({ command: '' })], - ['one', server({ maxMessageBytes: 0 })], - ['one', server({ maxStderrBytes: 1.5 })], - ['one', server({ shutdownTimeoutMs: 2_147_483_648 })], - ] as const) { - const fixture = pluginContext(new FakeRemote(), subprocess()) - await expect(apply(fixture.ctx, { servers: { [id]: config } })).rejects.toThrow() - } - }) - - it('registers the package-owned invariant companion', async () => { - const ctx = new Context() - await ctx.plugin(InvariantService, { enabled: true }) - const fiber = await ctx.plugin(E2BLspInvariant).await() - await fiber.dispose() - }) -}) diff --git a/packages/e2b/lsp-e2b/tests/transport.spec.ts b/packages/e2b/lsp-e2b/tests/transport.spec.ts deleted file mode 100644 index ae5106bc09..0000000000 --- a/packages/e2b/lsp-e2b/tests/transport.spec.ts +++ /dev/null @@ -1,181 +0,0 @@ -import { once } from 'node:events' -import { PassThrough } from 'node:stream' -import { describe, expect, it } from 'vitest' -import { encodeE2BFrame } from '@deepseek-ai/dsh-e2b' -import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess' -import { E2BLspTransport } from '@deepseek-ai/dsh-lsp-e2b' - -class FakeHandle implements SubprocessHandle { - readonly pid = 321 - readonly stdin: PassThrough | undefined - readonly stdout: PassThrough | undefined - readonly stderr = undefined - readonly collected: SubprocessHandle['collected'] - readonly result = Promise.withResolvers() - readonly done = this.result.promise - terminated = 0 - waitResult = true - - constructor(options: { stdin?: boolean; stdout?: boolean; diagnostic?: string } = {}) { - this.stdin = options.stdin === false ? undefined : new PassThrough() - this.stdout = options.stdout === false ? undefined : new PassThrough() - this.collected = options.diagnostic === undefined - ? {} - : { stderr: { readFrom: () => ({ text: options.diagnostic as string, nextOffset: 0, lossy: false }) } } - } - - terminate(): void { - this.terminated += 1 - } - - async waitForExit(): Promise { - return this.waitResult - } - - resolve(outcome: SubprocessOutcome = { exitCode: 0, signal: null }): void { - this.stdout?.end() - this.result.resolve(outcome) - } - - reject(error: unknown): void { - this.stdout?.end() - this.result.reject(error) - } -} - -function proxyFrame(type: 'stdout' | 'stderr', data: string | Buffer): string { - return encodeE2BFrame({ type, data: Buffer.from(data).toString('base64') }) -} - -describe('E2BLspTransport', () => { - it('frames stdin, decodes split byte output, and exposes handle lifecycle', async () => { - const inner = new FakeHandle({ diagnostic: 'proxy diagnostic' }) - const transport = new E2BLspTransport(inner, 1_024, 64) - let stdin = '' - inner.stdin?.on('data', (chunk) => { stdin += String(chunk) }) - let stdout = Buffer.alloc(0) - transport.stdout.on('data', (chunk) => { stdout = Buffer.concat([stdout, chunk]) }) - - transport.stdin.write(Buffer.from([0, 0xff])) - await new Promise(resolve => setImmediate(resolve)) - const encodedInput = stdin.trim() - const input = JSON.parse(Buffer.from(encodedInput, 'base64').toString('utf8')) as Record - expect(input).toEqual({ type: 'stdin', data: 'AP8=' }) - - const frames = proxyFrame('stdout', Buffer.from([0, 0xff])) - + proxyFrame('stderr', 'server diagnostic') - + encodeE2BFrame({ type: 'exit', code: 0, signal: null }) - inner.stdout?.write(frames.slice(0, 7)) - inner.stdout?.write(frames.slice(7)) - inner.stdout?.write(proxyFrame('stderr', '')) - inner.stdout?.write(encodeE2BFrame({ type: 'exit', code: null, signal: 'SIGTERM' })) - inner.resolve() - - await expect(transport.done).resolves.toEqual({ exitCode: 0, signal: null }) - expect(stdout).toEqual(Buffer.from([0, 0xff])) - expect(transport.collected.stderr?.readFrom(0).text).toBe('server diagnosticproxy diagnostic') - expect(transport.pid).toBe(321) - inner.waitResult = false - await expect(transport.waitForExit()).resolves.toBe(false) - transport.terminate() - expect(inner.terminated).toBe(1) - }) - - it('ends the inner stdin and retains a bounded byte tail with independent offsets', async () => { - const inner = new FakeHandle() - const transport = new E2BLspTransport(inner, 1_024, 4) - const finished = once(inner.stdin!, 'finish') - transport.stdin.end() - await finished - - inner.stdout?.write(proxyFrame('stderr', 'ab')) - inner.stdout?.write(proxyFrame('stderr', 'cdef')) - const reader = transport.collected.stderr! - expect(reader.readFrom(0)).toEqual({ text: 'cdef', nextOffset: 6, lossy: true }) - expect(reader.readFrom(2)).toEqual({ text: 'cdef', nextOffset: 6, lossy: false }) - expect(reader.readFrom(5)).toEqual({ text: 'f', nextOffset: 6, lossy: false }) - expect(reader.readFrom(99)).toEqual({ text: '', nextOffset: 6, lossy: false }) - expect(() => reader.readFrom(-1)).toThrow('non-negative safe integer') - expect(() => reader.readFrom(1.5)).toThrow('non-negative safe integer') - inner.resolve() - await transport.done - - const partialInner = new FakeHandle() - const partial = new E2BLspTransport(partialInner, 1_024, 4) - partialInner.stdout?.write(proxyFrame('stderr', 'abcdef')) - expect(partial.collected.stderr?.readFrom(0)).toEqual({ text: 'cdef', nextOffset: 6, lossy: true }) - partialInner.resolve() - await partial.done - }) - - it.each([ - ['non-object', encodeE2BFrame(null), 'malformed frame'], - ['wrong type', encodeE2BFrame({ type: 'other', data: '' }), 'malformed frame'], - ['invalid exit', encodeE2BFrame({ type: 'exit', code: 'zero', signal: null }), 'malformed frame'], - ['missing data', encodeE2BFrame({ type: 'stdout' }), 'malformed frame'], - ['invalid base64', encodeE2BFrame({ type: 'stdout', data: 'abc' }), 'invalid base64'], - ['invalid outer frame', 'not-base64\n', 'invalid base64'], - ['non-ASCII outer frame', 'é', 'non-ASCII'], - ])('fails %s proxy traffic and terminates the group', async (_name, frame, message) => { - const inner = new FakeHandle() - const transport = new E2BLspTransport(inner, 1_024, 32) - inner.stdout?.write(frame) - inner.stdout?.write(proxyFrame('stdout', 'ignored')) - inner.stdout?.emit('error', new Error('second failure')) - inner.resolve() - await expect(transport.done).rejects.toThrow(message) - expect(inner.terminated).toBe(1) - }) - - it('fails truncated frames and preserves inner spawn rejection', async () => { - const truncatedInner = new FakeHandle() - const truncated = new E2BLspTransport(truncatedInner, 1_024, 32) - truncatedInner.stdout?.write('YQ==') - truncatedInner.resolve() - await expect(truncated.done).rejects.toThrow('mid-frame') - - const rejectedInner = new FakeHandle({ diagnostic: 'tail' }) - const rejected = new E2BLspTransport(rejectedInner, 1_024, 32) - rejectedInner.reject(new Error('spawn failed')) - await expect(rejected.done).rejects.toThrow('spawn failed') - expect(rejected.collected.stderr?.readFrom(0).text).toBe('tail') - }) - - it('forwards output and input stream errors without an unhandled inner error', async () => { - const outputInner = new FakeHandle() - const output = new E2BLspTransport(outputInner, 1_024, 32) - outputInner.stdout?.emit('error', new Error('proxy stdout failed')) - outputInner.resolve() - await expect(output.done).rejects.toThrow('proxy stdout failed') - - const inputInner = new FakeHandle() - const input = new E2BLspTransport(inputInner, 1_024, 32) - const outerError = once(input.stdin, 'error') - inputInner.stdin?.emit('error', new Error('proxy stdin failed')) - await expect(outerError).resolves.toMatchObject([{ message: 'proxy stdin failed' }]) - inputInner.resolve() - await input.done - }) - - it('normalizes a non-Error decoder throw', async () => { - const inner = new FakeHandle() - const transport = new E2BLspTransport(inner, 1_024, 32) - const internal = transport as unknown as { - decoder: { push(chunk: string): unknown[] } - onProxyData(chunk: Buffer): void - } - internal.decoder = { push: () => { throw 'raw decoder failure' } } - internal.onProxyData(Buffer.from('x')) - inner.resolve() - await expect(transport.done).rejects.toThrow('raw decoder failure') - }) - - it('rejects a subprocess that drops either required pipe', () => { - const missingStdin = new FakeHandle({ stdin: false }) - const missingStdout = new FakeHandle({ stdout: false }) - expect(() => new E2BLspTransport(missingStdin, 10, 10)).toThrow('dropped a piped stream') - expect(() => new E2BLspTransport(missingStdout, 10, 10)).toThrow('dropped a piped stream') - expect(missingStdin.terminated).toBe(1) - expect(missingStdout.terminated).toBe(1) - }) -}) diff --git a/packages/e2b/lsp-e2b/tsconfig.json b/packages/e2b/lsp-e2b/tsconfig.json deleted file mode 100644 index a6c0fd8ece..0000000000 --- a/packages/e2b/lsp-e2b/tsconfig.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "extends": "../../../tsconfig.base.json", - "compilerOptions": { - "rootDir": "src", - "outDir": "lib/types" - }, - "include": ["src"], - "references": [ - { "path": "../../../vendor/cosmokit" }, - { "path": "../../../vendor/cordis" }, - { "path": "../../../vendor/schemastery" }, - { "path": "../e2b" }, - { "path": "../../lsp/lsp" }, - { "path": "../../lsp/lsp-local" }, - { "path": "../../subprocess/subprocess" }, - { "path": "../subprocess-e2b" }, - { "path": "../../util/timeout" }, - { "path": "../../support/invariants" } - ] -} diff --git a/packages/e2b/pty-e2b/README.i18n.yaml b/packages/e2b/pty-e2b/README.i18n.yaml index ddae4c21db..580f874e62 100644 --- a/packages/e2b/pty-e2b/README.i18n.yaml +++ b/packages/e2b/pty-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/pty-e2b/README.md -README.md: d6d8e05f13701b85cee9a9646bfdd82ac07769da -README.zh.md: 265257c72e7bcbd2dfd805b343db5bab7173ab51 +README.md: d1b1731cd7b65577a18814a90363acdc66d7d262 +README.zh.md: 4aa2cc8d661e61e80ef61693422a7b6f73c2cb67 diff --git a/packages/e2b/pty-e2b/README.md b/packages/e2b/pty-e2b/README.md index d6d8e05f13..d1b1731cd7 100644 --- a/packages/e2b/pty-e2b/README.md +++ b/packages/e2b/pty-e2b/README.md @@ -26,7 +26,7 @@ Numeric values are positive safe integers, `backendType` is non-empty, and `maxR The backend uses E2B's byte-oriented PTY callback with a streaming fatal UTF-8 decoder, then the backend-neutral line sanitizer and bounded buffers from `dsh-pty`. It installs a controlled Bash prompt marker and waits for printable prompt text; when that marker is unavailable, observed output plus the configured silence bound yields `inferred_idle`. Startup with no output reaches the absolute timeout and fails instead of publishing an empty session. -Each send writes UTF-8 bytes and an optional carriage-return submit sequence. Cancellation and explicit signals resolve the remote terminal's foreground process group through `ps`, then signal that group; `SIGKILL` refuses to target the shell itself. The backend records the terminal's POSIX session id at startup. Close sends `SIGTERM` to every process group still in that session, escalates survivors to `SIGKILL`, verifies that the session is empty, and does not resolve until the SDK handle reports exit. A startup failure closes the unpublished PTY, and `PtyBackendCleanupError` preserves a concurrent cleanup failure. +Each send writes UTF-8 bytes and an optional carriage-return submit sequence. Cancellation and explicit signals resolve the remote terminal's foreground process group through `ps`, then signal that group; cancellation rechecks the originating send after lookup so a settled operation cannot signal or fail its successor, and `SIGKILL` refuses to target the shell itself. The backend records the terminal's POSIX session id at startup. Close sends `SIGTERM` to every process group still in that session, escalates survivors to `SIGKILL`, verifies that the session is empty, and does not resolve until the SDK handle reports exit. A startup failure closes the unpublished PTY, and `PtyBackendCleanupError` preserves a concurrent cleanup failure. The remote PTY process and its child processes live in E2B. Prompt/readiness state, scrollback, operation handles, owner authority, and SDK event delivery remain in host memory. diff --git a/packages/e2b/pty-e2b/README.zh.md b/packages/e2b/pty-e2b/README.zh.md index 265257c72e..4aa2cc8d66 100644 --- a/packages/e2b/pty-e2b/README.zh.md +++ b/packages/e2b/pty-e2b/README.zh.md @@ -26,7 +26,7 @@ 该后端为 E2B 面向字节的 PTY 回调配备流式、遇到无效序列即失败的 UTF-8 解码器，随后使用 `dsh-pty` 提供的后端无关行清理器与有界缓冲区。它会安装受控的 Bash 提示符标记，并等待可打印的提示符文本；若该标记不可用，系统会在已经观察到输出且达到已配置的静默上限时得出 `inferred_idle`。零输出的启动过程会达到绝对超时并失败，不会发布空会话。 -每次发送都会写入 UTF-8 字节，并可选写入回车提交序列。取消与显式信号会通过 `ps` 确定远程终端的前台进程组，再向该组发送信号；发送 `SIGKILL` 时拒绝以 shell 本身为目标。后端会在启动时记录终端的 POSIX 会话 id。关闭操作会向该会话内仍存在的每个进程组发送 `SIGTERM`，对存活者升级为 `SIGKILL`，验证会话已经清空，并且直到 SDK 句柄报告退出才结算。如果启动失败，系统会关闭尚未发布的 PTY；若清理同时失败，`PtyBackendCleanupError` 会保留这项失败。 +每次发送都会写入 UTF-8 字节，并可选写入回车提交序列。取消与显式信号会通过 `ps` 确定远程终端的前台进程组，再向该组发送信号；取消处理会在查找后重新检查原发送操作是否仍为当前操作，以免已结算的操作向后继操作发送信号或令其失败；发送 `SIGKILL` 时拒绝以 shell 本身为目标。后端会在启动时记录终端的 POSIX 会话 id。关闭操作会向该会话内仍存在的每个进程组发送 `SIGTERM`，对存活者升级为 `SIGKILL`，验证会话已经清空，并且直到 SDK 句柄报告退出才结算。如果启动失败，系统会关闭尚未发布的 PTY；若清理同时失败，`PtyBackendCleanupError` 会保留这项失败。 远程 PTY 进程及其子进程位于 E2B。提示符／就绪状态、scrollback、操作句柄、所有者权限和 SDK 事件交付仍保留在宿主内存中。 diff --git a/packages/e2b/pty-e2b/src/session.ts b/packages/e2b/pty-e2b/src/session.ts index bf8780103d..7742cd44f0 100644 --- a/packages/e2b/pty-e2b/src/session.ts +++ b/packages/e2b/pty-e2b/src/session.ts @@ -226,6 +226,10 @@ export class E2BPtySession implements PtyBackendSession { /* jscpd:ignore-start -- Signal, status, and close methods preserve the seam shape around remote identities. */ async signal(signal: PtySignal): Promise { const pgid = await this.foregroundPgid() + return await this.deliverSignal(signal, pgid) + } + + private async deliverSignal(signal: PtySignal, pgid: number): Promise { if (signal === 'SIGKILL' && pgid === this.pid) { throw new Error('refusing to SIGKILL the E2B PTY shell; use terminal_close') } @@ -302,7 +306,15 @@ export class E2BPtySession implements PtyBackendSession { private interrupt(operation: E2BSendOperation): void { if (this.active !== operation) return - void this.signal('SIGINT').catch((error: unknown) => { this.failActive(error) }) + void this.interruptActive(operation).catch((error: unknown) => { + if (this.active === operation) this.failActive(error) + }) + } + + private async interruptActive(operation: E2BSendOperation): Promise { + const pgid = await this.foregroundPgid() + if (this.active !== operation) return + await this.deliverSignal('SIGINT', pgid) } private async foregroundPgid(): Promise { diff --git a/packages/e2b/pty-e2b/tests/session.spec.ts b/packages/e2b/pty-e2b/tests/session.spec.ts index 554a4a3590..1fedf6bbac 100644 --- a/packages/e2b/pty-e2b/tests/session.spec.ts +++ b/packages/e2b/pty-e2b/tests/session.spec.ts @@ -68,6 +68,7 @@ class FakeSandbox { sendError: unknown signalError: unknown killError: unknown + foregroundLookup: Promise | undefined onTerm: (() => void) | undefined onGroupKill: (() => void) | undefined onKill: (() => void) | undefined @@ -88,7 +89,9 @@ class FakeSandbox { commands: { run: async (command: string): Promise => { this.commands.push(command) - if (command.startsWith('ps -o tpgid')) return { exitCode: 0, stdout: this.pgid, stderr: '' } + if (command.startsWith('ps -o tpgid')) { + return await (this.foregroundLookup ?? Promise.resolve({ exitCode: 0, stdout: this.pgid, stderr: '' })) + } if (command.startsWith('ps -eo sid=')) { return { exitCode: 0, stdout: this.sessionGroups.map(value => `${value}\n`).join(''), stderr: '' } } @@ -273,6 +276,45 @@ describe('E2BPtySession readiness, output, and signals', () => { expect(sendInput).toHaveBeenCalled() }) + it('does not let a stale interrupt signal or fail a successor send', async () => { + vi.useFakeTimers() + const fake = new FakeSandbox() + const handle = new FakePtyHandle() + const session = new E2BPtySession(fake.sandbox, handle.asHandle(), 123, config()) + await initialize(session) + + const lookup = Promise.withResolvers() + fake.foregroundLookup = lookup.promise + const stale = session.startSend({ text: 'old', submit: true }) + expect(stale.cancel()).toBe(true) + session.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) + await vi.advanceTimersByTimeAsync(10) + await stale.done + + const successor = session.startSend({ text: 'new', submit: true }) + fake.signalError = new Error('late interrupt failure') + lookup.resolve({ exitCode: 0, stdout: '789\n', stderr: '' }) + await vi.advanceTimersByTimeAsync(0) + expect(fake.commands).not.toContain('kill -INT -- -789') + session.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) + await vi.advanceTimersByTimeAsync(10) + await expect(successor.done).resolves.toMatchObject({ waitReason: 'stdin_read' }) + + const failedLookup = Promise.withResolvers() + fake.foregroundLookup = failedLookup.promise + const staleFailure = session.startSend({ text: 'old failure', submit: true }) + expect(staleFailure.cancel()).toBe(true) + session.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) + await vi.advanceTimersByTimeAsync(10) + await staleFailure.done + const finalSuccessor = session.startSend({ text: 'new after failure', submit: true }) + failedLookup.reject(new Error('late lookup failure')) + await vi.advanceTimersByTimeAsync(0) + session.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) + await vi.advanceTimersByTimeAsync(10) + await expect(finalSuccessor.done).resolves.toMatchObject({ waitReason: 'stdin_read' }) + }) + it('preserves startup abort reasons and classifies invalid UTF-8 transport failures', async () => { const fake = new FakeSandbox() const abortHandle = new FakePtyHandle()