DeepSeek/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness synced 2026-08-15 21:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(e2b): close remote lifecycle gaps

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-28 16:26:20 +08:00
parent e64d40837c
commit 3dea36f1ce
22 changed files with 568 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
packages/e2b/code-runtime-e2b/src/index.ts
View File

@@ -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<Config>
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<PreparedRuntime>
private readonly live = new Set<LiveRun>()
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<typeof this.ready>
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<PreparedRuntime | undefined> {
if (signal === undefined) return this.ready
return new Promise<PreparedRuntime | undefined>((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<PreparedRuntime> {
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<string, CodeBindingNamespace>,
runtime: { node: string; runner: string },
runtime: PreparedRuntime,
): Promise<CodeRunResult> {
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<void>((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 */
}

142
packages/e2b/code-runtime-e2b/tests/runtime.spec.ts
View File

@@ -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<boolean> | 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<boolean>
} = {},
) {
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<boolean> {
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<Sandbox>()
const fixture = await setup([], {}, {}, () => gate.promise)
const running = fixture.runtime.run(request())
await (fixture.runtime as unknown as { teardown(): Promise<void> }).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<Sandbox>()
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<void> }).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<boolean>()
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 })