feat(ai): reliable bulk data import via code-interpreter (#22209)

## Summary Makes AI-assisted bulk data import (CSV/Excel/spreadsheets) reliable and token-efficient by letting an entire import run inside a single code-interpreter call, with a persistent sandbox session and server-side bulk helpers. Also includes supporting improvements to attachment handling, upsert reporting, and field-permission error messages. ## Changes ### Code interpreter - **Persistent per-session kernel** in `LocalDriver`: a long-lived Python process per `sessionId` keeps variables, imports, and files alive across calls (matching E2B behavior). Falls back to the existing ephemeral per-call path when no session is provided. CAN BE REMOVED, INTERESTING FOR DEV X - Idle watchdog that self-terminates the kernel, configurable via the new `CODE_INTERPRETER_IDLE_TIMEOUT_MS` config variable; the process also exits on parent shutdown (EOF on control fd). CAN BE REMOVED, INTERESTING FOR DEV X - New `bulk_upsert` and `lookup_by` helpers on the sandbox `twenty` object for idempotent batched writes (≤200/batch) and bounded relation-ID resolution. ### Records - `upsert_many_*` now reports a `created` / `updated` / `total` split in its result and log line (new `isFreshlyCreatedRecord` util). ### AI chat - `replaceUnsupportedFileParts`: user-attached files whose MIME type the model can't handle natively (and that aren't code-interpreter-supported) are downgraded to a descriptive text note instead of being sent as unsupported file parts. Modality→MIME mapping drives native support detection. - Finalize dangling tool parts before `convertToModelMessages` to avoid malformed model messages. - Extracted shared types/constants for code-interpreter file extraction. ### Permissions - Field permission-denied exceptions now include the field name and entity name for easier debugging. ### Skill docs - Added the bulk-import recipe ## To do in following PR - [ ] Skill command migration ## Test plan - [x] Unit tests for `getNativeMimeTypesForModalities` and `replaceUnsupportedFileParts` pass - [x] Run a bulk import (>50 rows) end-to-end through the code interpreter and verify a single sandbox call handles read → resolve relations → upsert → summary - [x] Verify session persistence: define a variable in one call, use it in the next within the same session - [x] Verify the kernel self-terminates after `CODE_INTERPRETER_IDLE_TIMEOUT_MS` - [x] Verify unsupported attachments are replaced with a text note for models lacking the modality - [x] Verify `upsert_many_*` returns correct created/updated counts - [x] Verify field-restricted role triggers a permission error naming the field and entity - [ ] Test with [hotel_business.xlsx](https://github.com/user-attachments/files/29376307/hotel_business.xlsx) and simple "import record" prompt  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22209?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-06-29 11:54:27 +02:00
parent 02d62c4175
commit 56deba351b
21 changed files with 1059 additions and 88 deletions
@@ -52,7 +52,12 @@ export class CodeInterpreterDriverFactory extends DriverFactoryBase<CodeInterpre
          );
        }

