fix: inject environment variables at dequeue time for self-hosted deployments

This commit fixes the issue where environment variables configured in the
Trigger.dev portal were not being injected into runner pods/containers in
self-hosted deployments (GitHub issue #2792).

Root Cause:
Environment variables were only being fetched during the 'startRunAttempt'
phase, which occurs AFTER the runner pod/container has already been created.
This meant that pods were created without access to portal-configured
environment variables, causing tasks to fail immediately.

Solution:
Move environment variable fetching from the 'startRunAttempt' phase to the
'dequeue' phase, so that env vars are available when the supervisor creates
the runner workload.

Changes:
1. Added optional 'envVars' field to DequeuedMessage schema
2. Modified AuthenticatedWorkerInstance.dequeue() to fetch and include
   environment variables for each dequeued message
3. Added 'envVars' field to WorkloadManagerCreateOptions interface
4. Updated supervisor to pass envVars when creating workloads
5. Modified KubernetesWorkloadManager to inject envVars into pod spec
6. Modified DockerWorkloadManager to inject envVars into container

Testing:
- Tested in self-hosted Kubernetes deployment (EKS)
- Verified environment variables appear in runner pods
- Confirmed tasks execute successfully with portal-configured env vars

Fixes #2792

Author: shawnfultz

apps/supervisor/src/index.ts

@@ -267,6 +267,7 @@ class ManagedSupervisor {
           snapshotId: message.snapshot.id,
           snapshotFriendlyId: message.snapshot.friendlyId,
           placementTags: message.placementTags,
+          envVars: message.envVars,
         });
 
         // Disabled for now

apps/supervisor/src/workloadManager/docker.ts

@@ -110,6 +110,12 @@ export class DockerWorkloadManager implements WorkloadManager {
       });
     }
 
+    if (opts.envVars) {
+      Object.entries(opts.envVars).forEach(([key, value]) => {
+        envVars.push(`${key}=${value}`);
+      });
+    }
+
     const hostConfig: Docker.HostConfig = {
       AutoRemove: !!this.opts.dockerAutoremove,
     };

apps/supervisor/src/workloadManager/kubernetes.ts

@@ -225,6 +225,12 @@ export class KubernetesWorkloadManager implements WorkloadManager {
                         value: value,
                       }))
                     : []),
+                  ...(opts.envVars
+                    ? Object.entries(opts.envVars).map(([key, value]) => ({
+                        name: key,
+                        value: value,
+                      }))
+                    : []),
                 ],
               },
             ],

apps/supervisor/src/workloadManager/types.ts

@@ -24,6 +24,7 @@ export interface WorkloadManagerCreateOptions {
   nextAttemptNumber?: number;
   dequeuedAt: Date;
   placementTags?: PlacementTag[];
+  envVars?: Record<string, string>;
   // identifiers
   envId: string;
   envType: EnvironmentType;

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

@@ -370,12 +370,44 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
   }
 
   async dequeue({ runnerId }: { runnerId?: string }): Promise<DequeuedMessage[]> {
-    return await this._engine.dequeueFromWorkerQueue({
+    const messages = await this._engine.dequeueFromWorkerQueue({
       consumerId: this.workerInstanceId,
       workerQueue: this.masterQueue,
       workerId: this.workerInstanceId,
       runnerId,
     });
+
+    // Fetch and inject environment variables for each message
+    const messagesWithEnvVars = await Promise.all(
+      messages.map(async (message) => {
+        const defaultMachinePreset = machinePresetFromName(defaultMachine);
+
+        const environment = await this._prisma.runtimeEnvironment.findFirst({
+          where: {
+            id: message.environment.id,
+          },
+          include: {
+            parentEnvironment: true,
+          },
+        });
+
+        const envVars = environment
+          ? await this.getEnvVars(
+              environment,
+              message.run.id,
+              message.run.machine ?? defaultMachinePreset,
+              environment.parentEnvironment ?? undefined
+            )
+          : {};
+
+        return {
+          ...message,
+          envVars,
+        };
+      })
+    );
+
+    return messagesWithEnvVars;
   }
 
   async heartbeatWorkerInstance() {

packages/core/src/v3/schemas/runEngine.ts

@@ -270,5 +270,6 @@ export const DequeuedMessage = z.object({
     id: z.string(),
   }),
   placementTags: z.array(PlacementTag).optional(),
+  envVars: z.record(z.string()).optional(),
 });
 export type DequeuedMessage = z.infer<typeof DequeuedMessage>;