Initial dev process keep alive. moved resource attributes to span attributes because resource attributes are global and immutable

apps/webapp/app/v3/otlpExporter.server.ts

@@ -183,7 +183,7 @@ class OTLPExporter {
 function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<CreatableEvent> {
   const resourceAttributes = resourceLog.resource?.attributes ?? [];
 
-  const resourceProperties = extractResourceProperties(resourceAttributes);
+  const resourceProperties = extractEventProperties(resourceAttributes);
 
   return resourceLog.scopeLogs.flatMap((scopeLog) => {
     return scopeLog.logRecords
@@ -194,6 +194,11 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
           return;
         }
 
+        const logProperties = extractEventProperties(
+          log.attributes ?? [],
+          SemanticInternalAttributes.METADATA
+        );
+
         return {
           traceId: binaryToHex(log.traceId),
           spanId: eventRepository.generateSpanId(),
@@ -212,11 +217,6 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
               SemanticInternalAttributes.SPAN_ID,
               SemanticInternalAttributes.SPAN_PARTIAL,
             ]),
-            ...convertKeyValueItemsToMap(
-              resourceAttributes,
-              [SemanticInternalAttributes.TRIGGER],
-              SemanticInternalAttributes.METADATA
-            ),
           },
           style: convertKeyValueItemsToMap(
             pickAttributes(log.attributes ?? [], SemanticInternalAttributes.STYLE),
@@ -236,7 +236,35 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
             ),
             SemanticInternalAttributes.PAYLOAD
           ),
