Add relative path translation for alert_task.workspace_path in job tasks (#4836)

## Summary
- Add `alert_task.workspace_path` to the job task path rewrite patterns
so relative paths (e.g. `./my_alert.dbalert.json`) are translated to
full workspace paths
- Covers both regular tasks and `for_each_task` nested tasks via
`TranslateModeFile`

## Test plan
- [x] Unit tests for path visitor (`TestVisitJobPaths`,
`TestVisitJobPaths_foreach`)
- [x] Unit test for path translation
(`TestTranslatePathsInSubdirectories`)
- [x] Acceptance test (`acceptance/bundle/resources/jobs/alert-task`) —
deploys a job with `alert_task.workspace_path`, reads it back via `jobs
get`, and asserts the fully qualified workspace path. Runs on both local
(mock server) and cloud, on both terraform and direct engines.

This pull request was AI-assisted by Isaac.

Author: shreyas-goenka

acceptance/bundle/resources/jobs/alert-task/databricks.yml.tmpl

@@ -0,0 +1,12 @@
+bundle:
+  name: alert-task-$UNIQUE_NAME
+
+resources:
+  jobs:
+    my_job:
+      name: alert-task-$UNIQUE_NAME
+      tasks:
+        - task_key: alert_task
+          alert_task:
+            workspace_path: ./my_alert.dbalert.json
+            warehouse_id: $TEST_DEFAULT_WAREHOUSE_ID

acceptance/bundle/resources/jobs/alert-task/my_alert.dbalert.json

@@ -0,0 +1,19 @@
+{
+  "query_lines": ["SELECT 1"],
+  "schedule": {
+    "quartz_cron_schedule": "0 0 * * * ?",
+    "timezone_id": "UTC"
+  },
+  "evaluation": {
+    "comparison_operator": "EQUAL",
+    "source": {
+      "name": "1",
+      "aggregation": "MAX"
+    },
+    "threshold": {
+      "value": {
+        "double_value": 1
+      }
+    }
+  }
+}

acceptance/bundle/resources/jobs/alert-task/out.test.toml

@@ -0,0 +1,21 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/alert-task-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] jobs get [JOB_ID]
+{
+  "warehouse_id": "[TEST_DEFAULT_WAREHOUSE_ID]",
+  "workspace_path": "/Workspace/Users/[USERNAME]/.bundle/alert-task-[UNIQUE_NAME]/default/files/my_alert.dbalert.json"
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/alert-task-[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/jobs/alert-task/output.txt

@@ -0,0 +1,13 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+job_id=$($CLI bundle summary -o json | jq -r '.resources.jobs.my_job.id')
+echo "$job_id:JOB_ID" >> ACC_REPLS
+
+trace $CLI jobs get $job_id | jq '.settings.tasks[0].alert_task'

acceptance/bundle/resources/jobs/alert-task/script

@@ -0,0 +1,7 @@
+Local = true
+Cloud = true
+RecordRequests = false
+Ignore = ["databricks.yml", ".databricks"]
+
+[Env]
+MSYS_NO_PATHCONV = "1"

acceptance/bundle/resources/jobs/alert-task/test.toml

@@ -36,6 +36,11 @@ func jobTaskRewritePatterns(base dyn.Pattern) []jobRewritePattern {
 			TranslateModeFile,
 			noSkipRewrite,
 		},
+		{
+			base.Append(dyn.Key("alert_task"), dyn.Key("workspace_path")),
+			TranslateModeFile,
+			noSkipRewrite,
+		},
 		{
 			base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("requirements")),
 			TranslateModeFile,

bundle/config/mutator/paths/job_paths_visitor.go

@@ -49,6 +49,11 @@ func TestVisitJobPaths(t *testing.T) {
 			{Requirements: "requirements.txt"},
 		},
 	}
+	task7 := jobs.Task{
+		AlertTask: &jobs.AlertTask{
+			WorkspacePath: "abc",
+		},
+	}
 
 	job0 := &resources.Job{
 		JobSettings: jobs.JobSettings{
@@ -60,6 +65,7 @@ func TestVisitJobPaths(t *testing.T) {
 				task4,
 				task5,
 				task6,
+				task7,
 			},
 		},
 	}
@@ -79,6 +85,7 @@ func TestVisitJobPaths(t *testing.T) {
 		dyn.MustPathFromString("resources.jobs.job0.tasks[2].dbt_task.project_directory"),
 		dyn.MustPathFromString("resources.jobs.job0.tasks[3].sql_task.file.path"),
 		dyn.MustPathFromString("resources.jobs.job0.tasks[6].libraries[0].requirements"),
+		dyn.MustPathFromString("resources.jobs.job0.tasks[7].alert_task.workspace_path"),
 	}
 
 	assert.ElementsMatch(t, expected, actual)
@@ -125,10 +132,20 @@ func TestVisitJobPaths_foreach(t *testing.T) {
 			},
 		},
 	}
+	task1 := jobs.Task{
+		ForEachTask: &jobs.ForEachTask{
+			Task: jobs.Task{
+				AlertTask: &jobs.AlertTask{
+					WorkspacePath: "abc",
+				},
+			},
+		},
+	}
 	job0 := &resources.Job{
 		JobSettings: jobs.JobSettings{
 			Tasks: []jobs.Task{
 				task0,
+				task1,
 			},
 		},
 	}
@@ -144,6 +161,7 @@ func TestVisitJobPaths_foreach(t *testing.T) {
 	actual := collectVisitedPaths(t, root, VisitJobPaths)
 	expected := []dyn.Path{
 		dyn.MustPathFromString("resources.jobs.job0.tasks[0].for_each_task.task.notebook_task.notebook_path"),
+		dyn.MustPathFromString("resources.jobs.job0.tasks[1].for_each_task.task.alert_task.workspace_path"),
 	}
 
 	assert.ElementsMatch(t, expected, actual)

bundle/config/mutator/paths/job_paths_visitor_test.go

@@ -290,6 +290,7 @@ func TestTranslatePathsInSubdirectories(t *testing.T) {
 	touchEmptyFile(t, filepath.Join(dir, "pipeline", "my_python_file.py"))
 	touchEmptyFile(t, filepath.Join(dir, "job", "my_sql_file.sql"))
 	touchEmptyFile(t, filepath.Join(dir, "job", "my_dbt_project", "dbt_project.yml"))
+	touchEmptyFile(t, filepath.Join(dir, "job", "my_alert.dbalert.json"))
 
 	b := &bundle.Bundle{
 		SyncRootPath:   dir,
@@ -329,6 +330,11 @@ func TestTranslatePathsInSubdirectories(t *testing.T) {
 										ProjectDirectory: "./my_dbt_project",
 									},
 								},
+								{
+									AlertTask: &jobs.AlertTask{
+										WorkspacePath: "./my_alert.dbalert.json",
+									},
+								},
 							},
 						},
 					},
@@ -376,6 +382,11 @@ func TestTranslatePathsInSubdirectories(t *testing.T) {
 		"/bundle/job/my_dbt_project",
 		b.Config.Resources.Jobs["job"].Tasks[3].DbtTask.ProjectDirectory,
 	)
+	assert.Equal(
+		t,
+		"/bundle/job/my_alert.dbalert.json",
+		b.Config.Resources.Jobs["job"].Tasks[4].AlertTask.WorkspacePath,
+	)
 
 	assert.Equal(
 		t,