diff --git a/acceptance/bundle/paths/fallback_metric/output.txt b/acceptance/bundle/paths/fallback_metric/output.txt
index 6b8ef00e47..a98e275088 100644
--- a/acceptance/bundle/paths/fallback_metric/output.txt
+++ b/acceptance/bundle/paths/fallback_metric/output.txt
@@ -23,11 +23,11 @@ Deployment complete!
     "value": false
   },
   {
-    "key": "skip_artifact_cleanup",
+    "key": "python_wheel_wrapper_is_set",
     "value": false
   },
   {
-    "key": "python_wheel_wrapper_is_set",
+    "key": "skip_artifact_cleanup",
     "value": false
   },
   {
@@ -57,11 +57,11 @@ Deployment complete!
     "value": true
   },
   {
-    "key": "skip_artifact_cleanup",
+    "key": "python_wheel_wrapper_is_set",
     "value": false
   },
   {
-    "key": "python_wheel_wrapper_is_set",
+    "key": "skip_artifact_cleanup",
     "value": false
   },
   {
diff --git a/acceptance/bundle/plan/no_upload/databricks.yml b/acceptance/bundle/plan/no_upload/databricks.yml
new file mode 100644
index 0000000000..76eed50823
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/databricks.yml
@@ -0,0 +1,19 @@
+bundle:
+  name: plan-no-upload
+
+resources:
+  jobs:
+    my_job:
+      name: my-job
+      tasks:
+        - task_key: task1
+          spark_python_task:
+            python_file: "./my_script.py"
+          environment_key: "env"
+
+      environments:
+        - environment_key: "env"
+          spec:
+            client: "1"
+            dependencies:
+              - "*.whl"
diff --git a/acceptance/bundle/plan/no_upload/my_script.py b/acceptance/bundle/plan/no_upload/my_script.py
new file mode 100644
index 0000000000..7df869a15e
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/my_script.py
@@ -0,0 +1 @@
+print("Hello, World!")
diff --git a/acceptance/bundle/plan/no_upload/out.test.toml b/acceptance/bundle/plan/no_upload/out.test.toml
new file mode 100644
index 0000000000..8f3575be7b
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/out.test.toml
@@ -0,0 +1,5 @@
+Local = true
+Cloud = false
+
+[EnvMatrix]
+  DATABRICKS_CLI_DEPLOYMENT = ["terraform", "direct-exp"]
diff --git a/acceptance/bundle/plan/no_upload/output.txt b/acceptance/bundle/plan/no_upload/output.txt
new file mode 100644
index 0000000000..8997a8b2f2
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/output.txt
@@ -0,0 +1,5 @@
+
+>>> [CLI] bundle plan
+create jobs.my_job
+
+>>> jq -s .[] | select(.method != "GET") out.requests.txt
diff --git a/acceptance/bundle/plan/no_upload/script b/acceptance/bundle/plan/no_upload/script
new file mode 100644
index 0000000000..02d44d657b
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/script
@@ -0,0 +1,6 @@
+trace $CLI bundle plan
+
+# Expect no non-GET requests
+trace jq -s '.[] | select(.method != "GET")' out.requests.txt
+
+rm out.requests.txt
diff --git a/acceptance/bundle/plan/no_upload/test.toml b/acceptance/bundle/plan/no_upload/test.toml
new file mode 100644
index 0000000000..9efba53719
--- /dev/null
+++ b/acceptance/bundle/plan/no_upload/test.toml
@@ -0,0 +1,3 @@
+Local = true
+Cloud = false
+RecordRequests = true
diff --git a/acceptance/bundle/plan/no_upload/test.whl b/acceptance/bundle/plan/no_upload/test.whl
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/acceptance/bundle/resource_deps/missing_ingestion_definition/out.requests.txt b/acceptance/bundle/resource_deps/missing_ingestion_definition/out.requests.txt
index 9d772ac786..08f0d3dcd4 100644
--- a/acceptance/bundle/resource_deps/missing_ingestion_definition/out.requests.txt
+++ b/acceptance/bundle/resource_deps/missing_ingestion_definition/out.requests.txt
@@ -29,18 +29,3 @@
   "method": "GET",
   "path": "/api/2.0/workspace/get-status"
 }