-          ...resourceProperties,
+          metadata: logProperties.metadata ?? resourceProperties.metadata,
+          serviceName: logProperties.serviceName ?? resourceProperties.serviceName ?? "unknown",
+          serviceNamespace:
+            logProperties.serviceNamespace ?? resourceProperties.serviceNamespace ?? "unknown",
+          environmentId:
+            logProperties.environmentId ?? resourceProperties.environmentId ?? "unknown",
+          environmentType:
+            logProperties.environmentType ?? resourceProperties.environmentType ?? "DEVELOPMENT",
+          organizationId:
+            logProperties.organizationId ?? resourceProperties.organizationId ?? "unknown",
+          projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown",
+          projectRef: logProperties.projectRef ?? resourceProperties.projectRef ?? "unknown",
+          runId: logProperties.runId ?? resourceProperties.runId ?? "unknown",
+          runIsTest: logProperties.runIsTest ?? resourceProperties.runIsTest ?? false,
+          taskSlug: logProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
+          taskPath: logProperties.taskPath ?? resourceProperties.taskPath ?? "unknown",
+          workerId: logProperties.workerId ?? resourceProperties.workerId ?? "unknown",
+          workerVersion:
+            logProperties.workerVersion ?? resourceProperties.workerVersion ?? "unknown",
+          queueId: logProperties.queueId ?? resourceProperties.queueId ?? "unknown",
+          queueName: logProperties.queueName ?? resourceProperties.queueName ?? "unknown",
+          batchId: logProperties.batchId ?? resourceProperties.batchId,
+          idempotencyKey: logProperties.idempotencyKey ?? resourceProperties.idempotencyKey,
+          machinePreset: logProperties.machinePreset ?? resourceProperties.machinePreset,
+          machinePresetCpu: logProperties.machinePresetCpu ?? resourceProperties.machinePresetCpu,
+          machinePresetMemory:
+            logProperties.machinePresetMemory ?? resourceProperties.machinePresetMemory,
+          machinePresetCentsPerMs:
+            logProperties.machinePresetCentsPerMs ?? resourceProperties.machinePresetCentsPerMs,
           attemptId:
             extractStringAttribute(
               log.attributes ?? [],
@@ -258,7 +286,10 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
 function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<CreatableEvent> {
   const resourceAttributes = resourceSpan.resource?.attributes ?? [];
 
-  const resourceProperties = extractResourceProperties(resourceAttributes);
+  const resourceProperties = extractEventProperties(
+    resourceAttributes,
+    SemanticInternalAttributes.METADATA
+  );
 
   return resourceSpan.scopeSpans.flatMap((scopeSpan) => {
     return scopeSpan.spans
@@ -269,6 +300,11 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
           return;
         }
 
+        const spanProperties = extractEventProperties(
+          span.attributes ?? [],
+          SemanticInternalAttributes.METADATA
+        );
+
         return {
           traceId: binaryToHex(span.traceId),
           spanId: isPartial
@@ -294,11 +330,6 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
               SemanticInternalAttributes.SPAN_ID,
               SemanticInternalAttributes.SPAN_PARTIAL,
             ]),
-            ...convertKeyValueItemsToMap(
-              resourceAttributes,
-              [SemanticInternalAttributes.TRIGGER],
-              SemanticInternalAttributes.METADATA
-            ),
           },
           style: convertKeyValueItemsToMap(
             pickAttributes(span.attributes ?? [], SemanticInternalAttributes.STYLE),
@@ -327,7 +358,35 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
               span.attributes ?? [],
               SemanticInternalAttributes.PAYLOAD_TYPE
             ) ?? "application/json",
-          ...resourceProperties,
+          metadata: spanProperties.metadata ?? resourceProperties.metadata,
+          serviceName: spanProperties.serviceName ?? resourceProperties.serviceName ?? "unknown",
+          serviceNamespace:
+            spanProperties.serviceNamespace ?? resourceProperties.serviceNamespace ?? "unknown",
+          environmentId:
+            spanProperties.environmentId ?? resourceProperties.environmentId ?? "unknown",
+          environmentType:
+            spanProperties.environmentType ?? resourceProperties.environmentType ?? "DEVELOPMENT",
+          organizationId:
+            spanProperties.organizationId ?? resourceProperties.organizationId ?? "unknown",
+          projectId: spanProperties.projectId ?? resourceProperties.projectId ?? "unknown",
+          projectRef: spanProperties.projectRef ?? resourceProperties.projectRef ?? "unknown",
+          runId: spanProperties.runId ?? resourceProperties.runId ?? "unknown",
+          runIsTest: spanProperties.runIsTest ?? resourceProperties.runIsTest ?? false,
+          taskSlug: spanProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
+          taskPath: spanProperties.taskPath ?? resourceProperties.taskPath ?? "unknown",
+          workerId: spanProperties.workerId ?? resourceProperties.workerId ?? "unknown",
+          workerVersion:
+            spanProperties.workerVersion ?? resourceProperties.workerVersion ?? "unknown",
+          queueId: spanProperties.queueId ?? resourceProperties.queueId ?? "unknown",
+          queueName: spanProperties.queueName ?? resourceProperties.queueName ?? "unknown",
+          batchId: spanProperties.batchId ?? resourceProperties.batchId,
+          idempotencyKey: spanProperties.idempotencyKey ?? resourceProperties.idempotencyKey,
+          machinePreset: spanProperties.machinePreset ?? resourceProperties.machinePreset,
+          machinePresetCpu: spanProperties.machinePresetCpu ?? resourceProperties.machinePresetCpu,
+          machinePresetMemory:
+            spanProperties.machinePresetMemory ?? resourceProperties.machinePresetMemory,
+          machinePresetCentsPerMs:
+            spanProperties.machinePresetCentsPerMs ?? resourceProperties.machinePresetCentsPerMs,
           attemptId:
             extractStringAttribute(
               span.attributes ?? [],
@@ -359,67 +418,81 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
   });
 }
 
