diff --git a/cli/internal/fs/testdata/invalid-env-1/package.json b/cli/internal/fs/testdata/invalid-env-1/package.json
new file mode 100644
index 0000000000000..9d8f737ea004b
--- /dev/null
+++ b/cli/internal/fs/testdata/invalid-env-1/package.json
@@ -0,0 +1,3 @@
+{
+  "name": "test-repo"
+}
diff --git a/cli/internal/fs/testdata/invalid-env-1/turbo.json b/cli/internal/fs/testdata/invalid-env-1/turbo.json
new file mode 100644
index 0000000000000..e4a6517a00487
--- /dev/null
+++ b/cli/internal/fs/testdata/invalid-env-1/turbo.json
@@ -0,0 +1,8 @@
+{
+  "pipeline": {
+    "task1": {
+      // all invalid value
+      "env": ["$A", "$B"]
+    }
+  }
+}
diff --git a/cli/internal/fs/testdata/invalid-env-2/package.json b/cli/internal/fs/testdata/invalid-env-2/package.json
new file mode 100644
index 0000000000000..9d8f737ea004b
--- /dev/null
+++ b/cli/internal/fs/testdata/invalid-env-2/package.json
@@ -0,0 +1,3 @@
+{
+  "name": "test-repo"
+}
diff --git a/cli/internal/fs/testdata/invalid-env-2/turbo.json b/cli/internal/fs/testdata/invalid-env-2/turbo.json
new file mode 100644
index 0000000000000..92eec96f607a9
--- /dev/null
+++ b/cli/internal/fs/testdata/invalid-env-2/turbo.json
@@ -0,0 +1,8 @@
+{
+  "pipeline": {
+    "task1": {
+      // Mixed values
+      "env": ["$A", "B"]
+    }
+  }
+}
diff --git a/cli/internal/fs/testdata/legacy-env/package.json b/cli/internal/fs/testdata/legacy-env/package.json
new file mode 100644
index 0000000000000..9d8f737ea004b
--- /dev/null
+++ b/cli/internal/fs/testdata/legacy-env/package.json
@@ -0,0 +1,3 @@
+{
+  "name": "test-repo"
+}
diff --git a/cli/internal/fs/testdata/legacy-env/turbo.json b/cli/internal/fs/testdata/legacy-env/turbo.json
new file mode 100644
index 0000000000000..df9ca80d9db2e
--- /dev/null
+++ b/cli/internal/fs/testdata/legacy-env/turbo.json
@@ -0,0 +1,31 @@
+// mocked test comment
+{
+  "pipeline": {
+    // Only legacy declaration
+    "task1": {
+      "dependsOn": ["$A"]
+    },
+    // Only new declaration
+    "task2": {
+      "env": ["A"]
+    },
+    // Same var declared in both
+    "task3": {
+      "dependsOn": ["$A"],
+      "env": ["A"]
+    },
+    // Different vars declared in both
+    "task4": {
+      "dependsOn": ["$A"],
+      "env": ["B"]
+    },
+
+    // some edge cases
+    "task6": { "env": ["A", "B", "C"], "dependsOn": ["$D", "$E", "$F"] },
+    "task7": { "env": ["A", "B", "C"], "dependsOn": ["$A", "$B", "$C"] },
+    "task8": { "env": ["A", "B", "C"], "dependsOn": ["A", "B", "C"] },
+    "task9": { "env": [], "dependsOn": ["$A"]  },
+    "task10": { "env": ["A", "A"], "dependsOn": ["$A", "$A"] },
+    "task11": { "env": ["A", "A"], "dependsOn": ["$B", "$B"] }
+  }
+}
diff --git a/cli/internal/fs/turbo_json.go b/cli/internal/fs/turbo_json.go
index 27ebbecb5c0be..81b974a8023ea 100644
--- a/cli/internal/fs/turbo_json.go
+++ b/cli/internal/fs/turbo_json.go
@@ -43,6 +43,7 @@ type pipelineJSON struct {
 	DependsOn  []string            `json:"dependsOn,omitempty"`
 	Inputs     []string            `json:"inputs,omitempty"`
 	OutputMode util.TaskOutputMode `json:"outputMode,omitempty"`
+	Env        []string            `json:"env,omitempty"`
 }
 
 // Pipeline is a struct for deserializing .pipeline in configFile
@@ -108,7 +109,6 @@ func readTurboJSON(path turbopath.AbsolutePath) (*TurboJSON, error) {
 	}
 	err = jsonc.Unmarshal(data, &turboJSON)
 	if err != nil {
-		println("error unmarshalling", err.Error())
 		return nil, err
 	}
 	return turboJSON, nil
@@ -162,18 +162,33 @@ func (c *TaskDefinition) UnmarshalJSON(data []byte) error {
 	} else {
 		c.ShouldCache = *rawPipeline.Cache
 	}
-	c.EnvVarDependencies = []string{}
+
+	envVarDependencies := make(util.Set)
 	c.TopologicalDependencies = []string{}
 	c.TaskDependencies = []string{}
+
 	for _, dependency := range rawPipeline.DependsOn {
 		if strings.HasPrefix(dependency, envPipelineDelimiter) {
-			c.EnvVarDependencies = append(c.EnvVarDependencies, strings.TrimPrefix(dependency, envPipelineDelimiter))
+			envVarDependencies.Add(strings.TrimPrefix(dependency, envPipelineDelimiter))
 		} else if strings.HasPrefix(dependency, topologicalPipelineDelimiter) {
 			c.TopologicalDependencies = append(c.TopologicalDependencies, strings.TrimPrefix(dependency, topologicalPipelineDelimiter))
 		} else {
 			c.TaskDependencies = append(c.TaskDependencies, dependency)
 		}
 	}
