Allow artifact path to be located outside the sync root (#2128)

## Changes

We perform a check during path translation that the path being
referenced is contained in the bundle's sync root. If it isn't, it's not
a valid remote reference. However, this doesn't apply to paths that are
_always_ local, such as the artifact path. An artifact's build command
is executed in its path. Files created by the artifact build (e.g.
wheels or JARs) don't need to be in the sync root because they have a
dedicated and different upload path into `${workspace.artifact_path}`.

Therefore, this check that a path is contained in the bundle's sync root
doesn't apply to artifact paths. This change modifies the structure of
path translation to allow opting out of this check.

Fixes #1927.

## Tests

* Existing and new tests pass.
* Manually confirmed that building and using a wheel built outside the
sync root path works as expected.
* No acceptance tests because we don't run build as part of validate.

bundle/config/mutator/translate_paths.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io/fs"
 	"net/url"
+	"os"
 	"path"
 	"path/filepath"
 	"strings"
@@ -17,6 +18,47 @@ import (
 	"github.com/databricks/cli/libs/notebook"
 )
 
+// TranslateMode specifies how a path should be translated.
+type TranslateMode int
+
+const (
+	// TranslateModeNotebook translates a path to a remote notebook.
+	TranslateModeNotebook TranslateMode = iota
+
+	// TranslateModeFile translates a path to a remote regular file.
+	TranslateModeFile
+
+	// TranslateModeDirectory translates a path to a remote directory.
+	TranslateModeDirectory
+
+	// TranslateModeLocalAbsoluteFile translates a path to the local absolute file path.
+	// It returns an error if the path does not exist or is a directory.
+	TranslateModeLocalAbsoluteFile
+
+	// TranslateModeLocalAbsoluteDirectory translates a path to the local absolute directory path.
+	// It returns an error if the path does not exist or is not a directory.
+	TranslateModeLocalAbsoluteDirectory
+
+	// TranslateModeLocalRelative translates a path to be relative to the bundle sync root path.
+	// It does not check if the path exists, nor care if it is a file or directory.
+	TranslateModeLocalRelative
+
+	// TranslateModeLocalRelativeWithPrefix translates a path to be relative to the bundle sync root path.
+	// It a "./" prefix to the path if it does not already have one.
+	// This allows for disambiguating between paths and PyPI package names.
+	TranslateModeLocalRelativeWithPrefix
+)
+
+// translateOptions control path translation behavior.
+type translateOptions struct {
+	// Mode specifies how the path should be translated.
+	Mode TranslateMode
+
+	// AllowPathOutsideSyncRoot can be set for paths that are not tied to the sync root path.
+	// This is the case for artifact paths, for example.
+	AllowPathOutsideSyncRoot bool
+}
+
 type ErrIsNotebook struct {
 	path string
 }
@@ -44,8 +86,6 @@ func (m *translatePaths) Name() string {
 	return "TranslatePaths"
 }
 
-type rewriteFunc func(literal, localFullPath, localRelPath, remotePath string) (string, error)
-
 // translateContext is a context for rewriting paths in a config.
 // It is freshly instantiated on every mutator apply call.
 // It provides access to the underlying bundle object such that
@@ -56,74 +96,90 @@ type translateContext struct {
 	// seen is a map of local paths to their corresponding remote paths.
 	// If a local path has already been successfully resolved, we do not need to resolve it again.
 	seen map[string]string
+
+	// remoteRoot is the root path of the remote workspace.
+	// It is equal to ${workspace.file_path} for regular deployments.
+	// It points to the source root path for source-linked deployments.
+	remoteRoot string
 }
 
 // rewritePath converts a given relative path from the loaded config to a new path based on the passed rewriting function
 //
 // It takes these arguments:
-//   - The argument `dir` is the directory relative to which the given relative path is.
-//   - The given relative path is both passed and written back through `*p`.
-//   - The argument `fn` is a function that performs the actual rewriting logic.
-//     This logic is different between regular files or notebooks.
+//   - The context in which the function is called.
+//   - The argument `dir` is the directory relative to which the relative path should be interpreted.
+//   - The argument `input` is the relative path to rewrite.
+//   - The argument `opts` is a struct that specifies how the path should be rewritten.
+//     It contains a `Mode` field that specifies how the path should be rewritten.
 //
-// The function returns an error if it is impossible to rewrite the given relative path.
+// The function returns the rewritten path if successful, or an error if the path could not be rewritten.
+// The returned path is an empty string if the path was not rewritten.
 func (t *translateContext) rewritePath(
+	ctx context.Context,
 	dir string,
-	p *string,
-	fn rewriteFunc,
-) error {
+	input string,
+	opts translateOptions,
+) (string, error) {
 	// We assume absolute paths point to a location in the workspace
-	if path.IsAbs(*p) {
-		return nil
+	if path.IsAbs(input) {
+		return "", nil
 	}
 
-	url, err := url.Parse(*p)
+	url, err := url.Parse(input)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	// If the file path has scheme, it's a full path and we don't need to transform it
 	if url.Scheme != "" {
-		return nil
+		return "", nil
 	}
 
 	// Local path is relative to the directory the resource was defined in.
-	localPath := filepath.Join(dir, filepath.FromSlash(*p))
+	localPath := filepath.Join(dir, filepath.FromSlash(input))
 	if interp, ok := t.seen[localPath]; ok {
-		*p = interp
-		return nil
+		return interp, nil
 	}
 
 	// Local path must be contained in the sync root.
 	// If it isn't, it won't be synchronized into the workspace.
 	localRelPath, err := filepath.Rel(t.b.SyncRootPath, localPath)
 	if err != nil {
-		return err
+		return "", err
 	}
-	if strings.HasPrefix(localRelPath, "..") {
-		return fmt.Errorf("path %s is not contained in sync root path", localPath)
+	if !opts.AllowPathOutsideSyncRoot && !filepath.IsLocal(localRelPath) {
+		return "", fmt.Errorf("path %s is not contained in sync root path", localPath)
 	}
 
-	var workspacePath string
-	if config.IsExplicitlyEnabled(t.b.Config.Presets.SourceLinkedDeployment) {
-		workspacePath = t.b.SyncRootPath
-	} else {
-		workspacePath = t.b.Config.Workspace.FilePath
-	}
-	remotePath := path.Join(workspacePath, filepath.ToSlash(localRelPath))
-
 	// Convert local path into workspace path via specified function.
-	interp, err := fn(*p, localPath, localRelPath, remotePath)
+	var interp string
+	switch opts.Mode {
+	case TranslateModeNotebook:
+		interp, err = t.translateNotebookPath(ctx, input, localPath, localRelPath)
+	case TranslateModeFile:
+		interp, err = t.translateFilePath(ctx, input, localPath, localRelPath)
+	case TranslateModeDirectory:
+		interp, err = t.translateDirectoryPath(ctx, input, localPath, localRelPath)
+	case TranslateModeLocalAbsoluteFile:
+		interp, err = t.translateLocalAbsoluteFilePath(ctx, input, localPath, localRelPath)
+	case TranslateModeLocalAbsoluteDirectory:
+		interp, err = t.translateLocalAbsoluteDirectoryPath(ctx, input, localPath, localRelPath)
+	case TranslateModeLocalRelative:
+		interp, err = t.translateLocalRelativePath(ctx, input, localPath, localRelPath)
+	case TranslateModeLocalRelativeWithPrefix:
+		interp, err = t.translateLocalRelativeWithPrefixPath(ctx, input, localPath, localRelPath)
+	default:
+		return "", fmt.Errorf("unsupported translate mode: %d", opts.Mode)
+	}
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	*p = interp
 	t.seen[localPath] = interp
-	return nil
+	return interp, nil
 }
 
-func (t *translateContext) translateNotebookPath(literal, localFullPath, localRelPath, remotePath string) (string, error) {
+func (t *translateContext) translateNotebookPath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
 	nb, _, err := notebook.DetectWithFS(t.b.SyncRoot, filepath.ToSlash(localRelPath))
 	if errors.Is(err, fs.ErrNotExist) {
 		if filepath.Ext(localFullPath) != notebook.ExtensionNone {
@@ -162,10 +218,11 @@ to contain one of the following file extensions: [%s]`, literal, strings.Join(ex
 	}
 
 	// Upon import, notebooks are stripped of their extension.
-	return strings.TrimSuffix(remotePath, filepath.Ext(localFullPath)), nil
+	localRelPathNoExt := strings.TrimSuffix(localRelPath, filepath.Ext(localRelPath))
+	return path.Join(t.remoteRoot, filepath.ToSlash(localRelPathNoExt)), nil
 }
 
-func (t *translateContext) translateFilePath(literal, localFullPath, localRelPath, remotePath string) (string, error) {
+func (t *translateContext) translateFilePath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
 	nb, _, err := notebook.DetectWithFS(t.b.SyncRoot, filepath.ToSlash(localRelPath))
 	if errors.Is(err, fs.ErrNotExist) {
 		return "", fmt.Errorf("file %s not found", literal)
@@ -176,10 +233,10 @@ func (t *translateContext) translateFilePath(literal, localFullPath, localRelPat
 	if nb {
 		return "", ErrIsNotebook{localFullPath}
 	}
-	return remotePath, nil
+	return path.Join(t.remoteRoot, filepath.ToSlash(localRelPath)), nil
 }
 
-func (t *translateContext) translateDirectoryPath(literal, localFullPath, localRelPath, remotePath string) (string, error) {
+func (t *translateContext) translateDirectoryPath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
 	info, err := t.b.SyncRoot.Stat(filepath.ToSlash(localRelPath))
 	if err != nil {
 		return "", err
@@ -187,14 +244,10 @@ func (t *translateContext) translateDirectoryPath(literal, localFullPath, localR
 	if !info.IsDir() {
 		return "", fmt.Errorf("%s is not a directory", localFullPath)
 	}
-	return remotePath, nil
+	return path.Join(t.remoteRoot, filepath.ToSlash(localRelPath)), nil
 }
 
-func (t *translateContext) translateNoOp(literal, localFullPath, localRelPath, remotePath string) (string, error) {
-	return localRelPath, nil
-}
-
-func (t *translateContext) retainLocalAbsoluteFilePath(literal, localFullPath, localRelPath, remotePath string) (string, error) {
+func (t *translateContext) translateLocalAbsoluteFilePath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
 	info, err := t.b.SyncRoot.Stat(filepath.ToSlash(localRelPath))
 	if errors.Is(err, fs.ErrNotExist) {
 		return "", fmt.Errorf("file %s not found", literal)
@@ -208,16 +261,33 @@ func (t *translateContext) retainLocalAbsoluteFilePath(literal, localFullPath, l
 	return localFullPath, nil
 }
 
-func (t *translateContext) translateNoOpWithPrefix(literal, localFullPath, localRelPath, remotePath string) (string, error) {
+func (t *translateContext) translateLocalAbsoluteDirectoryPath(ctx context.Context, literal, localFullPath, _ string) (string, error) {
+	info, err := os.Stat(localFullPath)
+	if errors.Is(err, fs.ErrNotExist) {
+		return "", fmt.Errorf("directory %s not found", literal)
+	}
+	if err != nil {
+		return "", fmt.Errorf("unable to determine if %s is a directory: %w", localFullPath, err)
+	}
+	if !info.IsDir() {
+		return "", fmt.Errorf("expected %s to be a directory but found a file", literal)
+	}
+	return localFullPath, nil
+}
+
+func (t *translateContext) translateLocalRelativePath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
+	return localRelPath, nil
+}
+
+func (t *translateContext) translateLocalRelativeWithPrefixPath(ctx context.Context, literal, localFullPath, localRelPath string) (string, error) {
 	if !strings.HasPrefix(localRelPath, ".") {
 		localRelPath = "." + string(filepath.Separator) + localRelPath
 	}
 	return localRelPath, nil
 }
 
-func (t *translateContext) rewriteValue(p dyn.Path, v dyn.Value, fn rewriteFunc, dir string) (dyn.Value, error) {
-	out := v.MustString()
-	err := t.rewritePath(dir, &out, fn)
+func (t *translateContext) rewriteValue(ctx context.Context, p dyn.Path, v dyn.Value, dir string, opts translateOptions) (dyn.Value, error) {
+	out, err := t.rewritePath(ctx, dir, v.MustString(), opts)
 	if err != nil {
 		if target := (&ErrIsNotebook{}); errors.As(err, target) {
 			return dyn.InvalidValue, fmt.Errorf(`expected a file for "%s" but got a notebook: %w`, p, target)
@@ -228,43 +298,38 @@ func (t *translateContext) rewriteValue(p dyn.Path, v dyn.Value, fn rewriteFunc,
 		return dyn.InvalidValue, err
 	}
 
+	// If the path was not rewritten, return the original value.
+	if out == "" {
+		return v, nil
+	}
+
 	return dyn.NewValue(out, v.Locations()), nil
 }
 
-func (t *translateContext) rewriteRelativeTo(p dyn.Path, v dyn.Value, fn rewriteFunc, dir, fallback string) (dyn.Value, error) {
-	nv, err := t.rewriteValue(p, v, fn, dir)
-	if err == nil {
-		return nv, nil
-	}
-
-	// If we failed to rewrite the path, try to rewrite it relative to the fallback directory.
-	if fallback != "" {
-		nv, nerr := t.rewriteValue(p, v, fn, fallback)
-		if nerr == nil {
-			// TODO: Emit a warning that this path should be rewritten.
-			return nv, nil
-		}
-	}
-
-	return dyn.InvalidValue, err
-}
-
-func (m *translatePaths) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+func (m *translatePaths) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	t := &translateContext{
 		b:    b,
 		seen: make(map[string]string),
 	}
 
+	// Set the remote root to the sync root if source-linked deployment is enabled.
+	// Otherwise, set it to the workspace file path.
+	if config.IsExplicitlyEnabled(t.b.Config.Presets.SourceLinkedDeployment) {
+		t.remoteRoot = t.b.SyncRootPath
+	} else {
+		t.remoteRoot = t.b.Config.Workspace.FilePath
+	}
+
 	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
 		var err error
-		for _, fn := range []func(dyn.Value) (dyn.Value, error){
+		for _, fn := range []func(context.Context, dyn.Value) (dyn.Value, error){
 			t.applyJobTranslations,
 			t.applyPipelineTranslations,
 			t.applyArtifactTranslations,
 			t.applyDashboardTranslations,
 			t.applyAppsTranslations,
 		} {
-			v, err = fn(v)
+			v, err = fn(ctx, v)
 			if err != nil {
 				return dyn.InvalidValue, err
 			}
@@ -275,6 +340,8 @@ func (m *translatePaths) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnos
 	return diag.FromErr(err)
 }
 
+// gatherFallbackPaths collects the fallback paths for relative paths in the configuration.
+// Read more about the motivation for this functionality in the "fallback" path translation tests.
 func gatherFallbackPaths(v dyn.Value, typ string) (map[string]string, error) {
 	fallback := make(map[string]string)
 	pattern := dyn.NewPattern(dyn.Key("resources"), dyn.Key(typ), dyn.AnyKey())

bundle/config/mutator/translate_paths_apps.go

@@ -1,12 +1,13 @@
 package mutator
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/databricks/cli/libs/dyn"
 )
 
-func (t *translateContext) applyAppsTranslations(v dyn.Value) (dyn.Value, error) {
+func (t *translateContext) applyAppsTranslations(ctx context.Context, v dyn.Value) (dyn.Value, error) {
 	// Convert the `source_code_path` field to a remote absolute path.
 	// We use this path for app deployment to point to the source code.
 	pattern := dyn.NewPattern(
@@ -16,6 +17,10 @@ func (t *translateContext) applyAppsTranslations(v dyn.Value) (dyn.Value, error)
 		dyn.Key("source_code_path"),
 	)
 
+	opts := translateOptions{
+		Mode: TranslateModeDirectory,
+	}
+
 	return dyn.MapByPattern(v, pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
 		key := p[2].Key()
 		dir, err := v.Location().Directory()
@@ -23,6 +28,6 @@ func (t *translateContext) applyAppsTranslations(v dyn.Value) (dyn.Value, error)
 			return dyn.InvalidValue, fmt.Errorf("unable to determine directory for app %s: %w", key, err)
 		}
 
-		return t.rewriteRelativeTo(p, v, t.translateDirectoryPath, dir, "")
+		return t.rewriteValue(ctx, p, v, dir, opts)
 	})
 }

bundle/config/mutator/translate_paths_artifacts.go

@@ -1,6 +1,7 @@
 package mutator
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/databricks/cli/libs/dyn"
@@ -8,7 +9,7 @@ import (
 
 type artifactRewritePattern struct {
 	pattern dyn.Pattern
-	fn      rewriteFunc
+	opts    translateOptions
 }
 
 func (t *translateContext) artifactRewritePatterns() []artifactRewritePattern {
@@ -22,12 +23,18 @@ func (t *translateContext) artifactRewritePatterns() []artifactRewritePattern {
 	return []artifactRewritePattern{
 		{
 			base.Append(dyn.Key("path")),
-			t.translateNoOp,
+			translateOptions{
+				Mode: TranslateModeLocalAbsoluteDirectory,
+
+				// Artifact paths may be outside the sync root.
+				// They are the working directory for artifact builds.
+				AllowPathOutsideSyncRoot: true,
+			},
 		},
 	}
 }
 
-func (t *translateContext) applyArtifactTranslations(v dyn.Value) (dyn.Value, error) {
+func (t *translateContext) applyArtifactTranslations(ctx context.Context, v dyn.Value) (dyn.Value, error) {
 	var err error
 
 	for _, rewritePattern := range t.artifactRewritePatterns() {
@@ -38,7 +45,7 @@ func (t *translateContext) applyArtifactTranslations(v dyn.Value) (dyn.Value, er
 				return dyn.InvalidValue, fmt.Errorf("unable to determine directory for artifact %s: %w", key, err)
 			}
 
-			return t.rewriteRelativeTo(p, v, rewritePattern.fn, dir, "")
+			return t.rewriteValue(ctx, p, v, dir, rewritePattern.opts)
 		})
 		if err != nil {
 			return dyn.InvalidValue, err

bundle/config/mutator/translate_paths_artifacts_test.go

@@ -0,0 +1,83 @@
+package mutator_test
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/mutator"
+	"github.com/databricks/cli/bundle/internal/bundletest"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/vfs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTranslatePathsArtifacts_InsideSyncRoot(t *testing.T) {
+	tmp := t.TempDir()
+	dir := filepath.Join(tmp, "bundle")
+	lib := filepath.Join(dir, "my_lib")
+	_ = os.MkdirAll(lib, 0o755)
+	_ = os.MkdirAll(dir, 0o755)
+
+	b := &bundle.Bundle{
+		SyncRootPath: dir,
+		SyncRoot:     vfs.MustNew(dir),
+		Config: config.Root{
+			Artifacts: map[string]*config.Artifact{
+				"my_artifact": {
+					Type: "wheel",
+
+					// Assume this is defined in a subdir to the sync root.
+					Path: "../my_lib",
+				},
+			},
+		},
+	}
+
+	bundletest.SetLocation(b, "artifacts", []dyn.Location{{
+		File: filepath.Join(dir, "config/artifacts.yml"),
+	}})
+
+	diags := bundle.Apply(context.Background(), b, mutator.TranslatePaths())
+	require.NoError(t, diags.Error())
+
+	// Assert that the artifact path has been converted to a local absolute path.
+	assert.Equal(t, lib, b.Config.Artifacts["my_artifact"].Path)
+}
+
+func TestTranslatePathsArtifacts_OutsideSyncRoot(t *testing.T) {
+	tmp := t.TempDir()
+	lib := filepath.Join(tmp, "my_lib")
+	dir := filepath.Join(tmp, "bundle")
+	_ = os.MkdirAll(lib, 0o755)
+	_ = os.MkdirAll(dir, 0o755)
+
+	b := &bundle.Bundle{
+		SyncRootPath: dir,
+		SyncRoot:     vfs.MustNew(dir),
+		Config: config.Root{
+			Artifacts: map[string]*config.Artifact{
+				"my_artifact": {
+					Type: "wheel",
+
+					// Assume this is defined in a subdir of the bundle root.
+					Path: "../../my_lib",
+				},
+			},
+		},
+	}
+
+	bundletest.SetLocation(b, "artifacts", []dyn.Location{{
+		File: filepath.Join(dir, "config/artifacts.yml"),
+	}})
+
+	diags := bundle.Apply(context.Background(), b, mutator.TranslatePaths())
+	require.NoError(t, diags.Error())
+
+	// Assert that the artifact path has been converted to a local absolute path.
+	assert.Equal(t, lib, b.Config.Artifacts["my_artifact"].Path)
+}

bundle/config/mutator/translate_paths_dashboards.go

@@ -1,12 +1,13 @@
 package mutator
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/databricks/cli/libs/dyn"
 )
 
-func (t *translateContext) applyDashboardTranslations(v dyn.Value) (dyn.Value, error) {
+func (t *translateContext) applyDashboardTranslations(ctx context.Context, v dyn.Value) (dyn.Value, error) {
 	// Convert the `file_path` field to a local absolute path.
 	// We load the file at this path and use its contents for the dashboard contents.
 	pattern := dyn.NewPattern(
@@ -16,6 +17,10 @@ func (t *translateContext) applyDashboardTranslations(v dyn.Value) (dyn.Value, e
 		dyn.Key("file_path"),
 	)
 
+	opts := translateOptions{
+		Mode: TranslateModeLocalAbsoluteFile,
+	}
+
 	return dyn.MapByPattern(v, pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
 		key := p[2].Key()
 		dir, err := v.Location().Directory()
@@ -23,6 +28,6 @@ func (t *translateContext) applyDashboardTranslations(v dyn.Value) (dyn.Value, e
 			return dyn.InvalidValue, fmt.Errorf("unable to determine directory for dashboard %s: %w", key, err)
 		}
 
-		return t.rewriteRelativeTo(p, v, t.retainLocalAbsoluteFilePath, dir, "")
+		return t.rewriteValue(ctx, p, v, dir, opts)
 	})
 }

bundle/config/mutator/translate_paths_jobs.go

@@ -1,6 +1,7 @@
 package mutator
 
 import (
+	"context"
 	"fmt"
 	"slices"
 
@@ -9,7 +10,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 )
 
-func (t *translateContext) applyJobTranslations(v dyn.Value) (dyn.Value, error) {
+func (t *translateContext) applyJobTranslations(ctx context.Context, v dyn.Value) (dyn.Value, error) {
 	var err error
 
 	fallback, err := gatherFallbackPaths(v, "jobs")
@@ -38,28 +39,48 @@ func (t *translateContext) applyJobTranslations(v dyn.Value) (dyn.Value, error)
 			return dyn.InvalidValue, fmt.Errorf("unable to determine directory for job %s: %w", key, err)
 		}
 
-		rewritePatternFn, err := t.getRewritePatternFn(kind)
+		mode, err := getJobTranslateMode(kind)
 		if err != nil {
 			return dyn.InvalidValue, err
 		}
 
-		return t.rewriteRelativeTo(p, v, rewritePatternFn, dir, fallback[key])
+		opts := translateOptions{
+			Mode: mode,
+		}
+
+		// Try to rewrite the path relative to the directory of the configuration file where the value was defined.
+		nv, err := t.rewriteValue(ctx, p, v, dir, opts)
+		if err == nil {
+			return nv, nil
+		}
+
+		// If we failed to rewrite the path, try to rewrite it relative to the fallback directory.
+		// We only do this for jobs and pipelines because of the comment in [gatherFallbackPaths].
+		if fallback[key] != "" {
+			nv, nerr := t.rewriteValue(ctx, p, v, fallback[key], opts)
+			if nerr == nil {
+				// TODO: Emit a warning that this path should be rewritten.
+				return nv, nil
+			}
+		}
+
+		return dyn.InvalidValue, err
 	})
 }
 
-func (t *translateContext) getRewritePatternFn(kind paths.PathKind) (rewriteFunc, error) {
+func getJobTranslateMode(kind paths.PathKind) (TranslateMode, error) {
 	switch kind {
 	case paths.PathKindLibrary:
-		return t.translateNoOp, nil
+		return TranslateModeLocalRelative, nil
 	case paths.PathKindNotebook:
-		return t.translateNotebookPath, nil
+		return TranslateModeNotebook, nil
 	case paths.PathKindWorkspaceFile:
-		return t.translateFilePath, nil
+		return TranslateModeFile, nil
 	case paths.PathKindDirectory:
-		return t.translateDirectoryPath, nil
+		return TranslateModeDirectory, nil
 	case paths.PathKindWithPrefix:
-		return t.translateNoOpWithPrefix, nil
+		return TranslateModeLocalRelativeWithPrefix, nil
 	}
 
-	return nil, fmt.Errorf("unsupported path kind: %d", kind)
+	return TranslateMode(0), fmt.Errorf("unsupported path kind: %d", kind)
 }

bundle/config/mutator/translate_paths_pipelines.go

@@ -1,6 +1,7 @@
 package mutator
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/databricks/cli/libs/dyn"
@@ -8,7 +9,7 @@ import (
 
 type pipelineRewritePattern struct {
 	pattern dyn.Pattern
-	fn      rewriteFunc
+	opts    translateOptions
 }
 
 func (t *translateContext) pipelineRewritePatterns() []pipelineRewritePattern {
@@ -25,16 +26,16 @@ func (t *translateContext) pipelineRewritePatterns() []pipelineRewritePattern {
 	return []pipelineRewritePattern{
 		{
 			base.Append(dyn.Key("notebook"), dyn.Key("path")),
-			t.translateNotebookPath,
+			translateOptions{Mode: TranslateModeNotebook},
 		},
 		{
 			base.Append(dyn.Key("file"), dyn.Key("path")),
-			t.translateFilePath,
+			translateOptions{Mode: TranslateModeFile},
 		},
 	}
 }
 
-func (t *translateContext) applyPipelineTranslations(v dyn.Value) (dyn.Value, error) {
+func (t *translateContext) applyPipelineTranslations(ctx context.Context, v dyn.Value) (dyn.Value, error) {
 	var err error
 
 	fallback, err := gatherFallbackPaths(v, "pipelines")
@@ -50,7 +51,23 @@ func (t *translateContext) applyPipelineTranslations(v dyn.Value) (dyn.Value, er
 				return dyn.InvalidValue, fmt.Errorf("unable to determine directory for pipeline %s: %w", key, err)
 			}
 
-			return t.rewriteRelativeTo(p, v, rewritePattern.fn, dir, fallback[key])
+			// Try to rewrite the path relative to the directory of the configuration file where the value was defined.
+			nv, err := t.rewriteValue(ctx, p, v, dir, rewritePattern.opts)
+			if err == nil {
+				return nv, nil
+			}
+
+			// If we failed to rewrite the path, try to rewrite it relative to the fallback directory.
+			// We only do this for jobs and pipelines because of the comment in [gatherFallbackPaths].
+			if fallback[key] != "" {
+				nv, nerr := t.rewriteValue(ctx, p, v, fallback[key], rewritePattern.opts)
+				if nerr == nil {
+					// TODO: Emit a warning that this path should be rewritten.
+					return nv, nil
+				}
+			}
+
+			return dyn.InvalidValue, err
 		})
 		if err != nil {
 			return dyn.InvalidValue, err

bundle/tests/relative_path_with_includes_test.go

@@ -17,8 +17,8 @@ func TestRelativePathsWithIncludes(t *testing.T) {
 	diags := bundle.Apply(context.Background(), b, m)
 	assert.NoError(t, diags.Error())
 
-	assert.Equal(t, "artifact_a", b.Config.Artifacts["test_a"].Path)
-	assert.Equal(t, filepath.Join("subfolder", "artifact_b"), b.Config.Artifacts["test_b"].Path)
+	assert.Equal(t, filepath.Join(b.SyncRootPath, "artifact_a"), b.Config.Artifacts["test_a"].Path)
+	assert.Equal(t, filepath.Join(b.SyncRootPath, "subfolder", "artifact_b"), b.Config.Artifacts["test_b"].Path)
 
 	assert.ElementsMatch(
 		t,