-function extractResourceProperties(attributes: KeyValue[]) {
+function extractEventProperties(attributes: KeyValue[], prefix?: string) {
   return {
     metadata: convertKeyValueItemsToMap(attributes, [SemanticInternalAttributes.TRIGGER]),
-    serviceName: extractStringAttribute(
-      attributes,
-      SemanticResourceAttributes.SERVICE_NAME,
-      "unknown"
-    ),
+    serviceName: extractStringAttribute(attributes, SemanticResourceAttributes.SERVICE_NAME),
     serviceNamespace: extractStringAttribute(
       attributes,
-      SemanticResourceAttributes.SERVICE_NAMESPACE,
-      "unknown"
+      SemanticResourceAttributes.SERVICE_NAMESPACE
     ),
-    environmentId: extractStringAttribute(
-      attributes,
+    environmentId: extractStringAttribute(attributes, [
+      prefix,
       SemanticInternalAttributes.ENVIRONMENT_ID,
-      "unknown"
-    ),
-    environmentType: extractStringAttribute(
-      attributes,
+    ]),
+    environmentType: extractStringAttribute(attributes, [
+      prefix,
       SemanticInternalAttributes.ENVIRONMENT_TYPE,
-      "unknown"
-    ) as CreatableEventEnvironmentType,
-    organizationId: extractStringAttribute(
-      attributes,
+    ]) as CreatableEventEnvironmentType,
+    organizationId: extractStringAttribute(attributes, [
+      prefix,
       SemanticInternalAttributes.ORGANIZATION_ID,
-      "unknown"
-    ),
-    projectId: extractStringAttribute(attributes, SemanticInternalAttributes.PROJECT_ID, "unknown"),
-    projectRef: extractStringAttribute(
-      attributes,
+    ]),
+    projectId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.PROJECT_ID]),
+    projectRef: extractStringAttribute(attributes, [
+      prefix,
       SemanticInternalAttributes.PROJECT_REF,
+    ]),
+    runId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.RUN_ID]),
+    runIsTest: extractBooleanAttribute(
+      attributes,
+      [prefix, SemanticInternalAttributes.RUN_IS_TEST],
+      false
+    ),
+    attemptId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.ATTEMPT_ID]),
+    attemptNumber: extractNumberAttribute(attributes, [
+      prefix,
+      SemanticInternalAttributes.ATTEMPT_NUMBER,
+    ]),
+    taskSlug: extractStringAttribute(
+      attributes,
+      [prefix, SemanticInternalAttributes.TASK_SLUG],
       "unknown"
     ),
-    runId: extractStringAttribute(attributes, SemanticInternalAttributes.RUN_ID, "unknown"),
-    runIsTest: extractBooleanAttribute(attributes, SemanticInternalAttributes.RUN_IS_TEST, false),
-    attemptId: extractStringAttribute(attributes, SemanticInternalAttributes.ATTEMPT_ID),
-    attemptNumber: extractNumberAttribute(attributes, SemanticInternalAttributes.ATTEMPT_NUMBER),
-    taskSlug: extractStringAttribute(attributes, SemanticInternalAttributes.TASK_SLUG, "unknown"),
-    taskPath: extractStringAttribute(attributes, SemanticInternalAttributes.TASK_PATH),
+    taskPath: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_PATH]),
     taskExportName: "@deprecated",
-    workerId: extractStringAttribute(attributes, SemanticInternalAttributes.WORKER_ID),
-    workerVersion: extractStringAttribute(attributes, SemanticInternalAttributes.WORKER_VERSION),
-    queueId: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_ID),
-    queueName: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_NAME),
-    batchId: extractStringAttribute(attributes, SemanticInternalAttributes.BATCH_ID),
-    idempotencyKey: extractStringAttribute(attributes, SemanticInternalAttributes.IDEMPOTENCY_KEY),
-    machinePreset: extractStringAttribute(
-      attributes,
-      SemanticInternalAttributes.MACHINE_PRESET_NAME
-    ),
+    workerId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.WORKER_ID]),
+    workerVersion: extractStringAttribute(attributes, [
+      prefix,
+      SemanticInternalAttributes.WORKER_VERSION,
+    ]),
+    queueId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.QUEUE_ID]),
+    queueName: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.QUEUE_NAME]),
+    batchId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.BATCH_ID]),
+    idempotencyKey: extractStringAttribute(attributes, [
+      prefix,
+      SemanticInternalAttributes.IDEMPOTENCY_KEY,
+    ]),
+    machinePreset: extractStringAttribute(attributes, [
+      prefix,
+      SemanticInternalAttributes.MACHINE_PRESET_NAME,
+    ]),
     machinePresetCpu:
