Fix trampoline notebook filename collision (#5508)

## Why

Found during a full-repo review of the CLI. The python wheel wrapper
(trampoline) names its generated notebook
`notebook_<job_key>_<task_key>`. Both keys can contain underscores, so
distinct pairs can produce the same name: job `a_b` with task `c` and
job `a` with task `b_c` both map to `notebook_a_b_c`. The second wrapper
overwrites the first, and both tasks point at the same notebook, so one
task silently runs the other task's wrapper.

## Changes

Before, two different (job key, task key) pairs could collide on one
wrapper filename; now every pair gets a unique name. The name is now
`notebook_<job_key_length>_<job_key>_<task_key>` (in
`bundle/trampoline/trampoline.go`). The length prefix makes the encoding
unambiguous for any keys, with no new dependencies. The example above
becomes `notebook_3_a_b_c` vs `notebook_1_a_b_c`.

Existing wrapper notebooks get a new path on the next deploy. This only
affects bundles with `experimental.python_wheel_wrapper: true`, and the
wrapper is regenerated and the job spec updated on every deploy, so no
action is needed.

## Test plan

- [x] New unit test `TestGenerateTrampolineWithCollidingKeys` covering
the colliding pair, asserting two distinct wrapper files and notebook
paths (fails without the fix)
- [x] Updated existing expectations in `bundle/trampoline` tests for the
new naming
- [x] `go test ./bundle/trampoline/ -count=1` passes
- [x] `./task fmt-q`, `./task lint-q`, and `./task checks` pass

This pull request and its description were written by Isaac.

Author: simonfaltum

bundle/trampoline/conditional_transform_test.go

@@ -108,7 +108,7 @@ func TestTransformWithExperimentalSettingSetToTrue(t *testing.T) {
 	task := b.Config.Resources.Jobs["job1"].Tasks[0]
 	require.Nil(t, task.PythonWheelTask)
 	require.NotNil(t, task.NotebookTask)
-	require.Equal(t, "/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_job1_key1", task.NotebookTask.NotebookPath)
+	require.Equal(t, "/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_4_job1_key1", task.NotebookTask.NotebookPath)
 
 	require.Len(t, task.Libraries, 1)
 	require.Equal(t, "/Workspace/Users/test@test.com/bundle/dist/test.jar", task.Libraries[0].Jar)

bundle/trampoline/trampoline.go

@@ -59,7 +59,9 @@ func (m *trampoline) generateNotebookWrapper(ctx context.Context, b *bundle.Bund
 		return err
 	}
 
-	notebookName := fmt.Sprintf("notebook_%s_%s", task.JobKey, task.Task.TaskKey)
+	// Keys may contain underscores, so joining with "_" alone is ambiguous ("a_b"+"c"
+	// vs "a"+"b_c"); the job key length prefix keeps the filename unique per task.
+	notebookName := fmt.Sprintf("notebook_%d_%s_%s", len(task.JobKey), task.JobKey, task.Task.TaskKey)
 	localNotebookPath := filepath.Join(internalDir, notebookName+".py")
 
 	err = os.MkdirAll(filepath.Dir(localNotebookPath), 0o755)

bundle/trampoline/trampoline_test.go

@@ -17,11 +17,13 @@ type functions struct{}
 
 func (f *functions) GetTasks(b *bundle.Bundle) []TaskWithJobKey {
 	var tasks []TaskWithJobKey
-	for k := range b.Config.Resources.Jobs["test"].Tasks {
-		tasks = append(tasks, TaskWithJobKey{
-			JobKey: "test",
-			Task:   &b.Config.Resources.Jobs["test"].Tasks[k],
-		})
+	for k, job := range b.Config.Resources.Jobs {
+		for i := range job.Tasks {
+			tasks = append(tasks, TaskWithJobKey{
+				JobKey: k,
+				Task:   &job.Tasks[i],
+			})
+		}
 	}
 
 	return tasks
@@ -85,14 +87,74 @@ func TestGenerateTrampoline(t *testing.T) {
 
 	dir, err := b.InternalDir(ctx)
 	require.NoError(t, err)
-	filename := filepath.Join(dir, "notebook_test_to_trampoline.py")
+	filename := filepath.Join(dir, "notebook_4_test_to_trampoline.py")
 
 	bytes, err := os.ReadFile(filename)
 	require.NoError(t, err)
 
 	require.Equal(t, "Hello from Trampoline", string(bytes))
 
 	task := b.Config.Resources.Jobs["test"].Tasks[0]
-	require.Equal(t, "/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_test_to_trampoline", task.NotebookTask.NotebookPath)
+	require.Equal(t, "/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_4_test_to_trampoline", task.NotebookTask.NotebookPath)
 	require.Nil(t, task.PythonWheelTask)
 }
+
+func TestGenerateTrampolineWithCollidingKeys(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	newJob := func(taskKey string) *resources.Job {
+		return &resources.Job{
+			JobSettings: jobs.JobSettings{
+				Tasks: []jobs.Task{
+					{
+						TaskKey: taskKey,
+						PythonWheelTask: &jobs.PythonWheelTask{
+							PackageName: "test",
+							EntryPoint:  "run",
+						},
+					},
+				},
+			},
+		}
+	}
+
+	b := &bundle.Bundle{
+		BundleRootPath: filepath.Join(tmpDir, "parent", "my_bundle"),
+		SyncRootPath:   filepath.Join(tmpDir, "parent"),
+		Config: config.Root{
+			Workspace: config.Workspace{
+				FilePath: "/Workspace/files",
+			},
+			Bundle: config.Bundle{
+				Target: "development",
+			},
+			Resources: config.Resources{
+				Jobs: map[string]*resources.Job{
+					"a_b": newJob("c"),
+					"a":   newJob("b_c"),
+				},
+			},
+		},
+	}
+	ctx := t.Context()
+
+	funcs := functions{}
+	trampoline := NewTrampoline("test_trampoline", &funcs, "Hello from {{.MyName}}")
+	diags := bundle.Apply(ctx, b, trampoline)
+	require.NoError(t, diags.Error())
+
+	dir, err := b.InternalDir(ctx)
+	require.NoError(t, err)
+
+	for _, name := range []string{"notebook_3_a_b_c", "notebook_1_a_b_c"} {
+		_, err := os.Stat(filepath.Join(dir, name+".py"))
+		require.NoError(t, err)
+	}
+
+	require.Equal(t,
+		"/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_3_a_b_c",
+		b.Config.Resources.Jobs["a_b"].Tasks[0].NotebookTask.NotebookPath)
+	require.Equal(t,
+		"/Workspace/files/my_bundle/.databricks/bundle/development/.internal/notebook_1_a_b_c",
+		b.Config.Resources.Jobs["a"].Tasks[0].NotebookTask.NotebookPath)
+}