refactor(e2b): delegate terminal lifecycle
This commit is contained in:
@@ -591,7 +591,7 @@ Owns one lazily consumable E2B SDK handle and its final kill/pause/leave decisio
|
||||
async getSandbox(): Promise<Sandbox>
|
||||
```
|
||||
|
||||
Source: [`packages/e2b/e2b/src/index.ts:119`](../../packages/e2b/e2b/src/index.ts)
|
||||
Source: [`packages/e2b/e2b/src/index.ts:97`](../../packages/e2b/e2b/src/index.ts)
|
||||
|
||||
## `ctx.fs` — `FileSystem` (abstract seam)
|
||||
|
||||
@@ -1180,7 +1180,7 @@ list(owner: Agent): PtySessionSnapshot[]
|
||||
|
||||
Types: [Agent](../core-data-structures/core.md) · [PtyBackend](../core-data-structures/pty.md) · [PtyReadRequest](../core-data-structures/pty.md) · [PtyReadResult](../core-data-structures/pty.md) · [PtySendOperation](../core-data-structures/pty.md) · [PtySendRequest](../core-data-structures/pty.md) · [PtySessionId](../core-data-structures/pty.md) · [PtySessionSnapshot](../core-data-structures/pty.md) · [PtySignal](../core-data-structures/pty.md) · [PtySignalResult](../core-data-structures/pty.md) · [PtySpawnRequest](../core-data-structures/pty.md) · [PtySpawnResult](../core-data-structures/pty.md)
|
||||
|
||||
Source: [`packages/pty/pty/src/index.ts:114`](../../packages/pty/pty/src/index.ts)
|
||||
Source: [`packages/pty/pty/src/index.ts:105`](../../packages/pty/pty/src/index.ts)
|
||||
|
||||
## `ctx.sandbox` — `SandboxProvider` (abstract seam)
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
quoteE2BShellArg,
|
||||
} from '@deepseek-ai/dsh-e2b'
|
||||
import type { CommandHandle, CommandResult, Sandbox } from '@deepseek-ai/dsh-e2b'
|
||||
import { SENSITIVE_ENV_PATTERN } from '@deepseek-ai/dsh-subprocess'
|
||||
import { SENSITIVE_ENV_PATTERN, SubprocessTerminalLifecycle } from '@deepseek-ai/dsh-subprocess'
|
||||
import type {
|
||||
SubprocessOutcome,
|
||||
SubprocessTerminalForeground,
|
||||
@@ -127,9 +127,8 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle {
|
||||
readonly done: Promise<SubprocessOutcome>
|
||||
|
||||
private topLevelExited = false
|
||||
private termination: Promise<void> | undefined
|
||||
private readonly lifecycle: SubprocessTerminalLifecycle
|
||||
private terminationSignal: NodeJS.Signals | null = null
|
||||
private removeAbort: (() => void) | undefined
|
||||
|
||||
constructor(
|
||||
private readonly sandbox: Sandbox,
|
||||
@@ -143,13 +142,11 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle {
|
||||
) {
|
||||
this.pid = handle.pid
|
||||
this.done = this.waitForCommand()
|
||||
void this.done.then(() => { this.terminate() }, () => { this.terminate() })
|
||||
if (signal !== undefined) {
|
||||
const onAbort = (): void => { this.terminate() }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
this.removeAbort = () => { signal.removeEventListener('abort', onAbort) }
|
||||
if (signal.aborted) this.terminate()
|
||||
}
|
||||
this.lifecycle = new SubprocessTerminalLifecycle({
|
||||
done: this.done,
|
||||
cleanup: () => this.closeOnce(),
|
||||
signal,
|
||||
})
|
||||
}
|
||||
|
||||
/** @inheritdoc */
|
||||
@@ -192,36 +189,12 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle {
|
||||
|
||||
/** @inheritdoc */
|
||||
terminate(): void {
|
||||
this.termination ??= this.closeOnce().catch((error: unknown) => {
|
||||
this.termination = undefined
|
||||
throw error
|
||||
})
|
||||
void this.termination.catch(() => {})
|
||||
this.lifecycle.terminate()
|
||||
}
|
||||
|
||||
/** @inheritdoc */
|
||||
async waitForExit(signal?: AbortSignal): Promise<boolean> {
|
||||
const quiescence = this.termination ?? this.done.then(
|
||||
() => { this.terminate(); return this.termination },
|
||||
() => { this.terminate(); return this.termination },
|
||||
)
|
||||
if (signal === undefined) {
|
||||
await quiescence
|
||||
return true
|
||||
}
|
||||
if (signal.aborted) return false
|
||||
return await new Promise<boolean>((resolve, reject) => {
|
||||
const onAbort = (): void => { cleanup(); resolve(false) }
|
||||
const cleanup = (): void => { signal.removeEventListener('abort', onAbort) }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
void quiescence.then(
|
||||
() => { cleanup(); resolve(true) },
|
||||
(error: unknown) => {
|
||||
cleanup()
|
||||
reject(error instanceof Error ? error : new Error(String(error)))
|
||||
},
|
||||
)
|
||||
})
|
||||
return await this.lifecycle.waitForExit(signal)
|
||||
}
|
||||
|
||||
private async waitForCommand(): Promise<SubprocessOutcome> {
|
||||
@@ -261,7 +234,6 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle {
|
||||
}
|
||||
|
||||
private async signalGroups(groups: number[], signal: 'TERM' | 'KILL'): Promise<void> {
|
||||
if (groups.length === 0) return
|
||||
try {
|
||||
await this.sandbox.commands.run(`kill -${signal} -- ${groups.map(group => `-${group}`).join(' ')}`)
|
||||
} catch (error: unknown) {
|
||||
@@ -301,8 +273,6 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle {
|
||||
if (!this.topLevelExited) {
|
||||
throw new Error(`subprocess-e2b: terminal cleanup failed; surviving pid: ${this.pid}`)
|
||||
}
|
||||
this.removeAbort?.()
|
||||
this.removeAbort = undefined
|
||||
await this.handle.disconnect()
|
||||
await this.sandbox.files.remove(this.stateDir).catch(() => {})
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { Buffer } from 'node:buffer'
|
||||
import { once } from 'node:events'
|
||||
import { PassThrough } from 'node:stream'
|
||||
import { Context } from 'cordis'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
@@ -12,7 +13,7 @@ import {
|
||||
import type E2BSandboxService from '@deepseek-ai/dsh-e2b'
|
||||
import type { SubprocessTerminalSpawnSpec } from '@deepseek-ai/dsh-subprocess'
|
||||
import E2BSubprocessService from '@deepseek-ai/dsh-subprocess-e2b'
|
||||
import { spawnE2BTerminal } from '../src/terminal.ts'
|
||||
import { E2BTerminalHandle, spawnE2BTerminal } from '../src/terminal.ts'
|
||||
|
||||
function commandError(exitCode: number): CommandExitError {
|
||||
return new CommandExitError({ exitCode, stdout: '', stderr: '', error: `exit ${exitCode}` })
|
||||
@@ -80,6 +81,7 @@ class FakeTerminalSandbox {
|
||||
createOptions: Parameters<Sandbox['pty']['create']>[0] | undefined
|
||||
ambient = 'KEEP=visible\0NPM_TOKEN=secret\0DSH_STALE=old\0BROKEN\0=bad\0'
|
||||
ready: string | Error = 'ready\n'
|
||||
readyMisses = 0
|
||||
sessionId = '123\n'
|
||||
foreground = '456\n'
|
||||
groups = [123]
|
||||
@@ -106,6 +108,10 @@ class FakeTerminalSandbox {
|
||||
return files.map(() => ({}))
|
||||
},
|
||||
read: async (): Promise<string> => {
|
||||
if (this.readyMisses > 0) {
|
||||
this.readyMisses -= 1
|
||||
throw new FileNotFoundError('not ready')
|
||||
}
|
||||
if (this.ready instanceof Error) throw this.ready
|
||||
return this.ready
|
||||
},
|
||||
@@ -191,6 +197,7 @@ function spec(overrides: Partial<SubprocessTerminalSpawnSpec> = {}): SubprocessT
|
||||
describe('E2B terminal allocation', () => {
|
||||
it('boots the requested argv through a private runner and preserves buffered bytes', async () => {
|
||||
const fake = new FakeTerminalSandbox()
|
||||
fake.readyMisses = 1
|
||||
const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/terminal-one')
|
||||
let output = ''
|
||||
terminal.output.on('data', (chunk) => { output += String(chunk) })
|
||||
@@ -281,6 +288,9 @@ describe('E2B terminal allocation', () => {
|
||||
await expect(spawnE2BTerminal(runtime(invalidSession), spec(), '/runtime/session'))
|
||||
.rejects.toThrow('cannot resolve process session')
|
||||
expect(invalidSession.handle.sdkKills).toBe(1)
|
||||
const lateData = invalidSession.createOptions?.onData
|
||||
if (lateData === undefined) throw new Error('missing captured terminal callback')
|
||||
expect(lateData(Buffer.from('late bytes'))).toBeUndefined()
|
||||
|
||||
const cleanupFailed = new FakeTerminalSandbox()
|
||||
cleanupFailed.handle.pid = 0
|
||||
@@ -321,6 +331,26 @@ describe('E2B terminal lifecycle', () => {
|
||||
await expect(terminal.write(Buffer.from('late'))).rejects.toThrow('exited')
|
||||
fake.foregroundFailure = commandError(1)
|
||||
await expect(terminal.inspectForeground()).resolves.toBeUndefined()
|
||||
await expect(terminal.signalForeground('SIGINT')).rejects.toThrow('cannot resolve foreground process group')
|
||||
})
|
||||
|
||||
it('starts cleanup when the lifetime signal is already aborted at handle publication', async () => {
|
||||
const fake = new FakeTerminalSandbox()
|
||||
const controller = new AbortController()
|
||||
controller.abort(new Error('publication cancelled'))
|
||||
const terminal = new E2BTerminalHandle(
|
||||
fake.sandbox,
|
||||
fake.handle.asHandle(),
|
||||
new PassThrough(),
|
||||
fake.handle.wait(),
|
||||
123,
|
||||
'/runtime/pre-aborted',
|
||||
1,
|
||||
controller.signal,
|
||||
)
|
||||
|
||||
await expect(terminal.done).resolves.toEqual({ exitCode: null, signal: 'SIGTERM' })
|
||||
await expect(terminal.waitForExit()).resolves.toBe(true)
|
||||
})
|
||||
|
||||
it.each([
|
||||
@@ -404,6 +434,27 @@ describe('E2B terminal lifecycle', () => {
|
||||
await terminal.done
|
||||
terminal.terminate()
|
||||
await expect(terminal.waitForExit()).resolves.toBe(true)
|
||||
|
||||
const alreadyExited = new FakeTerminalSandbox()
|
||||
alreadyExited.termFailure = commandError(1)
|
||||
const tolerant = await spawnE2BTerminal(runtime(alreadyExited), spec({ graceMs: 1 }), '/runtime/group-exited')
|
||||
tolerant.terminate()
|
||||
await expect(tolerant.done).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
|
||||
await expect(tolerant.waitForExit()).resolves.toBe(true)
|
||||
})
|
||||
|
||||
it('normalizes a non-Error cleanup rejection for an observing wait', async () => {
|
||||
const fake = new FakeTerminalSandbox()
|
||||
const terminal = await spawnE2BTerminal(runtime(fake), spec(), '/runtime/non-error-cleanup')
|
||||
fake.commandFailure = 'cleanup transport gone'
|
||||
terminal.terminate()
|
||||
await expect(terminal.waitForExit(new AbortController().signal)).rejects.toThrow('cleanup transport gone')
|
||||
|
||||
fake.groups = []
|
||||
fake.handle.succeed(0)
|
||||
await terminal.done
|
||||
terminal.terminate()
|
||||
await expect(terminal.waitForExit()).resolves.toBe(true)
|
||||
})
|
||||
|
||||
it('keeps command rejection authoritative while cleanup is already waiting', async () => {
|
||||
|
||||
Reference in New Issue
Block a user