direct: treat tasks as a map identified by task_key

acceptance/bundle/artifacts/whl_dynamic/out.plan_update.direct.json

@@ -150,10 +150,10 @@
           "environments[0].spec.dependencies[0]": {
             "action": "update"
           },
-          "tasks[1].for_each_task.task.libraries[0].whl": {
+          "tasks[task_key='TestTask'].for_each_task.task.libraries[0].whl": {
             "action": "update"
           },
-          "tasks[1].libraries[0].whl": {
+          "tasks[task_key='TestTask'].libraries[0].whl": {
             "action": "update"
           }
         },

acceptance/bundle/migrate/basic/out.plan_update.json

@@ -67,7 +67,7 @@
             "action": "skip",
             "reason": "server_side_default"
           },
-          "tasks[0].notebook_task.source": {
+          "tasks[task_key='main'].notebook_task.source": {
             "action": "skip",
             "reason": "server_side_default"
           },

acceptance/bundle/resources/jobs/delete_task/out.plan_update.direct.json

@@ -96,7 +96,7 @@
       },
       "changes": {
         "local": {
-          "tasks": {
+          "tasks[task_key='TestTask1']": {
             "action": "update"
           }
         },

acceptance/bundle/resources/jobs/tasks-reorder-locally/databricks.yml

@@ -0,0 +1,13 @@
+bundle:
+  name: test-bundle
+
+resources:
+  jobs:
+    foo:
+      tasks:
+        - task_key: task_bb
+          notebook_task:
+            notebook_path: task_bb.py
+        - task_key: task_aa
+          notebook_task:
+            notebook_path: task_aa.py

acceptance/bundle/resources/jobs/tasks-reorder-locally/foo.py

@@ -0,0 +1 @@
+# Databricks notebook source

acceptance/bundle/resources/jobs/tasks-reorder-locally/output.txt

@@ -0,0 +1,26 @@
+
+>>> [CLI] bundle plan
+create jobs.foo
+
+Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged

acceptance/bundle/resources/jobs/tasks-reorder-locally/script

@@ -0,0 +1,12 @@
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace $CLI bundle plan | contains.py "0 to add, 0 to change, 0 to delete"
+
+update_file.py databricks.yml task_aa task_XX
+update_file.py databricks.yml task_bb task_aa
+update_file.py databricks.yml task_XX task_bb
+
+trace $CLI bundle plan
+
+trace $CLI bundle deploy
+trace $CLI bundle plan | contains.py "0 to add, 0 to change, 0 to delete"

acceptance/bundle/resources/jobs/tasks-reorder-locally/task_aa.py

@@ -0,0 +1 @@
+# Databricks notebook source

acceptance/bundle/resources/jobs/tasks-reorder-locally/task_bb.py

@@ -0,0 +1 @@
+# Databricks notebook source

acceptance/bundle/resources/jobs/tasks-reorder-locally/test.toml

@@ -0,0 +1,2 @@
+RecordRequests = false
+Ignore = [".databricks"]

bundle/direct/dresources/job.go

@@ -29,6 +29,16 @@ func (*ResourceJob) RemapState(jobs *jobs.Job) *jobs.JobSettings {
 	return jobs.Settings
 }
 
+func getTaskKey(x jobs.Task) (string, string) {
+	return "task_key", x.TaskKey
+}
+
+func (*ResourceJob) KeyedSlices() map[string]any {
+	return map[string]any{
+		"tasks": getTaskKey,
+	}
+}
+
 func (r *ResourceJob) DoRead(ctx context.Context, id string) (*jobs.Job, error) {
 	idInt, err := parseJobID(id)
 	if err != nil {