-{
-  "method": "POST",
-  "path": "/api/2.0/workspace/delete",
-  "body": {
-    "path": "/Workspace/Users/[USERNAME]/.bundle/test-bundle/default/artifacts/.internal",
-    "recursive": true
-  }
-}
-{
-  "method": "POST",
-  "path": "/api/2.0/workspace/mkdirs",
-  "body": {
-    "path": "/Workspace/Users/[USERNAME]/.bundle/test-bundle/default/artifacts/.internal"
-  }
-}
diff --git a/acceptance/bundle/telemetry/deploy-compute-type/output.txt b/acceptance/bundle/telemetry/deploy-compute-type/output.txt
index 9cb418ea1d..0466986023 100644
--- a/acceptance/bundle/telemetry/deploy-compute-type/output.txt
+++ b/acceptance/bundle/telemetry/deploy-compute-type/output.txt
@@ -22,11 +22,11 @@ Deployment complete!
     "value": false
   },
   {
-    "key": "skip_artifact_cleanup",
+    "key": "python_wheel_wrapper_is_set",
     "value": false
   },
   {
-    "key": "python_wheel_wrapper_is_set",
+    "key": "skip_artifact_cleanup",
     "value": false
   },
   {
@@ -52,11 +52,11 @@ Deployment complete!
     "value": false
   },
   {
-    "key": "skip_artifact_cleanup",
+    "key": "python_wheel_wrapper_is_set",
     "value": false
   },
   {
-    "key": "python_wheel_wrapper_is_set",
+    "key": "skip_artifact_cleanup",
     "value": false
   },
   {
diff --git a/acceptance/bundle/telemetry/deploy-experimental/output.txt b/acceptance/bundle/telemetry/deploy-experimental/output.txt
index e5e2ea254c..4a88862cd8 100644
--- a/acceptance/bundle/telemetry/deploy-experimental/output.txt
+++ b/acceptance/bundle/telemetry/deploy-experimental/output.txt
@@ -21,11 +21,11 @@ Deployment complete!
       "value": false
     },
     {
-      "key": "skip_artifact_cleanup",
+      "key": "python_wheel_wrapper_is_set",
       "value": false
     },
     {
-      "key": "python_wheel_wrapper_is_set",
+      "key": "skip_artifact_cleanup",
       "value": false
     },
     {
diff --git a/acceptance/bundle/telemetry/deploy-name-prefix/custom/output.txt b/acceptance/bundle/telemetry/deploy-name-prefix/custom/output.txt
index 35c1c4070e..e79e408218 100644
--- a/acceptance/bundle/telemetry/deploy-name-prefix/custom/output.txt
+++ b/acceptance/bundle/telemetry/deploy-name-prefix/custom/output.txt
@@ -17,11 +17,11 @@ Deployment complete!
       "value": true
     },
     {
-      "key": "skip_artifact_cleanup",
+      "key": "python_wheel_wrapper_is_set",
       "value": false
     },
     {
-      "key": "python_wheel_wrapper_is_set",
+      "key": "skip_artifact_cleanup",
       "value": false
     },
     {
diff --git a/acceptance/bundle/telemetry/deploy-name-prefix/mode-development/output.txt b/acceptance/bundle/telemetry/deploy-name-prefix/mode-development/output.txt
index c54466ab89..d28b22deea 100644
--- a/acceptance/bundle/telemetry/deploy-name-prefix/mode-development/output.txt
+++ b/acceptance/bundle/telemetry/deploy-name-prefix/mode-development/output.txt
@@ -17,11 +17,11 @@ Deployment complete!
       "value": true
     },
     {
-      "key": "skip_artifact_cleanup",
+      "key": "python_wheel_wrapper_is_set",
       "value": false
     },
     {
-      "key": "python_wheel_wrapper_is_set",
+      "key": "skip_artifact_cleanup",
       "value": false
     },
     {
diff --git a/acceptance/bundle/telemetry/deploy-whl-artifacts/output.txt b/acceptance/bundle/telemetry/deploy-whl-artifacts/output.txt
index 941b1aa988..207ee71d24 100644
--- a/acceptance/bundle/telemetry/deploy-whl-artifacts/output.txt
+++ b/acceptance/bundle/telemetry/deploy-whl-artifacts/output.txt
@@ -25,11 +25,11 @@ Deployment complete!
       "value": false
     },
     {
-      "key": "skip_artifact_cleanup",
+      "key": "python_wheel_wrapper_is_set",
       "value": false
     },
     {
-      "key": "python_wheel_wrapper_is_set",
+      "key": "skip_artifact_cleanup",
       "value": false
     },
     {
@@ -61,11 +61,11 @@ Deployment complete!
       "value": true
     },
     {
-      "key": "skip_artifact_cleanup",
+      "key": "python_wheel_wrapper_is_set",
       "value": true
     },
     {
-      "key": "python_wheel_wrapper_is_set",
+      "key": "skip_artifact_cleanup",
       "value": true
     },
     {
diff --git a/acceptance/bundle/telemetry/deploy/out.telemetry.txt b/acceptance/bundle/telemetry/deploy/out.telemetry.txt
index 8d4afad759..0c4a1d5d62 100644
--- a/acceptance/bundle/telemetry/deploy/out.telemetry.txt
+++ b/acceptance/bundle/telemetry/deploy/out.telemetry.txt
@@ -51,11 +51,11 @@
               "value": false
             },
             {
-              "key": "skip_artifact_cleanup",
+              "key": "python_wheel_wrapper_is_set",
               "value": false
             },
             {
-              "key": "python_wheel_wrapper_is_set",
+              "key": "skip_artifact_cleanup",
               "value": false
             },
             {
diff --git a/bundle/libraries/filer.go b/bundle/libraries/filer.go
index 73ca442735..4b62da5935 100644
--- a/bundle/libraries/filer.go
+++ b/bundle/libraries/filer.go
@@ -3,6 +3,7 @@ package libraries
 import (
 	"context"
 	"path"
+	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/libs/diag"
@@ -24,6 +25,7 @@ func GetFilerForLibraries(ctx context.Context, b *bundle.Bundle) (filer.Filer, s
 	}
 
 	uploadPath := path.Join(b.Config.Workspace.ArtifactPath, InternalDirName)
+	uploadPath = ensureWorkspaceOrVolumesPrefix(uploadPath)
 
 	switch {
 	case IsVolumesPath(artifactPath):
@@ -40,11 +42,23 @@ func GetFilerForLibrariesCleanup(ctx context.Context, b *bundle.Bundle) (filer.F
 		return nil, "", diag.Errorf("remote artifact path not configured")
 	}
 
+	artifactPath = ensureWorkspaceOrVolumesPrefix(artifactPath)
+
 	switch {
 	case IsVolumesPath(artifactPath):
-		return filerForVolume(b, b.Config.Workspace.ArtifactPath)
+		return filerForVolume(b, artifactPath)
 
 	default:
-		return filerForWorkspace(b, b.Config.Workspace.ArtifactPath)
+		return filerForWorkspace(b, artifactPath)
+	}
+}
+
+// If the remote path does not start with /Workspace or /Volumes, prepend /Workspace
+// Some of the bundle configuration might use workspace paths like /Users or /Shared.
+// While this is still a valid workspace path, the backend converts it to /Workspace/Users or /Workspace/Shared.
+func ensureWorkspaceOrVolumesPrefix(path string) string {
+	if !strings.HasPrefix(path, "/Workspace") && !strings.HasPrefix(path, "/Volumes") {
+		path = "/Workspace" + path
 	}
+	return path
 }
diff --git a/bundle/libraries/filer_test.go b/bundle/libraries/filer_test.go
index b4799fcf36..258a88dad1 100644
--- a/bundle/libraries/filer_test.go
+++ b/bundle/libraries/filer_test.go
@@ -28,7 +28,7 @@ func TestGetFilerForLibrariesValidWsfs(t *testing.T) {
 
 	client, uploadPath, diags := GetFilerForLibraries(context.Background(), b)
 	require.NoError(t, diags.Error())
-	assert.Equal(t, "/foo/bar/artifacts/.internal", uploadPath)
+	assert.Equal(t, "/Workspace/foo/bar/artifacts/.internal", uploadPath)
 
 	assert.IsType(t, &filer.WorkspaceFilesClient{}, client)
 }
@@ -48,7 +48,7 @@ func TestGetFilerForLibrariesCleanupValidWsfs(t *testing.T) {
 
 	client, uploadPath, diags := GetFilerForLibrariesCleanup(context.Background(), b)
 	require.NoError(t, diags.Error())
-	assert.Equal(t, "/foo/bar/artifacts", uploadPath)
+	assert.Equal(t, "/Workspace/foo/bar/artifacts", uploadPath)
 
 	assert.IsType(t, &filer.WorkspaceFilesClient{}, client)
 }
diff --git a/bundle/libraries/remote_path.go b/bundle/libraries/remote_path.go
new file mode 100644
index 0000000000..d24387653a
--- /dev/null
+++ b/bundle/libraries/remote_path.go
@@ -0,0 +1,141 @@
+package libraries
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"path/filepath"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/utils"
+)
+
+// ReplaceWithRemotePath updates all the libraries paths to point to the remote location
+// where the libraries will be uploaded later.
+func ReplaceWithRemotePath(ctx context.Context, b *bundle.Bundle) (map[string][]LocationToUpdate, diag.Diagnostics) {
+	_, uploadPath, diags := GetFilerForLibraries(ctx, b)
+	if diags.HasError() {
+		return nil, diags
+	}
+
+	libs, err := collectLocalLibraries(b)
+	if err != nil {
+		return nil, diag.FromErr(err)
+	}
+
+	sources := utils.SortedKeys(libs)
+
+	// Update all the config paths to point to the uploaded location
+	err = b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+		for _, source := range sources {
+			locations := libs[source]
+			remotePath := path.Join(uploadPath, filepath.Base(source))
+
+			for _, location := range locations {
+				v, err = dyn.SetByPath(v, location.configPath, dyn.NewValue(remotePath, []dyn.Location{location.location}))
+				if err != nil {
+					return v, fmt.Errorf("internal error: failed to update path %#v to %#v: %w", source, remotePath, err)
+				}
+			}
+		}
+
+		return v, nil
+	})
+	if err != nil {
+		diags = diags.Extend(diag.FromErr(err))
+	}
+
+	return libs, diags
+}
+
+// Collect all libraries from the bundle configuration and their config paths.
+// By this stage all glob references are expanded and we have a list of all libraries that need to be uploaded.
+// We collect them from task libraries, foreach task libraries, environment dependencies, and artifacts.
+// We return a map of library source to a list of config paths and locations where the library is used.
+// We use map so we don't upload the same library multiple times.
+// This map is later used to upload the libraries to the remote location and update the config paths to point to the uploaded location.
+func collectLocalLibraries(b *bundle.Bundle) (map[string][]LocationToUpdate, error) {
+	libs := make(map[string]([]LocationToUpdate))
+
+	patterns := []dyn.Pattern{
+		taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
+		taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
+		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
+		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
+		envDepsPattern.Append(dyn.AnyIndex()),
+		pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
+	}
+
+	for _, pattern := range patterns {
+		err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+			return dyn.MapByPattern(v, pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+				source, ok := v.AsString()
+				if !ok {
+					return v, fmt.Errorf("expected string, got %s", v.Kind())
+				}
+
+				if !IsLibraryLocal(source) {
+					return v, nil
+				}
+
+				source = filepath.Join(b.SyncRootPath, source)
+				libs[source] = append(libs[source], LocationToUpdate{
+					configPath: p,
+					location:   v.Location(),
+				})
+
+				return v, nil
+			})
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	artifactPattern := dyn.NewPattern(
+		dyn.Key("artifacts"),
+		dyn.AnyKey(),
+		dyn.Key("files"),
+		dyn.AnyIndex(),
+	)
+
+	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+		return dyn.MapByPattern(v, artifactPattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+			file, ok := v.AsMap()
+			if !ok {
+				return v, fmt.Errorf("expected map, got %s", v.Kind())
+			}
+
+			sv, ok := file.GetByString("source")
+			if !ok {
+				return v, nil
+			}
+
+			source, ok := sv.AsString()
+			if !ok {
+				return v, fmt.Errorf("expected string, got %s", v.Kind())
+			}
+
+			if sv, ok = file.GetByString("patched"); ok {
+				patched, ok := sv.AsString()
+				if ok && patched != "" {
+					source = patched
+				}
+			}
+
+			libs[source] = append(libs[source], LocationToUpdate{
+				configPath: p.Append(dyn.Key("remote_path")),
+				location:   v.Location(),
+			})
+
+			return v, nil
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return libs, nil
+}
diff --git a/bundle/libraries/upload.go b/bundle/libraries/upload.go
index ed69d27f88..590adda4ff 100644
--- a/bundle/libraries/upload.go
+++ b/bundle/libraries/upload.go
@@ -5,9 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
-	"path"
 	"path/filepath"
-	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/libs/cmdio"
@@ -25,117 +23,31 @@ import (
 // avoid hitting the rate limit.
 var maxFilesRequestsInFlight = 5
 
-func Upload() bundle.Mutator {
-	return &upload{}
+func Upload(libs map[string][]LocationToUpdate) bundle.Mutator {
+	return &upload{
+		libs: libs,
+	}
 }
 
-func UploadWithClient(client filer.Filer) bundle.Mutator {
+func UploadWithClient(libs map[string][]LocationToUpdate, client filer.Filer) bundle.Mutator {
 	return &upload{
+		libs:   libs,
 		client: client,
 	}
 }
 
 type upload struct {
 	client filer.Filer
+	libs   map[string][]LocationToUpdate
 }
 
-type configLocation struct {
+type LocationToUpdate struct {
 	configPath dyn.Path
 	location   dyn.Location
 }
 
-// Collect all libraries from the bundle configuration and their config paths.
-// By this stage all glob references are expanded and we have a list of all libraries that need to be uploaded.
-// We collect them from task libraries, foreach task libraries, environment dependencies, and artifacts.
-// We return a map of library source to a list of config paths and locations where the library is used.
-// We use map so we don't upload the same library multiple times.
-// Instead we upload it once and update all the config paths to point to the uploaded location.
-func collectLocalLibraries(b *bundle.Bundle) (map[string][]configLocation, error) {
-	libs := make(map[string]([]configLocation))
-
-	patterns := []dyn.Pattern{
-		taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
-		taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
-		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
-		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
-		envDepsPattern.Append(dyn.AnyIndex()),
-		pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
-	}
-
-	for _, pattern := range patterns {
-		err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
-			return dyn.MapByPattern(v, pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
-				source, ok := v.AsString()
-				if !ok {
-					return v, fmt.Errorf("expected string, got %s", v.Kind())
-				}
-
-				if !IsLibraryLocal(source) {
-					return v, nil
-				}
-
-				source = filepath.Join(b.SyncRootPath, source)
-				libs[source] = append(libs[source], configLocation{
-					configPath: p,
-					location:   v.Location(),
-				})
-
-				return v, nil
-			})
-		})
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	artifactPattern := dyn.NewPattern(
-		dyn.Key("artifacts"),
-		dyn.AnyKey(),
-		dyn.Key("files"),
-		dyn.AnyIndex(),
-	)
-
-	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
-		return dyn.MapByPattern(v, artifactPattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
-			file, ok := v.AsMap()
-			if !ok {
-				return v, fmt.Errorf("expected map, got %s", v.Kind())
-			}
-
-			sv, ok := file.GetByString("source")
-			if !ok {
-				return v, nil
-			}
-
-			source, ok := sv.AsString()
-			if !ok {
-				return v, fmt.Errorf("expected string, got %s", v.Kind())
-			}
-
-			if sv, ok = file.GetByString("patched"); ok {
-				patched, ok := sv.AsString()
-				if ok && patched != "" {
-					source = patched
-				}
-			}
-
-			libs[source] = append(libs[source], configLocation{
-				configPath: p.Append(dyn.Key("remote_path")),
-				location:   v.Location(),
-			})
-
-			return v, nil
-		})
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return libs, nil
-}
-
 func (u *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
-	client, uploadPath, diags := GetFilerForLibraries(ctx, b)
+	client, _, diags := GetFilerForLibraries(ctx, b)
 	if diags.HasError() {
 		return diags
 	}
@@ -146,12 +58,7 @@ func (u *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 		u.client = client
 	}
 
-	libs, err := collectLocalLibraries(b)
-	if err != nil {
-		return diag.FromErr(err)
-	}
-
-	sources := utils.SortedKeys(libs)
+	sources := utils.SortedKeys(u.libs)
 
 	errs, errCtx := errgroup.WithContext(ctx)
 	errs.SetLimit(maxFilesRequestsInFlight)
@@ -173,30 +80,6 @@ func (u *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 		return diag.FromErr(err)
 	}
 
-	// Update all the config paths to point to the uploaded location
-	for _, source := range sources {
-		locations := libs[source]
-		err = b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
-			remotePath := path.Join(uploadPath, filepath.Base(source))
-
-			// If the remote path does not start with /Workspace or /Volumes, prepend /Workspace
-			if !strings.HasPrefix(remotePath, "/Workspace") && !strings.HasPrefix(remotePath, "/Volumes") {
-				remotePath = "/Workspace" + remotePath
-			}
-			for _, location := range locations {
-				v, err = dyn.SetByPath(v, location.configPath, dyn.NewValue(remotePath, []dyn.Location{location.location}))
-				if err != nil {
-					return v, err
-				}
-			}
-
-			return v, nil
-		})
-		if err != nil {
-			diags = diags.Extend(diag.FromErr(err))
-		}
-	}
-
 	return diags
 }
 
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index 6439c71222..ad8dda876c 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -8,7 +8,6 @@ import (
 	"github.com/databricks/cli/bundle/apps"
 	"github.com/databricks/cli/bundle/artifacts"
 	"github.com/databricks/cli/bundle/config"
-	"github.com/databricks/cli/bundle/config/mutator"
 	"github.com/databricks/cli/bundle/deploy"
 	"github.com/databricks/cli/bundle/deploy/files"
 	"github.com/databricks/cli/bundle/deploy/lock"
@@ -20,7 +19,6 @@ import (
 	"github.com/databricks/cli/bundle/permissions"
 	"github.com/databricks/cli/bundle/scripts"
 	"github.com/databricks/cli/bundle/statemgmt"
-	"github.com/databricks/cli/bundle/trampoline"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
@@ -151,36 +149,13 @@ func deployCore(ctx context.Context, b *bundle.Bundle) {
 	}
 }
 
-// deployPrepare is common set of mutators between "bundle plan" and "bundle deploy".
-// Ideally it should not modify the remote, only in-memory bundle config.
-// TODO: currently it does affect remote, via artifacts.CleanUp() and libraries.Upload().
-// We should refactor deployment so that it consists of two stages:
-// 1. Preparation: only local config is changed. This will be used by both "bundle deploy" and "bundle plan"
-// 2. Deployment: this does all the uploads. Only used by "deploy", not "plan".
-func deployPrepare(ctx context.Context, b *bundle.Bundle) {
+// uploadLibraries uploads libraries to the workspace.
+// It also cleans up the artifacts directory and transforms wheel tasks.
+// It is called by only "bundle deploy".
+func uploadLibraries(ctx context.Context, b *bundle.Bundle, libs map[string][]libraries.LocationToUpdate) {
 	bundle.ApplySeqContext(ctx, b,
-		statemgmt.StatePull(),
-		terraform.CheckDashboardsModifiedRemotely(),
-		deploy.StatePull(),
-		mutator.ValidateGitDetails(),
-		terraform.CheckRunningResource(),
-
-		// artifacts.CleanUp() is there because I'm not sure if it's safe to move to later stage.
 		artifacts.CleanUp(),
-
-		// libraries.CheckForSameNameLibraries() needs to be run after we expand glob references so we
-		// know what are the actual library paths.
-		// libraries.ExpandGlobReferences() has to be run after the libraries are built and thus this
-		// mutator is part of the deploy step rather than validate.
-		libraries.ExpandGlobReferences(),
-		libraries.CheckForSameNameLibraries(),
-		// SwitchToPatchedWheels must be run after ExpandGlobReferences and after build phase because it Artifact.Source and Artifact.Patched populated
-		libraries.SwitchToPatchedWheels(),
-
-		// libraries.Upload() not just uploads but also replaces local paths with remote paths.
-		// TransformWheelTask depends on it and planning also depends on it.
-		libraries.Upload(),
-		trampoline.TransformWheelTask(),
+		libraries.Upload(libs),
 	)
 }
 
@@ -205,7 +180,12 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDeploy))
 	}()
 
