From bb6da7b7d1d7d3b785808df69481d3f611a01041 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?F=C3=A9lix=20Malfait?= Date: Wed, 17 Jun 2026 10:12:24 +0200 Subject: [PATCH] feat(code-interpreter): reuse a warm sandbox per conversation (E2B) (#21664) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- .yarnrc.yml | 5 - packages/twenty-server/package.json | 2 +- .../commands/database-command.module.ts | 2 + .../code-interpreter-driver.factory.ts | 8 +- .../code-interpreter.service.ts | 71 ++++- ...r-session-cleanup-cron-pattern.constant.ts | 1 + .../session-sandbox-metadata-key.constant.ts | 1 + ...code-interpreter-session-cleanup.module.ts | 13 + ...nterpreter-session-cleanup.cron.command.ts | 33 +++ ...de-interpreter-session-cleanup.cron.job.ts | 36 +++ .../code-interpreter/drivers/e2b.driver.ts | 105 ++++++-- .../code-interpreter-driver.interface.ts | 3 + ...get-or-create-session-sandbox.util.spec.ts | 250 ++++++++++++++++++ .../release-session-sandboxes.util.spec.ts | 75 ++++++ ...eep-expired-session-sandboxes.util.spec.ts | 73 +++++ .../get-or-create-session-sandbox.util.ts | 144 ++++++++++ .../utils/release-session-sandboxes.util.ts | 37 +++ .../sweep-expired-session-sandboxes.util.ts | 42 +++ .../engine/core-modules/core-engine.module.ts | 2 + .../interfaces/tool-provider-context.type.ts | 1 + .../providers/action-tool.provider.ts | 1 + .../services/tool-registry.service.ts | 1 + .../tool-provider/types/tool-context.type.ts | 1 + .../code-interpreter-tool.ts | 12 +- .../tool/types/tool-execution-context.type.ts | 1 + .../twenty-config/config-variables.ts | 20 ++ .../ai/ai-chat/jobs/stream-agent-chat.job.ts | 1 + .../ai/ai-chat/services/agent-chat.service.ts | 21 ++ .../services/chat-execution.service.ts | 3 + yarn.lock | 79 ++++-- 30 files changed, 986 insertions(+), 58 deletions(-) create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/crons/commands/code-interpreter-session-cleanup.cron.command.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/get-or-create-session-sandbox.util.spec.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/release-session-sandboxes.util.spec.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/sweep-expired-session-sandboxes.util.spec.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util.ts create mode 100644 packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util.ts diff --git a/.yarnrc.yml b/.yarnrc.yml index 7b39d64eb8..1696d1178d 100644 --- a/.yarnrc.yml +++ b/.yarnrc.yml @@ -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" diff --git a/packages/twenty-server/package.json b/packages/twenty-server/package.json index 77a64fd212..723a3308de 100644 --- a/packages/twenty-server/package.json +++ b/packages/twenty-server/package.json @@ -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", diff --git a/packages/twenty-server/src/database/commands/database-command.module.ts b/packages/twenty-server/src/database/commands/database-command.module.ts index 25b8649fcb..dcc12ab671 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -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, diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/code-interpreter-driver.factory.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/code-interpreter-driver.factory.ts index 06452dcbe3..295986a9ff 100644 --- a/packages/twenty-server/src/engine/core-modules/code-interpreter/code-interpreter-driver.factory.ts +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/code-interpreter-driver.factory.ts @@ -64,7 +64,13 @@ export class CodeInterpreterDriverFactory extends DriverFactoryBase>(); + 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 { + await this.codeInterpreterDriverFactory + .getCurrentDriver() + .releaseSession?.(`${workspaceId}:${threadId}`); + } + + async sweepExpiredSandboxes(): Promise { + 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 { - 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 { + return this.codeInterpreterDriverFactory + .getCurrentDriver() + .execute(code, files, context, callbacks); + } + + private async runSerializedPerSession( + sessionId: string, + task: () => Promise, + ): Promise { + 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); + } + } } } diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant.ts new file mode 100644 index 0000000000..6a65d9ad7d --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/code-interpreter-session-cleanup-cron-pattern.constant.ts @@ -0,0 +1 @@ +export const CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN = '0 * * * *'; diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant.ts new file mode 100644 index 0000000000..d375a50a74 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/constants/session-sandbox-metadata-key.constant.ts @@ -0,0 +1 @@ +export const SESSION_SANDBOX_METADATA_KEY = 'twentySessionId'; diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module.ts new file mode 100644 index 0000000000..05bfc57adf --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/code-interpreter-session-cleanup.module.ts @@ -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 {} diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/commands/code-interpreter-session-cleanup.cron.command.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/commands/code-interpreter-session-cleanup.cron.command.ts new file mode 100644 index 0000000000..caa6a66360 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/commands/code-interpreter-session-cleanup.cron.command.ts @@ -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 { + await this.messageQueueService.addCron({ + jobName: CodeInterpreterSessionCleanupCronJob.name, + data: undefined, + options: { + repeat: { + pattern: CODE_INTERPRETER_SESSION_CLEANUP_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job.ts new file mode 100644 index 0000000000..764f3f69c9 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/crons/jobs/code-interpreter-session-cleanup.cron.job.ts @@ -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 { + const reclaimedCount = + await this.codeInterpreterService.sweepExpiredSandboxes(); + + if (reclaimedCount > 0) { + this.logger.log( + `Reclaimed ${reclaimedCount} expired code interpreter sandbox(es)`, + ); + } + } +} diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/e2b.driver.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/e2b.driver.ts index e5e342fc48..448f93a346 100644 --- a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/e2b.driver.ts +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/e2b.driver.ts @@ -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 { - 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 { + 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 { + await releaseSessionSandboxes(Sandbox, this.options.apiKey, sessionId); + } + + async sweepExpiredSessions(maxAgeMs: number): Promise { + return sweepExpiredSessionSandboxes(Sandbox, this.options.apiKey, maxAgeMs); + } } diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/interfaces/code-interpreter-driver.interface.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/interfaces/code-interpreter-driver.interface.ts index be2f5a51b7..53aa801886 100644 --- a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/interfaces/code-interpreter-driver.interface.ts +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/interfaces/code-interpreter-driver.interface.ts @@ -20,6 +20,7 @@ export type CodeExecutionResult = { export type ExecutionContext = { env?: Record; + sessionId?: string; }; export type StreamCallbacks = { @@ -35,4 +36,6 @@ export interface CodeInterpreterDriver { context?: ExecutionContext, callbacks?: StreamCallbacks, ): Promise; + releaseSession?(sessionId: string): Promise; + sweepExpiredSessions?(maxAgeMs: number): Promise; } diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/get-or-create-session-sandbox.util.spec.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/get-or-create-session-sandbox.util.spec.ts new file mode 100644 index 0000000000..4e6039d385 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/get-or-create-session-sandbox.util.spec.ts @@ -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 }; + +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 = {}, +): { 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); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/release-session-sandboxes.util.spec.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/release-session-sandboxes.util.spec.ts new file mode 100644 index 0000000000..c1e183f688 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/release-session-sandboxes.util.spec.ts @@ -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; + 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(); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/sweep-expired-session-sandboxes.util.spec.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/sweep-expired-session-sandboxes.util.spec.ts new file mode 100644 index 0000000000..b4b81c32fa --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/__tests__/sweep-expired-session-sandboxes.util.spec.ts @@ -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; + 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'] }, + }); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util.ts new file mode 100644 index 0000000000..30599d1c0d --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/get-or-create-session-sandbox.util.ts @@ -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> = []; + + 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 => { + 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 }; +}; diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util.ts new file mode 100644 index 0000000000..37075bb1e2 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/release-session-sandboxes.util.ts @@ -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 => { + const paginator = sandboxApi.list({ + apiKey, + query: { + state: ['running', 'paused'], + metadata: { [SESSION_SANDBOX_METADATA_KEY]: sessionId }, + }, + }); + + const sandboxes: Awaited> = []; + + 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), + ), + ); +}; diff --git a/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util.ts b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util.ts new file mode 100644 index 0000000000..c99ec017d0 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/code-interpreter/drivers/utils/sweep-expired-session-sandboxes.util.ts @@ -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 => { + const paginator = sandboxApi.list({ + apiKey, + query: { state: ['running', 'paused'] }, + }); + + const sandboxes: Awaited> = []; + + 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; +}; diff --git a/packages/twenty-server/src/engine/core-modules/core-engine.module.ts b/packages/twenty-server/src/engine/core-modules/core-engine.module.ts index 4734e9c303..cf8208006a 100644 --- a/packages/twenty-server/src/engine/core-modules/core-engine.module.ts +++ b/packages/twenty-server/src/engine/core-modules/core-engine.module.ts @@ -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, diff --git a/packages/twenty-server/src/engine/core-modules/tool-provider/interfaces/tool-provider-context.type.ts b/packages/twenty-server/src/engine/core-modules/tool-provider/interfaces/tool-provider-context.type.ts index 4f4dcc27fe..11eab6e673 100644 --- a/packages/twenty-server/src/engine/core-modules/tool-provider/interfaces/tool-provider-context.type.ts +++ b/packages/twenty-server/src/engine/core-modules/tool-provider/interfaces/tool-provider-context.type.ts @@ -12,5 +12,6 @@ export type ToolProviderContext = { actorContext?: ActorMetadata; userId?: string; userWorkspaceId?: string; + threadId?: string; onCodeExecutionUpdate?: CodeExecutionStreamEmitter; }; diff --git a/packages/twenty-server/src/engine/core-modules/tool-provider/providers/action-tool.provider.ts b/packages/twenty-server/src/engine/core-modules/tool-provider/providers/action-tool.provider.ts index 736a6fdc2f..8cc3a4584e 100644 --- a/packages/twenty-server/src/engine/core-modules/tool-provider/providers/action-tool.provider.ts +++ b/packages/twenty-server/src/engine/core-modules/tool-provider/providers/action-tool.provider.ts @@ -144,6 +144,7 @@ export class ActionToolProvider implements ToolProvider { workspaceId: context.workspaceId, userId: context.userId, userWorkspaceId: context.userWorkspaceId, + threadId: context.threadId, onCodeExecutionUpdate: context.onCodeExecutionUpdate, }); } diff --git a/packages/twenty-server/src/engine/core-modules/tool-provider/services/tool-registry.service.ts b/packages/twenty-server/src/engine/core-modules/tool-provider/services/tool-registry.service.ts index ad17ad572f..8fe5eecccd 100644 --- a/packages/twenty-server/src/engine/core-modules/tool-provider/services/tool-registry.service.ts +++ b/packages/twenty-server/src/engine/core-modules/tool-provider/services/tool-registry.service.ts @@ -389,6 +389,7 @@ export class ToolRegistryService { authContext: context.authContext, userId: context.userId, userWorkspaceId: context.userWorkspaceId, + threadId: context.threadId, onCodeExecutionUpdate: context.onCodeExecutionUpdate, }; } diff --git a/packages/twenty-server/src/engine/core-modules/tool-provider/types/tool-context.type.ts b/packages/twenty-server/src/engine/core-modules/tool-provider/types/tool-context.type.ts index ab915b6e25..da88923e91 100644 --- a/packages/twenty-server/src/engine/core-modules/tool-provider/types/tool-context.type.ts +++ b/packages/twenty-server/src/engine/core-modules/tool-provider/types/tool-context.type.ts @@ -11,5 +11,6 @@ export type ToolContext = { actorContext?: ActorMetadata; userId?: string; userWorkspaceId?: string; + threadId?: string; onCodeExecutionUpdate?: CodeExecutionStreamEmitter; }; diff --git a/packages/twenty-server/src/engine/core-modules/tool/tools/code-interpreter-tool/code-interpreter-tool.ts b/packages/twenty-server/src/engine/core-modules/tool/tools/code-interpreter-tool/code-interpreter-tool.ts index de8370090b..f4482a99c5 100644 --- a/packages/twenty-server/src/engine/core-modules/tool/tools/code-interpreter-tool/code-interpreter-tool.ts +++ b/packages/twenty-server/src/engine/core-modules/tool/tools/code-interpreter-tool/code-interpreter-tool.ts @@ -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 { - 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) => { diff --git a/packages/twenty-server/src/engine/core-modules/tool/types/tool-execution-context.type.ts b/packages/twenty-server/src/engine/core-modules/tool/types/tool-execution-context.type.ts index e36ddd2dd0..aed3c2a2a8 100644 --- a/packages/twenty-server/src/engine/core-modules/tool/types/tool-execution-context.type.ts +++ b/packages/twenty-server/src/engine/core-modules/tool/types/tool-execution-context.type.ts @@ -4,5 +4,6 @@ export type ToolExecutionContext = { workspaceId: string; userId?: string; userWorkspaceId?: string; + threadId?: string; onCodeExecutionUpdate?: CodeExecutionStreamEmitter; }; diff --git a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts index 4c03094e84..eb6191506d 100644 --- a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts +++ b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts @@ -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', diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts index b0e70e8369..b55ab0753b 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts @@ -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, diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts index 5436ca2bfb..6750de2a47 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts @@ -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, 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({ diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts index ec794b2cf2..8d0aeec329 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts @@ -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[]; 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, }; diff --git a/yarn.lock b/yarn.lock index 8263e6b8e4..a6170073b6 100644 --- a/yarn.lock +++ b/yarn.lock @@ -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"