runtime.spec.ts 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598
  1. import { PassThrough, Writable } from 'node:stream'
  2. import { Context } from 'cordis'
  3. import { describe, expect, it, vi } from 'vitest'
  4. import type { Sandbox } from '@deepseek-ai/dsh-e2b'
  5. import {
  6. E2BFrameDecoder,
  7. encodeE2BFrame,
  8. } from '@deepseek-ai/dsh-e2b'
  9. import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
  10. import type {
  11. SubprocessHandle,
  12. SubprocessOutcome,
  13. SubprocessSpawnSpec,
  14. } from '@deepseek-ai/dsh-subprocess'
  15. import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b'
  16. import {
  17. encodeWorkerJson,
  18. } from '@deepseek-ai/dsh-code-runtime-worker'
  19. import E2BCodeRuntime from '@deepseek-ai/dsh-code-runtime-e2b'
  20. import * as E2BCodeRuntimeInvariant from '../src/invariant.ts'
  21. import { CODE_RUNNER_SOURCE } from '../src/runner-source.ts'
  22. import InvariantService from '@deepseek-ai/dsh-invariants'
  23. class FakeHandle implements SubprocessHandle {
  24. readonly pid = 123
  25. readonly stdin: Writable | undefined
  26. readonly stdout: PassThrough | undefined
  27. readonly stderr = undefined
  28. readonly collected: SubprocessHandle['collected']
  29. readonly done: Promise<SubprocessOutcome>
  30. readonly writes: unknown[] = []
  31. readonly result = Promise.withResolvers<SubprocessOutcome>()
  32. terminated = 0
  33. waitCalls = 0
  34. private readonly decoder = new E2BFrameDecoder(10_000_000)
  35. private readonly waitError: Error | undefined
  36. private readonly waitResult: Promise<boolean> | undefined
  37. private settled = false
  38. constructor(
  39. private readonly onMessage: (message: unknown, handle: FakeHandle) => void = () => {},
  40. options: {
  41. stdin?: boolean
  42. stdout?: boolean
  43. stderr?: string
  44. writeError?: Error
  45. waitError?: Error
  46. waitResult?: Promise<boolean>
  47. } = {},
  48. ) {
  49. this.waitError = options.waitError
  50. this.waitResult = options.waitResult
  51. this.stdin = options.stdin === false
  52. ? undefined
  53. : options.writeError === undefined
  54. ? new PassThrough()
  55. : new Writable({ write: (_chunk, _encoding, callback) => { callback(options.writeError) } })
  56. this.stdout = options.stdout === false ? undefined : new PassThrough()
  57. this.collected = options.stderr === undefined
  58. ? {}
  59. : { stderr: { readFrom: () => ({ text: options.stderr as string, nextOffset: 0, lossy: false }) } }
  60. this.done = this.result.promise
  61. this.stdin?.on('data', (chunk: Buffer) => {
  62. for (const message of this.decoder.push(chunk.toString('ascii'))) {
  63. this.writes.push(message)
  64. this.onMessage(message, this)
  65. }
  66. })
  67. }
  68. emit(message: unknown): void {
  69. this.stdout?.write(encodeE2BFrame(message))
  70. }
  71. emitRaw(text: string): void {
  72. this.stdout?.write(text)
  73. }
  74. exit(outcome: SubprocessOutcome = { exitCode: 0, signal: null }): void {
  75. if (this.settled) return
  76. this.settled = true
  77. this.stdout?.end()
  78. this.result.resolve(outcome)
  79. }
  80. crash(error: unknown): void {
  81. if (this.settled) return
  82. this.settled = true
  83. this.stdout?.end()
  84. this.result.reject(error)
  85. }
  86. terminate(): void {
  87. this.terminated += 1
  88. this.exit({ exitCode: null, signal: 'SIGTERM' })
  89. }
  90. async waitForExit(): Promise<boolean> {
  91. this.waitCalls += 1
  92. if (this.waitError !== undefined) throw this.waitError
  93. if (this.waitResult !== undefined) return await this.waitResult
  94. return true
  95. }
  96. }
  97. interface RuntimeFixture {
  98. ctx: Context
  99. fiber: Awaited<ReturnType<Context['plugin']>>
  100. runtime: E2BCodeRuntime
  101. sandbox: Sandbox
  102. spawn: ReturnType<typeof vi.fn<(spec: SubprocessSpawnSpec) => SubprocessHandle>>
  103. write: ReturnType<typeof vi.fn>
  104. run: ReturnType<typeof vi.fn>
  105. }
  106. async function setup(
  107. handles: FakeHandle[] = [],
  108. config: Record<string, number> = {},
  109. sandboxOverrides: Partial<Sandbox> = {},
  110. getSandbox?: () => Promise<Sandbox>,
  111. ): Promise<RuntimeFixture> {
  112. const write = vi.fn().mockResolvedValue([])
  113. const run = vi.fn().mockImplementation(async (command: string) => ({
  114. exitCode: 0,
  115. stdout: command.startsWith('command -v') ? '/usr/bin/node\n' : '',
  116. stderr: '',
  117. }))
  118. const sandbox = {
  119. files: { write },
  120. commands: { run },
  121. ...sandboxOverrides,
  122. } as unknown as Sandbox
  123. const e2b = {
  124. cwd: '/workspace',
  125. runtimeRoot: '/workspace/.dsh-e2b',
  126. getSandbox: getSandbox ?? (async () => sandbox),
  127. } as unknown as E2BSandboxService
  128. const spawn = vi.fn<(spec: SubprocessSpawnSpec) => SubprocessHandle>(() => {
  129. const handle = handles.shift()
  130. if (handle === undefined) throw new Error('no fake handle queued')
  131. return handle
  132. })
  133. const subprocess = Object.create(E2BSubprocessService.prototype) as E2BSubprocessService
  134. Object.defineProperty(subprocess, 'spawn', { value: spawn })
  135. const ctx = new Context()
  136. ctx.provide('e2b', e2b)
  137. ctx.provide('subprocess', subprocess)
  138. const fiber = await ctx.plugin(E2BCodeRuntime, config)
  139. return { ctx, fiber, runtime: ctx.codeRuntime as E2BCodeRuntime, sandbox, spawn, write, run }
  140. }
  141. function request(program = 'return 1') {
  142. return { program, bindings: [] }
  143. }
  144. describe('E2BCodeRuntime', () => {
  145. it('prepares the remote runner and returns logs and a lossless completion', async () => {
  146. const handle = new FakeHandle((message, current) => {
  147. if ((message as { type?: string }).type !== 'boot') return
  148. current.emit({ type: 'log', text: 'remote 你好' })
  149. current.emitRaw(
  150. encodeE2BFrame({ type: 'done', value: encodeWorkerJson({ answer: 42 }) })
  151. + encodeE2BFrame({ type: 'log', text: 'ignored after done' }),
  152. )
  153. current.emit({ type: 'log', text: 'also ignored after done' })
  154. })
  155. const fixture = await setup([handle])
  156. await expect(fixture.runtime.run(request('const answer: number = 42; return { answer }')))
  157. .resolves.toEqual({ logs: ['remote 你好'], value: { answer: 42 } })
  158. expect(fixture.runtime.language).toBe('typescript')
  159. expect(fixture.runtime.isolation).toBe('container')
  160. expect(fixture.write).toHaveBeenCalledWith([{ path: '/workspace/.dsh-e2b/code-runtime-runner.mjs', data: CODE_RUNNER_SOURCE }])
  161. expect(fixture.run).toHaveBeenCalledWith("chmod 600 -- '/workspace/.dsh-e2b/code-runtime-runner.mjs'")
  162. expect(fixture.spawn).toHaveBeenCalledWith(expect.objectContaining({
  163. argv: ['/usr/bin/node', '/workspace/.dsh-e2b/code-runtime-runner.mjs'],
  164. cwd: '/workspace',
  165. env: {},
  166. }))
  167. expect(handle.terminated).toBe(1)
  168. expect(handle.waitCalls).toBe(1)
  169. await fixture.fiber.dispose()
  170. })
  171. it('bridges binding success, host rejection, unknown members, and invalid values', async () => {
  172. const replies: unknown[] = []
  173. const handle = new FakeHandle((message, current) => {
  174. const record = message as { type?: string; id?: number; ok?: boolean }
  175. if (record.type === 'boot') {
  176. current.emit({ type: 'call', id: 1, global: 'bridge', name: 'double', args: encodeWorkerJson({ value: 4 }) })
  177. current.emit({ type: 'call', id: 2, global: 'bridge', name: 'fail', args: encodeWorkerJson(null) })
  178. current.emit({ type: 'call', id: 3, global: 'bridge', name: 'missing', args: encodeWorkerJson(null) })
  179. current.emit({ type: 'call', id: 4, global: 'bridge', name: 'double', args: [] })
  180. current.emit({ type: 'call', id: 5, global: 'bridge', name: 'invalid', args: encodeWorkerJson(null) })
  181. current.emit({ type: 'call', id: 6, global: 'bridge', name: 'throwing', args: encodeWorkerJson(null) })
  182. current.emit({ type: 'call', id: 1, global: 'bridge', name: 'double', args: encodeWorkerJson({ value: 99 }) })
  183. return
  184. }
  185. if (record.type === 'reply') {
  186. replies.push(message)
  187. if (replies.length === 6) current.emit({ type: 'done', value: encodeWorkerJson('done') })
  188. }
  189. })
  190. const fixture = await setup([handle])
  191. const result = await fixture.runtime.run({
  192. program: 'return await bridge.double({ value: 4 })',
  193. bindings: [
  194. {
  195. global: 'bridge',
  196. errorClass: { name: 'BridgeError', memberNameProperty: 'member' },
  197. functions: {
  198. double: async args => (args as { value: number }).value * 2,
  199. fail: async () => { throw 'nope' },
  200. invalid: (async () => undefined) as never,
  201. throwing: async () => Object.defineProperty({}, 'value', {
  202. enumerable: true,
  203. get: () => { throw new Error('getter failed') },
  204. }),
  205. },
  206. },
  207. { global: 'plain', functions: {} },
  208. ],
  209. })
  210. expect(result).toEqual({ logs: [], value: 'done' })
  211. expect(replies.sort((left, right) => (left as { id: number }).id - (right as { id: number }).id)).toEqual([
  212. { type: 'reply', id: 1, ok: true, value: encodeWorkerJson(8) },
  213. { type: 'reply', id: 2, ok: false, message: 'nope' },
  214. { type: 'reply', id: 3, ok: false, message: 'unknown binding "bridge.missing"' },
  215. { type: 'reply', id: 4, ok: false, message: 'binding arguments must be lossless JSON' },
  216. { type: 'reply', id: 5, ok: false, message: 'binding resolution must be lossless JSON' },
  217. { type: 'reply', id: 6, ok: false, message: 'binding resolution must be lossless JSON' },
  218. ])
  219. await fixture.fiber.dispose()
  220. })
  221. it('ignores malformed runner traffic and classifies terminal runner messages', async () => {
  222. const ignored = [
  223. null, 1, {}, { type: 'log' }, { type: 'call' },
  224. { type: 'call', id: 0, global: 'x', name: 'y', args: [] },
  225. { type: 'call', id: 1, global: 1, name: 'y', args: [] },
  226. { type: 'call', id: 1, global: 'x', name: 1, args: [] },
  227. { type: 'call', id: 1, global: 'x', name: 'y', args: {} },
  228. { type: 'done', error: null },
  229. { type: 'done', error: { kind: 'invented', message: 'x' } },
  230. { type: 'done', error: { kind: 'exception', message: 1 } },
  231. ]
  232. const handles = [
  233. new FakeHandle((message, current) => {
  234. if ((message as { type?: string }).type !== 'boot') return
  235. for (const item of ignored) current.emit(item)
  236. current.emit({ type: 'done' })
  237. }),
  238. new FakeHandle((message, current) => {
  239. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'done', error: { kind: 'exception', message: 'boom' } })
  240. }),
  241. new FakeHandle((message, current) => {
  242. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'done', value: [] })
  243. }),
  244. new FakeHandle((message, current) => {
  245. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'output-limit' })
  246. }),
  247. ]
  248. const fixture = await setup(handles, { maxOutputBytes: 64, maxFrameBytes: 128 })
  249. await expect(fixture.runtime.run(request())).resolves.toEqual({ logs: [] })
  250. await expect(fixture.runtime.run(request())).resolves.toEqual({ logs: [], error: { kind: 'exception', message: 'boom' } })
  251. await expect(fixture.runtime.run(request())).resolves.toEqual({ logs: [], error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' } })
  252. await expect(fixture.runtime.run(request())).resolves.toEqual({ logs: [], error: { kind: 'output-limit', message: 'outer output exceeded 64 bytes' } })
  253. await fixture.fiber.dispose()
  254. })
  255. it('enforces the host output ledger and catches malformed bridge output', async () => {
  256. const handles = [
  257. new FakeHandle((message, current) => {
  258. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'log', text: 'x'.repeat(1_000) })
  259. }),
  260. new FakeHandle((message, current) => {
  261. if ((message as { type?: string }).type === 'boot') current.emitRaw('not-base64\n')
  262. }),
  263. new FakeHandle((message, current) => {
  264. if ((message as { type?: string }).type === 'boot') current.emitRaw('é')
  265. }),
  266. new FakeHandle((message, current) => {
  267. if ((message as { type?: string }).type === 'boot') current.stdout?.emit('error', new Error('stdout broke'))
  268. }),
  269. ]
  270. const fixture = await setup(handles, { maxOutputBytes: 128, maxFrameBytes: 4_096 })
  271. expect((await fixture.runtime.run(request())).error?.kind).toBe('output-limit')
  272. const malformed = (await fixture.runtime.run(request())).error
  273. expect(malformed?.kind).toBe('worker-exit')
  274. expect(malformed?.message).toContain('bridge failed')
  275. expect((await fixture.runtime.run(request())).error?.message).toContain('non-ASCII')
  276. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime stdout failed: stdout broke' })
  277. await fixture.fiber.dispose()
  278. })
  279. it('enforces the outbound frame bound on boot and binding replies', async () => {
  280. const oversizedBoot = new FakeHandle()
  281. const oversizedReply = new FakeHandle((message, current) => {
  282. if ((message as { type?: string }).type === 'boot') {
  283. current.emit({ type: 'call', id: 1, global: 'bridge', name: 'large', args: encodeWorkerJson(null) })
  284. }
  285. })
  286. const fixture = await setup([oversizedBoot, oversizedReply], { maxOutputBytes: 128, maxFrameBytes: 512 })
  287. const bootResult = await fixture.runtime.run(request(`return ${JSON.stringify('x'.repeat(1_000))}`))
  288. expect(bootResult.error).toMatchObject({ kind: 'worker-exit' })
  289. expect(bootResult.error?.message).toContain('frame exceeded its byte limit')
  290. expect(oversizedBoot.writes).toHaveLength(0)
  291. const replyResult = await fixture.runtime.run({
  292. program: 'return await bridge.large(null)',
  293. bindings: [{ global: 'bridge', functions: { large: async () => 'x'.repeat(1_000) } }],
  294. })
  295. expect(replyResult.error).toMatchObject({ kind: 'worker-exit' })
  296. expect(replyResult.error?.message).toContain('frame exceeded its byte limit')
  297. expect(oversizedReply.writes).toHaveLength(1)
  298. await fixture.fiber.dispose()
  299. })
  300. it('contains stdin errors, process exits, spawn failures, and missing pipes', async () => {
  301. const writeError = new FakeHandle(() => {}, { writeError: new Error('write callback broke') })
  302. const stdinError = new FakeHandle((message, current) => {
  303. if ((message as { type?: string }).type === 'boot') current.stdin?.emit('error', new Error('stdin broke'))
  304. })
  305. const earlyExit = new FakeHandle(() => {}, { stderr: 'remote diagnostic' })
  306. const quietExit = new FakeHandle()
  307. const emptyStderrExit = new FakeHandle(() => {}, { stderr: '' })
  308. const spawnFailure = new FakeHandle()
  309. const missingStdin = new FakeHandle(() => {}, { stdin: false })
  310. const missingStdout = new FakeHandle(() => {}, { stdout: false, waitError: new Error('missing-stream process query failed') })
  311. const truncated = new FakeHandle((message, current) => {
  312. if ((message as { type?: string }).type === 'boot') {
  313. current.emitRaw('YQ==')
  314. setImmediate(() => { current.exit() })
  315. }
  316. })
  317. const cleanupFailure = new FakeHandle((message, current) => {
  318. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'done' })
  319. }, { waitError: new Error('process query failed') })
  320. const fixture = await setup([
  321. writeError, stdinError, earlyExit, quietExit, emptyStderrExit,
  322. spawnFailure, missingStdin, missingStdout, truncated, cleanupFailure,
  323. ])
  324. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime bridge write failed: write callback broke' })
  325. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime stdin failed: stdin broke' })
  326. setImmediate(() => { earlyExit.exit() })
  327. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime exited before completing: remote diagnostic' })
  328. setImmediate(() => { quietExit.exit() })
  329. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime exited before completing' })
  330. setImmediate(() => { emptyStderrExit.exit() })
  331. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime exited before completing' })
  332. setImmediate(() => { spawnFailure.crash('spawn rejected') })
  333. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime spawn failed: spawn rejected' })
  334. expect((await fixture.runtime.run(request())).error?.message).toContain('dropped a piped runtime stream')
  335. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime cleanup failed: missing-stream process query failed' })
  336. expect(missingStdin.terminated).toBe(1)
  337. expect(missingStdin.waitCalls).toBe(1)
  338. expect(missingStdout.terminated).toBe(1)
  339. expect(missingStdout.waitCalls).toBe(1)
  340. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B frame stream ended mid-frame' })
  341. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'worker-exit', message: 'E2B runtime cleanup failed: process query failed' })
  342. await fixture.fiber.dispose()
  343. })
  344. it('reports wall timeout, abort, pre-abort, type-strip failure, and disposal', async () => {
  345. const timeout = new FakeHandle()
  346. const abort = new FakeHandle()
  347. const disposing = new FakeHandle()
  348. const fixture = await setup([timeout, abort, disposing], { maxWallMs: 20 })
  349. expect((await fixture.runtime.run(request())).error).toEqual({ kind: 'timeout', message: 'wall-clock ceiling reached (20ms)' })
  350. const controller = new AbortController()
  351. const aborting = fixture.runtime.run({ ...request(), signal: controller.signal })
  352. controller.abort('stop')
  353. expect((await aborting).error).toEqual({ kind: 'abort', message: 'stop' })
  354. expect((await fixture.runtime.run({ ...request(), signal: AbortSignal.abort('already') })).error)
  355. .toEqual({ kind: 'abort', message: 'already' })
  356. expect((await fixture.runtime.run(request('enum E { A }'))).error?.kind).toBe('exception')
  357. const live = fixture.runtime.run(request())
  358. await new Promise(resolve => setImmediate(resolve))
  359. await fixture.fiber.dispose()
  360. expect((await live).error).toEqual({ kind: 'abort', message: 'runtime disposed' })
  361. await expect(fixture.runtime.run(request())).rejects.toThrow('after disposal')
  362. })
  363. it('drops binding replies that settle after abort', async () => {
  364. const controller = new AbortController()
  365. const resolution = Promise.withResolvers<string>()
  366. const invoked = Promise.withResolvers<undefined>()
  367. const handle = new FakeHandle((message, current) => {
  368. if ((message as { type?: string }).type === 'boot') {
  369. current.emit({ type: 'call', id: 1, global: 'bridge', name: 'late', args: encodeWorkerJson(null) })
  370. }
  371. })
  372. const fixture = await setup([handle])
  373. const running = fixture.runtime.run({
  374. program: 'return await bridge.late(null)',
  375. bindings: [{
  376. global: 'bridge',
  377. functions: {
  378. late: async () => {
  379. invoked.resolve(undefined)
  380. return await resolution.promise
  381. },
  382. },
  383. }],
  384. signal: controller.signal,
  385. })
  386. await invoked.promise
  387. controller.abort('stop')
  388. expect((await running).error).toEqual({ kind: 'abort', message: 'stop' })
  389. resolution.resolve('late')
  390. await new Promise(resolve => setImmediate(resolve))
  391. expect(handle.writes).toHaveLength(1)
  392. await fixture.fiber.dispose()
  393. })
  394. it('validates binding and runtime configuration before remote execution', async () => {
  395. const fixture = await setup([])
  396. const invalidRequests = [
  397. { global: 'not-valid!', functions: {} },
  398. { global: 'await', functions: {} },
  399. { global: 'console', functions: {} },
  400. { global: 'same', functions: {} },
  401. { global: 'same', functions: {} },
  402. { global: 'ok', functions: {}, errorClass: { name: 'not-valid!', memberNameProperty: 'member' } },
  403. { global: 'ok', functions: {}, errorClass: { name: 'await', memberNameProperty: 'member' } },
  404. { global: 'Clash', functions: {}, errorClass: { name: 'Clash', memberNameProperty: 'member' } },
  405. { global: 'one', functions: {}, errorClass: { name: 'Err', memberNameProperty: 'member' } },
  406. { global: 'two', functions: {}, errorClass: { name: 'Err', memberNameProperty: 'member' } },
  407. { global: 'ok', functions: {}, errorClass: { name: 'Err', memberNameProperty: '' } },
  408. { global: 'ok', functions: {}, errorClass: { name: 'Err', memberNameProperty: 'message' } },
  409. ]
  410. for (const bindings of [
  411. [invalidRequests[0]], [invalidRequests[1]], [invalidRequests[2]],
  412. invalidRequests.slice(3, 5), [invalidRequests[5]], [invalidRequests[6]],
  413. [invalidRequests[7]], invalidRequests.slice(8, 10), [invalidRequests[10]], [invalidRequests[11]],
  414. ]) {
  415. await expect(fixture.runtime.run({ program: 'return 1', bindings: bindings as never })).rejects.toThrow()
  416. }
  417. await fixture.fiber.dispose()
  418. for (const config of [
  419. { computeMs: 0 }, { computeMs: 1.5 }, { maxOutputBytes: 3 },
  420. { maxWallMs: 2_147_483_648 }, { maxFrameBytes: 10, maxOutputBytes: 20 },
  421. ]) {
  422. const ctx = new Context()
  423. const subprocess = Object.create(E2BSubprocessService.prototype) as E2BSubprocessService
  424. ctx.provide('e2b', { getSandbox: async () => ({}) } as never)
  425. ctx.provide('subprocess', subprocess)
  426. await expect(ctx.plugin(E2BCodeRuntime, config)).rejects.toThrow()
  427. }
  428. const wrong = new Context()
  429. wrong.provide('e2b', { getSandbox: async () => ({}) } as never)
  430. wrong.provide('subprocess', {} as never)
  431. await expect(wrong.plugin(E2BCodeRuntime, {})).rejects.toThrow('dsh-subprocess-e2b')
  432. })
  433. it('turns asynchronous runtime preparation failure into a run result', async () => {
  434. const sandbox = {
  435. files: { write: vi.fn().mockRejectedValue(new Error('upload failed')) },
  436. commands: { run: vi.fn() },
  437. } as unknown as Sandbox
  438. const fixture = await setup([], {}, sandbox)
  439. expect((await fixture.runtime.run(request())).error).toEqual({
  440. kind: 'worker-exit',
  441. message: 'E2B runtime setup failed: upload failed',
  442. })
  443. await fixture.fiber.dispose()
  444. })
  445. it('returns disposal when remote preparation completes after teardown', async () => {
  446. const gate = Promise.withResolvers<Sandbox>()
  447. const fixture = await setup([], {}, {}, () => gate.promise)
  448. const running = fixture.runtime.run(request())
  449. const disposing = fixture.fiber.dispose()
  450. let disposed = false
  451. void disposing.then(() => { disposed = true })
  452. await new Promise(resolve => setImmediate(resolve))
  453. const disposedBeforeSetup = disposed
  454. gate.resolve(fixture.sandbox)
  455. await disposing
  456. expect(disposedBeforeSetup).toBe(false)
  457. expect((await running).error).toEqual({ kind: 'abort', message: 'runtime disposed' })
  458. expect(fixture.write).not.toHaveBeenCalled()
  459. })
  460. it('observes abort while runtime preparation is pending', async () => {
  461. const gate = Promise.withResolvers<Sandbox>()
  462. const fixture = await setup([], {}, {}, () => gate.promise)
  463. const controller = new AbortController()
  464. const running = fixture.runtime.run({ ...request(), signal: controller.signal })
  465. controller.abort('stop during setup')
  466. const early = await Promise.race([
  467. running.then(result => ({ kind: 'result' as const, result })),
  468. new Promise<{ kind: 'pending' }>((resolve) => { setImmediate(() => { resolve({ kind: 'pending' }) }) }),
  469. ])
  470. expect(fixture.spawn).not.toHaveBeenCalled()
  471. gate.resolve(fixture.sandbox)
  472. expect(early).toMatchObject({ kind: 'result', result: { error: { kind: 'abort', message: 'stop during setup' } } })
  473. await running
  474. await fixture.fiber.dispose()
  475. })
  476. it('classifies an abort that races synchronous subprocess spawn', async () => {
  477. const fixture = await setup()
  478. const controller = new AbortController()
  479. fixture.spawn.mockImplementationOnce(() => {
  480. controller.abort('stop at spawn')
  481. throw new Error('aborted before spawn')
  482. })
  483. expect((await fixture.runtime.run({ ...request(), signal: controller.signal })).error)
  484. .toEqual({ kind: 'abort', message: 'stop at spawn' })
  485. fixture.spawn.mockImplementationOnce(() => { throw new Error('synchronous spawn failure') })
  486. expect((await fixture.runtime.run(request())).error).toEqual({
  487. kind: 'worker-exit',
  488. message: 'E2B runtime spawn failed: synchronous spawn failure',
  489. })
  490. await fixture.fiber.dispose()
  491. const disposingFixture = await setup()
  492. disposingFixture.spawn.mockImplementationOnce(() => {
  493. void (disposingFixture.runtime as unknown as { teardown(): Promise<void> }).teardown()
  494. throw new Error('spawn raced disposal')
  495. })
  496. expect((await disposingFixture.runtime.run(request())).error)
  497. .toEqual({ kind: 'abort', message: 'runtime disposed' })
  498. await disposingFixture.fiber.dispose()
  499. })
  500. it('closes both abort races around runtime readiness and live-run publication', async () => {
  501. let preparationAborted = false
  502. const preparationSignal = {
  503. get aborted() { return preparationAborted },
  504. reason: 'preparation race',
  505. addEventListener() { preparationAborted = true },
  506. removeEventListener() {},
  507. } as unknown as AbortSignal
  508. const liveHandle = new FakeHandle()
  509. const fixture = await setup([liveHandle])
  510. expect((await fixture.runtime.run({ ...request(), signal: preparationSignal })).error)
  511. .toEqual({ kind: 'abort', message: 'preparation race' })
  512. expect(fixture.spawn).not.toHaveBeenCalled()
  513. let liveAborted = false
  514. let registrations = 0
  515. const liveSignal = {
  516. get aborted() { return liveAborted },
  517. reason: 'live publication race',
  518. addEventListener() {
  519. registrations += 1
  520. if (registrations === 2) liveAborted = true
  521. },
  522. removeEventListener() {},
  523. } as unknown as AbortSignal
  524. expect((await fixture.runtime.run({ ...request(), signal: liveSignal })).error)
  525. .toEqual({ kind: 'abort', message: 'live publication race' })
  526. await fixture.fiber.dispose()
  527. })
  528. it('retains a live run until remote cleanup reaches quiescence', async () => {
  529. const cleanup = Promise.withResolvers<boolean>()
  530. const handle = new FakeHandle((message, current) => {
  531. if ((message as { type?: string }).type === 'boot') current.emit({ type: 'done' })
  532. }, { waitResult: cleanup.promise })
  533. const fixture = await setup([handle])
  534. const running = fixture.runtime.run(request())
  535. await vi.waitFor(() => { expect(handle.waitCalls).toBe(1) })
  536. const disposing = fixture.fiber.dispose()
  537. let disposed = false
  538. void disposing.then(() => { disposed = true })
  539. await new Promise(resolve => setImmediate(resolve))
  540. const disposedBeforeCleanup = disposed
  541. cleanup.resolve(true)
  542. await expect(running).resolves.toEqual({ logs: [] })
  543. await expect(disposing).resolves.toBeUndefined()
  544. expect(disposedBeforeCleanup).toBe(false)
  545. })
  546. it('registers the package-owned invariant companion', async () => {
  547. const ctx = new Context()
  548. await ctx.plugin(InvariantService, { enabled: true })
  549. const fiber = await ctx.plugin(E2BCodeRuntimeInvariant).await()
  550. await fiber.dispose()
  551. })
  552. })