-      extractDoubleAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_CPU) ??
-      extractNumberAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_CPU),
+      extractDoubleAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_PRESET_CPU]) ??
+      extractNumberAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_PRESET_CPU]),
     machinePresetMemory:
-      extractDoubleAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_MEMORY) ??
-      extractNumberAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_MEMORY),
-    machinePresetCentsPerMs: extractDoubleAttribute(
-      attributes,
-      SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS
-    ),
+      extractDoubleAttribute(attributes, [
+        prefix,
+        SemanticInternalAttributes.MACHINE_PRESET_MEMORY,
+      ]) ??
+      extractNumberAttribute(attributes, [
+        prefix,
+        SemanticInternalAttributes.MACHINE_PRESET_MEMORY,
+      ]),
+    machinePresetCentsPerMs: extractDoubleAttribute(attributes, [
+      prefix,
+      SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS,
+    ]),
   };
 }
 
@@ -643,56 +716,92 @@ function convertUnixNanoToDate(unixNano: bigint | number): Date {
   return new Date(Number(BigInt(unixNano) / BigInt(1_000_000)));
 }
 
-function extractStringAttribute(attributes: KeyValue[], name: string): string | undefined;
-function extractStringAttribute(attributes: KeyValue[], name: string, fallback: string): string;
 function extractStringAttribute(
   attributes: KeyValue[],
-  name: string,
+  name: string | Array<string | undefined>
+): string | undefined;
+function extractStringAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
+  fallback: string
+): string;
+function extractStringAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
   fallback?: string
 ): string | undefined {
-  const attribute = attributes.find((attribute) => attribute.key === name);
+  const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
+
+  const attribute = attributes.find((attribute) => attribute.key === key);
 
   if (!attribute) return fallback;
 
   return isStringValue(attribute?.value) ? attribute.value.stringValue : fallback;
 }
 
-function extractNumberAttribute(attributes: KeyValue[], name: string): number | undefined;
-function extractNumberAttribute(attributes: KeyValue[], name: string, fallback: number): number;
 function extractNumberAttribute(
   attributes: KeyValue[],
-  name: string,
+  name: string | Array<string | undefined>
+): number | undefined;
+function extractNumberAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
+  fallback: number
+): number;
+function extractNumberAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
   fallback?: number
 ): number | undefined {
-  const attribute = attributes.find((attribute) => attribute.key === name);
+  const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
+
+  const attribute = attributes.find((attribute) => attribute.key === key);
 
   if (!attribute) return fallback;
 
   return isIntValue(attribute?.value) ? Number(attribute.value.intValue) : fallback;
 }
 
-function extractDoubleAttribute(attributes: KeyValue[], name: string): number | undefined;
-function extractDoubleAttribute(attributes: KeyValue[], name: string, fallback: number): number;
 function extractDoubleAttribute(
   attributes: KeyValue[],
-  name: string,
+  name: string | Array<string | undefined>
+): number | undefined;
+function extractDoubleAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
+  fallback: number
+): number;
+function extractDoubleAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
   fallback?: number
 ): number | undefined {
-  const attribute = attributes.find((attribute) => attribute.key === name);
+  const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
+
+  const attribute = attributes.find((attribute) => attribute.key === key);
 
   if (!attribute) return fallback;
 
   return isDoubleValue(attribute?.value) ? Number(attribute.value.doubleValue) : fallback;
 }
 