-	deployPrepare(ctx, b)
+	libs := deployPrepare(ctx, b)
+	if logdiag.HasError(ctx) {
+		return
+	}
+
+	uploadLibraries(ctx, b, libs)
 	if logdiag.HasError(ctx) {
 		return
 	}
diff --git a/bundle/phases/plan.go b/bundle/phases/plan.go
new file mode 100644
index 0000000000..2b0f09e2f1
--- /dev/null
+++ b/bundle/phases/plan.go
@@ -0,0 +1,48 @@
+package phases
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/mutator"
+	"github.com/databricks/cli/bundle/deploy"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/libraries"
+	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/bundle/trampoline"
+	"github.com/databricks/cli/libs/logdiag"
+)
+
+// deployPrepare is common set of mutators between "bundle plan" and "bundle deploy".
+// This function does not make any mutations in the workspace remotely, only in-memory bundle config mutations
+func deployPrepare(ctx context.Context, b *bundle.Bundle) map[string][]libraries.LocationToUpdate {
+	bundle.ApplySeqContext(ctx, b,
+		statemgmt.StatePull(),
+		terraform.CheckDashboardsModifiedRemotely(),
+		deploy.StatePull(),
+		mutator.ValidateGitDetails(),
+		terraform.CheckRunningResource(),
+
+		// libraries.CheckForSameNameLibraries() needs to be run after we expand glob references so we
+		// know what are the actual library paths.
+		// libraries.ExpandGlobReferences() has to be run after the libraries are built and thus this
+		// mutator is part of the deploy step rather than validate.
+		libraries.ExpandGlobReferences(),
+		libraries.CheckForSameNameLibraries(),
+		// SwitchToPatchedWheels must be run after ExpandGlobReferences and after build phase because it Artifact.Source and Artifact.Patched populated
+		libraries.SwitchToPatchedWheels(),
+	)
+
+	libs, diags := libraries.ReplaceWithRemotePath(ctx, b)
+	for _, diag := range diags {
+		logdiag.LogDiag(ctx, diag)
+	}
+
+	bundle.ApplySeqContext(ctx, b,
+		// TransformWheelTask must be run after ReplaceWithRemotePath so we can use correct remote path in the
+		// transformed notebook
+		trampoline.TransformWheelTask(),
+	)
+
+	return libs
+}