feat(code-interpreter): reuse a warm sandbox per conversation (E2B) (#21664)
## What
The E2B code-interpreter driver created a **fresh sandbox on every
execution** and killed it in `finally`, so every call in a conversation
paid full cold-start and started blank. This PR keeps **one warm sandbox
per conversation** and, on idle, **pauses** it rather than killing it.
## How
- **Discovery without a registry:** the sandbox is tagged with the chat
`threadId` (scoped `workspaceId:threadId`) via E2B **metadata**, found
with `Sandbox.list({ query: { state: ['running','paused'], metadata }
})` and resumed with `Sandbox.connect()` (which auto-resumes a paused
sandbox). E2B is the source of truth — no Redis/DB mapping.
- **Pause/resume (E2B 2.x):** session sandboxes are created with
`lifecycle: { onTimeout: 'pause', autoResume: true }`. When idle they
**pause** — compute billing stops, filesystem **and** kernel/memory
state are preserved — and resume in ~1s on the next call. This replaces
the earlier keepalive approach.
- **No premature pause mid-run:** the sandbox is kept alive for
`max(execution timeout, idle window)`, so a long execution is never
paused underneath itself.
- **Tenant isolation:** discovery filters by the `twentySessionId` tag
and **re-checks it client-side**, so a loose server-side match can never
hand one conversation's warm sandbox (with its files, kernel state,
token) to another.
- **Concurrency:** executions sharing a session are serialized
in-process (one active stream per thread, run as a single job — the chat
resolver queues concurrent messages), so parallel tool calls can't race
the shared kernel.
- **Output isolation:** `/home/user/output` is reset at the start of
each reused run, so a call only returns the artifacts it actually
produced; durable state lives elsewhere and persists.
## SDK upgrade
`@e2b/code-interpreter` **`^1.0.4` → `^2.6.0`** (pulls `e2b@2.x`). The
typed pause/resume API, `lifecycle`, and the `state`/`metadata` list
filter only exist in the 2.x line; 1.x exposed them only as untyped
OpenAPI internals. `Sandbox.list()` is now a paginator (handled).
## Config
| Var | Default | Purpose |
|---|---|---|
| `CODE_INTERPRETER_TIMEOUT_MS` | `300000` | Max single-execution
duration. |
| `CODE_INTERPRETER_IDLE_TIMEOUT_MS` | `300000` | Idle window before the
warm sandbox auto-pauses. |
Reuse is always-on when a session id is present (chat path). The
workflow-agent path and the dev-only `LocalDriver` are unaffected.
## ⚠️ Open item before merge: paused-sandbox GC
E2B retains paused sandboxes **indefinitely** (no TTL). Unlike the old
keepalive path (which auto-killed on idle), pause means a conversation's
sandbox persists after the chat ends — so without garbage collection,
paused sandboxes accumulate (≈ one per historical conversation) and
consume storage. A GC policy is required; the approach + retention
window are being decided (see PR discussion). Also: the E2B runtime path
can't run in CI, so this still needs a **live smoke test** (reuse hit,
idle→pause, resume) and confirmation of paused-storage pricing before
rollout.
## Tests / checks
- Resolver unit tests (`getOrCreateSessionSandbox`): reuse+extend,
create-when-absent, duplicate reaping, connect-failure fallback,
keep-first-connectable-when-earlier-dead, **ignore cross-tenant
metadata**, and **kill-on-timeout-refresh-failure**.
- `nx typecheck twenty-server` (against e2b 2.x), `oxlint --type-aware`,
`oxfmt --check` all clean.
---------
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Félix Malfait
@@ -13,11 +13,6 @@ npmMinimalAgeGate: 3d
|
||||
npmPreapprovedPackages:
|
||||
- twenty-sdk
|
||||
- twenty-client-sdk
|
||||
# esbuild 0.28.1 (+ its @esbuild/* platform binaries) is the patched release for
|
||||
# GHSA-gv7w-rqvm-qjhr (Deno-module binary-integrity RCE, >=0.17.0 <0.28.1) and the earlier
|
||||
# GHSA-g7r4-m6w7-qqqr (dev-server path-traversal), both published 2026-06-11. Preapproved —
|
||||
# scoped to this exact version — to bypass the 3d age gate so the security fix can land
|
||||
# immediately. Safe to remove once 0.28.1 is older than npmMinimalAgeGate.
|
||||
- esbuild@0.28.1
|
||||
- "@esbuild/*@0.28.1"
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@
|
||||
"@azure/msal-node": "^5.2.3",
|
||||
"@blocknote/server-util": "^0.51.4",
|
||||
"@clickhouse/client": "^1.18.1",
|
||||
"@e2b/code-interpreter": "^1.0.4",
|
||||
"@e2b/code-interpreter": "^2.6.0",
|
||||
"@envelop/core": "4.0.3",
|
||||
"@faker-js/faker": "9.8.0",
|
||||
"@file-type/pdf": "^0.2.0",
|
||||
|
||||
@@ -38,6 +38,7 @@ import { FieldMetadataModule } from 'src/engine/metadata-modules/field-metadata/
|
||||
import { ObjectMetadataModule } from 'src/engine/metadata-modules/object-metadata/object-metadata.module';
|
||||
import { RoleEntity } from 'src/engine/metadata-modules/role/role.entity';
|
||||
import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/provide-workspace-scoped-repository';
|
||||
import { CodeInterpreterSessionCleanupModule } from 'src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module';
|
||||
import { TrashCleanupModule } from 'src/engine/trash-cleanup/trash-cleanup.module';
|
||||
import { WorkspaceCacheStorageModule } from 'src/engine/workspace-cache-storage/workspace-cache-storage.module';
|
||||
import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
|
||||
@@ -75,6 +76,7 @@ import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/au
|
||||
WorkspaceCleanerModule,
|
||||
WorkspaceMigrationModule,
|
||||
TrashCleanupModule,
|
||||
CodeInterpreterSessionCleanupModule,
|
||||
PublicDomainModule,
|
||||
EventLogCleanupModule,
|
||||
EnterpriseModule,
|
||||
|
||||
+7
-1
@@ -64,7 +64,13 @@ export class CodeInterpreterDriverFactory extends DriverFactoryBase<CodeInterpre
|
||||
);
|
||||
}
|
||||
|
||||
return new E2BDriver({ apiKey, timeoutMs });
|
||||
return new E2BDriver({
|
||||
apiKey,
|
||||
timeoutMs,
|
||||
idleTimeoutMs: this.twentyConfigService.get(
|
||||
'CODE_INTERPRETER_IDLE_TIMEOUT_MS',
|
||||
),
|
||||
});
|
||||
}
|
||||
|
||||
default:
|
||||
|
||||
+69
-2
@@ -1,5 +1,7 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import { CodeInterpreterDriverFactory } from 'src/engine/core-modules/code-interpreter/code-interpreter-driver.factory';
|
||||
import { CodeInterpreterDriverType } from 'src/engine/core-modules/code-interpreter/code-interpreter.interface';
|
||||
import {
|
||||
@@ -13,6 +15,10 @@ import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twent
|
||||
|
||||
@Injectable()
|
||||
export class CodeInterpreterService implements CodeInterpreterDriver {
|
||||
// One active stream per thread (the chat resolver queues the rest), so
|
||||
// in-process chaining is enough to serialize a session — no distributed lock.
|
||||
private readonly sessionExecutionTails = new Map<string, Promise<void>>();
|
||||
|
||||
constructor(
|
||||
private readonly codeInterpreterDriverFactory: CodeInterpreterDriverFactory,
|
||||
private readonly twentyConfigService: TwentyConfigService,
|
||||
@@ -25,14 +31,75 @@ export class CodeInterpreterService implements CodeInterpreterDriver {
|
||||
);
|
||||
}
|
||||
|
||||
async releaseThreadSandbox(
|
||||
workspaceId: string,
|
||||
threadId: string,
|
||||
): Promise<void> {
|
||||
await this.codeInterpreterDriverFactory
|
||||
.getCurrentDriver()
|
||||
.releaseSession?.(`${workspaceId}:${threadId}`);
|
||||
}
|
||||
|
||||
async sweepExpiredSandboxes(): Promise<number> {
|
||||
const maxAgeMs = this.twentyConfigService.get(
|
||||
'CODE_INTERPRETER_SESSION_MAX_AGE_MS',
|
||||
);
|
||||
|
||||
return (
|
||||
(await this.codeInterpreterDriverFactory
|
||||
.getCurrentDriver()
|
||||
.sweepExpiredSessions?.(maxAgeMs)) ?? 0
|
||||
);
|
||||
}
|
||||
|
||||
execute(
|
||||
code: string,
|
||||
files?: InputFile[],
|
||||
context?: ExecutionContext,
|
||||
callbacks?: StreamCallbacks,
|
||||
): Promise<CodeExecutionResult> {
|
||||
const driver = this.codeInterpreterDriverFactory.getCurrentDriver();
|
||||
const sessionId = context?.sessionId;
|
||||
|
||||
return driver.execute(code, files, context, callbacks);
|
||||
if (isDefined(sessionId)) {
|
||||
return this.runSerializedPerSession(sessionId, () =>
|
||||
this.runOnDriver(code, files, context, callbacks),
|
||||
);
|
||||
}
|
||||
|
||||
return this.runOnDriver(code, files, context, callbacks);
|
||||
}
|
||||
|
||||
private runOnDriver(
|
||||
code: string,
|
||||
files?: InputFile[],
|
||||
context?: ExecutionContext,
|
||||
callbacks?: StreamCallbacks,
|
||||
): Promise<CodeExecutionResult> {
|
||||
return this.codeInterpreterDriverFactory
|
||||
.getCurrentDriver()
|
||||
.execute(code, files, context, callbacks);
|
||||
}
|
||||
|
||||
private async runSerializedPerSession<T>(
|
||||
sessionId: string,
|
||||
task: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const previous =
|
||||
this.sessionExecutionTails.get(sessionId) ?? Promise.resolve();
|
||||
const result = previous.then(task, task);
|
||||
const tail = result.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
|
||||
this.sessionExecutionTails.set(sessionId, tail);
|
||||
|
||||
try {
|
||||
return await result;
|
||||
} finally {
|
||||
if (this.sessionExecutionTails.get(sessionId) === tail) {
|
||||
this.sessionExecutionTails.delete(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+1
@@ -0,0 +1 @@
|
||||
export const CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN = '0 * * * *';
|
||||
+1
@@ -0,0 +1 @@
|
||||
export const SESSION_SANDBOX_METADATA_KEY = 'twentySessionId';
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
|
||||
import { CodeInterpreterSessionCleanupCronCommand } from 'src/engine/core-modules/code-interpreter/crons/commands/code-interpreter-session-cleanup.cron.command';
|
||||
import { CodeInterpreterSessionCleanupCronJob } from 'src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job';
|
||||
|
||||
@Module({
|
||||
providers: [
|
||||
CodeInterpreterSessionCleanupCronJob,
|
||||
CodeInterpreterSessionCleanupCronCommand,
|
||||
],
|
||||
exports: [CodeInterpreterSessionCleanupCronCommand],
|
||||
})
|
||||
export class CodeInterpreterSessionCleanupModule {}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
import { Command, CommandRunner } from 'nest-commander';
|
||||
|
||||
import { CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant';
|
||||
import { CodeInterpreterSessionCleanupCronJob } from 'src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job';
|
||||
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
|
||||
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
|
||||
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
|
||||
|
||||
@Command({
|
||||
name: 'cron:code-interpreter:session-cleanup',
|
||||
description:
|
||||
'Starts a cron job to reclaim expired (abandoned) code interpreter sandboxes',
|
||||
})
|
||||
export class CodeInterpreterSessionCleanupCronCommand extends CommandRunner {
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.cronQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
|
||||
async run(): Promise<void> {
|
||||
await this.messageQueueService.addCron<undefined>({
|
||||
jobName: CodeInterpreterSessionCleanupCronJob.name,
|
||||
data: undefined,
|
||||
options: {
|
||||
repeat: {
|
||||
pattern: CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN,
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { CodeInterpreterService } from 'src/engine/core-modules/code-interpreter/code-interpreter.service';
|
||||
import { CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant';
|
||||
import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
|
||||
import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
|
||||
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
|
||||
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
|
||||
|
||||
@Injectable()
|
||||
@Processor(MessageQueue.cronQueue)
|
||||
export class CodeInterpreterSessionCleanupCronJob {
|
||||
private readonly logger = new Logger(
|
||||
CodeInterpreterSessionCleanupCronJob.name,
|
||||
);
|
||||
|
||||
constructor(
|
||||
private readonly codeInterpreterService: CodeInterpreterService,
|
||||
) {}
|
||||
|
||||
@Process(CodeInterpreterSessionCleanupCronJob.name)
|
||||
@SentryCronMonitor(
|
||||
CodeInterpreterSessionCleanupCronJob.name,
|
||||
CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN,
|
||||
)
|
||||
async handle(): Promise<void> {
|
||||
const reclaimedCount =
|
||||
await this.codeInterpreterService.sweepExpiredSandboxes();
|
||||
|
||||
if (reclaimedCount > 0) {
|
||||
this.logger.log(
|
||||
`Reclaimed ${reclaimedCount} expired code interpreter sandbox(es)`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
+84
-21
@@ -2,8 +2,14 @@ import { promises as fs } from 'fs';
|
||||
import { join } from 'path';
|
||||
|
||||
import { Sandbox } from '@e2b/code-interpreter';
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import { DEFAULT_CODE_INTERPRETER_TIMEOUT_MS } from 'src/engine/core-modules/code-interpreter/code-interpreter.constants';
|
||||
import { getOrCreateSessionSandbox } from 'src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util';
|
||||
import { releaseSessionSandboxes } from 'src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util';
|
||||
import { sweepExpiredSessionSandboxes } from 'src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util';
|
||||
import { getMimeType } from 'src/engine/core-modules/code-interpreter/utils/get-mime-type.util';
|
||||
|
||||
import {
|
||||
@@ -18,12 +24,13 @@ import {
|
||||
export type E2BDriverOptions = {
|
||||
apiKey: string;
|
||||
timeoutMs?: number;
|
||||
idleTimeoutMs?: number;
|
||||
};
|
||||
|
||||
const SANDBOX_SCRIPTS_PATH = join(__dirname, '..', 'sandbox-scripts');
|
||||
|
||||
async function uploadDirectoryToSandbox(
|
||||
sbx: Sandbox,
|
||||
sandbox: Sandbox,
|
||||
localPath: string,
|
||||
remotePath: string,
|
||||
) {
|
||||
@@ -34,17 +41,19 @@ async function uploadDirectoryToSandbox(
|
||||
const remoteEntryPath = `${remotePath}/${entry.name}`;
|
||||
|
||||
if (entry.isDirectory()) {
|
||||
await uploadDirectoryToSandbox(sbx, localEntryPath, remoteEntryPath);
|
||||
await uploadDirectoryToSandbox(sandbox, localEntryPath, remoteEntryPath);
|
||||
} else {
|
||||
const content = await fs.readFile(localEntryPath);
|
||||
const arrayBuffer = new Uint8Array(content).buffer;
|
||||
|
||||
await sbx.files.write(remoteEntryPath, arrayBuffer);
|
||||
await sandbox.files.write(remoteEntryPath, arrayBuffer);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export class E2BDriver implements CodeInterpreterDriver {
|
||||
private readonly logger = new Logger(E2BDriver.name);
|
||||
|
||||
constructor(private options: E2BDriverOptions) {}
|
||||
|
||||
async execute(
|
||||
@@ -53,27 +62,52 @@ export class E2BDriver implements CodeInterpreterDriver {
|
||||
context?: ExecutionContext,
|
||||
callbacks?: StreamCallbacks,
|
||||
): Promise<CodeExecutionResult> {
|
||||
const sbx = await Sandbox.create({
|
||||
apiKey: this.options.apiKey,
|
||||
timeoutMs: this.options.timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS,
|
||||
});
|
||||
const { apiKey } = this.options;
|
||||
const sessionId = context?.sessionId;
|
||||
const idleTimeoutMs =
|
||||
this.options.idleTimeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS;
|
||||
const timeoutMs =
|
||||
this.options.timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS;
|
||||
|
||||
let sandbox: Sandbox;
|
||||
let isReused = false;
|
||||
let keepWarm = false;
|
||||
|
||||
if (isDefined(sessionId)) {
|
||||
keepWarm = true;
|
||||
({ sandbox, isReused } = await getOrCreateSessionSandbox({
|
||||
sandboxApi: Sandbox,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
}));
|
||||
} else {
|
||||
sandbox = await Sandbox.create({ apiKey, timeoutMs });
|
||||
}
|
||||
|
||||
try {
|
||||
// Upload pre-installed scripts to sandbox
|
||||
try {
|
||||
await uploadDirectoryToSandbox(
|
||||
sbx,
|
||||
SANDBOX_SCRIPTS_PATH,
|
||||
'/home/user/scripts',
|
||||
);
|
||||
} catch {
|
||||
// Scripts directory might not exist
|
||||
// A reused sandbox already has the scripts from its first run.
|
||||
if (!isReused) {
|
||||
try {
|
||||
await uploadDirectoryToSandbox(
|
||||
sandbox,
|
||||
SANDBOX_SCRIPTS_PATH,
|
||||
'/home/user/scripts',
|
||||
);
|
||||
} catch {
|
||||
// Scripts directory might not exist
|
||||
}
|
||||
}
|
||||
|
||||
if (isReused) {
|
||||
await this.resetOutputDirectory(sandbox);
|
||||
}
|
||||
|
||||
for (const file of files ?? []) {
|
||||
const arrayBuffer = new Uint8Array(file.content).buffer;
|
||||
|
||||
await sbx.files.write(`/home/user/${file.filename}`, arrayBuffer);
|
||||
await sandbox.files.write(`/home/user/${file.filename}`, arrayBuffer);
|
||||
}
|
||||
|
||||
const envSetup = context?.env
|
||||
@@ -93,7 +127,7 @@ export class E2BDriver implements CodeInterpreterDriver {
|
||||
const outputFiles: OutputFile[] = [];
|
||||
let chartCounter = 0;
|
||||
|
||||
const execution = await sbx.runCode(envSetup + code, {
|
||||
const execution = await sandbox.runCode(envSetup + code, {
|
||||
onStdout: (data) => callbacks?.onStdout?.(data.line),
|
||||
onStderr: (data) => callbacks?.onStderr?.(data.line),
|
||||
onResult: async (result) => {
|
||||
@@ -111,12 +145,13 @@ export class E2BDriver implements CodeInterpreterDriver {
|
||||
});
|
||||
|
||||
try {
|
||||
const outputDir = await sbx.files.list('/home/user/output');
|
||||
const outputDir = await sandbox.files.list('/home/user/output');
|
||||
|
||||
for (const file of outputDir) {
|
||||
if (file.type === 'file') {
|
||||
const content = await sbx.files.read(
|
||||
const content = await sandbox.files.read(
|
||||
`/home/user/output/${file.name}`,
|
||||
{ format: 'bytes' },
|
||||
);
|
||||
|
||||
const outputFile: OutputFile = {
|
||||
@@ -141,7 +176,35 @@ export class E2BDriver implements CodeInterpreterDriver {
|
||||
error: execution.error?.value,
|
||||
};
|
||||
} finally {
|
||||
await sbx.kill();
|
||||
if (!keepWarm) {
|
||||
await sandbox.kill();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async resetOutputDirectory(sandbox: Sandbox): Promise<void> {
|
||||
const outputDirectory = '/home/user/output';
|
||||
|
||||
try {
|
||||
if (await sandbox.files.exists(outputDirectory)) {
|
||||
await sandbox.files.remove(outputDirectory);
|
||||
}
|
||||
|
||||
await sandbox.files.makeDir(outputDirectory);
|
||||
} catch (error) {
|
||||
const reason = error instanceof Error ? error.message : String(error);
|
||||
|
||||
this.logger.warn(
|
||||
`Failed to reset reused sandbox output directory; results may include stale files: ${reason}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async releaseSession(sessionId: string): Promise<void> {
|
||||
await releaseSessionSandboxes(Sandbox, this.options.apiKey, sessionId);
|
||||
}
|
||||
|
||||
async sweepExpiredSessions(maxAgeMs: number): Promise<number> {
|
||||
return sweepExpiredSessionSandboxes(Sandbox, this.options.apiKey, maxAgeMs);
|
||||
}
|
||||
}
|
||||
|
||||
+3
@@ -20,6 +20,7 @@ export type CodeExecutionResult = {
|
||||
|
||||
export type ExecutionContext = {
|
||||
env?: Record<string, string>;
|
||||
sessionId?: string;
|
||||
};
|
||||
|
||||
export type StreamCallbacks = {
|
||||
@@ -35,4 +36,6 @@ export interface CodeInterpreterDriver {
|
||||
context?: ExecutionContext,
|
||||
callbacks?: StreamCallbacks,
|
||||
): Promise<CodeExecutionResult>;
|
||||
releaseSession?(sessionId: string): Promise<void>;
|
||||
sweepExpiredSessions?(maxAgeMs: number): Promise<number>;
|
||||
}
|
||||
|
||||
+250
@@ -0,0 +1,250 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
import { getOrCreateSessionSandbox } from 'src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util';
|
||||
|
||||
type SandboxApiMock = {
|
||||
list: jest.Mock;
|
||||
connect: jest.Mock;
|
||||
create: jest.Mock;
|
||||
kill: jest.Mock;
|
||||
};
|
||||
|
||||
type ListedSandbox = { sandboxId: string; metadata: Record<string, string> };
|
||||
|
||||
const apiKey = 'test-api-key';
|
||||
const sessionId = 'workspace-1:thread-1';
|
||||
const timeoutMs = 600_000;
|
||||
const idleTimeoutMs = 300_000;
|
||||
// The resolver keeps the sandbox alive for the larger of the two.
|
||||
const aliveTimeoutMs = 600_000;
|
||||
|
||||
// Sandbox.list() returns a paginator; emit the tagged sandboxes as a single page.
|
||||
const paginatorOf = (items: ListedSandbox[]) => {
|
||||
let fetched = false;
|
||||
|
||||
return {
|
||||
get hasNext() {
|
||||
return !fetched;
|
||||
},
|
||||
nextItems: async () => {
|
||||
fetched = true;
|
||||
|
||||
return items;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const buildSandboxApi = (
|
||||
overrides: Partial<SandboxApiMock> = {},
|
||||
): { api: typeof Sandbox; mock: SandboxApiMock } => {
|
||||
const mock: SandboxApiMock = {
|
||||
list: jest.fn(() => paginatorOf([])),
|
||||
connect: jest.fn(),
|
||||
create: jest.fn(),
|
||||
kill: jest.fn().mockResolvedValue(true),
|
||||
...overrides,
|
||||
};
|
||||
|
||||
return { api: mock as unknown as typeof Sandbox, mock };
|
||||
};
|
||||
|
||||
const buildFakeSandbox = (): Sandbox =>
|
||||
({
|
||||
setTimeout: jest.fn().mockResolvedValue(undefined),
|
||||
kill: jest.fn().mockResolvedValue(undefined),
|
||||
}) as unknown as Sandbox;
|
||||
|
||||
const listedSandbox = (
|
||||
sandboxId: string,
|
||||
tag: string = sessionId,
|
||||
): ListedSandbox => ({
|
||||
sandboxId,
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: tag },
|
||||
});
|
||||
|
||||
describe('getOrCreateSessionSandbox', () => {
|
||||
it('reuses and extends the existing sandbox for the session', async () => {
|
||||
const sandbox = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() => paginatorOf([listedSandbox('sbx-1')])),
|
||||
connect: jest.fn().mockResolvedValue(sandbox),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox, isReused: true });
|
||||
expect(mock.list).toHaveBeenCalledWith({
|
||||
apiKey,
|
||||
query: {
|
||||
state: ['running', 'paused'],
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
},
|
||||
});
|
||||
expect(mock.connect).toHaveBeenCalledWith('sbx-1', { apiKey });
|
||||
expect(sandbox.setTimeout).toHaveBeenCalledWith(aliveTimeoutMs);
|
||||
expect(mock.create).not.toHaveBeenCalled();
|
||||
expect(mock.kill).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates a new tagged auto-pausing sandbox when none exists for the session', async () => {
|
||||
const sandbox = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() => paginatorOf([])),
|
||||
create: jest.fn().mockResolvedValue(sandbox),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox, isReused: false });
|
||||
expect(mock.create).toHaveBeenCalledWith({
|
||||
apiKey,
|
||||
timeoutMs: aliveTimeoutMs,
|
||||
lifecycle: { onTimeout: 'pause', autoResume: true },
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
});
|
||||
expect(mock.connect).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('reaps duplicate sandboxes and reuses a single one', async () => {
|
||||
const keep = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() =>
|
||||
paginatorOf([
|
||||
listedSandbox('sbx-keep'),
|
||||
listedSandbox('sbx-dup-1'),
|
||||
listedSandbox('sbx-dup-2'),
|
||||
]),
|
||||
),
|
||||
connect: jest.fn().mockResolvedValue(keep),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox: keep, isReused: true });
|
||||
expect(mock.connect).toHaveBeenCalledTimes(1);
|
||||
expect(mock.connect).toHaveBeenCalledWith('sbx-keep', { apiKey });
|
||||
expect(mock.kill).toHaveBeenCalledWith('sbx-dup-1', { apiKey });
|
||||
expect(mock.kill).toHaveBeenCalledWith('sbx-dup-2', { apiKey });
|
||||
expect(mock.kill).not.toHaveBeenCalledWith('sbx-keep', { apiKey });
|
||||
expect(mock.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates a fresh sandbox when connecting to the existing one fails', async () => {
|
||||
const created = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() => paginatorOf([listedSandbox('sbx-dead')])),
|
||||
connect: jest.fn().mockRejectedValue(new Error('sandbox not found')),
|
||||
create: jest.fn().mockResolvedValue(created),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox: created, isReused: false });
|
||||
expect(mock.create).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('keeps the first connectable sandbox instead of cold-creating when an earlier one is dead', async () => {
|
||||
const live = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() =>
|
||||
paginatorOf([listedSandbox('sbx-dead'), listedSandbox('sbx-live')]),
|
||||
),
|
||||
connect: jest.fn((sandboxId: string) =>
|
||||
sandboxId === 'sbx-live'
|
||||
? Promise.resolve(live)
|
||||
: Promise.reject(new Error('sandbox not found')),
|
||||
),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox: live, isReused: true });
|
||||
expect(live.setTimeout).toHaveBeenCalledWith(aliveTimeoutMs);
|
||||
expect(mock.kill).toHaveBeenCalledWith('sbx-dead', { apiKey });
|
||||
expect(mock.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('never reuses a sandbox whose metadata does not exactly match the session', async () => {
|
||||
const created = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
// A sandbox from a different session/tenant that a loose server-side match
|
||||
// could surface — it must be ignored, never connected to or reaped.
|
||||
list: jest.fn(() =>
|
||||
paginatorOf([listedSandbox('sbx-foreign', 'workspace-2:thread-9')]),
|
||||
),
|
||||
create: jest.fn().mockResolvedValue(created),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox: created, isReused: false });
|
||||
expect(mock.connect).not.toHaveBeenCalled();
|
||||
expect(mock.kill).not.toHaveBeenCalled();
|
||||
expect(mock.create).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('kills a connected sandbox and creates fresh when refreshing its timeout fails', async () => {
|
||||
const stale = {
|
||||
setTimeout: jest
|
||||
.fn()
|
||||
.mockRejectedValue(new Error('timeout refresh failed')),
|
||||
kill: jest.fn().mockResolvedValue(undefined),
|
||||
} as unknown as Sandbox;
|
||||
const created = buildFakeSandbox();
|
||||
const { api, mock } = buildSandboxApi({
|
||||
list: jest.fn(() => paginatorOf([listedSandbox('sbx-stale')])),
|
||||
connect: jest.fn().mockResolvedValue(stale),
|
||||
create: jest.fn().mockResolvedValue(created),
|
||||
});
|
||||
|
||||
const result = await getOrCreateSessionSandbox({
|
||||
sandboxApi: api,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sandbox: created, isReused: false });
|
||||
expect(stale.setTimeout).toHaveBeenCalledWith(aliveTimeoutMs);
|
||||
expect(stale.kill).toHaveBeenCalledTimes(1);
|
||||
expect(mock.create).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
+75
@@ -0,0 +1,75 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
import { releaseSessionSandboxes } from 'src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util';
|
||||
|
||||
const apiKey = 'test-api-key';
|
||||
const sessionId = 'workspace-1:thread-1';
|
||||
|
||||
type ListedSandbox = {
|
||||
sandboxId: string;
|
||||
metadata: Record<string, string>;
|
||||
startedAt: string;
|
||||
};
|
||||
|
||||
const paginatorOf = (items: ListedSandbox[]) => {
|
||||
let fetched = false;
|
||||
|
||||
return {
|
||||
get hasNext() {
|
||||
return !fetched;
|
||||
},
|
||||
nextItems: async () => {
|
||||
fetched = true;
|
||||
|
||||
return items;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const buildApi = (items: ListedSandbox[]) => {
|
||||
const kill = jest.fn().mockResolvedValue(true);
|
||||
const list = jest.fn(() => paginatorOf(items));
|
||||
|
||||
return { api: { list, kill } as unknown as typeof Sandbox, kill, list };
|
||||
};
|
||||
|
||||
const sandboxInfo = (
|
||||
sandboxId: string,
|
||||
{ tag = sessionId }: { tag?: string | null } = {},
|
||||
): ListedSandbox => ({
|
||||
sandboxId,
|
||||
metadata: tag === null ? {} : { [SESSION_SANDBOX_METADATA_KEY]: tag },
|
||||
startedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
describe('releaseSessionSandboxes', () => {
|
||||
it('kills every sandbox tagged for the session', async () => {
|
||||
const { api, kill, list } = buildApi([
|
||||
sandboxInfo('sbx-1'),
|
||||
sandboxInfo('sbx-2'),
|
||||
]);
|
||||
|
||||
await releaseSessionSandboxes(api, apiKey, sessionId);
|
||||
|
||||
expect(list).toHaveBeenCalledWith({
|
||||
apiKey,
|
||||
query: {
|
||||
state: ['running', 'paused'],
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
},
|
||||
});
|
||||
expect(kill).toHaveBeenCalledWith('sbx-1', { apiKey });
|
||||
expect(kill).toHaveBeenCalledWith('sbx-2', { apiKey });
|
||||
});
|
||||
|
||||
it('ignores a sandbox whose tag does not match the session', async () => {
|
||||
const { api, kill } = buildApi([
|
||||
sandboxInfo('sbx-other', { tag: 'workspace-2:thread-9' }),
|
||||
]);
|
||||
|
||||
await releaseSessionSandboxes(api, apiKey, sessionId);
|
||||
|
||||
expect(kill).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
+73
@@ -0,0 +1,73 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
import { sweepExpiredSessionSandboxes } from 'src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util';
|
||||
|
||||
const apiKey = 'test-api-key';
|
||||
const sessionId = 'workspace-1:thread-1';
|
||||
const dayMs = 24 * 60 * 60 * 1000;
|
||||
|
||||
type ListedSandbox = {
|
||||
sandboxId: string;
|
||||
metadata: Record<string, string>;
|
||||
startedAt: string;
|
||||
};
|
||||
|
||||
const paginatorOf = (items: ListedSandbox[]) => {
|
||||
let fetched = false;
|
||||
|
||||
return {
|
||||
get hasNext() {
|
||||
return !fetched;
|
||||
},
|
||||
nextItems: async () => {
|
||||
fetched = true;
|
||||
|
||||
return items;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const buildApi = (items: ListedSandbox[]) => {
|
||||
const kill = jest.fn().mockResolvedValue(true);
|
||||
const list = jest.fn(() => paginatorOf(items));
|
||||
|
||||
return { api: { list, kill } as unknown as typeof Sandbox, kill, list };
|
||||
};
|
||||
|
||||
const sandboxInfo = (
|
||||
sandboxId: string,
|
||||
{ tag = sessionId, ageMs = 0 }: { tag?: string | null; ageMs?: number } = {},
|
||||
): ListedSandbox => ({
|
||||
sandboxId,
|
||||
metadata: tag === null ? {} : { [SESSION_SANDBOX_METADATA_KEY]: tag },
|
||||
startedAt: new Date(Date.now() - ageMs).toISOString(),
|
||||
});
|
||||
|
||||
describe('sweepExpiredSessionSandboxes', () => {
|
||||
it('kills only session sandboxes older than the max age', async () => {
|
||||
const { api, kill } = buildApi([
|
||||
sandboxInfo('sbx-old', { ageMs: 2 * dayMs }),
|
||||
sandboxInfo('sbx-fresh', { ageMs: 60 * 1000 }),
|
||||
sandboxInfo('sbx-untagged', { tag: null, ageMs: 5 * dayMs }),
|
||||
]);
|
||||
|
||||
const killed = await sweepExpiredSessionSandboxes(api, apiKey, dayMs);
|
||||
|
||||
expect(killed).toBe(1);
|
||||
expect(kill).toHaveBeenCalledWith('sbx-old', { apiKey });
|
||||
expect(kill).not.toHaveBeenCalledWith('sbx-fresh', { apiKey });
|
||||
expect(kill).not.toHaveBeenCalledWith('sbx-untagged', { apiKey });
|
||||
});
|
||||
|
||||
it('lists both running and paused sandboxes', async () => {
|
||||
const { api, list } = buildApi([]);
|
||||
|
||||
await sweepExpiredSessionSandboxes(api, apiKey, 1000);
|
||||
|
||||
expect(list).toHaveBeenCalledWith({
|
||||
apiKey,
|
||||
query: { state: ['running', 'paused'] },
|
||||
});
|
||||
});
|
||||
});
|
||||
+144
@@ -0,0 +1,144 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
|
||||
type SandboxApi = typeof Sandbox;
|
||||
|
||||
type GetOrCreateSessionSandboxArgs = {
|
||||
sandboxApi: SandboxApi;
|
||||
apiKey: string;
|
||||
sessionId: string;
|
||||
timeoutMs: number;
|
||||
idleTimeoutMs: number;
|
||||
};
|
||||
|
||||
const listSandboxesForSession = async (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
sessionId: string,
|
||||
) => {
|
||||
const paginator = sandboxApi.list({
|
||||
apiKey,
|
||||
query: {
|
||||
state: ['running', 'paused'],
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
},
|
||||
});
|
||||
|
||||
const sandboxes: Awaited<ReturnType<typeof paginator.nextItems>> = [];
|
||||
|
||||
while (paginator.hasNext) {
|
||||
sandboxes.push(...(await paginator.nextItems()));
|
||||
}
|
||||
|
||||
// Re-check the tag client-side so a loose server-side match can never reuse
|
||||
// another tenant's sandbox.
|
||||
return sandboxes.filter(
|
||||
(sandbox) => sandbox.metadata?.[SESSION_SANDBOX_METADATA_KEY] === sessionId,
|
||||
);
|
||||
};
|
||||
|
||||
const connectAndKeepAlive = async (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
sandboxId: string,
|
||||
timeoutMs: number,
|
||||
): Promise<Sandbox | undefined> => {
|
||||
const sandbox = await sandboxApi
|
||||
.connect(sandboxId, { apiKey })
|
||||
.catch(() => undefined);
|
||||
|
||||
if (!isDefined(sandbox)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
try {
|
||||
await sandbox.setTimeout(timeoutMs);
|
||||
|
||||
return sandbox;
|
||||
} catch {
|
||||
// Couldn't refresh the timeout — kill it instead of leaking a running sandbox.
|
||||
await sandbox.kill().catch(() => undefined);
|
||||
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
const killSandboxById = (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
sandboxId: string,
|
||||
) => sandboxApi.kill(sandboxId, { apiKey }).catch(() => undefined);
|
||||
|
||||
const createSessionSandbox = (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
sessionId: string,
|
||||
timeoutMs: number,
|
||||
) =>
|
||||
sandboxApi.create({
|
||||
apiKey,
|
||||
timeoutMs,
|
||||
lifecycle: { onTimeout: 'pause', autoResume: true },
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
});
|
||||
|
||||
export const getOrCreateSessionSandbox = async ({
|
||||
sandboxApi,
|
||||
apiKey,
|
||||
sessionId,
|
||||
timeoutMs,
|
||||
idleTimeoutMs,
|
||||
}: GetOrCreateSessionSandboxArgs): Promise<{
|
||||
sandbox: Sandbox;
|
||||
isReused: boolean;
|
||||
}> => {
|
||||
// The sandbox must outlive a single execution and the idle window before it
|
||||
// auto-pauses, so take the larger of the two.
|
||||
const aliveTimeoutMs = Math.max(timeoutMs, idleTimeoutMs);
|
||||
|
||||
const sessionSandboxes = await listSandboxesForSession(
|
||||
sandboxApi,
|
||||
apiKey,
|
||||
sessionId,
|
||||
);
|
||||
|
||||
let reusedSandbox: Sandbox | undefined;
|
||||
let reusedSandboxId: string | undefined;
|
||||
|
||||
for (const { sandboxId } of sessionSandboxes) {
|
||||
const sandbox = await connectAndKeepAlive(
|
||||
sandboxApi,
|
||||
apiKey,
|
||||
sandboxId,
|
||||
aliveTimeoutMs,
|
||||
);
|
||||
|
||||
if (isDefined(sandbox)) {
|
||||
reusedSandbox = sandbox;
|
||||
reusedSandboxId = sandboxId;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (isDefined(reusedSandbox)) {
|
||||
await Promise.all(
|
||||
sessionSandboxes
|
||||
.filter(({ sandboxId }) => sandboxId !== reusedSandboxId)
|
||||
.map(({ sandboxId }) => killSandboxById(sandboxApi, apiKey, sandboxId)),
|
||||
);
|
||||
|
||||
return { sandbox: reusedSandbox, isReused: true };
|
||||
}
|
||||
|
||||
const sandbox = await createSessionSandbox(
|
||||
sandboxApi,
|
||||
apiKey,
|
||||
sessionId,
|
||||
aliveTimeoutMs,
|
||||
);
|
||||
|
||||
return { sandbox, isReused: false };
|
||||
};
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
|
||||
type SandboxApi = typeof Sandbox;
|
||||
|
||||
// Kills every sandbox tagged for a session — used when a conversation is deleted.
|
||||
export const releaseSessionSandboxes = async (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
sessionId: string,
|
||||
): Promise<void> => {
|
||||
const paginator = sandboxApi.list({
|
||||
apiKey,
|
||||
query: {
|
||||
state: ['running', 'paused'],
|
||||
metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId },
|
||||
},
|
||||
});
|
||||
|
||||
const sandboxes: Awaited<ReturnType<typeof paginator.nextItems>> = [];
|
||||
|
||||
while (paginator.hasNext) {
|
||||
sandboxes.push(...(await paginator.nextItems()));
|
||||
}
|
||||
|
||||
await Promise.all(
|
||||
sandboxes
|
||||
.filter(
|
||||
(sandbox) =>
|
||||
sandbox.metadata?.[SESSION_SANDBOX_METADATA_KEY] === sessionId,
|
||||
)
|
||||
.map((sandbox) =>
|
||||
sandboxApi.kill(sandbox.sandboxId, { apiKey }).catch(() => undefined),
|
||||
),
|
||||
);
|
||||
};
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
import { type Sandbox } from '@e2b/code-interpreter';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import { SESSION_SANDBOX_METADATA_KEY } from 'src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant';
|
||||
|
||||
type SandboxApi = typeof Sandbox;
|
||||
|
||||
// E2B retains paused sandboxes indefinitely, so abandoned sessions need an age bound.
|
||||
export const sweepExpiredSessionSandboxes = async (
|
||||
sandboxApi: SandboxApi,
|
||||
apiKey: string,
|
||||
maxAgeMs: number,
|
||||
): Promise<number> => {
|
||||
const paginator = sandboxApi.list({
|
||||
apiKey,
|
||||
query: { state: ['running', 'paused'] },
|
||||
});
|
||||
|
||||
const sandboxes: Awaited<ReturnType<typeof paginator.nextItems>> = [];
|
||||
|
||||
while (paginator.hasNext) {
|
||||
sandboxes.push(...(await paginator.nextItems()));
|
||||
}
|
||||
|
||||
const expiredBefore = Date.now() - maxAgeMs;
|
||||
|
||||
const expiredSandboxes = sandboxes.filter(
|
||||
(sandbox) =>
|
||||
isDefined(sandbox.metadata?.[SESSION_SANDBOX_METADATA_KEY]) &&
|
||||
// startedAt is typed as Date but comes back as an ISO string at runtime.
|
||||
new Date(sandbox.startedAt).getTime() < expiredBefore,
|
||||
);
|
||||
|
||||
await Promise.all(
|
||||
expiredSandboxes.map((sandbox) =>
|
||||
sandboxApi.kill(sandbox.sandboxId, { apiKey }).catch(() => undefined),
|
||||
),
|
||||
);
|
||||
|
||||
return expiredSandboxes.length;
|
||||
};
|
||||
@@ -69,6 +69,7 @@ import { PageLayoutModule } from 'src/engine/metadata-modules/page-layout/page-l
|
||||
import { RoleModule } from 'src/engine/metadata-modules/role/role.module';
|
||||
import { RowLevelPermissionModule } from 'src/engine/metadata-modules/row-level-permission-predicate/row-level-permission.module';
|
||||
import { SubscriptionsModule } from 'src/engine/subscriptions/subscriptions.module';
|
||||
import { CodeInterpreterSessionCleanupModule } from 'src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module';
|
||||
import { TrashCleanupModule } from 'src/engine/trash-cleanup/trash-cleanup.module';
|
||||
import { WorkspaceEventEmitterModule } from 'src/engine/workspace-event-emitter/workspace-event-emitter.module';
|
||||
import { ChannelSyncModule } from 'src/modules/connected-account/channel-sync/channel-sync.module';
|
||||
@@ -156,6 +157,7 @@ import { FileModule } from './file/file.module';
|
||||
PageLayoutModule,
|
||||
ImpersonationModule,
|
||||
TrashCleanupModule,
|
||||
CodeInterpreterSessionCleanupModule,
|
||||
DashboardModule,
|
||||
EventLogsViewerModule,
|
||||
PreInstalledAppsModule,
|
||||
|
||||
+1
@@ -12,5 +12,6 @@ export type ToolProviderContext = {
|
||||
actorContext?: ActorMetadata;
|
||||
userId?: string;
|
||||
userWorkspaceId?: string;
|
||||
threadId?: string;
|
||||
onCodeExecutionUpdate?: CodeExecutionStreamEmitter;
|
||||
};
|
||||
|
||||
+1
@@ -144,6 +144,7 @@ export class ActionToolProvider implements ToolProvider {
|
||||
workspaceId: context.workspaceId,
|
||||
userId: context.userId,
|
||||
userWorkspaceId: context.userWorkspaceId,
|
||||
threadId: context.threadId,
|
||||
onCodeExecutionUpdate: context.onCodeExecutionUpdate,
|
||||
});
|
||||
}
|
||||
|
||||
+1
@@ -389,6 +389,7 @@ export class ToolRegistryService {
|
||||
authContext: context.authContext,
|
||||
userId: context.userId,
|
||||
userWorkspaceId: context.userWorkspaceId,
|
||||
threadId: context.threadId,
|
||||
onCodeExecutionUpdate: context.onCodeExecutionUpdate,
|
||||
};
|
||||
}
|
||||
|
||||
+1
@@ -11,5 +11,6 @@ export type ToolContext = {
|
||||
actorContext?: ActorMetadata;
|
||||
userId?: string;
|
||||
userWorkspaceId?: string;
|
||||
threadId?: string;
|
||||
onCodeExecutionUpdate?: CodeExecutionStreamEmitter;
|
||||
};
|
||||
|
||||
+9
-3
@@ -42,7 +42,7 @@ export class CodeInterpreterTool implements Tool {
|
||||
private readonly logger = new Logger(CodeInterpreterTool.name);
|
||||
|
||||
description =
|
||||
'Execute Python code in a sandboxed environment for data analysis, CSV processing, calculations, and chart generation. Returns stdout, stderr, and generated files. Input files are available at /home/user/{filename}. Save output files (charts, reports) to /home/user/output/ using plt.savefig() for matplotlib charts.';
|
||||
'Execute Python code in a sandboxed environment for data analysis, CSV processing, calculations, and chart generation. Returns stdout, stderr, and generated files. Input files are available at /home/user/{filename}. Save output files (charts, reports) to /home/user/output/ using plt.savefig() for matplotlib charts. The sandbox is reused across calls within the same conversation: variables, imports, and any files you write outside /home/user/output persist between calls, so build on earlier results instead of recomputing or re-downloading them. /home/user/output is cleared at the start of every call, so write the files you want returned in the same call that produces them.';
|
||||
|
||||
inputSchema = CodeInterpreterInputZodSchema;
|
||||
|
||||
@@ -82,8 +82,13 @@ export class CodeInterpreterTool implements Tool {
|
||||
parameters: ToolInput,
|
||||
context: ToolExecutionContext,
|
||||
): Promise<ToolOutput> {
|
||||
const { workspaceId, userId, userWorkspaceId, onCodeExecutionUpdate } =
|
||||
context;
|
||||
const {
|
||||
workspaceId,
|
||||
userId,
|
||||
userWorkspaceId,
|
||||
threadId,
|
||||
onCodeExecutionUpdate,
|
||||
} = context;
|
||||
const { code, files } = parameters as CodeInterpreterInput;
|
||||
const executionId = v4();
|
||||
const startTime = Date.now();
|
||||
@@ -128,6 +133,7 @@ export class CodeInterpreterTool implements Tool {
|
||||
TWENTY_SERVER_URL: serverUrl,
|
||||
TWENTY_API_TOKEN: sessionToken,
|
||||
},
|
||||
sessionId: threadId ? `${workspaceId}:${threadId}` : undefined,
|
||||
},
|
||||
{
|
||||
onStdout: (line) => {
|
||||
|
||||
+1
@@ -4,5 +4,6 @@ export type ToolExecutionContext = {
|
||||
workspaceId: string;
|
||||
userId?: string;
|
||||
userWorkspaceId?: string;
|
||||
threadId?: string;
|
||||
onCodeExecutionUpdate?: CodeExecutionStreamEmitter;
|
||||
};
|
||||
|
||||
@@ -701,6 +701,26 @@ export class ConfigVariables {
|
||||
@CastToPositiveNumber()
|
||||
CODE_INTERPRETER_TIMEOUT_MS = 300_000;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.CODE_INTERPRETER_CONFIG,
|
||||
description:
|
||||
'Idle lifetime in milliseconds for a reused code interpreter sandbox; refreshed on each execution and reclaimed after this period of inactivity (default: 300000)',
|
||||
type: ConfigVariableType.NUMBER,
|
||||
})
|
||||
@IsOptional()
|
||||
@CastToPositiveNumber()
|
||||
CODE_INTERPRETER_IDLE_TIMEOUT_MS = 300_000;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.CODE_INTERPRETER_CONFIG,
|
||||
description:
|
||||
'Maximum age in milliseconds a reused code interpreter sandbox is kept before garbage collection reclaims it; abandoned conversations are reclaimed after this period (default: 86400000)',
|
||||
type: ConfigVariableType.NUMBER,
|
||||
})
|
||||
@IsOptional()
|
||||
@CastToPositiveNumber()
|
||||
CODE_INTERPRETER_SESSION_MAX_AGE_MS = 86_400_000;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.ANALYTICS_CONFIG,
|
||||
description: 'Enable or disable analytics for telemetry',
|
||||
|
||||
+1
@@ -230,6 +230,7 @@ export class StreamAgentChatJob {
|
||||
await this.chatExecutionService.streamChat({
|
||||
workspace,
|
||||
userWorkspaceId: data.userWorkspaceId,
|
||||
threadId: data.threadId,
|
||||
messages: data.messages,
|
||||
browsingContext: data.browsingContext,
|
||||
modelId: data.modelId,
|
||||
|
||||
+21
@@ -6,6 +6,7 @@ import type { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialE
|
||||
|
||||
import type { UIDataTypes, UIMessagePart, UITools } from 'ai';
|
||||
|
||||
import { CodeInterpreterService } from 'src/engine/core-modules/code-interpreter/code-interpreter.service';
|
||||
import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
|
||||
import { AgentMessagePartEntity } from 'src/engine/metadata-modules/ai/ai-agent-execution/entities/agent-message-part.entity';
|
||||
import {
|
||||
@@ -66,6 +67,7 @@ export class AgentChatService {
|
||||
private readonly fileRepository: WorkspaceScopedRepository<FileEntity>,
|
||||
private readonly titleGenerationService: AgentTitleGenerationService,
|
||||
private readonly workspaceEventBroadcaster: WorkspaceEventBroadcaster,
|
||||
private readonly codeInterpreterService: CodeInterpreterService,
|
||||
) {}
|
||||
|
||||
async createThread({
|
||||
@@ -512,6 +514,8 @@ export class AgentChatService {
|
||||
|
||||
await this.broadcastThreadUpdated(thread, ['deletedAt'], userWorkspaceId);
|
||||
|
||||
this.releaseThreadSandboxBestEffort(workspaceId, threadId);
|
||||
|
||||
return thread;
|
||||
}
|
||||
|
||||
@@ -598,6 +602,23 @@ export class AgentChatService {
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
this.releaseThreadSandboxBestEffort(workspaceId, threadId);
|
||||
}
|
||||
|
||||
private releaseThreadSandboxBestEffort(
|
||||
workspaceId: string,
|
||||
threadId: string,
|
||||
): void {
|
||||
void this.codeInterpreterService
|
||||
.releaseThreadSandbox(workspaceId, threadId)
|
||||
.catch((error) =>
|
||||
this.logger.warn(
|
||||
`Failed to release code interpreter sandbox for thread ${threadId}: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async notifyThreadActivityUpdated({
|
||||
|
||||
+3
@@ -68,6 +68,7 @@ import { SkillService } from 'src/engine/metadata-modules/skill/skill.service';
|
||||
export type ChatExecutionOptions = {
|
||||
workspace: WorkspaceEntity;
|
||||
userWorkspaceId: string;
|
||||
threadId?: string;
|
||||
messages: UIMessage<unknown, UIDataTypes, UITools>[];
|
||||
browsingContext: BrowsingContextType | null;
|
||||
onCodeExecutionUpdate?: CodeExecutionStreamEmitter;
|
||||
@@ -105,6 +106,7 @@ export class ChatExecutionService {
|
||||
async streamChat({
|
||||
workspace,
|
||||
userWorkspaceId,
|
||||
threadId,
|
||||
messages,
|
||||
browsingContext,
|
||||
onCodeExecutionUpdate,
|
||||
@@ -125,6 +127,7 @@ export class ChatExecutionService {
|
||||
actorContext,
|
||||
userId,
|
||||
userWorkspaceId,
|
||||
threadId,
|
||||
onCodeExecutionUpdate,
|
||||
};
|
||||
|
||||
|
||||
@@ -5426,12 +5426,12 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"@e2b/code-interpreter@npm:^1.0.4":
|
||||
version: 1.5.1
|
||||
resolution: "@e2b/code-interpreter@npm:1.5.1"
|
||||
"@e2b/code-interpreter@npm:^2.6.0":
|
||||
version: 2.6.0
|
||||
resolution: "@e2b/code-interpreter@npm:2.6.0"
|
||||
dependencies:
|
||||
e2b: "npm:^1.4.0"
|
||||
checksum: 10c0/edb0aa6643749e91c2446aa5e124304066703cce2c41adb8647ad5a124340332f6f547758bf196c9a5400a171f1a295a13eb0946d4462d76b06106aed65913c2
|
||||
e2b: "npm:^2.28.0"
|
||||
checksum: 10c0/b772ae32285003669afe0cf07dadc5952f7c60aa120194c1b5876af2e034b195a51b148aa06c1b2b25f517d1ae9de9b5dae72bf386bd8f20fcd21abe52318e31
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
@@ -31462,6 +31462,16 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"dockerfile-ast@npm:^0.7.1":
|
||||
version: 0.7.1
|
||||
resolution: "dockerfile-ast@npm:0.7.1"
|
||||
dependencies:
|
||||
vscode-languageserver-textdocument: "npm:^1.0.8"
|
||||
vscode-languageserver-types: "npm:^3.17.3"
|
||||
checksum: 10c0/7afbef27bb7d59d677d3d0537016093207633ada351c83edcae225dda832de64df8271ae3df3a9b5199dfed37081849453b0984c71190a9d49675dbe386412df
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"doctrine@npm:^3.0.0":
|
||||
version: 3.0.0
|
||||
resolution: "doctrine@npm:3.0.0"
|
||||
@@ -31763,17 +31773,22 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"e2b@npm:^1.4.0":
|
||||
version: 1.13.2
|
||||
resolution: "e2b@npm:1.13.2"
|
||||
"e2b@npm:^2.28.0":
|
||||
version: 2.29.1
|
||||
resolution: "e2b@npm:2.29.1"
|
||||
dependencies:
|
||||
"@bufbuild/protobuf": "npm:^2.6.2"
|
||||
"@connectrpc/connect": "npm:2.0.0-rc.3"
|
||||
"@connectrpc/connect-web": "npm:2.0.0-rc.3"
|
||||
chalk: "npm:^5.3.0"
|
||||
compare-versions: "npm:^6.1.0"
|
||||
openapi-fetch: "npm:^0.9.7"
|
||||
dockerfile-ast: "npm:^0.7.1"
|
||||
glob: "npm:^11.1.0"
|
||||
openapi-fetch: "npm:^0.14.1"
|
||||
platform: "npm:^1.3.6"
|
||||
checksum: 10c0/5c6f6b15cf380331dae5561017bf4653aed38e7e7102542984997b2888390732c525a399aac06da5ad26bfe08cf99073b7174825c8f6d54a1b78a327456fccd6
|
||||
tar: "npm:^7.5.11"
|
||||
undici: "npm:^7.25.0"
|
||||
checksum: 10c0/c75522c9497986eafe7b748e1bc8038b9fb2bdf67ab2cbdc741ee7b67b5543514ba735a5e66f4d7d058e4d2150c16070e2f25c7915356c00e1caa2e238bf244d
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
@@ -44194,6 +44209,15 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"openapi-fetch@npm:^0.14.1":
|
||||
version: 0.14.1
|
||||
resolution: "openapi-fetch@npm:0.14.1"
|
||||
dependencies:
|
||||
openapi-typescript-helpers: "npm:^0.0.15"
|
||||
checksum: 10c0/e1b5375f0461969ff08e951b6308a115a582cca84ed3e0ecc2e41919d44a86038144a7a0b2e217ddb1bb7160a6779f459a2ca0224ff7634a9e89f19afe2a83db
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"openapi-fetch@npm:^0.17.0":
|
||||
version: 0.17.0
|
||||
resolution: "openapi-fetch@npm:0.17.0"
|
||||
@@ -44203,15 +44227,6 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"openapi-fetch@npm:^0.9.7":
|
||||
version: 0.9.8
|
||||
resolution: "openapi-fetch@npm:0.9.8"
|
||||
dependencies:
|
||||
openapi-typescript-helpers: "npm:^0.0.8"
|
||||
checksum: 10c0/a96590c3df4f3e6800003f1128bafd50a01113073f00ea32a6e21e8c1ec38f3dc88141448569e52e791c58088fdbee61ea4fc44b6856cc24da2768fc43fddb82
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"openapi-types@npm:12.1.3":
|
||||
version: 12.1.3
|
||||
resolution: "openapi-types@npm:12.1.3"
|
||||
@@ -44219,10 +44234,10 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"openapi-typescript-helpers@npm:^0.0.8":
|
||||
version: 0.0.8
|
||||
resolution: "openapi-typescript-helpers@npm:0.0.8"
|
||||
checksum: 10c0/b31a326c67d65f947732e32126f79976ecc67597f911f7ed43ea9ea23b1e2df7a012026327b6e39df5897a70781f9c57f9fcdd86f8a05c0f67ab2b23ee5d708a
|
||||
"openapi-typescript-helpers@npm:^0.0.15":
|
||||
version: 0.0.15
|
||||
resolution: "openapi-typescript-helpers@npm:0.0.15"
|
||||
checksum: 10c0/5eb68d487b787e3e31266470b1a310726549dd45a1079655ab18066ab291b0b3c343fdf629991013706a2329b86964f8798d56ef0272b94b931fe6c19abd7a88
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
@@ -52290,7 +52305,7 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"tar@npm:^7.5.16, tar@npm:^7.5.4":
|
||||
"tar@npm:^7.5.11, tar@npm:^7.5.16, tar@npm:^7.5.4":
|
||||
version: 7.5.16
|
||||
resolution: "tar@npm:7.5.16"
|
||||
dependencies:
|
||||
@@ -53659,7 +53674,7 @@ __metadata:
|
||||
"@azure/msal-node": "npm:^5.2.3"
|
||||
"@blocknote/server-util": "npm:^0.51.4"
|
||||
"@clickhouse/client": "npm:^1.18.1"
|
||||
"@e2b/code-interpreter": "npm:^1.0.4"
|
||||
"@e2b/code-interpreter": "npm:^2.6.0"
|
||||
"@envelop/core": "npm:4.0.3"
|
||||
"@faker-js/faker": "npm:9.8.0"
|
||||
"@file-type/pdf": "npm:^0.2.0"
|
||||
@@ -55906,6 +55921,13 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"vscode-languageserver-textdocument@npm:^1.0.8":
|
||||
version: 1.0.13
|
||||
resolution: "vscode-languageserver-textdocument@npm:1.0.13"
|
||||
checksum: 10c0/1de174f1de3bfa9e1660a4b3c1b1c711497196d761e33f4d00785fdbfc556ae5a1f440db04771cbc0f6bd66c2a6f303062bc17d8dc46e76820e39fa2d5626564
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"vscode-languageserver-types@npm:^3.17.1":
|
||||
version: 3.17.5
|
||||
resolution: "vscode-languageserver-types@npm:3.17.5"
|
||||
@@ -55913,6 +55935,13 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"vscode-languageserver-types@npm:^3.17.3":
|
||||
version: 3.18.0
|
||||
resolution: "vscode-languageserver-types@npm:3.18.0"
|
||||
checksum: 10c0/d8c51cd286cc55b4c1f333e87fa3cefc66d2aae8c65f6209e7ecf88032d3326308b333d62b05243c194be8a3304760445b87630b298116c6c612607f011b4de0
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"vscode-uri@npm:^3.0.8, vscode-uri@npm:^3.1.0":
|
||||
version: 3.1.0
|
||||
resolution: "vscode-uri@npm:3.1.0"
|
||||
|
||||
Reference in New Issue
Block a user