fix(core): ensure task environments are processed properly in dte (#30862)

<!-- Please make sure you have read the submission guidelines before
posting an PR -->
<!--
https://github.com/nrwl/nx/blob/master/CONTRIBUTING.md#-submitting-a-pr
-->

<!-- Please make sure that your commit message follows our format -->
<!-- Example: `fix(nx): must begin with lowercase` -->

<!-- If this is a particularly complex change or feature addition, you
can request a dedicated Nx release for this pull request branch. Mention
someone from the Nx team or the `@nrwl/nx-pipelines-reviewers` and they
will confirm if the PR warrants its own release for testing purposes,
and generate it for you if appropriate. -->

## Current Behavior
<!-- This is the behavior we have today -->

The new DTE APIs are missing a few things related to task environments.
Firstly, even though it processes task envs.. they are not stored on the
`processedTasks` map. Thus, the task env is not actually used to run
tasks. Secondly, some environment variables are not set during dte..
which used to.

## Expected Behavior
<!-- This is the behavior we should expect with the changes in this PR
-->

The.2 things that are missing are now back. The map is populated with
the task envs and the environment variables are set based on the args.

## Related Issue(s)
<!-- Please link the issue being fixed so it gets closed when this is
merged. -->

Fixes #

packages/nx/src/tasks-runner/init-tasks-runner.ts

@@ -1,3 +1,4 @@
+import type { NxJsonConfiguration } from '../config/nx-json';
 import { readNxJson } from '../config/nx-json';
 import { NxArgs } from '../utils/command-line-utils';
 import { createProjectGraphAsync } from '../project-graph/project-graph';
@@ -6,6 +7,7 @@ import {
   constructLifeCycles,
   getRunner,
   invokeTasksRunner,
+  setEnvVarsBasedOnArgs,
 } from './run-command';
 import { InvokeRunnerTerminalOutputLifeCycle } from './life-cycles/invoke-runner-terminal-output-life-cycle';
 import { performance } from 'perf_hooks';
@@ -15,7 +17,6 @@ import { CompositeLifeCycle, LifeCycle, TaskResult } from './life-cycle';
 import { TaskOrchestrator } from './task-orchestrator';
 import { createTaskHasher } from '../hasher/create-task-hasher';
 import type { ProjectGraph } from '../config/project-graph';
-import type { NxJsonConfiguration } from '../config/nx-json';
 import { daemonClient } from '../daemon/client/client';
 import { RunningTask } from './running-tasks/running-task';
 import { TaskResultsLifeCycle } from './life-cycles/task-results-life-cycle';
@@ -133,6 +134,13 @@ async function createOrchestrator(
     }, {} as any),
   };
 
+  const nxArgs = {
+    ...options,
+    parallel: tasks.length,
+    lifeCycle: compositedLifeCycle,
+  };
+  setEnvVarsBasedOnArgs(nxArgs, true);
+
   const orchestrator = new TaskOrchestrator(
     hasher,
     null,
@@ -140,7 +148,7 @@ async function createOrchestrator(
     projectGraph,
     taskGraph,
     nxJson,
-    { ...options, parallel: tasks.length, lifeCycle: compositedLifeCycle },
+    nxArgs,
     false,
     daemonClient,
     undefined,
@@ -149,7 +157,7 @@ async function createOrchestrator(
 
   await orchestrator.init();
 
-  await Promise.all(tasks.map((task) => orchestrator.processTask(task.id)));
+  orchestrator.processTasks(tasks.map((task) => task.id));
 
   return orchestrator;
 }

packages/nx/src/tasks-runner/run-command.ts

@@ -802,7 +802,10 @@ async function confirmRunningTasksWithSyncFailures(): Promise<void> {
   }
 }
 
-function setEnvVarsBasedOnArgs(nxArgs: NxArgs, loadDotEnvFiles: boolean) {
+export function setEnvVarsBasedOnArgs(
+  nxArgs: NxArgs,
+  loadDotEnvFiles: boolean
+) {
   if (
     nxArgs.outputStyle == 'stream' ||
     process.env.NX_BATCH_MODE === 'true' ||

packages/nx/src/tasks-runner/task-orchestrator.ts

@@ -199,8 +199,17 @@ export class TaskOrchestrator {
     );
   }
 
+  processTasks(taskIds: string[]) {
+    for (const taskId of taskIds) {
+      // Task is already handled or being handled
+      if (!this.processedTasks.has(taskId)) {
+        this.processedTasks.set(taskId, this.processTask(taskId));
+      }
+    }
+  }
+
   // region Processing Scheduled Tasks
-  async processTask(taskId: string): Promise<NodeJS.ProcessEnv> {
+  private async processTask(taskId: string): Promise<NodeJS.ProcessEnv> {
     const task = this.taskGraph.tasks[taskId];
     const taskSpecificEnv = getTaskSpecificEnv(task);
 
@@ -245,12 +254,7 @@ export class TaskOrchestrator {
     for (const batch of scheduledBatches) {
       this.processedBatches.set(batch, this.processScheduledBatch(batch));
     }
-    for (const taskId of scheduledTasks) {
+    this.processTasks(scheduledTasks);
-      // Task is already handled or being handled
-      if (!this.processedTasks.has(taskId)) {
-        this.processedTasks.set(taskId, this.processTask(taskId));
-      }
-    }
   }
 
   // endregion Processing Scheduled Tasks