From 3dea36f1ce43d72413fb2b6e8fc3144b7e358d86 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:26:20 +0800 Subject: [PATCH] fix(e2b): close remote lifecycle gaps --- .../tests/fixtures/e2b/e2b/bin.ts | 71 ++++++- .../tests/fixtures/e2b/e2b/cordis.yml | 4 +- packages/e2b/code-runtime-e2b/src/index.ts | 91 ++++++-- .../code-runtime-e2b/tests/runtime.spec.ts | 142 ++++++++++++- packages/e2b/e2b/src/frame.ts | 23 +- packages/e2b/e2b/src/index.ts | 8 +- packages/e2b/e2b/tests/composition.e2e.ts | 10 +- packages/e2b/e2b/tests/e2b.spec.ts | 25 +++ packages/e2b/fs-e2b/src/index.ts | 9 + packages/e2b/fs-e2b/tests/filesystem.spec.ts | 25 ++- packages/e2b/lsp-e2b/src/index.ts | 11 + packages/e2b/lsp-e2b/tests/provider.spec.ts | 5 + packages/e2b/pty-e2b/src/index.ts | 5 +- packages/e2b/pty-e2b/tests/index.spec.ts | 4 +- packages/e2b/subprocess-e2b/README.i18n.yaml | 4 +- packages/e2b/subprocess-e2b/README.md | 2 +- packages/e2b/subprocess-e2b/README.zh.md | 2 +- packages/e2b/subprocess-e2b/src/process.ts | 47 +++-- .../subprocess-e2b/tests/subprocess.spec.ts | 47 ++++- packages/lsp/lsp-local/src/index.ts | 199 ++++++++---------- packages/lsp/lsp-local/src/instance.ts | 13 +- packages/lsp/lsp-local/tests/instance.spec.ts | 2 + 22 files changed, 568 insertions(+), 181 deletions(-) diff --git a/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts b/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts index 33e9530708..89b09a6818 100644 --- a/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts +++ b/examples/headless-agent/tests/fixtures/e2b/e2b/bin.ts @@ -1,5 +1,5 @@ import { readFile } from 'node:fs/promises' -import { resolve } from 'node:path' +import { posix, resolve } from 'node:path' import { boot } from '@deepseek-ai/dsh-app-boot' import { AgentMessageId } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent' @@ -48,20 +48,68 @@ try { const fromBash = await ctx.fs.resolve('from-bash.txt') const fsRead = await ctx.fs.readText(fromBash) + const environmentHandle = ctx.subprocess.spawn({ + argv: ['env'], + cwd: process.cwd(), + stdio: { stdin: 'ignore', stdout: { maxBytes: 65_536 }, stderr: { maxBytes: 4_096 } }, + graceMs: 500, + env: { + 'FOO-BAR': 'hyphen-value', + DSH_EXPLICIT: 'managed-value', + TOKEN_EXPLICIT: 'credential-value', + }, + }) + const environmentOutcome = await environmentHandle.done + const environmentText = environmentHandle.collected.stdout?.readFrom(0).text + if (environmentOutcome.exitCode !== 0 || environmentText === undefined) { + throw new Error(`E2B subprocess environment probe failed: ${JSON.stringify(environmentOutcome)}`) + } + const environmentLines = new Set(environmentText.trimEnd().split('\n')) + const explicitEnvironment = [ + 'FOO-BAR=hyphen-value', + 'DSH_EXPLICIT=managed-value', + 'TOKEN_EXPLICIT=credential-value', + ].every(entry => environmentLines.has(entry)) + if (!explicitEnvironment) throw new Error(`E2B subprocess dropped an explicit environment entry: ${environmentText}`) + + const spillHandle = ctx.subprocess.spawn({ + argv: ['bash', '-c', "printf '0123456789'; sleep 30"], + cwd: process.cwd(), + stdio: { stdin: 'ignore', stdout: { maxBytes: 4, spill: { maxBytes: 6 } }, stderr: { maxBytes: 4_096 } }, + graceMs: 500, + env: {}, + }) + const spillReader = spillHandle.collected.stdout + if (spillReader === undefined) throw new Error('E2B subprocess omitted its configured stdout collector') + const spillDeadline = Date.now() + 5_000 + while (spillReader.readFrom(0).nextOffset < 10) { + if (Date.now() >= spillDeadline) throw new Error('E2B subprocess did not stream the spill probe output') + await new Promise(resolveDelay => setTimeout(resolveDelay, 20)) + } + const spillPath = posix.join((spillHandle as unknown as { stateDir: string }).stateDir, 'stdout.log') + const liveSpillBytes = (await (await ctx.e2b.getSandbox()).files.getInfo(spillPath)).size + spillHandle.terminate() + const spillOutcome = await spillHandle.done + const spillExited = await spillHandle.waitForExit(AbortSignal.timeout(5_000)) + const spillRead = spillReader.readFrom(0) + if (liveSpillBytes !== 6 || !spillExited || spillRead.spillPath !== undefined) { + throw new Error(`E2B subprocess spill bound failed: ${JSON.stringify({ liveSpillBytes, spillExited, spillRead })}`) + } + const lspFixture = await readFile(new URL('./fixture-lsp.mjs', import.meta.url), 'utf8') const remoteLspFixture = await ctx.fs.resolve('fixture-lsp.mjs') await ctx.fs.writeText(remoteLspFixture, lspFixture, { kind: 'createIfAbsent' }) - const remoteSource = await ctx.fs.resolve('multibyte.ts') + const remoteSource = await ctx.fs.resolve('multibyte # file.ts') await ctx.fs.writeText(remoteSource, 'const café = "你好"\nconsole.log(café)\n', { kind: 'createIfAbsent' }) const hover = await ctx.lsp.query({ operation: 'hover', - filePath: 'multibyte.ts', + filePath: 'multibyte # file.ts', position: { line: 0, character: 7 }, workspaceRoot: process.cwd(), }) const definition = await ctx.lsp.query({ operation: 'goToDefinition', - filePath: 'multibyte.ts', + filePath: 'multibyte # file.ts', position: { line: 0, character: 7 }, workspaceRoot: process.cwd(), }) @@ -143,6 +191,17 @@ try { }) setTimeout(() => { abortController.abort('live abort') }, 50) const aborted = await aborting + const oversizedBoot = await ctx.codeRuntime.run({ + program: `return ${JSON.stringify('x'.repeat(40_000))}`, + bindings: [], + }) + const oversizedReply = await ctx.codeRuntime.run({ + program: 'return await bridge.large(null)', + bindings: [{ + global: 'bridge', + functions: { large: async () => 'x'.repeat(40_000) }, + }], + }) const remoteProcesses = await (await ctx.e2b.getSandbox()).commands.list() const lingeringCodeRunners = remoteProcesses.filter(processInfo => JSON.stringify([processInfo.cmd, processInfo.args]).includes('code-runtime-runner.mjs'), @@ -152,6 +211,8 @@ try { sandboxId: await ctx.e2b.sandboxId, bashRead: bashRead.stdout.text, fsRead, + explicitEnvironment, + spill: { liveBytes: liveSpillBytes, outcome: spillOutcome, read: spillRead }, hover, definition, terminal: { @@ -165,6 +226,8 @@ try { hostileOutput, timedOut, aborted, + oversizedBoot, + oversizedReply, lingeringCodeRunners: lingeringCodeRunners.length, })}\n`) } finally { diff --git a/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml b/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml index f5cf403c4b..6b3350e49c 100644 --- a/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml +++ b/examples/headless-agent/tests/fixtures/e2b/e2b/cordis.yml @@ -2,7 +2,7 @@ name: '@deepseek-ai/dsh-e2b' config: cwd: !!js process.cwd() - timeoutMs: 60000 + timeoutMs: 120000 onTimeout: kill onDispose: kill @@ -55,5 +55,5 @@ maxWallMs: 5000 maxOutputBytes: 4096 maxOldGenerationSizeMb: 128 - maxFrameBytes: 4194304 + maxFrameBytes: 32768 killGraceMs: 500 diff --git a/packages/e2b/code-runtime-e2b/src/index.ts b/packages/e2b/code-runtime-e2b/src/index.ts index fbfbdea91f..a567a2bb6d 100644 --- a/packages/e2b/code-runtime-e2b/src/index.ts +++ b/packages/e2b/code-runtime-e2b/src/index.ts @@ -14,7 +14,7 @@ import type { } from '@deepseek-ai/dsh-code-runtime' import { E2BFrameDecoder, - encodeE2BFrame, + encodeBoundedE2BFrame, quoteE2BShellArg, resolveE2BExecutable, } from '@deepseek-ai/dsh-e2b' @@ -25,6 +25,7 @@ import { } from '@deepseek-ai/dsh-code-runtime-worker' import type { WorkerJsonWire } from '@deepseek-ai/dsh-code-runtime-worker' import { snapshotJsonValue } from '@deepseek-ai/dsh-session' +import type { SubprocessHandle } from '@deepseek-ai/dsh-subprocess' import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import { CODE_RUNNER_SOURCE } from './runner-source.ts' @@ -46,6 +47,7 @@ export interface Config { } type ResolvedConfig = Required +type PreparedRuntime = { node: string; runner: string } interface LiveRun { settle(failure: CodeRunFailure): void @@ -130,7 +132,7 @@ export class E2BCodeRuntime extends CodeRuntime { readonly isolation = 'container' private readonly config: ResolvedConfig - private readonly ready: Promise<{ node: string; runner: string }> + private readonly ready: Promise private readonly live = new Set() private readonly subprocess: E2BSubprocessService private disposed = false @@ -176,25 +178,61 @@ export class E2BCodeRuntime extends CodeRuntime { } catch (error: unknown) { return this.failure({ kind: 'exception', message: messageOf(error) }) } - let runtime: Awaited + let runtime: PreparedRuntime | undefined try { - runtime = await this.ready + runtime = await this.awaitPreparation(request.signal) } catch (error: unknown) { + // Disposal can race the awaited setup despite the synchronous precheck. + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (this.disposed) return this.failure({ kind: 'abort', message: 'runtime disposed' }) return this.failure({ kind: 'worker-exit', message: `E2B runtime setup failed: ${messageOf(error)}` }) } + if (runtime === undefined) { + return this.failure({ kind: 'abort', message: String(request.signal?.reason) }) + } // Disposal can race the awaited remote setup after the pre-await check. + /* v8 ignore start -- requires disposal between promise resolution and its awaiting continuation. */ // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition if (this.disposed) return this.failure({ kind: 'abort', message: 'runtime disposed' }) + /* v8 ignore stop */ return await this.execute(request, code, bindings, runtime) } /* jscpd:ignore-end */ - private async prepare(): Promise<{ node: string; runner: string }> { + private awaitPreparation(signal: AbortSignal | undefined): Promise { + if (signal === undefined) return this.ready + return new Promise((resolve, reject) => { + const onAbort = (): void => { cleanup(); resolve(undefined) } + const cleanup = (): void => { signal.removeEventListener('abort', onAbort) } + signal.addEventListener('abort', onAbort, { once: true }) + if (signal.aborted) { + onAbort() + return + } + void this.ready.then( + (runtime) => { cleanup(); resolve(runtime) }, + (error: unknown) => { + cleanup() + reject(error instanceof Error ? error : new Error(String(error))) + }, + ) + }) + } + + private assertPreparationActive(): void { + if (this.disposed) throw new Error('code-runtime-e2b: runtime disposed during setup') + } + + private async prepare(): Promise { const sandbox = await this.ctx.e2b.getSandbox() + this.assertPreparationActive() const runner = posix.join(this.ctx.e2b.runtimeRoot, 'code-runtime-runner.mjs') await sandbox.files.write([{ path: runner, data: CODE_RUNNER_SOURCE }]) + this.assertPreparationActive() await sandbox.commands.run(`chmod 600 -- ${quoteE2BShellArg(runner)}`) + this.assertPreparationActive() const node = await resolveE2BExecutable(sandbox, 'node') + this.assertPreparationActive() return { node, runner } } @@ -237,16 +275,25 @@ export class E2BCodeRuntime extends CodeRuntime { request: CodeRunRequest, code: string, bindings: Map, - runtime: { node: string; runner: string }, + runtime: PreparedRuntime, ): Promise { - const handle = this.subprocess.spawn({ - argv: [runtime.node, runtime.runner], - cwd: this.ctx.e2b.cwd, - stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: this.config.maxOutputBytes } }, - graceMs: this.config.killGraceMs, - ...request.signal === undefined ? {} : { signal: request.signal }, - env: {}, - }) + let handle: SubprocessHandle + try { + handle = this.subprocess.spawn({ + argv: [runtime.node, runtime.runner], + cwd: this.ctx.e2b.cwd, + stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: this.config.maxOutputBytes } }, + graceMs: this.config.killGraceMs, + ...request.signal === undefined ? {} : { signal: request.signal }, + env: {}, + }) + } catch (error: unknown) { + if (this.disposed) return this.failure({ kind: 'abort', message: 'runtime disposed' }) + if (request.signal?.aborted === true) { + return this.failure({ kind: 'abort', message: String(request.signal.reason) }) + } + return this.failure({ kind: 'worker-exit', message: `E2B runtime spawn failed: ${messageOf(error)}` }) + } if (handle.stdin === undefined || handle.stdout === undefined) { handle.terminate() await Promise.allSettled([handle.done]) @@ -279,7 +326,6 @@ export class E2BCodeRuntime extends CodeRuntime { settled = true clearTimeout(wallTimer.current) request.signal?.removeEventListener('abort', onAbort) - this.live.delete(live) void new Promise((resume) => { setImmediate(resume) }).then(async () => { handle.terminate() await handle.done.catch(() => {}) @@ -298,6 +344,7 @@ export class E2BCodeRuntime extends CodeRuntime { result = output.failure(logs, { kind: 'worker-exit', message: `E2B runtime cleanup failed: ${messageOf(cleanupError)}` }) } const final = typeof result === 'function' ? result() : result + this.live.delete(live) finishResolve() resolve(final) }) @@ -305,7 +352,14 @@ export class E2BCodeRuntime extends CodeRuntime { const sendReply = (message: unknown): void => { if (settled) return - stdin.write(encodeE2BFrame(message), (error?: Error | null) => { + let frame: string + try { + frame = encodeBoundedE2BFrame(message, this.config.maxFrameBytes) + } catch (error: unknown) { + finish(() => output.failure(logs, { kind: 'worker-exit', message: `E2B runtime bridge failed: ${messageOf(error)}` })) + return + } + stdin.write(frame, (error?: Error | null) => { if (error !== undefined && error !== null) { finish(() => output.failure(logs, { kind: 'worker-exit', message: `E2B runtime bridge write failed: ${error.message}` })) } @@ -433,7 +487,10 @@ export class E2BCodeRuntime extends CodeRuntime { this.disposed = true const runs = [...this.live] for (const run of runs) run.settle({ kind: 'abort', message: 'runtime disposed' }) - await Promise.all(runs.map(run => run.finished)) + await Promise.all([ + this.ready.then(() => {}, () => {}), + ...runs.map(run => run.finished), + ]) } /* jscpd:ignore-end */ } diff --git a/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts b/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts index ff1c5dd84a..9824fb2cf0 100644 --- a/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts +++ b/packages/e2b/code-runtime-e2b/tests/runtime.spec.ts @@ -34,13 +34,22 @@ class FakeHandle implements SubprocessHandle { waitCalls = 0 private readonly decoder = new E2BFrameDecoder(10_000_000) private readonly waitError: Error | undefined + private readonly waitResult: Promise | undefined private settled = false constructor( private readonly onMessage: (message: unknown, handle: FakeHandle) => void = () => {}, - options: { stdin?: boolean; stdout?: boolean; stderr?: string; writeError?: Error; waitError?: Error } = {}, + options: { + stdin?: boolean + stdout?: boolean + stderr?: string + writeError?: Error + waitError?: Error + waitResult?: Promise + } = {}, ) { this.waitError = options.waitError + this.waitResult = options.waitResult this.stdin = options.stdin === false ? undefined : options.writeError === undefined @@ -89,6 +98,7 @@ class FakeHandle implements SubprocessHandle { async waitForExit(): Promise { this.waitCalls += 1 if (this.waitError !== undefined) throw this.waitError + if (this.waitResult !== undefined) return await this.waitResult return true } } @@ -286,6 +296,30 @@ describe('E2BCodeRuntime', () => { await fixture.fiber.dispose() }) + it('enforces the outbound frame bound on boot and binding replies', async () => { + const oversizedBoot = new FakeHandle() + const oversizedReply = new FakeHandle((message, current) => { + if ((message as { type?: string }).type === 'boot') { + current.emit({ type: 'call', id: 1, global: 'bridge', name: 'large', args: encodeWorkerJson(null) }) + } + }) + const fixture = await setup([oversizedBoot, oversizedReply], { maxOutputBytes: 128, maxFrameBytes: 512 }) + + const bootResult = await fixture.runtime.run(request(`return ${JSON.stringify('x'.repeat(1_000))}`)) + expect(bootResult.error).toMatchObject({ kind: 'worker-exit' }) + expect(bootResult.error?.message).toContain('frame exceeded its byte limit') + expect(oversizedBoot.writes).toHaveLength(0) + + const replyResult = await fixture.runtime.run({ + program: 'return await bridge.large(null)', + bindings: [{ global: 'bridge', functions: { large: async () => 'x'.repeat(1_000) } }], + }) + expect(replyResult.error).toMatchObject({ kind: 'worker-exit' }) + expect(replyResult.error?.message).toContain('frame exceeded its byte limit') + expect(oversizedReply.writes).toHaveLength(1) + await fixture.fiber.dispose() + }) + it('contains stdin errors, process exits, spawn failures, and missing pipes', async () => { const writeError = new FakeHandle(() => {}, { writeError: new Error('write callback broke') }) const stdinError = new FakeHandle((message, current) => { @@ -445,12 +479,116 @@ describe('E2BCodeRuntime', () => { const gate = Promise.withResolvers() const fixture = await setup([], {}, {}, () => gate.promise) const running = fixture.runtime.run(request()) - await (fixture.runtime as unknown as { teardown(): Promise }).teardown() + const disposing = fixture.fiber.dispose() + let disposed = false + void disposing.then(() => { disposed = true }) + await new Promise(resolve => setImmediate(resolve)) + const disposedBeforeSetup = disposed gate.resolve(fixture.sandbox) + await disposing + expect(disposedBeforeSetup).toBe(false) expect((await running).error).toEqual({ kind: 'abort', message: 'runtime disposed' }) + expect(fixture.write).not.toHaveBeenCalled() + }) + + it('observes abort while runtime preparation is pending', async () => { + const gate = Promise.withResolvers() + const fixture = await setup([], {}, {}, () => gate.promise) + const controller = new AbortController() + const running = fixture.runtime.run({ ...request(), signal: controller.signal }) + + controller.abort('stop during setup') + const early = await Promise.race([ + running.then(result => ({ kind: 'result' as const, result })), + new Promise<{ kind: 'pending' }>((resolve) => { setImmediate(() => { resolve({ kind: 'pending' }) }) }), + ]) + expect(fixture.spawn).not.toHaveBeenCalled() + + gate.resolve(fixture.sandbox) + expect(early).toMatchObject({ kind: 'result', result: { error: { kind: 'abort', message: 'stop during setup' } } }) + await running await fixture.fiber.dispose() }) + it('classifies an abort that races synchronous subprocess spawn', async () => { + const fixture = await setup() + const controller = new AbortController() + fixture.spawn.mockImplementationOnce(() => { + controller.abort('stop at spawn') + throw new Error('aborted before spawn') + }) + + expect((await fixture.runtime.run({ ...request(), signal: controller.signal })).error) + .toEqual({ kind: 'abort', message: 'stop at spawn' }) + + fixture.spawn.mockImplementationOnce(() => { throw new Error('synchronous spawn failure') }) + expect((await fixture.runtime.run(request())).error).toEqual({ + kind: 'worker-exit', + message: 'E2B runtime spawn failed: synchronous spawn failure', + }) + await fixture.fiber.dispose() + + const disposingFixture = await setup() + disposingFixture.spawn.mockImplementationOnce(() => { + void (disposingFixture.runtime as unknown as { teardown(): Promise }).teardown() + throw new Error('spawn raced disposal') + }) + expect((await disposingFixture.runtime.run(request())).error) + .toEqual({ kind: 'abort', message: 'runtime disposed' }) + await disposingFixture.fiber.dispose() + }) + + it('closes both abort races around runtime readiness and live-run publication', async () => { + let preparationAborted = false + const preparationSignal = { + get aborted() { return preparationAborted }, + reason: 'preparation race', + addEventListener() { preparationAborted = true }, + removeEventListener() {}, + } as unknown as AbortSignal + const liveHandle = new FakeHandle() + const fixture = await setup([liveHandle]) + expect((await fixture.runtime.run({ ...request(), signal: preparationSignal })).error) + .toEqual({ kind: 'abort', message: 'preparation race' }) + expect(fixture.spawn).not.toHaveBeenCalled() + + let liveAborted = false + let registrations = 0 + const liveSignal = { + get aborted() { return liveAborted }, + reason: 'live publication race', + addEventListener() { + registrations += 1 + if (registrations === 2) liveAborted = true + }, + removeEventListener() {}, + } as unknown as AbortSignal + expect((await fixture.runtime.run({ ...request(), signal: liveSignal })).error) + .toEqual({ kind: 'abort', message: 'live publication race' }) + await fixture.fiber.dispose() + }) + + it('retains a live run until remote cleanup reaches quiescence', async () => { + const cleanup = Promise.withResolvers() + const handle = new FakeHandle((message, current) => { + if ((message as { type?: string }).type === 'boot') current.emit({ type: 'done' }) + }, { waitResult: cleanup.promise }) + const fixture = await setup([handle]) + const running = fixture.runtime.run(request()) + await vi.waitFor(() => { expect(handle.waitCalls).toBe(1) }) + + const disposing = fixture.fiber.dispose() + let disposed = false + void disposing.then(() => { disposed = true }) + await new Promise(resolve => setImmediate(resolve)) + const disposedBeforeCleanup = disposed + + cleanup.resolve(true) + await expect(running).resolves.toEqual({ logs: [] }) + await expect(disposing).resolves.toBeUndefined() + expect(disposedBeforeCleanup).toBe(false) + }) + it('registers the package-owned invariant companion', async () => { const ctx = new Context() await ctx.plugin(InvariantService, { enabled: true }) diff --git a/packages/e2b/e2b/src/frame.ts b/packages/e2b/e2b/src/frame.ts index 39800b4428..5b6fe536c0 100644 --- a/packages/e2b/e2b/src/frame.ts +++ b/packages/e2b/e2b/src/frame.ts @@ -10,9 +10,30 @@ const BASE64_LINE = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3 * @returns Base64-encoded UTF-8 JSON followed by one newline. */ export function encodeE2BFrame(value: unknown): string { + return encodeFrame(value) +} + +/** + * Encode one JSON-compatible value while enforcing the decoded frame bound. + * @param value - Value accepted by `JSON.stringify`. + * @param maxFrameBytes - Maximum UTF-8 JSON bytes in the encoded frame. + * @returns Base64-encoded UTF-8 JSON followed by one newline. + */ +export function encodeBoundedE2BFrame(value: unknown, maxFrameBytes: number): string { + if (!Number.isSafeInteger(maxFrameBytes) || maxFrameBytes <= 0) { + throw new Error('E2B frame maxFrameBytes must be a positive safe integer') + } + return encodeFrame(value, maxFrameBytes) +} + +function encodeFrame(value: unknown, maxFrameBytes?: number): string { const json: unknown = JSON.stringify(value) if (typeof json !== 'string') throw new Error('E2B frame value is not JSON-serializable') - return `${Buffer.from(json).toString('base64')}\n` + const bytes = Buffer.from(json) + if (maxFrameBytes !== undefined && bytes.length > maxFrameBytes) { + throw new Error('E2B frame exceeded its byte limit') + } + return `${bytes.toString('base64')}\n` } /** Incremental decoder for newline-delimited base64 JSON frames. */ diff --git a/packages/e2b/e2b/src/index.ts b/packages/e2b/e2b/src/index.ts index 4c4597f6ef..11625b3b21 100644 --- a/packages/e2b/e2b/src/index.ts +++ b/packages/e2b/e2b/src/index.ts @@ -10,7 +10,7 @@ import z from 'schemastery' import { Sandbox } from 'e2b' import type { Branded } from '@deepseek-ai/dsh-brand' -export { E2BFrameDecoder, encodeE2BFrame } from './frame.ts' +export { E2BFrameDecoder, encodeBoundedE2BFrame, encodeE2BFrame } from './frame.ts' export { CommandExitError, @@ -202,7 +202,11 @@ export class E2BSandboxService extends Service { */ async getSandbox(): Promise { if (this.disposed) throw new Error('E2B sandbox service is disposing') - return await this.ready + const sandbox = await this.ready + // Disposal can race the awaited sandbox readiness despite the synchronous precheck. + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (this.disposed) throw new Error('E2B sandbox service is disposing') + return sandbox } private validate(): void { diff --git a/packages/e2b/e2b/tests/composition.e2e.ts b/packages/e2b/e2b/tests/composition.e2e.ts index d056d4b851..5c9bd56fc9 100644 --- a/packages/e2b/e2b/tests/composition.e2e.ts +++ b/packages/e2b/e2b/tests/composition.e2e.ts @@ -24,7 +24,7 @@ describe.skipIf(!process.env.E2B_API_KEY)('E2B live Loader composition', () => { }, processTimeoutMs: 120_000, inspect: async (cwd) => { - for (const name of ['from-fs.txt', 'from-bash.txt', 'multibyte.ts', 'fixture-lsp.mjs']) { + for (const name of ['from-fs.txt', 'from-bash.txt', 'multibyte # file.ts', 'fixture-lsp.mjs']) { await expect(access(join(cwd, name))).rejects.toMatchObject({ code: 'ENOENT' }) } }, @@ -35,6 +35,12 @@ describe.skipIf(!process.env.E2B_API_KEY)('E2B live Loader composition', () => { expect(output).toMatchObject({ bashRead: 'written-by-fs\n', fsRead: 'written-by-bash\n', + explicitEnvironment: true, + spill: { + liveBytes: 6, + outcome: { exitCode: null, signal: 'SIGTERM' }, + read: { text: '6789', nextOffset: 10, lossy: true }, + }, hover: { kind: 'hover', hover: { contents: '**remote hover** 你好 café' }, @@ -51,6 +57,8 @@ describe.skipIf(!process.env.E2B_API_KEY)('E2B live Loader composition', () => { hostileOutput: { error: { kind: 'output-limit' } }, timedOut: { error: { kind: 'timeout' } }, aborted: { error: { kind: 'abort', message: 'live abort' } }, + oversizedBoot: { error: { kind: 'worker-exit' } }, + oversizedReply: { error: { kind: 'worker-exit' } }, lingeringCodeRunners: 0, }) expect((output.terminal as { motd: string }).motd.length).toBeGreaterThan(0) diff --git a/packages/e2b/e2b/tests/e2b.spec.ts b/packages/e2b/e2b/tests/e2b.spec.ts index 541e69df9c..58d2ffbd7a 100644 --- a/packages/e2b/e2b/tests/e2b.spec.ts +++ b/packages/e2b/e2b/tests/e2b.spec.ts @@ -4,6 +4,7 @@ import type { Sandbox as SandboxType } from 'e2b' import E2BSandboxService, { E2BFrameDecoder, E2BSandboxId, + encodeBoundedE2BFrame, encodeE2BFrame, quoteE2BShellArg, resolveE2BExecutable, @@ -91,6 +92,22 @@ describe('E2BSandboxService', () => { await expect(service.getSandbox()).rejects.toThrow(/disposing/) }) + it('rejects handle acquisition when disposal starts during setup', async () => { + const fixture = fakeSandbox() + const opening = Promise.withResolvers() + sdk.create.mockReturnValue(opening.promise) + const ctx = new Context() + const fiber = await ctx.plugin(E2BSandboxService, { apiKey: 'test-key' }) + + const acquisition = ctx.e2b.getSandbox() + const disposing = fiber.dispose() + opening.resolve(fixture.sandbox) + + await expect(acquisition).rejects.toThrow(/disposing/) + await expect(disposing).resolves.toBeUndefined() + expect(fixture.kill).toHaveBeenCalledOnce() + }) + it('creates from a template, honors timeout and pause policies, and reads the key from the environment', async () => { vi.stubEnv('E2B_API_KEY', 'environment-key') const fixture = fakeSandbox('template-sandbox') @@ -240,6 +257,14 @@ describe('E2B helpers and invariant companion', () => { expect(() => encodeE2BFrame(undefined)).toThrow('not JSON-serializable') }) + it('bounds outbound frames by decoded UTF-8 bytes', () => { + const exact = encodeBoundedE2BFrame({ text: '你' }, 14) + expect(new E2BFrameDecoder(14).push(exact)).toEqual([{ text: '你' }]) + expect(() => encodeBoundedE2BFrame({ text: '你' }, 13)).toThrow('byte limit') + expect(() => encodeBoundedE2BFrame(null, 0)).toThrow('positive safe integer') + expect(() => encodeBoundedE2BFrame(null, 1.5)).toThrow('positive safe integer') + }) + it('rejects malformed, oversized, and truncated frame streams', () => { expect(() => new E2BFrameDecoder(0)).toThrow('positive safe integer') expect(() => new E2BFrameDecoder(1.5)).toThrow('positive safe integer') diff --git a/packages/e2b/fs-e2b/src/index.ts b/packages/e2b/fs-e2b/src/index.ts index 020b0f4857..4f733f1fba 100644 --- a/packages/e2b/fs-e2b/src/index.ts +++ b/packages/e2b/fs-e2b/src/index.ts @@ -199,6 +199,7 @@ export class E2BFileSystem extends FileSystem { const reader = stream.getReader() const decoder = new TextDecoder('utf-8', { fatal: true }) let sampledBytes = 0 + let completed = false try { while (true) { assertNotAborted(signal, 'read') @@ -222,9 +223,17 @@ export class E2BFileSystem extends FileSystem { } catch (error: unknown) { throw new FsError(`cannot read "${displayPath}": invalid UTF-8 text`, 'FS_NOT_TEXT', { cause: error }) } + completed = true } catch (error: unknown) { throw mapError(error, 'read', displayPath, signal) } finally { + if (!completed) { + try { + await reader.cancel() + } catch (_streamCancellationFailure) { + // The primary read outcome owns the result; cancellation is best-effort after early stop. + } + } reader.releaseLock() } }, diff --git a/packages/e2b/fs-e2b/tests/filesystem.spec.ts b/packages/e2b/fs-e2b/tests/filesystem.spec.ts index b8eb5447d6..177bb9e59c 100644 --- a/packages/e2b/fs-e2b/tests/filesystem.spec.ts +++ b/packages/e2b/fs-e2b/tests/filesystem.spec.ts @@ -12,7 +12,7 @@ import { FsVersion } from '@deepseek-ai/dsh-fs' import E2BFileSystem from '@deepseek-ai/dsh-fs-e2b' import * as E2BFsInvariant from '../src/invariant.ts' import InvariantService from '@deepseek-ai/dsh-invariants' -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' interface RemoteNode { type: FileType @@ -38,6 +38,8 @@ class FakeRemote { readonly removals: string[] = [] readonly commands: string[] = [] streamChunks: Uint8Array[] | undefined + streamKeepOpen = false + readonly streamCancel = vi.fn() nextCommandError: unknown nextInfoError: unknown nextListError: unknown @@ -148,10 +150,11 @@ class FakeRemote { if (options.format === 'bytes') return data.slice() const chunks = this.streamChunks ?? [data.slice()] return new ReadableStream({ - start(controller) { + start: (controller) => { for (const chunk of chunks) controller.enqueue(chunk) - controller.close() + if (!this.streamKeepOpen) controller.close() }, + cancel: () => { this.streamCancel() }, }) }, list: async (path: string, options?: { depth?: number; signal?: AbortSignal }): Promise => { @@ -303,6 +306,22 @@ describe('E2BFileSystem identity, metadata, and reads', () => { expect(initiallyBuffered).toBe('€') }) + it('cancels a remote stream when its consumer stops early', async () => { + const remote = new FakeRemote() + remote.file('/workspace/text.txt', 'ab') + remote.streamChunks = [bytes('a'), bytes('b')] + remote.streamKeepOpen = true + const { fs } = await setup(remote) + const stream = await fs.streamText(await fs.resolve('text.txt')) + + for await (const chunk of stream) { + expect(chunk).toBe('a') + break + } + + expect(remote.streamCancel).toHaveBeenCalledOnce() + }) + it('matches local binary sampling while edits still reject any NUL byte', async () => { const remote = new FakeRemote() remote.file('/workspace/late-nul.txt', `${'a'.repeat(8192)}\0tail`) diff --git a/packages/e2b/lsp-e2b/src/index.ts b/packages/e2b/lsp-e2b/src/index.ts index eb9596db5f..b378ebfc9a 100644 --- a/packages/e2b/lsp-e2b/src/index.ts +++ b/packages/e2b/lsp-e2b/src/index.ts @@ -200,6 +200,16 @@ export async function readE2BSource( 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 { @@ -298,6 +308,7 @@ export class E2BLspProvider implements LspProvider { 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({ diff --git a/packages/e2b/lsp-e2b/tests/provider.spec.ts b/packages/e2b/lsp-e2b/tests/provider.spec.ts index 0b41be12a3..d746abe7f7 100644 --- a/packages/e2b/lsp-e2b/tests/provider.spec.ts +++ b/packages/e2b/lsp-e2b/tests/provider.spec.ts @@ -69,6 +69,7 @@ import { E2BLspProvider, apply, canonicalizeE2BWorkspace, + e2bFileUri, readE2BSource, } from '@deepseek-ai/dsh-lsp-e2b' import type { LspE2BServerConfig } from '@deepseek-ai/dsh-lsp-e2b' @@ -234,6 +235,10 @@ describe('E2BLspProvider pooling and lifecycle', () => { 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 } }, diff --git a/packages/e2b/pty-e2b/src/index.ts b/packages/e2b/pty-e2b/src/index.ts index fca16cf5c7..8fd7a519a0 100644 --- a/packages/e2b/pty-e2b/src/index.ts +++ b/packages/e2b/pty-e2b/src/index.ts @@ -71,9 +71,10 @@ export class E2BPtyBackend implements PtyBackend { } const session = new E2BPtySession(sandbox, handle, this.config) created.session = session - for (const data of pending) session.onData(data) try { - await session.initialize(spec.signal) + const initializing = session.initialize(spec.signal) + for (const data of pending) session.onData(data) + await initializing return session } catch (error: unknown) { try { diff --git a/packages/e2b/pty-e2b/tests/index.spec.ts b/packages/e2b/pty-e2b/tests/index.spec.ts index 567b3b9215..e306d8b86b 100644 --- a/packages/e2b/pty-e2b/tests/index.spec.ts +++ b/packages/e2b/pty-e2b/tests/index.spec.ts @@ -52,7 +52,7 @@ describe('E2BPtyBackend and plugin', () => { const backend = new E2BPtyBackend(ctx, config(), async (_sandbox, received) => { options = received void received.onData(Buffer.from('banner\n')) - setTimeout(() => { void received.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) }, 0) + void received.onData(Buffer.from('\x1b]133;D;0\x07dsh> ')) return created }) const pending = backend.spawn({ @@ -62,7 +62,7 @@ describe('E2BPtyBackend and plugin', () => { await vi.advanceTimersByTimeAsync(2) const session = await pending - expect(session.motd).toBe('dsh> ') + expect(session.motd).toBe('banner\ndsh> ') expect(options).toMatchObject({ rows: 24, cols: 80, cwd: '/workspace/project', timeoutMs: 0 }) expect(options?.envs).toMatchObject({ TERM: 'dumb', PAGER: 'cat', GIT_PAGER: 'cat', PS1: 'dsh> ', diff --git a/packages/e2b/subprocess-e2b/README.i18n.yaml b/packages/e2b/subprocess-e2b/README.i18n.yaml index 26bc8f0b51..8b8a70838c 100644 --- a/packages/e2b/subprocess-e2b/README.i18n.yaml +++ b/packages/e2b/subprocess-e2b/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/e2b/subprocess-e2b/README.md -README.md: ce61de1100791be6e5c4db74c73ff43566281ed9 -README.zh.md: a4c619f0c002cc1d36310c5a9b4a3f7ae655ad1f +README.md: 3b3bfa88e7e6483decfcdec11355942ae4ff7403 +README.zh.md: 3ff9a51c60636dea5789f9dd11b04aa902b91d0c diff --git a/packages/e2b/subprocess-e2b/README.md b/packages/e2b/subprocess-e2b/README.md index ce61de1100..3b3bfa88e7 100644 --- a/packages/e2b/subprocess-e2b/README.md +++ b/packages/e2b/subprocess-e2b/README.md @@ -11,7 +11,7 @@ E2B implementation of the [`@deepseek-ai/dsh-subprocess`](../../subprocess/subpr - **Environment boundary** — the wrapper starts from the sandbox command environment, removes ambient `DSH_*` and credential-shaped (`*KEY*`, `*SECRET*`, `*TOKEN*`) names, then restores every `spec.env` entry as an explicit caller opt-in. Host ambient variables never enter the sandbox implicitly. - **Stdio projection** — pipe mode forwards E2B callbacks into host Node streams; inherit mode forwards them to the harness process streams; collect mode retains a bounded host tail with offset reads. Optional complete spill files are written remotely and advertised only while within their cap. Batch and streaming stdin use the SDK handle. -The base E2B image supplies the Bash/GNU utilities this adapter invokes: `bash`, `setsid`, `ps`, `tr`, `env`, `chmod`, `tee`, and `kill`. A custom template must retain compatible commands. +The base E2B image supplies the Bash/GNU utilities this adapter invokes: `bash`, `setsid`, `ps`, `tr`, `env`, `chmod`, `tee`, `head`, and `kill`. A custom template must retain compatible commands. ## Model Experience diff --git a/packages/e2b/subprocess-e2b/README.zh.md b/packages/e2b/subprocess-e2b/README.zh.md index a4c619f0c0..3ff9a51c60 100644 --- a/packages/e2b/subprocess-e2b/README.zh.md +++ b/packages/e2b/subprocess-e2b/README.zh.md @@ -11,7 +11,7 @@ - **环境边界**：包装层从沙箱命令环境开始，移除环境中的 `DSH_*` 和形似凭据的名称（`*KEY*`、`*SECRET*`、`*TOKEN*`），再把每个 `spec.env` 条目恢复为调用方显式选择。宿主环境变量绝不会隐式进入沙箱。 - **stdio 投影**：pipe 模式把 E2B 回调转发到宿主 Node 流；inherit 模式把回调转发到 harness 进程流；collect 模式保留有界的宿主尾部，并支持基于偏移量读取。可选的完整 spill 文件写在远程，并且只有未超过其上限时才会对外公布。批量 stdin 和流式 stdin 都使用 SDK 句柄。 -基础 E2B 镜像提供该适配器调用的 Bash/GNU 工具：`bash`、`setsid`、`ps`、`tr`、`env`、`chmod`、`tee` 和 `kill`。自定义模板必须保留兼容的命令。 +基础 E2B 镜像提供该适配器调用的 Bash/GNU 工具：`bash`、`setsid`、`ps`、`tr`、`env`、`chmod`、`tee`、`head` 和 `kill`。自定义模板必须保留兼容的命令。 ## 模型体验 diff --git a/packages/e2b/subprocess-e2b/src/process.ts b/packages/e2b/subprocess-e2b/src/process.ts index 507ea700c4..86bdade589 100644 --- a/packages/e2b/subprocess-e2b/src/process.ts +++ b/packages/e2b/subprocess-e2b/src/process.ts @@ -55,37 +55,38 @@ class DeferredStdin extends Writable { interface RemotePaths { pid: string status: string + environment: string stdout: string stderr: string } -function explicitEnvironmentNames(env: Readonly> | undefined): string { - return Object.keys(env ?? {}) - .map(quoteE2BShellArg) - .join(' ') +function explicitEnvironment(env: Readonly> | undefined): string { + return Object.entries(env ?? {}) + .map(([name, value]) => `${name}=${value}\0`) + .join('') } function commandText(spec: SubprocessSpawnSpec, paths: RemotePaths): string { const stdoutRedirect = hasSpill(spec.stdio.stdout) - ? `> >(tee -a -- ${quoteE2BShellArg(paths.stdout)})` + ? `> >(tee --output-error=warn-nopipe >(head -c ${spec.stdio.stdout.spill.maxBytes} > ${quoteE2BShellArg(paths.stdout)}))` : '' const stderrRedirect = hasSpill(spec.stdio.stderr) - ? `2> >(tee -a -- ${quoteE2BShellArg(paths.stderr)} >&2)` + ? `2> >(tee --output-error=warn-nopipe >(head -c ${spec.stdio.stderr.spill.maxBytes} > ${quoteE2BShellArg(paths.stderr)}) >&2)` : '' - const environmentNames = explicitEnvironmentNames(spec.env) const inner = [ 'set +e', 'umask 077', 'dsh_e2b_pgid="$(ps -o pgid= -p "$$" | tr -d " ")"', `printf '%s\\n' "$dsh_e2b_pgid" > ${quoteE2BShellArg(paths.pid)}`, + `mapfile -d '' -t dsh_e2b_explicit < ${quoteE2BShellArg(paths.environment)}`, + `: > ${quoteE2BShellArg(paths.environment)}`, 'dsh_e2b_env=()', - `dsh_e2b_explicit=(${environmentNames})`, - 'while IFS= read -r dsh_e2b_name; do', + "while IFS= read -r -d '' dsh_e2b_entry; do", + ' dsh_e2b_name="${dsh_e2b_entry%%=*}"', ' case "${dsh_e2b_name^^}" in DSH_*|*KEY*|*SECRET*|*TOKEN*) continue ;; esac', - ' dsh_e2b_env+=("$dsh_e2b_name=${!dsh_e2b_name}")', - 'done < <(compgen -e)', - 'for dsh_e2b_name in "${dsh_e2b_explicit[@]}"; do dsh_e2b_env+=("$dsh_e2b_name=${!dsh_e2b_name}"); done', - `env -i "\${dsh_e2b_env[@]}" "$@" ${stdoutRedirect} ${stderrRedirect}`.trimEnd(), + ' dsh_e2b_env+=("$dsh_e2b_entry")', + 'done < <(env -0)', + `env -i "\${dsh_e2b_env[@]}" "\${dsh_e2b_explicit[@]}" "$@" ${stdoutRedirect} ${stderrRedirect}`.trimEnd(), 'dsh_e2b_status=$?', 'wait', `printf '%s\\n' "$dsh_e2b_status" > ${quoteE2BShellArg(paths.status)}`, @@ -131,7 +132,6 @@ export class E2BSubprocessHandle implements SubprocessHandle { private readonly stderrReader: E2BOutputReader | undefined private readonly paths: RemotePaths private remotePid = -1 - private settled = false private terminationRequested = false private terminationSignal: NodeJS.Signals | null = null private termination: Promise | undefined @@ -150,6 +150,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { this.paths = { pid: posix.join(stateDir, 'pid'), status: posix.join(stateDir, 'exit-code'), + environment: posix.join(stateDir, 'environment'), stdout: posix.join(stateDir, 'stdout.log'), stderr: posix.join(stateDir, 'stderr.log'), } @@ -182,7 +183,7 @@ export class E2BSubprocessHandle implements SubprocessHandle { /** @inheritdoc */ terminate(): void { - if (this.terminationRequested || this.settled) return + if (this.terminationRequested) return this.terminationRequested = true this.termination = this.terminateRemote() void this.termination.catch(() => {}) @@ -240,7 +241,6 @@ export class E2BSubprocessHandle implements SubprocessHandle { cwd: this.spec.cwd, stdin: this.spec.stdio.stdin !== 'ignore', timeoutMs: 0, - ...(this.spec.env !== undefined ? { envs: this.spec.env } : {}), onStdout: async (data) => { await this.dispatchOutput('stdout', data) }, onStderr: async (data) => { await this.dispatchOutput('stderr', data) }, }, @@ -250,7 +250,12 @@ export class E2BSubprocessHandle implements SubprocessHandle { } const completion = handle.wait() void completion.catch(() => {}) - this.remotePid = await this.waitForProcessGroupId(sandbox, completion) + try { + this.remotePid = await this.waitForProcessGroupId(sandbox, completion) + } catch (error: unknown) { + await Promise.allSettled([handle.kill()]) + throw error + } this.readyState.resolve(handle) await this.writeBatchStdin(handle) const outcome = await this.waitForCommand(completion) @@ -260,7 +265,6 @@ export class E2BSubprocessHandle implements SubprocessHandle { this.readyState.reject(error) throw error } finally { - this.settled = true this.spec.signal?.removeEventListener('abort', this.onAbort) this.stdout?.end() this.stderr?.end() @@ -269,17 +273,16 @@ export class E2BSubprocessHandle implements SubprocessHandle { private async prepareState(sandbox: Sandbox): Promise { await sandbox.files.makeDir(this.stateDir) + await sandbox.commands.run(`chmod 700 -- ${quoteE2BShellArg(this.stateDir)}`) const files = [ { path: this.paths.pid, data: '' }, { path: this.paths.status, data: '' }, + { path: this.paths.environment, data: explicitEnvironment(this.spec.env) }, ...(hasSpill(this.spec.stdio.stdout) ? [{ path: this.paths.stdout, data: '' }] : []), ...(hasSpill(this.spec.stdio.stderr) ? [{ path: this.paths.stderr, data: '' }] : []), ] await sandbox.files.write(files) - await sandbox.commands.run([ - `chmod 700 -- ${quoteE2BShellArg(this.stateDir)}`, - `chmod 600 -- ${files.map(file => quoteE2BShellArg(file.path)).join(' ')}`, - ].join('\n')) + await sandbox.commands.run(`chmod 600 -- ${files.map(file => quoteE2BShellArg(file.path)).join(' ')}`) } private async writeBatchStdin(handle: CommandHandle): Promise { diff --git a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts index ff9d48073c..6be645a713 100644 --- a/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/subprocess.spec.ts @@ -80,6 +80,7 @@ class FakeSandbox { readonly handle = new FakeCommandHandle() readonly commandsSeen: string[] = [] readonly writtenFiles: string[][] = [] + readonly writtenFileData = new Map() readonly removed: string[] = [] readonly directories: string[] = [] startOptions: StartOptions | undefined @@ -129,6 +130,7 @@ class FakeSandbox { }, write: async (files: Array<{ path: string; data: string }>): Promise => { this.writtenFiles.push(files.map(file => file.path)) + for (const file of files) this.writtenFileData.set(file.path, file.data) return files.map(() => ({})) }, read: async (): Promise => this.processGroupReads.shift() ?? this.processGroupId, @@ -251,7 +253,7 @@ describe('E2BSubprocessHandle', () => { const handle = new E2BSubprocessHandle(runtime(fake), spec({ argv: ['tool', 'argument with spaces'], stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: 8, spill: { maxBytes: 32 } } }, - env: { PATH: '/bin', DEEPSEEK_API_KEY: 'explicit-secret', DSH_MODE: 'test' }, + env: { PATH: '/bin', 'FOO-BAR': 'hyphen-value', DEEPSEEK_API_KEY: 'explicit-secret', DSH_MODE: 'test' }, }), '/workspace/.dsh-e2b/processes/one') expect(handle.pid).toBe(-1) handle.stdin!.write('hello') @@ -261,17 +263,26 @@ describe('E2BSubprocessHandle', () => { expect(handle.pid).toBe(4343) expect(fake.handle.sent.map(value => String(value))).toEqual(['hello']) expect(fake.handle.closes).toBe(1) - expect(fake.startOptions?.envs).toEqual({ PATH: '/bin', DEEPSEEK_API_KEY: 'explicit-secret', DSH_MODE: 'test' }) + expect(fake.startOptions?.envs).toBeUndefined() const command = fake.commandsSeen.find(value => value.startsWith('exec setsid'))! expect(command).toContain('exec setsid --wait -- bash -c') - expect(command).toContain('DEEPSEEK_API_KEY') - expect(command).toContain('DSH_MODE') + expect(command).not.toContain('DEEPSEEK_API_KEY') + expect(command).not.toContain('DSH_MODE') + expect(command).not.toContain('FOO-BAR') expect(command).not.toContain('explicit-secret') + expect(command).not.toContain('hyphen-value') + expect(command).not.toContain('${!dsh_e2b_name}') + expect(command).toContain('env -0') + expect(command).toContain('mapfile -d') expect(fake.writtenFiles[0]).toEqual([ '/workspace/.dsh-e2b/processes/one/pid', '/workspace/.dsh-e2b/processes/one/exit-code', + '/workspace/.dsh-e2b/processes/one/environment', '/workspace/.dsh-e2b/processes/one/stderr.log', ]) + expect(fake.writtenFileData.get('/workspace/.dsh-e2b/processes/one/environment')).toBe( + 'PATH=/bin\0FOO-BAR=hyphen-value\0DEEPSEEK_API_KEY=explicit-secret\0DSH_MODE=test\0', + ) let piped = '' handle.stdout!.on('data', (chunk) => { piped += String(chunk) }) @@ -350,6 +361,11 @@ describe('E2BSubprocessHandle', () => { await handle.done expect(handle.collected.stdout!.readFrom(0)).toEqual({ text: 'cd', nextOffset: 4, lossy: true }) expect(fake.removed).toContain('/runtime/oversize/stdout.log') + const command = fake.commandsSeen.find(value => value.startsWith('exec setsid'))! + expect(command).toContain('head -c 3') + expect(command).toContain('/runtime/oversize/stdout.log') + expect(command).toContain('tee --output-error=warn-nopipe') + expect(command).not.toContain('tee -a') }) it('contains remote spill-removal failures and routes empty inherited output', async () => { @@ -415,6 +431,22 @@ describe('E2BSubprocessHandle', () => { await expect(handle.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' }) }) + it('can terminate a surviving process group after the command leader settles', async () => { + const fake = new FakeSandbox() + const handle = new E2BSubprocessHandle(runtime(fake), spec(), '/runtime/surviving-group') + await flush() + fake.handle.succeed(0) + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + expect(fake.alive).toBe(true) + + handle.terminate() + await flush() + const signaled = fake.commandsSeen.includes('kill -TERM -- -4242') + if (!signaled) fake.finish() + await expect(handle.waitForExit()).resolves.toBe(true) + expect(signaled).toBe(true) + }) + it('bounds waitForExit while startup or a live group is pending', async () => { const fake = new FakeSandbox() fake.deferStart() @@ -564,8 +596,14 @@ describe('E2BSubprocessHandle', () => { it('rejects invalid or absent process-group publication', async () => { const invalidGroup = new FakeSandbox() invalidGroup.processGroupId = 'not-a-pid\n' + vi.spyOn(invalidGroup.handle, 'kill').mockImplementation(async () => { + invalidGroup.handle.kills += 1 + invalidGroup.finish() + return true + }) const invalid = new E2BSubprocessHandle(runtime(invalidGroup), spec(), '/runtime/invalid-group') await expect(invalid.done).rejects.toThrow(/invalid process-group id/) + expect(invalidGroup.handle.kills).toBe(1) const absentGroup = new FakeSandbox() absentGroup.processGroupId = '' @@ -573,6 +611,7 @@ describe('E2BSubprocessHandle', () => { await flush() absentGroup.finish() await expect(absent.done).rejects.toThrow(/exited before publishing/) + expect(absentGroup.handle.kills).toBe(1) }) it('waits for delayed process-group publication', async () => { diff --git a/packages/lsp/lsp-local/src/index.ts b/packages/lsp/lsp-local/src/index.ts index fc7e50787f..3a2baa0c55 100644 --- a/packages/lsp/lsp-local/src/index.ts +++ b/packages/lsp/lsp-local/src/index.ts @@ -1,16 +1,19 @@ /** * Generic stdio language-server backend for `ctx.lsp`. One plugin instance configures a named table * of server commands and registers one isolated provider for each entry. Every provider lazily - * single-flights one server process per canonical workspace target, serves transient-open queries + * single-flights one server process per canonical workspace realpath, serves transient-open queries * through it, and replaces a selected transport that fails before or during the next read-only - * query. Providers read sources through `ctx.fs` and launch servers through - * `ctx.subprocess`, so both local and remote implementations share one host. + * query. Providers read sources through Node APIs in the host namespace (not `ctx.fs`) + * and trust their configured servers — no sandbox confinement. * * Namespace plugin (named exports, no default export). Lifecycle is effect-scoped: disposal * unregisters from `ctx.lsp` and tears down every live server. * @module @deepseek-ai/dsh-lsp-local */ +import { accessSync, constants, statSync } from 'node:fs' +import { delimiter, isAbsolute, join } from 'node:path' +import { pathToFileURL } from 'node:url' import type { Context } from 'cordis' import z from 'schemastery' import { LspError, LspProviderId } from '@deepseek-ai/dsh-lsp' @@ -22,9 +25,9 @@ import type { import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import { abortable, abortError } from './abort.ts' import { canonicalizeWorkspace, readHostSource } from './host.ts' -import type { HostWorkspace } from './host.ts' import { LspInstance } from './instance.ts' import type { ConnectionSpawner } from './connection.ts' +import { scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess' import type { InstanceSpec } from './instance.ts' export { canonicalizeWorkspace, readHostSource } from './host.ts' @@ -44,7 +47,10 @@ export { LspConnection } from './connection.ts' export const name = 'lsp-local' /** Services required by this plugin. */ -export const inject = ['fs', 'lsp', 'subprocess'] +export const inject = ['lsp', 'subprocess'] + +/** Credential-shaped ambient env vars are NOT forwarded to the child by default. */ + const DEFAULT_MAX_MESSAGE_BYTES = 16_000_000 const DEFAULT_MAX_STDERR_BYTES = 1_000_000 @@ -86,7 +92,6 @@ export interface Config { /** One server config after schemastery fills every default. */ type ResolvedServerConfig = Required -type WorkspaceKey = HostWorkspace['target']['targetKey'] const LspLocalServerConfig: z = z.object({ command: z.string().required(), @@ -106,67 +111,27 @@ export const Config: z = z.object({ servers: z.dict(LspLocalServerConfig).required(), }) -/** Propagate teardown failures only after every sibling has settled. */ -function throwTeardownFailures(results: readonly PromiseSettledResult[], message: string): void { - const failures: unknown[] = [] - for (const result of results) { - if (result.status === 'rejected') failures.push(result.reason) - } - if (failures.length === 1) throw failures[0] - if (failures.length > 1) throw new AggregateError(failures, message) -} - /** * Register the configured stdio LSP providers. Resolves every executable at load (after credential * scrubbing) before publishing any provider; each process launches lazily on its first matching * query. - * @param ctx - the plugin context carrying `fs`, `lsp`, and `subprocess`. + * @param ctx - the plugin context (must inject `lsp`). * @param config - the resolved plugin configuration (schemastery has filled every default). */ -export async function apply(ctx: Context, config: Config): Promise { +export function apply(ctx: Context, config: Config): void { const entries = Object.entries(config.servers) if (entries.length === 0) throw new Error('lsp-local: servers must contain at least one server') - const setupAbort = new AbortController() - const stopSetupCancellation = ctx.on('internal/plugin', (fiber) => { - // An async plugin callback must observe its own disposal before Cordis can - // run effect cleanup, because unload otherwise waits for this callback. - if (fiber === ctx.fiber && fiber.uid === null) { - setupAbort.abort(new Error('lsp-local setup disposed')) - } - }) - // Resolve every server-local setting before registration so a bad later command or bound cannot // publish an earlier provider. Registry-level mapping conflicts are rolled back below. - const providers = await (async () => { - const lookups = entries.map(async ([providerId, rawConfig]) => { - if (providerId.trim() === '') throw new Error('lsp-local: server ids must be non-empty strings') - const resolved = rawConfig as ResolvedServerConfig - validateServerConfig(providerId, resolved) - const executable = await ctx.subprocess.resolveExecutable( - resolved.command, - resolved.env, - setupAbort.signal, - ) - setupAbort.signal.throwIfAborted() - return new LocalLspProvider( - providerId, - ctx.fs, - resolved, - executable, - spec => ctx.subprocess.spawn(spec), - ) - }) - try { - return await Promise.all(lookups) - } catch (error: unknown) { - setupAbort.abort(error) - await Promise.allSettled(lookups) - throw error - } finally { - stopSetupCancellation() - } - })() + const providers = entries.map(([providerId, rawConfig]) => { + if (providerId.trim() === '') throw new Error('lsp-local: server ids must be non-empty strings') + const resolved = rawConfig as ResolvedServerConfig + validateServerConfig(providerId, resolved) + const childEnv = buildChildEnv(resolved.env) + const executable = resolveExecutable(resolved.command, childEnv) + return new LocalLspProvider(providerId, resolved, childEnv, executable, spec => ctx.subprocess.spawn(spec)) + }) ctx.effect(() => { const disposers: Array<() => void> = [] @@ -179,8 +144,7 @@ export async function apply(ctx: Context, config: Config): Promise { return async () => { // Remove every route before process teardown so no new query can enter a draining provider. for (const dispose of disposers.reverse()) dispose() - const results = await Promise.allSettled(providers.map(provider => provider.disposeAll())) - throwTeardownFailures(results, 'lsp-local provider teardown failed') + await Promise.all(providers.map(provider => provider.disposeAll())) } }, 'lsp-local.registerProviders') } @@ -217,19 +181,16 @@ function assertPositiveInteger(providerId: string, name: string, value: number): class LocalLspProvider implements LspProvider { readonly id: LspProviderId readonly extensionToLanguage: Readonly> - /** One live instance per stable canonical workspace identity. */ - private readonly instances = new Map() + /** One live instance per canonical workspace realpath. */ + private readonly instances = new Map() /** One complete source-read→open→query→close serialization tail per canonical workspace. */ - private readonly queues = new Map>() - /** Workspace canonicalizations that have not entered a provider-owned queue yet. */ - private readonly workspaceLookups = new Set>() - private readonly lifetime = new AbortController() + private readonly queues = new Map>() private disposed = false constructor( providerId: string, - private readonly fs: Context['fs'], private readonly config: ResolvedServerConfig, + private readonly childEnv: Record, private readonly executable: string, private readonly spawner: ConnectionSpawner, ) { @@ -250,60 +211,43 @@ class LocalLspProvider implements LspProvider { if (signal?.aborted) throw abortError(signal) } - /** Fuse caller cancellation with provider disposal for every filesystem and protocol await. */ - private querySignal(signal?: AbortSignal): AbortSignal { - return signal === undefined - ? this.lifetime.signal - : AbortSignal.any([signal, this.lifetime.signal]) - } - async query(request: LspProviderQuery, signal?: AbortSignal): Promise { - // Honor an already-aborted signal before provider I/O so a canceled request never starts a server. + // Honor an already-aborted signal before host I/O so a canceled request never starts a server. this.assertActive(signal) - const querySignal = this.querySignal(signal) - const workspaceResult = canonicalizeWorkspace(this.fs, request.workspaceRoot, querySignal) - const workspaceLookup = workspaceResult.then(() => undefined, () => undefined) - this.workspaceLookups.add(workspaceLookup) - let workspace: HostWorkspace - try { - workspace = await workspaceResult - } finally { - this.workspaceLookups.delete(workspaceLookup) - } - this.assertActive(querySignal) - const workspaceKey = workspace.target.targetKey - return this.enqueue(workspaceKey, querySignal, async () => { - this.assertActive(querySignal) + const workspace = await canonicalizeWorkspace(request.workspaceRoot, signal) + this.assertActive(signal) + return this.enqueue(workspace, signal, async () => { + this.assertActive(signal) // Read inside the workspace queue but before spawning: a queued query sees current bytes when // its turn starts, while an invalid source still cannot leave an idle process pooled. - const source = await readHostSource(this.fs, request.filePath, workspace, this.config.maxDocumentBytes, querySignal) + const source = await readHostSource(request.filePath, workspace, this.config.maxDocumentBytes, signal) // Disposal may have snapshotted the instance map while host I/O was pending. Re-check before a // synchronous get-or-create so every spawned process remains owned by teardown. - this.assertActive(querySignal) - let instance = this.instanceFor(workspaceKey, workspace) + this.assertActive(signal) + let instance = this.instanceFor(workspace) try { - return await instance.query(request, source, querySignal) + return await instance.query(request, source, signal) } catch (error) { // A selected child can have died while idle or fail during the next write. Queries are // read-only, so replace that transport once and retry transparently. if (!instance.isTransportFailure(error)) throw error await instance.dispose() - this.evictIfCurrent(workspaceKey, instance) - this.assertActive(querySignal) - instance = this.instanceFor(workspaceKey, workspace) - return await instance.query(request, source, querySignal) + this.evictIfCurrent(workspace, instance) + this.assertActive(signal) + instance = this.instanceFor(workspace) + return await instance.query(request, source, signal) } finally { // Reach quiescence before dropping a dead slot; a replacement must survive this ownership check. if (instance.dead) { await instance.dispose() - this.evictIfCurrent(workspaceKey, instance) + this.evictIfCurrent(workspace, instance) } } }) } /** Serialize one complete query lifecycle for a canonical workspace. */ - private enqueue(workspace: WorkspaceKey, signal: AbortSignal | undefined, run: () => Promise): Promise { + 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) // The tail follows the actual prior work even when this caller aborts its wait. It never rejects, @@ -317,34 +261,34 @@ class LocalLspProvider implements LspProvider { } /** Return or synchronously publish the one instance for a canonical workspace. */ - private instanceFor(workspaceKey: WorkspaceKey, workspace: HostWorkspace): LspInstance { + private instanceFor(workspace: string): LspInstance { this.assertActive() - const existing = this.instances.get(workspaceKey) + const existing = this.instances.get(workspace) if (existing !== undefined) return existing const created = this.createInstance(workspace) - this.instances.set(workspaceKey, created) + this.instances.set(workspace, created) return created } /** Drop the slot iff it still contains this instance. */ - private evictIfCurrent(workspace: WorkspaceKey, instance: LspInstance): void { + private evictIfCurrent(workspace: string, instance: LspInstance): void { /* v8 ignore next -- mismatch requires another query to replace the slot before this finally runs. */ if (this.instances.get(workspace) === instance) this.instances.delete(workspace) } - private createInstance(workspace: HostWorkspace): LspInstance { + private createInstance(workspace: string): LspInstance { const spec: InstanceSpec = { command: this.executable, args: this.config.args, - cwd: workspace.canonicalPath, - workspaceUri: workspace.fileUrl, - env: this.config.env, + cwd: workspace, + env: this.childEnv, configuration: this.config.configuration, initializationOptions: this.config.initializationOptions, maxMessageBytes: this.config.maxMessageBytes, maxStderrBytes: this.config.maxStderrBytes, shutdownTimeoutMs: this.config.shutdownTimeoutMs, killGraceMs: this.config.killGraceMs, + pathToFileUri: path => pathToFileURL(path).href, } return new LspInstance(spec, this.spawner) } @@ -352,18 +296,51 @@ class LocalLspProvider implements LspProvider { /** Dispose every live instance and block further queries. */ async disposeAll(): Promise { this.disposed = true - this.lifetime.abort(new LspError('lsp-local provider is disposed', 'LSP_DISPOSED')) const live = [...this.instances.values()] const draining = [...this.queues.values()] - const resolving = [...this.workspaceLookups] this.instances.clear() - const results = await Promise.allSettled([ + await Promise.all([ ...live.map(instance => instance.dispose()), ...draining, - ...resolving, ]) this.queues.clear() - this.workspaceLookups.clear() - throwTeardownFailures(results, 'lsp-local instance teardown failed') + } +} + +/** The seam's scrubbed parent env (credential-shaped and DSH_* names dropped), plus the config's explicit env. */ +function buildChildEnv(extra: Record): Record { + return { ...scrubbedParentEnv(), ...extra } +} + +/** + * Resolve the server executable to an absolute path: an absolute command is verified directly; a + * bare command is looked up on the child's PATH. Fails loudly when nothing is executable. + */ +function resolveExecutable(command: string, childEnv: Record): string { + if (isAbsolute(command)) { + // Verify an absolute command too, so an unavailable one fails at load, not on the first query. + if (!isExecutableFileSync(command)) { + throw new Error(`lsp-local: command "${command}" is not an executable file`) + } + return command + } + /* v8 ignore next -- buildChildEnv always sets PATH from the ambient env; the further fallbacks are defensive. */ + const pathValue = childEnv.PATH ?? process.env.PATH ?? '' + for (const dir of pathValue.split(delimiter)) { + if (dir === '') continue + const candidate = join(dir, command) + if (isExecutableFileSync(candidate)) return candidate + } + throw new Error(`lsp-local: command "${command}" was not found on PATH`) +} + +/** Synchronous regular-file and executable check used only at load-time resolution. */ +function isExecutableFileSync(path: string): boolean { + try { + if (!statSync(path).isFile()) return false + accessSync(path, constants.X_OK) + return true + } catch { + return false } } diff --git a/packages/lsp/lsp-local/src/instance.ts b/packages/lsp/lsp-local/src/instance.ts index d643db1528..5a12192c62 100644 --- a/packages/lsp/lsp-local/src/instance.ts +++ b/packages/lsp/lsp-local/src/instance.ts @@ -7,7 +7,6 @@ * @module @deepseek-ai/dsh-lsp-local/instance */ -import { pathToFileURL } from 'node:url' import { LspError } from '@deepseek-ai/dsh-lsp' import type { LspOperation, @@ -37,6 +36,12 @@ export interface InstanceSpec extends ConnectionSpec { readonly shutdownTimeoutMs: number /** PID advertised to the server; `null` when client and server do not share a process namespace. */ readonly clientProcessId?: number | null + /** + * Encode one implementation-native absolute path as a file URI. + * @param path - Canonical workspace or source path. + * @returns A file URI interpreted in the server's filesystem namespace. + */ + readonly pathToFileUri: (path: string) => string } /** @@ -111,8 +116,8 @@ export class LspInstance { private async initialize(): Promise { const initializeResult = await this.connection.request('initialize', { processId: this.spec.clientProcessId === undefined ? process.pid : this.spec.clientProcessId, - rootUri: pathToFileURL(this.spec.cwd).href, - workspaceFolders: [{ uri: pathToFileURL(this.spec.cwd).href, name: 'workspace' }], + rootUri: this.spec.pathToFileUri(this.spec.cwd), + workspaceFolders: [{ uri: this.spec.pathToFileUri(this.spec.cwd), name: 'workspace' }], capabilities: CLIENT_CAPABILITIES, initializationOptions: this.spec.initializationOptions, }) as WireInitializeResult @@ -149,7 +154,7 @@ export class LspInstance { throw new LspError('server does not support the transient textDocument/didOpen this host requires', 'LSP_UNSUPPORTED_OPERATION') } - const uri = pathToFileURL(source.canonicalPath).href + const uri = this.spec.pathToFileUri(source.canonicalPath) let opened = false try { /* v8 ignore next -- guards an abort landing between the ready wait and didOpen; not deterministically reproducible. */ diff --git a/packages/lsp/lsp-local/tests/instance.spec.ts b/packages/lsp/lsp-local/tests/instance.spec.ts index 939b58f0b1..fc1d05843c 100644 --- a/packages/lsp/lsp-local/tests/instance.spec.ts +++ b/packages/lsp/lsp-local/tests/instance.spec.ts @@ -56,6 +56,7 @@ function makeInstance( maxStderrBytes: 100_000, shutdownTimeoutMs: 200, killGraceMs: 200, + pathToFileUri: path => pathToFileURL(path).href, ...overrides, }, spawnSubprocess, writer) live.push(instance) @@ -91,6 +92,7 @@ function scriptInstance(script: string, overrides: Partial = {}): maxStderrBytes: 100_000, shutdownTimeoutMs: 150, killGraceMs: 150, + pathToFileUri: path => pathToFileURL(path).href, ...overrides, }, spawnSubprocess) live.push(instance)