-        return new LocalDriver({ timeoutMs });
+        return new LocalDriver({
+          timeoutMs,
+          idleTimeoutMs: this.twentyConfigService.get(
+            'CODE_INTERPRETER_IDLE_TIMEOUT_MS',
+          ),
+        });
      }

      case CodeInterpreterDriverType.E_2_B: {
@@ -0,0 +1,89 @@
+// Persistent Python "kernel" used by the LocalDriver to keep variables,
+// imports and files alive across calls within the same session — matching the
+// cross-call persistence the E2B driver provides.
+//
+// Protocol (newline-delimited JSON over dedicated file descriptors so user
+// stdout/stderr never collide with control messages):
+//   - reads requests from fd 3: { "code": "<base64-utf8>" }
+//   - writes one response per request to fd 4:
+//       { "stdout": "...", "stderr": "...", "exitCode": 0 | 1 }
+// The namespace dict is reused for every request, so user-defined variables and
+// imports survive between executions.
+//
+// A single mechanism guarantees the process is always destroyed: an idle
+// watchdog thread that self-terminates after KERNEL_IDLE_TIMEOUT_MS of
+// inactivity (0 disables it). As a free side effect of reading control input,
+// the read loop also ends on EOF (fd 3) when the parent dies, so the process
+// exits immediately on dev-server shutdown too.
+export const LOCAL_DRIVER_PERSISTENT_KERNEL_SCRIPT = `import sys
+import os
+import io
+import json
+import base64
+import contextlib
+import traceback
+import threading
+import time
+
+_last_activity = time.time()
+
+def _idle_watchdog(idle_ms):
+    while True:
+        time.sleep(5)
+        if (time.time() - _last_activity) * 1000 > idle_ms:
+            os._exit(0)
+
+def _main():
+    global _last_activity
+
+    idle_ms = int(os.environ.get('KERNEL_IDLE_TIMEOUT_MS', '0'))
+    if idle_ms > 0:
+        threading.Thread(target=_idle_watchdog, args=(idle_ms,), daemon=True).start()
+
+    control_in = os.fdopen(3, 'r')
+    control_out = os.fdopen(4, 'w')
+    namespace = {'__name__': '__main__'}
+
+    for line in control_in:
+        _last_activity = time.time()
+        line = line.strip()
+        if not line:
+            continue
+
+        try:
+            request = json.loads(line)
+        except Exception:
+            continue
+
+        try:
+            code = base64.b64decode(request.get('code', '')).decode('utf-8')
+        except Exception:
+            code = ''
+
+        stdout_buffer = io.StringIO()
+        stderr_buffer = io.StringIO()
+        error = None
+
+        with contextlib.redirect_stdout(stdout_buffer), contextlib.redirect_stderr(stderr_buffer):
+            try:
+                exec(compile(code, '<cell>', 'exec'), namespace)
+            except SystemExit:
+                pass
+            except BaseException:
+                error = traceback.format_exc()
+
+        stderr_value = stderr_buffer.getvalue()
+        if error:
+            stderr_value = stderr_value + error
+
+        response = {
+            'stdout': stdout_buffer.getvalue(),
+            'stderr': stderr_value,
+            'exitCode': 1 if error else 0,
+        }
+        control_out.write(json.dumps(response) + '\\n')
+        control_out.flush()
+        _last_activity = time.time()
+
+_main()
+`;
@@ -1,9 +1,14 @@
-import { spawn } from 'child_process';
+import { type ChildProcess, spawn } from 'child_process';
 import { promises as fs } from 'fs';
 import { tmpdir } from 'os';
 import { basename, join } from 'path';
+import { type Readable, type Writable } from 'stream';
+
+import { isDefined } from 'twenty-shared/utils';
+import { v4 as uuidv4 } from 'uuid';

 import { DEFAULT_CODE_INTERPRETER_TIMEOUT_MS } from 'src/engine/core-modules/code-interpreter/code-interpreter.constants';
+import { LOCAL_DRIVER_PERSISTENT_KERNEL_SCRIPT } from 'src/engine/core-modules/code-interpreter/drivers/local-driver-kernel.const';
 import { getMimeType } from 'src/engine/core-modules/code-interpreter/utils/get-mime-type.util';

 import {
@@ -17,10 +22,32 @@ import {

 export type LocalDriverOptions = {
  timeoutMs?: number;
+  idleTimeoutMs?: number;
 };

 const SANDBOX_SCRIPTS_PATH = join(__dirname, '..', 'sandbox-scripts');

+type KernelResponse = {
+  stdout: string;
+  stderr: string;
+  exitCode: number;
+};
+
+type LocalSession = {
+  workDir: string;
+  outputDir: string;
+  scriptsDir: string;
+  child: ChildProcess;
+  controlIn: Writable;
+  controlOut: Readable;
+  controlBuffer: string;
+  pending?: {
+    resolve: (response: KernelResponse) => void;
+    reject: (error: Error) => void;
+  };
+  hasExited: boolean;
+};
+
 async function copyDirectoryRecursive(src: string, dest: string) {
  await fs.mkdir(dest, { recursive: true });

@@ -38,9 +65,31 @@ async function copyDirectoryRecursive(src: string, dest: string) {
  }
 }

-// WARNING: This driver is UNSAFE and should only be used for development.
+const buildEnvSetup = (env?: Record<string, string>): string => {
+  if (!isDefined(env)) {
+    return '';
+  }
+
+  const assignments = Object.entries(env)
+    .map(([key, value]) => {
+      const escapedValue = value
+        .replace(/\\/g, '\\\\')
+        .replace(/'/g, "\\'")
+        .replace(/\n/g, '\\n')
+        .replace(/\r/g, '\\r');
+
+      return `os.environ['${key}'] = '${escapedValue}'`;
+    })
+    .join('\n');
+
+  return `import os\n${assignments}\n\n`;
+};
+
+// WARNING: This driver is UNSAFE and can be used for development only.
 // It executes arbitrary Python code on the server without any sandboxing.
 export class LocalDriver implements CodeInterpreterDriver {
+  private readonly sessions = new Map<string, LocalSession>();
+
  constructor(private options: LocalDriverOptions = {}) {}

  async execute(
@@ -48,6 +97,96 @@ export class LocalDriver implements CodeInterpreterDriver {
    files?: InputFile[],
    context?: ExecutionContext,
    callbacks?: StreamCallbacks,
+  ): Promise<CodeExecutionResult> {
+    if (isDefined(context?.sessionId)) {
+      return this.executeInSession(
+        context.sessionId,
+        code,
+        files,
+        context,
+        callbacks,
+      );
+    }
+
+    return this.executeEphemeral(code, files, context, callbacks);
+  }
+
+  // Persistent path: reuse a per-session process + work dir so state survives
+  // between calls (variables, imports, files).
+  private async executeInSession(
+    sessionId: string,
+    code: string,
+    files: InputFile[] | undefined,
+    context: ExecutionContext | undefined,
+    callbacks?: StreamCallbacks,
+  ): Promise<CodeExecutionResult> {
+    const session = await this.getOrCreateSession(sessionId, context?.env);
+
+    // /home/user/output is cleared at the start of every call (matching the E2B
+    // behavior and the tool contract).
+    await fs.rm(session.outputDir, { recursive: true, force: true });
+    await fs.mkdir(session.outputDir, { recursive: true });
+
+    for (const file of files ?? []) {
+      const safeFilename = basename(file.filename);
+
+      await fs.writeFile(join(session.workDir, safeFilename), file.content);
+    }
+
+    const rewrittenCode = this.rewriteSandboxPaths(
+      code,
+      session.workDir,
+      session.scriptsDir,
+      session.outputDir,
+    );
+
+    const submission = buildEnvSetup(context?.env) + rewrittenCode;
+    const timeoutMs =
+      this.options.timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS;
+
+    let response: KernelResponse;
+
+    try {
+      response = await this.runInSession(session, submission, timeoutMs);
+    } catch (error) {
+      // A timeout or a dead kernel: kill the (possibly wedged) process so the
+      // next call recreates a clean one. The 'exit' handler reclaims the work
+      // dir.
+      session.hasExited = true;
+      this.sessions.delete(sessionId);
+      session.child.kill('SIGKILL');
+
+      return {
+        stdout: '',
+        stderr: error instanceof Error ? error.message : String(error),
+        exitCode: 1,
+        files: [],
+        error: error instanceof Error ? error.message : String(error),
+      };
+    }
+
+    this.streamCaptured(response, callbacks);
+
+    const outputFiles = await this.collectOutputFiles(
+      session.outputDir,
+      callbacks,
+    );
+
+    return {
+      stdout: response.stdout,
+      stderr: response.stderr,
+      exitCode: response.exitCode,
+      files: outputFiles,
+    };
+  }
+
+  // Ephemeral path (no session): fresh work dir + process per call, cleaned up
+  // afterwards. State does NOT persist between calls here.
+  private async executeEphemeral(
+    code: string,
+    files: InputFile[] | undefined,
+    context: ExecutionContext | undefined,
+    callbacks?: StreamCallbacks,
  ): Promise<CodeExecutionResult> {
    const workDir = await fs.mkdtemp(join(tmpdir(), 'code-interpreter-'));
    const outputDir = join(workDir, 'output');
@@ -55,7 +194,6 @@ export class LocalDriver implements CodeInterpreterDriver {

    await fs.mkdir(outputDir);

-    // Copy pre-installed scripts to sandbox
    try {
      await copyDirectoryRecursive(SANDBOX_SCRIPTS_PATH, scriptsDir);
    } catch {
@@ -69,14 +207,12 @@ export class LocalDriver implements CodeInterpreterDriver {
        await fs.writeFile(join(workDir, safeFilename), file.content);
      }

-      // Rewrite E2B-style paths to local paths for compatibility
-      const rewrittenCode = code
-        .replace(/\/home\/user\/scripts\//g, `${scriptsDir}/`)
-        .replace(/\/home\/user\/scripts/g, scriptsDir)
-        .replace(/\/home\/user\/output\//g, `${outputDir}/`)
-        .replace(/\/home\/user\/output/g, outputDir)
-        .replace(/\/home\/user\//g, `${workDir}/`)
-        .replace(/\/home\/user/g, workDir);
+      const rewrittenCode = this.rewriteSandboxPaths(
+        code,
+        workDir,
+        scriptsDir,
+        outputDir,
+      );

      const scriptPath = join(workDir, 'script.py');

@@ -94,29 +230,7 @@ export class LocalDriver implements CodeInterpreterDriver {
        callbacks,
      );

-      const outputFiles: OutputFile[] = [];
-
-      try {
-        const outputEntries = await fs.readdir(outputDir, {
-          withFileTypes: true,
-        });
-
-        for (const entry of outputEntries) {
-          if (entry.isFile()) {
-            const content = await fs.readFile(join(outputDir, entry.name));
-            const outputFile: OutputFile = {
-              filename: entry.name,
-              content,
-              mimeType: getMimeType(entry.name),
-            };
-
-            outputFiles.push(outputFile);
-            await callbacks?.onResult?.(outputFile);
-          }
-        }
-      } catch {
-        // Output directory might be empty or not exist
-      }
+      const outputFiles = await this.collectOutputFiles(outputDir, callbacks);

      return {
        stdout,
@@ -130,6 +244,211 @@ export class LocalDriver implements CodeInterpreterDriver {
    }
  }

+  private rewriteSandboxPaths(
+    code: string,
+    workDir: string,
+    scriptsDir: string,
+    outputDir: string,
+  ): string {
+    // Rewrite E2B-style paths to local paths for compatibility
+    return code
+      .replace(/\/home\/user\/scripts\//g, `${scriptsDir}/`)
+      .replace(/\/home\/user\/scripts/g, scriptsDir)
+      .replace(/\/home\/user\/output\//g, `${outputDir}/`)
+      .replace(/\/home\/user\/output/g, outputDir)
+      .replace(/\/home\/user\//g, `${workDir}/`)
+      .replace(/\/home\/user/g, workDir);
+  }
+
+  private async collectOutputFiles(
+    outputDir: string,
+    callbacks?: StreamCallbacks,
+  ): Promise<OutputFile[]> {
+    const outputFiles: OutputFile[] = [];
+
+    try {
+      const outputEntries = await fs.readdir(outputDir, {
+        withFileTypes: true,
+      });
+
+      for (const entry of outputEntries) {
+        if (entry.isFile()) {
+          const content = await fs.readFile(join(outputDir, entry.name));
+          const outputFile: OutputFile = {
+            filename: entry.name,
+            content,
+            mimeType: getMimeType(entry.name),
+          };
+
+          outputFiles.push(outputFile);
+          await callbacks?.onResult?.(outputFile);
+        }
+      }
+    } catch {
+      // Output directory might be empty or not exist
+    }
+
+    return outputFiles;
+  }
+
+  private streamCaptured(
+    response: KernelResponse,
+    callbacks?: StreamCallbacks,
+  ): void {
+    for (const line of response.stdout.split('\n')) {
+      if (line) {
+        callbacks?.onStdout?.(line);
+      }
+    }
+
+    for (const line of response.stderr.split('\n')) {
+      if (line) {
+        callbacks?.onStderr?.(line);
+      }
+    }
+  }
+
+  private async getOrCreateSession(
+    sessionId: string,
+    env?: Record<string, string>,
+  ): Promise<LocalSession> {
+    const existing = this.sessions.get(sessionId);
+
+    if (isDefined(existing) && !existing.hasExited) {
+      return existing;
+    }
+
+    if (isDefined(existing)) {
+      this.sessions.delete(sessionId);
+    }
+
+    const sessionDirName = `code-interpreter-session-${sessionId}-${uuidv4()}`;
+    const workDir = join(tmpdir(), sessionDirName);
+    const outputDir = join(workDir, 'output');
+    const scriptsDir = join(workDir, 'scripts');
+
+    await fs.mkdir(outputDir, { recursive: true });
+
+    try {
+      await copyDirectoryRecursive(SANDBOX_SCRIPTS_PATH, scriptsDir);
+    } catch {
+      // Scripts directory might not exist in dev environment
+    }
+
+    const kernelPath = join(workDir, 'kernel.py');
+
+    await fs.writeFile(kernelPath, LOCAL_DRIVER_PERSISTENT_KERNEL_SCRIPT);
+
+    const child = spawn('python3', ['-u', kernelPath], {
+      cwd: workDir,
+      env: {
+        ...process.env,
+        OUTPUT_DIR: outputDir,
+        KERNEL_IDLE_TIMEOUT_MS: String(this.options.idleTimeoutMs ?? 0),
+        ...env,
+      },
+      stdio: ['ignore', 'ignore', 'pipe', 'pipe', 'pipe'],
+    });
+
+    const controlIn = child.stdio[3] as Writable;
+    const controlOut = child.stdio[4] as Readable;
+
+    const session: LocalSession = {
+      workDir,
+      outputDir,
+      scriptsDir,
+      child,
+      controlIn,
+      controlOut,
+      controlBuffer: '',
+      hasExited: false,
+    };
+
+    controlOut.on('data', (chunk: Buffer) => {
+      session.controlBuffer += chunk.toString();
+
+      let newlineIndex = session.controlBuffer.indexOf('\n');
+
+      while (newlineIndex >= 0) {
+        const line = session.controlBuffer.slice(0, newlineIndex);
+
+        session.controlBuffer = session.controlBuffer.slice(newlineIndex + 1);
+
+        if (line.trim().length > 0 && isDefined(session.pending)) {
+          const pending = session.pending;
+
+          session.pending = undefined;
+
+          try {
+            pending.resolve(JSON.parse(line) as KernelResponse);
+          } catch (error) {
+            pending.reject(
+              error instanceof Error ? error : new Error(String(error)),
+            );
+          }
+        }
+
+        newlineIndex = session.controlBuffer.indexOf('\n');
+      }
+    });
+
+    const markExited = () => {
+      session.hasExited = true;
+      this.sessions.delete(sessionId);
+      session.pending?.reject(new Error('Python kernel process exited'));
+      session.pending = undefined;
+      // The kernel self-terminates (idle watchdog) or dies on its own; reclaim
+      // its work dir here so it doesn't leak.
+      void fs.rm(workDir, { recursive: true, force: true }).catch(() => {});
+    };
+
+    child.on('exit', markExited);
+    child.on('error', markExited);
+
+    this.sessions.set(sessionId, session);
+
+    return session;
+  }
+
+  private runInSession(
+    session: LocalSession,
+    submission: string,
+    timeoutMs: number,
+  ): Promise<KernelResponse> {
+    return new Promise<KernelResponse>((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        if (session.pending) {
+          session.pending = undefined;
+          reject(new Error('Process timed out'));
+        }
+      }, timeoutMs);
+
+      session.pending = {
+        resolve: (response) => {
+          clearTimeout(timeout);
+          resolve(response);
+        },
+        reject: (error) => {
+          clearTimeout(timeout);
+          reject(error);
+        },
+      };
+
+      const payload =
+        JSON.stringify({
+          code: Buffer.from(submission, 'utf-8').toString('base64'),
+        }) + '\n';
+
+      session.controlIn.write(payload, (error) => {
+        if (isDefined(error) && session.pending) {
+          session.pending = undefined;
+          clearTimeout(timeout);
+          reject(error);
+        }
+      });
+    });
+  }
+
  private runPythonScript(
    scriptPath: string,
    workDir: string,