diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 33f357d2bf..7ce45c6817 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -591,7 +591,7 @@ Owns one lazily consumable E2B SDK handle and its final kill/pause/leave decisio async getSandbox(): Promise ``` -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) diff --git a/packages/e2b/subprocess-e2b/src/terminal.ts b/packages/e2b/subprocess-e2b/src/terminal.ts index 88bea44fde..1a916c0fd5 100644 --- a/packages/e2b/subprocess-e2b/src/terminal.ts +++ b/packages/e2b/subprocess-e2b/src/terminal.ts @@ -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 private topLevelExited = false - private termination: Promise | 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 { - 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((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 { @@ -261,7 +234,6 @@ export class E2BTerminalHandle implements SubprocessTerminalHandle { } private async signalGroups(groups: number[], signal: 'TERM' | 'KILL'): Promise { - 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(() => {}) } diff --git a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts index 9022e3d213..43cb88ea87 100644 --- a/packages/e2b/subprocess-e2b/tests/terminal.spec.ts +++ b/packages/e2b/subprocess-e2b/tests/terminal.spec.ts @@ -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[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 => { + 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 = {}): 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 () => {