-function extractBooleanAttribute(attributes: KeyValue[], name: string): boolean | undefined;
-function extractBooleanAttribute(attributes: KeyValue[], name: string, fallback: boolean): boolean;
 function extractBooleanAttribute(
   attributes: KeyValue[],
-  name: string,
+  name: string | Array<string | undefined>
+): boolean | undefined;
+function extractBooleanAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
+  fallback: boolean
+): boolean;
+function extractBooleanAttribute(
+  attributes: KeyValue[],
+  name: string | Array<string | undefined>,
   fallback?: boolean
 ): boolean | undefined {
-  const attribute = attributes.find((attribute) => attribute.key === name);
+  const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
+
+  const attribute = attributes.find((attribute) => attribute.key === key);
 
   if (!attribute) return fallback;
 

packages/cli-v3/src/dev/devSupervisor.ts

@@ -25,6 +25,7 @@ import {
 import pLimit from "p-limit";
 import { resolveLocalEnvVars } from "../utilities/localEnvVars.js";
 import type { Metafile } from "esbuild";
+import { TaskRunProcessPool } from "./taskRunProcessPool.js";
 
 export type WorkerRuntimeOptions = {
   name: string | undefined;
@@ -67,6 +68,7 @@ class DevSupervisor implements WorkerRuntime {
   private socketConnections = new Set<string>();
 
   private runLimiter?: ReturnType<typeof pLimit>;
+  private taskRunProcessPool?: TaskRunProcessPool;
 
   constructor(public readonly options: WorkerRuntimeOptions) {}
 
@@ -95,6 +97,31 @@ class DevSupervisor implements WorkerRuntime {
 
     this.runLimiter = pLimit(maxConcurrentRuns);
 
+    // Initialize the task run process pool
+    const env = await this.#getEnvVars();
+
+    const enableProcessReuse =
+      typeof this.options.config.experimental_processKeepAlive === "boolean"
+        ? this.options.config.experimental_processKeepAlive
+        : false;
+
+    if (enableProcessReuse) {
+      logger.debug("[DevSupervisor] Enabling process reuse", {
+        enableProcessReuse,
+      });
+    }
+
+    this.taskRunProcessPool = new TaskRunProcessPool({
+      env,
+      cwd: this.options.config.workingDir,
+      enableProcessReuse:
+        typeof this.options.config.experimental_processKeepAlive === "boolean"
+          ? this.options.config.experimental_processKeepAlive
+          : false,
+      maxPoolSize: 3,
+      maxExecutionsPerProcess: 50,
+    });
+
     this.socket = this.#createSocket();
 
     //start an SSE connection for presence
@@ -111,6 +138,11 @@ class DevSupervisor implements WorkerRuntime {
     } catch (error) {
       logger.debug("[DevSupervisor] shutdown, socket failed to close", { error });
     }
+
+    // Shutdown the task run process pool
+    if (this.taskRunProcessPool) {
+      await this.taskRunProcessPool.shutdown();
+    }
   }
 
   async initializeWorker(
@@ -293,12 +325,21 @@ class DevSupervisor implements WorkerRuntime {
           continue;
         }
 
+        if (!this.taskRunProcessPool) {
+          logger.debug(`[DevSupervisor] dequeueRuns. No task run process pool`, {
+            run: message.run.friendlyId,
+            worker,
+          });
+          continue;
+        }
+
         //new run
         runController = new DevRunController({
           runFriendlyId: message.run.friendlyId,
           worker: worker,
           httpClient: this.options.client,
           logLevel: this.options.args.logLevel,
+          taskRunProcessPool: this.taskRunProcessPool,
           onFinished: () => {
             logger.debug("[DevSupervisor] Run finished", { runId: message.run.friendlyId });
 

packages/cli-v3/src/dev/taskRunProcessPool.ts

@@ -0,0 +1,165 @@
+import {
+  MachinePresetResources,
+  ServerBackgroundWorker,
+  WorkerManifest,
+} from "@trigger.dev/core/v3";
+import { TaskRunProcess } from "../executions/taskRunProcess.js";
+import { logger } from "../utilities/logger.js";
+
+export type TaskRunProcessPoolOptions = {
+  env: Record<string, string>;
+  cwd: string;
+  enableProcessReuse: boolean;
+  maxPoolSize?: number;
+  maxExecutionsPerProcess?: number;
+};
+
+export class TaskRunProcessPool {
+  private availableProcesses: TaskRunProcess[] = [];
+  private busyProcesses: Set<TaskRunProcess> = new Set();
+  private readonly options: TaskRunProcessPoolOptions;
+  private readonly maxPoolSize: number;
+  private readonly maxExecutionsPerProcess: number;
+
+  constructor(options: TaskRunProcessPoolOptions) {
+    this.options = options;
+    this.maxPoolSize = options.maxPoolSize ?? 3;
+    this.maxExecutionsPerProcess = options.maxExecutionsPerProcess ?? 50;
+  }
+
+  async getProcess(
+    workerManifest: WorkerManifest,
+    serverWorker: ServerBackgroundWorker,
+    machineResources: MachinePresetResources,
+    env?: Record<string, string>
+  ): Promise<TaskRunProcess> {
+    // Try to reuse an existing process if enabled
+    if (this.options.enableProcessReuse) {
+      const reusableProcess = this.findReusableProcess();
+      if (reusableProcess) {
+        logger.debug("[TaskRunProcessPool] Reusing existing process", {
+          availableCount: this.availableProcesses.length,
+          busyCount: this.busyProcesses.size,
+        });
+
+        this.availableProcesses = this.availableProcesses.filter((p) => p !== reusableProcess);
+        this.busyProcesses.add(reusableProcess);
+        return reusableProcess;
+      } else {
+        logger.debug("[TaskRunProcessPool] No reusable process found", {
+          availableCount: this.availableProcesses.length,
+          busyCount: this.busyProcesses.size,
+        });
+      }
+    }
+
+    // Create new process
+    logger.debug("[TaskRunProcessPool] Creating new process", {
+      availableCount: this.availableProcesses.length,
+      busyCount: this.busyProcesses.size,
+    });
+
+    const newProcess = new TaskRunProcess({
+      workerManifest,
+      env: {
+        ...this.options.env,
+        ...env,
+      },
+      serverWorker,
+      machineResources,
+      cwd: this.options.cwd,
+    }).initialize();
+
+    this.busyProcesses.add(newProcess);
+    return newProcess;
+  }
+
+  async returnProcess(process: TaskRunProcess): Promise<void> {
+    this.busyProcesses.delete(process);
+
+    if (this.shouldReuseProcess(process)) {
+      logger.debug("[TaskRunProcessPool] Returning process to pool", {
+        availableCount: this.availableProcesses.length,
+        busyCount: this.busyProcesses.size,
+      });
+
+      // Clean up but don't kill the process
+      try {
+        await process.cleanup(false);
+        this.availableProcesses.push(process);
+      } catch (error) {
+        logger.debug("[TaskRunProcessPool] Failed to cleanup process for reuse, killing it", {
+          error,
+        });
+        await this.killProcess(process);
+      }
+    } else {
+      logger.debug("[TaskRunProcessPool] Killing process", {
+        availableCount: this.availableProcesses.length,
+        busyCount: this.busyProcesses.size,
+      });
+      await this.killProcess(process);
+    }
+  }
+
+  private findReusableProcess(): TaskRunProcess | undefined {
+    return this.availableProcesses.find((process) => this.isProcessHealthy(process));
+  }
+
+  private shouldReuseProcess(process: TaskRunProcess): boolean {
+    const isHealthy = this.isProcessHealthy(process);
+    const isBeingKilled = process.isBeingKilled;
+    const pid = process.pid;
+
+    logger.debug("[TaskRunProcessPool] Checking if process should be reused", {
+      isHealthy,
+      isBeingKilled,
+      pid,
+      availableCount: this.availableProcesses.length,
+      busyCount: this.busyProcesses.size,
+      maxPoolSize: this.maxPoolSize,
+    });
+
+    return (
+      this.options.enableProcessReuse &&
+      this.isProcessHealthy(process) &&
+      this.availableProcesses.length < this.maxPoolSize
+    );
+  }
+
+  private isProcessHealthy(process: TaskRunProcess): boolean {
+    // Basic health checks - we can expand this later
+    return !process.isBeingKilled && process.pid !== undefined;
+  }
+
+  private async killProcess(process: TaskRunProcess): Promise<void> {
+    try {
+      await process.cleanup(true);
+    } catch (error) {
+      logger.debug("[TaskRunProcessPool] Error killing process", { error });
+    }
+  }
+
+  async shutdown(): Promise<void> {
+    logger.debug("[TaskRunProcessPool] Shutting down pool", {
+      availableCount: this.availableProcesses.length,
+      busyCount: this.busyProcesses.size,
+    });
+
+    // Kill all available processes
+    await Promise.all(this.availableProcesses.map((process) => this.killProcess(process)));
+    this.availableProcesses = [];
+
+    // Kill all busy processes
+    await Promise.all(Array.from(this.busyProcesses).map((process) => this.killProcess(process)));
+    this.busyProcesses.clear();
+  }
+
+  getStats() {
+    return {
+      availableCount: this.availableProcesses.length,
+      busyCount: this.busyProcesses.size,
+      totalCount: this.availableProcesses.length + this.busyProcesses.size,
+    };
+  }
+}

packages/cli-v3/src/entryPoints/dev-run-controller.ts

@@ -19,6 +19,7 @@ import { sanitizeEnvVars } from "../utilities/sanitizeEnvVars.js";
 import { join } from "node:path";
 import { BackgroundWorker } from "../dev/backgroundWorker.js";
 import { eventBus } from "../utilities/eventBus.js";
+import { TaskRunProcessPool } from "../dev/taskRunProcessPool.js";
 
 type DevRunControllerOptions = {
   runFriendlyId: string;
@@ -26,6 +27,7 @@ type DevRunControllerOptions = {
   httpClient: CliApiClient;
   logLevel: LogLevel;
   heartbeatIntervalSeconds?: number;
+  taskRunProcessPool: TaskRunProcessPool;
   onSubscribeToRunNotifications: (run: Run, snapshot: Snapshot) => void;
   onUnsubscribeFromRunNotifications: (run: Run, snapshot: Snapshot) => void;
   onFinished: () => void;
@@ -605,25 +607,25 @@ export class DevRunController {
 
     this.snapshotPoller.start();
 
-    this.taskRunProcess = new TaskRunProcess({
-      workerManifest: this.opts.worker.manifest,
-      env: {
-        ...sanitizeEnvVars(envVars ?? {}),
-        ...sanitizeEnvVars(this.opts.worker.params.env),
-        TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"),
-        RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false",
-        TRIGGER_PROJECT_REF: execution.project.ref,
-      },
-      serverWorker: {
+    // Get process from pool instead of creating new one
+    this.taskRunProcess = await this.opts.taskRunProcessPool.getProcess(
+      this.opts.worker.manifest,
+      {
         id: "unmanaged",
         contentHash: this.opts.worker.build.contentHash,
         version: this.opts.worker.serverWorker?.version,
         engine: "V2",
       },
-      machineResources: execution.machine,
-    }).initialize();
+      execution.machine,
+      {
+        TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"),
+        RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false",
+      }
+    );
 
-    logger.debug("executing task run process", {
+    // Update the process environment for this specific run
+    // Note: We may need to enhance TaskRunProcess to support updating env vars
+    logger.debug("executing task run process from pool", {
       attemptNumber: execution.attempt.number,
       runId: execution.run.id,
     });
@@ -635,15 +637,21 @@ export class DevRunController {
         metrics,
       },
       messageId: run.friendlyId,
+      env: {
+        ...sanitizeEnvVars(envVars ?? {}),
+        ...sanitizeEnvVars(this.opts.worker.params.env),
+        TRIGGER_PROJECT_REF: execution.project.ref,
+      },
     });
 
     logger.debug("Completed run", completion);
 
+    // Return process to pool instead of killing it
     try {
-      await this.taskRunProcess.cleanup(true);
+      await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess);
       this.taskRunProcess = undefined;
     } catch (error) {
-      logger.debug("Failed to cleanup task run process, submitting completion anyway", {
+      logger.debug("Failed to return task run process to pool, submitting completion anyway", {
         error,
       });
     }
@@ -758,11 +766,14 @@ export class DevRunController {
   }
 
   private async runFinished() {
-    // Kill the run process
-    try {
-      await this.taskRunProcess?.kill("SIGKILL");
-    } catch (error) {
-      logger.debug("Failed to kill task run process", { error });
+    // Return the process to the pool instead of killing it directly
+    if (this.taskRunProcess) {
+      try {
+        await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess);
+        this.taskRunProcess = undefined;
+      } catch (error) {
+        logger.debug("Failed to return task run process to pool during runFinished", { error });
+      }
     }
 
     this.runHeartbeat.stop();
@@ -794,9 +805,10 @@ export class DevRunController {
 
     if (this.taskRunProcess && !this.taskRunProcess.isBeingKilled) {
       try {
-        await this.taskRunProcess.cleanup(true);
+        await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess);
+        this.taskRunProcess = undefined;
       } catch (error) {
-        logger.debug("Failed to cleanup task run process", { error });
+        logger.debug("Failed to return task run process to pool during stop", { error });
       }
     }
 

packages/core/src/v3/config.ts

@@ -234,6 +234,14 @@ export type TriggerConfig = {
     env?: Record<string, string>;
   };
 
+  /**
+   * @default false
+   * @description Keep the process alive after the task has finished running so the next task doesn't have to wait for the process to start up again.
+   *
+   * Note that the process could be killed at any time, and we don't make any guarantees about the process being alive for a certain amount of time
+   */
+  experimental_processKeepAlive?: boolean;
+
   /**
    * @deprecated Use `dirs` instead
    */

packages/core/src/v3/otel/tracingSDK.ts

@@ -114,7 +114,7 @@ export class TracingSDK {
       : {};
 
     const commonResources = detectResourcesSync({
-      detectors: [this.asyncResourceDetector, processDetectorSync],
+      detectors: [processDetectorSync],
     })
       .merge(
         new Resource({
@@ -123,6 +123,8 @@ export class TracingSDK {
             getEnvVar("OTEL_SERVICE_NAME") ?? "trigger.dev",
           [SemanticInternalAttributes.TRIGGER]: true,
           [SemanticInternalAttributes.CLI_VERSION]: VERSION,
+          [SemanticInternalAttributes.SDK_VERSION]: VERSION,
+          [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
         })
       )
       .merge(config.resource ?? new Resource({}))

packages/core/src/v3/taskContext/otelProcessors.ts

@@ -21,6 +21,7 @@ export class TaskContextSpanProcessor implements SpanProcessor {
       span.setAttributes(
         flattenAttributes(
           {
+            ...taskContext.attributes,
             [SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id,
             [SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number,
           },
@@ -109,6 +110,7 @@ export class TaskContextLogProcessor implements LogRecordProcessor {
       logRecord.setAttributes(
         flattenAttributes(
           {
+            ...taskContext.attributes,
             [SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id,
             [SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number,
           },

packages/core/src/v3/workers/taskExecutor.ts

@@ -112,11 +112,9 @@ export class TaskExecutor {
       runMetadata.enterWithMetadata(execution.run.metadata);
     }
 
-    this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
-      ...taskContext.attributes,
-      [SemanticInternalAttributes.SDK_VERSION]: VERSION,
-      [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
-    });
+    // this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
+    //   ...taskContext.attributes,
+    // });
 
     const result = await this._tracer.startActiveSpan(
       attemptMessage,

references/hello-world/trigger.config.ts

@@ -4,6 +4,7 @@ import { syncEnvVars } from "@trigger.dev/build/extensions/core";
 export default defineConfig({
   compatibilityFlags: ["run_engine_v2"],
   project: "proj_rrkpdguyagvsoktglnod",
+  experimental_processKeepAlive: true,
   logLevel: "log",
   maxDuration: 3600,
   retries: {