+
+	// Append env key into EnvVarDependencies
+	for _, value := range rawPipeline.Env {
+		if strings.HasPrefix(value, envPipelineDelimiter) {
+			// Hard error to help people specify this correctly during migration.
+			// TODO: Remove this error after we have run summary.
+			return fmt.Errorf("You specified \"%s\" in the \"env\" key. You should not prefix your environment variables with \"$\"", value)
+		}
+
+		envVarDependencies.Add(value)
+	}
+
+	c.EnvVarDependencies = envVarDependencies.UnsafeListOfStrings()
 	c.Inputs = rawPipeline.Inputs
 	c.OutputMode = rawPipeline.OutputMode
 	return nil
diff --git a/cli/internal/fs/turbo_json_test.go b/cli/internal/fs/turbo_json_test.go
index 7255b304c62f0..f97c657fd805e 100644
--- a/cli/internal/fs/turbo_json_test.go
+++ b/cli/internal/fs/turbo_json_test.go
@@ -2,6 +2,7 @@ package fs
 
 import (
 	"os"
+	"sort"
 	"strings"
 	"testing"
 
@@ -134,6 +135,70 @@ func Test_ReadTurboConfig_BothCorrectAndLegacy(t *testing.T) {
 	assert.Equal(t, rootPackageJSON.LegacyTurboConfig == nil, true)
 }
 
+func Test_ReadTurboConfig_InvalidEnvDeclarations1(t *testing.T) {
+	testDir := getTestDir(t, "invalid-env-1")
+
+	packageJSONPath := testDir.Join("package.json")
+	rootPackageJSON, pkgJSONReadErr := ReadPackageJSON(packageJSONPath)
+
+	if pkgJSONReadErr != nil {
+		t.Fatalf("invalid parse: %#v", pkgJSONReadErr)
+	}
+
+	_, turboJSONReadErr := ReadTurboConfig(testDir, rootPackageJSON)
+
+	expectedErrorMsg := "turbo.json: You specified \"$A\" in the \"env\" key. You should not prefix your environment variables with \"$\""
+
+	assert.EqualErrorf(t, turboJSONReadErr, expectedErrorMsg, "Error should be: %v, got: %v", expectedErrorMsg, turboJSONReadErr)
+}
+
+func Test_ReadTurboConfig_InvalidEnvDeclarations2(t *testing.T) {
+	testDir := getTestDir(t, "invalid-env-2")
+
+	packageJSONPath := testDir.Join("package.json")
+	rootPackageJSON, pkgJSONReadErr := ReadPackageJSON(packageJSONPath)
+
+	if pkgJSONReadErr != nil {
+		t.Fatalf("invalid parse: %#v", pkgJSONReadErr)
+	}
+
+	_, turboJSONReadErr := ReadTurboConfig(testDir, rootPackageJSON)
+
+	expectedErrorMsg := "turbo.json: You specified \"$A\" in the \"env\" key. You should not prefix your environment variables with \"$\""
+
+	assert.EqualErrorf(t, turboJSONReadErr, expectedErrorMsg, "Error should be: %v, got: %v", expectedErrorMsg, turboJSONReadErr)
+}
+
+func Test_ReadTurboConfig_EnvDeclarations(t *testing.T) {
+	testDir := getTestDir(t, "legacy-env")
+
+	packageJSONPath := testDir.Join("package.json")
+	rootPackageJSON, pkgJSONReadErr := ReadPackageJSON(packageJSONPath)
+
+	if pkgJSONReadErr != nil {
+		t.Fatalf("invalid parse: %#v", pkgJSONReadErr)
+	}
+
+	turboJSON, turboJSONReadErr := ReadTurboConfig(testDir, rootPackageJSON)
+
+	if turboJSONReadErr != nil {
+		t.Fatalf("invalid parse: %#v", turboJSONReadErr)
+	}
+
+	pipeline := turboJSON.Pipeline
+
+	assert.EqualValues(t, sortedArray(pipeline["task1"].EnvVarDependencies), sortedArray([]string{"A"}))
+	assert.EqualValues(t, sortedArray(pipeline["task2"].EnvVarDependencies), sortedArray([]string{"A"}))
+	assert.EqualValues(t, sortedArray(pipeline["task3"].EnvVarDependencies), sortedArray([]string{"A"}))
+	assert.EqualValues(t, sortedArray(pipeline["task4"].EnvVarDependencies), sortedArray([]string{"A", "B"}))
+	assert.EqualValues(t, sortedArray(pipeline["task6"].EnvVarDependencies), sortedArray([]string{"A", "B", "C", "D", "E", "F"}))
+	assert.EqualValues(t, sortedArray(pipeline["task7"].EnvVarDependencies), sortedArray([]string{"A", "B", "C"}))
+	assert.EqualValues(t, sortedArray(pipeline["task8"].EnvVarDependencies), sortedArray([]string{"A", "B", "C"}))
+	assert.EqualValues(t, sortedArray(pipeline["task9"].EnvVarDependencies), sortedArray([]string{"A"}))
+	assert.EqualValues(t, sortedArray(pipeline["task10"].EnvVarDependencies), sortedArray([]string{"A"}))
+	assert.EqualValues(t, sortedArray(pipeline["task11"].EnvVarDependencies), sortedArray([]string{"A", "B"}))
+}
+
 // Helpers
 func validateOutput(t *testing.T, actual Pipeline, expected map[string]TaskDefinition) {
 	// check top level keys
@@ -172,3 +237,8 @@ func getTestDir(t *testing.T, testName string) turbopath.AbsolutePath {
 
 	return cwd.Join("testdata", testName)
 }
+
+func sortedArray(arr []string) []string {
+	sort.Strings(arr)
+	return arr
+}