add support to prod v4 and v3 runs

Author: ericallam

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -22,6 +22,7 @@ import {
   TaskRunSuccessfulExecutionResult,
   parsePacket,
   serverWebsocketMessages,
+  SemanticInternalAttributes,
 } from "@trigger.dev/core/v3";
 import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
 import {
@@ -1648,6 +1649,7 @@ export const AttemptForExecutionGetPayload = {
         baseCostInCents: true,
         maxDurationInSeconds: true,
         tags: true,
+        taskEventStore: true,
       },
     },
     queue: {
@@ -1880,7 +1882,8 @@ class SharedQueueTasks {
     const variables = await this.#buildEnvironmentVariables(
       attempt.runtimeEnvironment,
       taskRun.id,
-      machinePreset
+      machinePreset,
+      taskRun.taskEventStore ?? undefined
     );
 
     const payload: V3ProdTaskRunExecutionPayload = {
@@ -2049,6 +2052,7 @@ class SharedQueueTasks {
           },
         },
         machinePreset: true,
+        taskEventStore: true,
       },
     });
 
@@ -2071,7 +2075,12 @@ class SharedQueueTasks {
     const machinePreset =
       machinePresetFromRun(run) ?? machinePresetFromConfig(run.lockedBy?.machineConfig ?? {});
 
-    const variables = await this.#buildEnvironmentVariables(environment, run.id, machinePreset);
+    const variables = await this.#buildEnvironmentVariables(
+      environment,
+      run.id,
+      machinePreset,
+      run.taskEventStore ?? undefined
+    );
 
     return {
       traceContext: run.traceContext as Record<string, unknown>,
@@ -2178,7 +2187,8 @@ class SharedQueueTasks {
   async #buildEnvironmentVariables(
     environment: RuntimeEnvironmentForEnvRepo,
     runId: string,
-    machinePreset: MachinePreset
+    machinePreset: MachinePreset,
+    taskEventStore?: string
   ): Promise<Array<EnvironmentVariable>> {
     const variables = await resolveVariablesForEnvironment(environment);
 
@@ -2187,6 +2197,14 @@ class SharedQueueTasks {
       machine_preset: machinePreset.name,
     });
 
+    if (taskEventStore) {
+      const resourceAttributes = JSON.stringify({
+        [SemanticInternalAttributes.TASK_EVENT_STORE]: taskEventStore,
+      });
+
+      variables.push(...[{ key: "OTEL_RESOURCE_ATTRIBUTES", value: resourceAttributes }]);
+    }
+
     return [
       ...variables,
       ...[

apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts

@@ -5,6 +5,7 @@ import {
   DequeuedMessage,
   ExecutionResult,
   MachinePreset,
+  SemanticInternalAttributes,
   StartRunAttemptResult,
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
@@ -444,7 +445,8 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
           environment,
           engineResult.run.id,
           engineResult.execution.machine ?? defaultMachinePreset,
-          environment.parentEnvironment ?? undefined
+          environment.parentEnvironment ?? undefined,
+          engineResult.run.taskEventStore ?? undefined
         )
       : {};
 
@@ -544,7 +546,8 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     environment: RuntimeEnvironment,
     runId: string,
     machinePreset: MachinePreset,
-    parentEnvironment?: RuntimeEnvironment
+    parentEnvironment?: RuntimeEnvironment,
+    taskEventStore?: string
   ): Promise<Record<string, string>> {
     const variables = await resolveVariablesForEnvironment(environment, parentEnvironment);
 
@@ -561,6 +564,19 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
       ]
     );
 
+    if (taskEventStore) {
+      const resourceAttributes = JSON.stringify({
+        [SemanticInternalAttributes.TASK_EVENT_STORE]: taskEventStore,
+      });
+
+      variables.push(
+        ...[
+          { key: "OTEL_RESOURCE_ATTRIBUTES", value: resourceAttributes },
+          { key: "TRIGGER_OTEL_RESOURCE_ATTRIBUTES", value: resourceAttributes },
+        ]
+      );
+    }
+
     return variables.reduce((acc: Record<string, string>, curr) => {
       acc[curr.key] = curr.value;
       return acc;