diff --git a/acceptance/bundle/deploy/immutable-no-artifacts/.gitignore b/acceptance/bundle/deploy/immutable-no-artifacts/.gitignore
new file mode 100644
index 00000000000..78caa68e38e
--- /dev/null
+++ b/acceptance/bundle/deploy/immutable-no-artifacts/.gitignore
@@ -0,0 +1,2 @@
+*.txt
+*.json
diff --git a/acceptance/bundle/deploy/immutable-no-artifacts/out.test.toml b/acceptance/bundle/deploy/immutable-no-artifacts/out.test.toml
index 0938e678987..57b0f616850 100644
--- a/acceptance/bundle/deploy/immutable-no-artifacts/out.test.toml
+++ b/acceptance/bundle/deploy/immutable-no-artifacts/out.test.toml
@@ -1,2 +1,3 @@
 Cloud = false
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.READPLAN = ["", "1"]
diff --git a/acceptance/bundle/deploy/immutable-no-artifacts/output.txt b/acceptance/bundle/deploy/immutable-no-artifacts/output.txt
index 317d6c9dd5e..55dde26d833 100644
--- a/acceptance/bundle/deploy/immutable-no-artifacts/output.txt
+++ b/acceptance/bundle/deploy/immutable-no-artifacts/output.txt
@@ -8,21 +8,114 @@ Workspace:
 
 Validation OK!
 
+>>> [CLI] bundle plan -o json
+
+>>> jq .plan["resources.jobs.my_job"].new_state.value.tasks plan.json
+[
+  {
+    "notebook_task": {
+      "base_parameters": {
+        "path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/some_path"
+      },
+      "notebook_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/notebook"
+    },
+    "task_key": "notebook_task"
+  },
+  {
+    "environment_key": "env",
+    "spark_python_task": {
+      "python_file": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py"
+    },
+    "task_key": "spark_python_task"
+  }
+]
+
+>>> jq .plan["resources.internal_immutable_snapshots.immutable"] plan.json
+{
+  "action": "create",
+  "new_state": {
+    "value": {
+      "remote_root": "/Workspace/Users/[UUID]/.snapshots",
+      "relative_path": "[UUID]/[SNAPSHOT_HASH]",
+      "full_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]",
+      "bundle_id": "[UUID]",
+      "acl": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ]
+    }
+  }
+}
+
 >>> [CLI] bundle deploy
-Uploading immutable bundle snapshot...
+Created internal_immutable_snapshots.immutable
 Created jobs.my_job
 Files: 0 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.tasks job.json
 "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py"
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.tasks job.json
 "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/notebook"
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.tasks job.json
 "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/some_path"
 
+>>> title Check snapshot plan after deploy: 
+
+=== Check snapshot plan after deploy: 
+>>> [CLI] bundle plan -o json
+
+>>> jq .plan["resources.internal_immutable_snapshots.immutable"] plan.json
+{
+  "action": "skip",
+  "remote_state": {
+    "relative_path": "[UUID]/[SNAPSHOT_HASH]",
+    "full_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]"
+  },
+  "changes": {
+    "acl": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ],
+      "new": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ],
+      "remote": null
+    },
+    "bundle_id": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": "[UUID]",
+      "new": "[UUID]",
+      "remote": ""
+    },
+    "remote_root": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": "/Workspace/Users/[UUID]/.snapshots",
+      "new": "/Workspace/Users/[UUID]/.snapshots",
+      "remote": ""
+    }
+  }
+}
+
+>>> [CLI] bundle deploy
+Files: 0 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+Snapshot did not change as expected
+
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
   delete resources.jobs.my_job
diff --git a/acceptance/bundle/deploy/immutable-no-artifacts/script b/acceptance/bundle/deploy/immutable-no-artifacts/script
index a7aad718750..6f8f5f97bd6 100644
--- a/acceptance/bundle/deploy/immutable-no-artifacts/script
+++ b/acceptance/bundle/deploy/immutable-no-artifacts/script
@@ -6,13 +6,33 @@ cleanup() {
 trap cleanup EXIT
 
 trace $CLI bundle validate
-trace $CLI bundle deploy
+trace $CLI bundle plan -o json > plan.json
+trace jq '.plan["resources.jobs.my_job"].new_state.value.tasks' plan.json
+trace jq '.plan["resources.internal_immutable_snapshots.immutable"]' plan.json
 
+trace $CLI bundle deploy $(readplanarg plan.json)
 
 # Get a job and check that task paths point into the snapshot
 JOB_ID=$($CLI bundle summary -o json | jq -r '.resources.jobs.my_job.id')
-trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file'
-trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.notebook_task != null) | .notebook_task.notebook_path'
-trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.notebook_task != null) | .notebook_task.base_parameters.path'
+$CLI jobs get $JOB_ID > job.json
+trace jq '.settings.tasks' job.json | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file'
+trace jq '.settings.tasks' job.json | jq '.[] | select(.notebook_task != null) | .notebook_task.notebook_path'
+trace jq '.settings.tasks' job.json | jq '.[] | select(.notebook_task != null) | .notebook_task.base_parameters.path'
+PYTHON_FILE_1=$(jq '.settings.tasks' job.json | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file')
 
+trace title "Check snapshot plan after deploy: "
+trace $CLI bundle plan -o json > plan.json
+trace jq '.plan["resources.internal_immutable_snapshots.immutable"]' plan.json
+
+trace $CLI bundle deploy $(readplanarg plan.json)
+rm plan.json
+rm job.json
+
+PYTHON_FILE_2=$($CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file')
+if [ "$PYTHON_FILE_1" == "$PYTHON_FILE_2" ]; then
+    echo "Snapshot did not change as expected"
+else
+    echo "Snapshot changed"
+    exit 1
+fi
 trace $CLI bundle destroy --auto-approve
diff --git a/acceptance/bundle/deploy/immutable-no-artifacts/test.toml b/acceptance/bundle/deploy/immutable-no-artifacts/test.toml
index d1aa5d2ddce..202e70986d8 100644
--- a/acceptance/bundle/deploy/immutable-no-artifacts/test.toml
+++ b/acceptance/bundle/deploy/immutable-no-artifacts/test.toml
@@ -1,5 +1,6 @@
 Cloud = false # Temporary disable cloud tests until the API is fully available
 RecordRequests = true
+EnvMatrix.READPLAN = ["", "1"]
 
 # immutable_folder only works with the direct engine.
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
@@ -17,3 +18,9 @@ Ignore = [
 [[Repls]]
 Old = '[0-9a-f]{64}'
 New = '[SNAPSHOT_HASH]'
+
+# When READPLAN=1, "bundle deploy" is called as "bundle deploy --plan plan.json".
+# Normalize so both variants produce identical output.
+[[Repls]]
+Old = ' --plan plan.json'
+New = ''
diff --git a/acceptance/bundle/deploy/immutable-permissions-change/output.txt b/acceptance/bundle/deploy/immutable-permissions-change/output.txt
index bfc43defff9..680d9579aae 100644
--- a/acceptance/bundle/deploy/immutable-permissions-change/output.txt
+++ b/acceptance/bundle/deploy/immutable-permissions-change/output.txt
@@ -1,10 +1,10 @@
 
 === Deploy without permissions
 >>> [CLI] bundle deploy
-Uploading immutable bundle snapshot...
+Created internal_immutable_snapshots.immutable
 Created jobs.my_job
 Files: 0 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
 First snapshot path: /Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py
 
 === Add permissions and redeploy
@@ -21,11 +21,11 @@ Consider using a adding a top-level permissions section such as the following:
 See https://docs.databricks.com/dev-tools/bundles/permissions.html to learn more about permission configuration.
   in databricks.yml:22:3
 
-Uploading immutable bundle snapshot...
+Recreated internal_immutable_snapshots.immutable
 Updated jobs.my_job
 Created jobs.my_job.permissions
 Files: 0 uploaded, 0 deleted
-Resources: 1 created, 1 changed, 0 deleted, 0 unchanged
+Resources: 2 created, 1 changed, 1 deleted, 0 unchanged
 Second snapshot path: /Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py
 
 === Verify snapshot path changed after permissions change
diff --git a/acceptance/bundle/deploy/immutable/output.txt b/acceptance/bundle/deploy/immutable/output.txt
index d02d2cd1d31..765e73a51bd 100644
--- a/acceptance/bundle/deploy/immutable/output.txt
+++ b/acceptance/bundle/deploy/immutable/output.txt
@@ -10,10 +10,12 @@ Validation OK!
 
 >>> [CLI] bundle plan -o json
 Building python_artifact...
+
+>>> jq .plan["resources.jobs.my_job"].new_state.value.tasks plan.json
 [
   {
     "notebook_task": {
-      "notebook_path": "${workspace.snapshot_path}/files/src/notebook"
+      "notebook_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/notebook"
     },
     "task_key": "notebook_task"
   },
@@ -28,30 +30,147 @@ Building python_artifact...
   {
     "environment_key": "env",
     "spark_python_task": {
-      "python_file": "${workspace.snapshot_path}/files/src/main.py"
+      "python_file": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py"
     },
     "task_key": "spark_python_task"
   }
 ]
 
+>>> jq .plan["resources.jobs.my_job"].new_state.value.environments plan.json
+[
+  {
+    "environment_key": "env",
+    "spec": {
+      "dependencies": [
+        "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/artifacts/.internal/immutable-0.0.1-py3-none-any.whl"
+      ],
+      "environment_version": "4"
+    }
+  }
+]
+
+>>> jq .plan["resources.internal_immutable_snapshots.immutable"] plan.json
+{
+  "action": "create",
+  "new_state": {
+    "value": {
+      "remote_root": "/Workspace/Users/[UUID]/.snapshots",
+      "relative_path": "[UUID]/[SNAPSHOT_HASH]",
+      "full_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]",
+      "bundle_id": "[UUID]",
+      "acl": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ]
+    }
+  }
+}
+
 >>> [CLI] bundle deploy
 Building python_artifact...
-Uploading immutable bundle snapshot...
+Created internal_immutable_snapshots.immutable
 Created jobs.my_job
 Files: 0 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.tasks job.json
 "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/main.py"
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.tasks job.json
 "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/src/notebook"
 
->>> [CLI] jobs get [NUMID]
+>>> jq .settings.environments[0].spec.dependencies job.json
 [
   "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/artifacts/.internal/immutable-0.0.1-py3-none-any.whl"
 ]
 
+>>> title Check snapshot plan after deploy
+
+=== Check snapshot plan after deploy
+>>> [CLI] bundle plan -o json
+Building python_artifact...
+
+>>> jq .plan["resources.internal_immutable_snapshots.immutable"] plan.json
+{
+  "action": "recreate",
+  "new_state": {
+    "value": {
+      "remote_root": "/Workspace/Users/[UUID]/.snapshots",
+      "relative_path": "[UUID]/[SNAPSHOT_HASH]",
+      "full_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]",
+      "bundle_id": "[UUID]",
+      "acl": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ]
+    }
+  },
+  "remote_state": {
+    "relative_path": "[UUID]/[SNAPSHOT_HASH]",
+    "full_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]"
+  },
+  "changes": {
+    "acl": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ],
+      "new": [
+        {
+          "user_name": "[USERNAME]",
+          "permission_level": "CAN_READ"
+        }
+      ],
+      "remote": null
+    },
+    "bundle_id": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": "[UUID]",
+      "new": "[UUID]",
+      "remote": ""
+    },
+    "full_path": {
+      "action": "recreate",
+      "reason": "immutable",
+      "old": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]",
+      "new": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]",
+      "remote": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]"
+    },
+    "relative_path": {
+      "action": "recreate",
+      "reason": "immutable",
+      "old": "[UUID]/[SNAPSHOT_HASH]",
+      "new": "[UUID]/[SNAPSHOT_HASH]",
+      "remote": "[UUID]/[SNAPSHOT_HASH]"
+    },
+    "remote_root": {
+      "action": "skip",
+      "reason": "missing_in_remote",
+      "old": "/Workspace/Users/[UUID]/.snapshots",
+      "new": "/Workspace/Users/[UUID]/.snapshots",
+      "remote": ""
+    }
+  }
+}
+
+>>> [CLI] bundle deploy
+Building python_artifact...
+Recreated internal_immutable_snapshots.immutable
+Updated jobs.my_job
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 1 changed, 1 deleted, 0 unchanged
+
+=== Check that job uses a new snapshot: Snapshot changed as expected
+
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
   delete resources.jobs.my_job
diff --git a/acceptance/bundle/deploy/immutable/script b/acceptance/bundle/deploy/immutable/script
index 85e9a908d1d..5babe793a9a 100644
--- a/acceptance/bundle/deploy/immutable/script
+++ b/acceptance/bundle/deploy/immutable/script
@@ -5,15 +5,37 @@ cleanup() {
 trap cleanup EXIT
 
 trace $CLI bundle validate
-trace $CLI bundle plan -o json | jq '.plan["resources.jobs.my_job"].new_state.value.tasks'
-trace $CLI bundle deploy
+trace $CLI bundle plan -o json > plan.json
+trace jq '.plan["resources.jobs.my_job"].new_state.value.tasks' plan.json
+trace jq '.plan["resources.jobs.my_job"].new_state.value.environments' plan.json
+trace jq '.plan["resources.internal_immutable_snapshots.immutable"]' plan.json
 
+trace $CLI bundle deploy
 
 # Get a job and check that task paths are immutable
 JOB_ID=$($CLI bundle summary -o json | jq -r '.resources.jobs.my_job.id')
-trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file'
-trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.notebook_task != null) | .notebook_task.notebook_path'
-trace $CLI jobs get $JOB_ID | jq '.settings.environments[0].spec.dependencies'
+$CLI jobs get $JOB_ID > job.json
+trace jq '.settings.tasks' job.json | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file'
+trace jq '.settings.tasks' job.json | jq '.[] | select(.notebook_task != null) | .notebook_task.notebook_path'
+trace jq '.settings.environments[0].spec.dependencies' job.json
+PYTHON_FILE_1=$(jq '.settings.tasks' job.json | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file')
+
+trace title "Check snapshot plan after deploy"
+trace $CLI bundle plan -o json > plan.json
+trace jq '.plan["resources.internal_immutable_snapshots.immutable"]' plan.json
+rm plan.json
+rm job.json
+
+trace $CLI bundle deploy
+
+title "Check that job uses a new snapshot: "
+PYTHON_FILE_2=$($CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.spark_python_task != null) | .spark_python_task.python_file')
+if [ "$PYTHON_FILE_1" != "$PYTHON_FILE_2" ]; then
+    echo "Snapshot changed as expected"
+else
+    echo "Snapshot did not change"
+    exit 1
+fi
 
 # Redirect run output to a log file — the real workspace produces different output than the local test server.
 $CLI bundle run my_job &> LOG.run
diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt
index 3c8ba93ce34..43e9c312252 100644
--- a/acceptance/bundle/refschema/out.fields.txt
+++ b/acceptance/bundle/refschema/out.fields.txt
@@ -837,6 +837,17 @@ resources.instance_pools.*.permissions[*].group_name	string	ALL
 resources.instance_pools.*.permissions[*].level	iam.PermissionLevel	ALL
 resources.instance_pools.*.permissions[*].service_principal_name	string	ALL
 resources.instance_pools.*.permissions[*].user_name	string	ALL
+resources.internal_immutable_snapshots.*.acl	[]snapshot.ACLEntry	INPUT	STATE
+resources.internal_immutable_snapshots.*.acl[*]	snapshot.ACLEntry	INPUT	STATE
+resources.internal_immutable_snapshots.*.acl[*].group_name	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.acl[*].permission_level	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.acl[*].service_principal_name	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.acl[*].user_name	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.bundle_id	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.full_path	string	REMOTE	STATE
+resources.internal_immutable_snapshots.*.relative_path	string	REMOTE	STATE
+resources.internal_immutable_snapshots.*.remote_root	string	INPUT	STATE
+resources.internal_immutable_snapshots.*.zip_content	string	INPUT
 resources.job_runs.*.dbt_commands	[]string	ALL
 resources.job_runs.*.dbt_commands[*]	string	ALL
 resources.job_runs.*.id	string	INPUT
diff --git a/acceptance/bundle/resources/apps/immutable/output.txt b/acceptance/bundle/resources/apps/immutable/output.txt
index 4b25ae82b8b..14c87a9eb2b 100644
--- a/acceptance/bundle/resources/apps/immutable/output.txt
+++ b/acceptance/bundle/resources/apps/immutable/output.txt
@@ -1,9 +1,9 @@
 
 >>> [CLI] bundle deploy
-Uploading immutable bundle snapshot...
 Created apps.my_app
+Created internal_immutable_snapshots.immutable
 Files: 0 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
 
 >>> [CLI] bundle run my_app
 ✓ Getting the status of the app my-immutable-app
@@ -37,6 +37,6 @@ You can access the app at my-immutable-app-123.cloud.databricksapps.com
   "path": "/api/2.0/apps/my-immutable-app/deployments",
   "body": {
     "mode": "SNAPSHOT",
-    "source_code_path": "${workspace.snapshot_path}/files/app"
+    "source_code_path": "/Workspace/Users/[UUID]/.snapshots/[UUID]/[SNAPSHOT_HASH]/files/app"
   }
 }
diff --git a/acceptance/bundle/resources/genie_spaces/recreate_when_gone/output.txt b/acceptance/bundle/resources/genie_spaces/recreate_when_gone/output.txt
index e5621c9677d..d7fdd89d8a1 100644
--- a/acceptance/bundle/resources/genie_spaces/recreate_when_gone/output.txt
+++ b/acceptance/bundle/resources/genie_spaces/recreate_when_gone/output.txt
@@ -16,4 +16,4 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
 >>> [CLI] bundle destroy --auto-approve
 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/recreate-gone-genie-space-[UNIQUE_NAME]/default
 
-Destroy: 1 deleted
+Destroy: 0 deleted
diff --git a/acceptance/bundle/resources/jobs/remote_delete/destroy/out.destroy.direct.txt b/acceptance/bundle/resources/jobs/remote_delete/destroy/out.destroy.direct.txt
index cbbde0249f1..ea08bea2059 100644
--- a/acceptance/bundle/resources/jobs/remote_delete/destroy/out.destroy.direct.txt
+++ b/acceptance/bundle/resources/jobs/remote_delete/destroy/out.destroy.direct.txt
@@ -2,4 +2,4 @@
 >>> errcode [CLI] bundle destroy --auto-approve
 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
 
-Destroy: 1 deleted
+Destroy: 0 deleted
diff --git a/acceptance/bundle/resources/permissions/genie_spaces/out_of_band_deletion/output.txt b/acceptance/bundle/resources/permissions/genie_spaces/out_of_band_deletion/output.txt
index ec85eecf2b4..e2a8af52883 100644
--- a/acceptance/bundle/resources/permissions/genie_spaces/out_of_band_deletion/output.txt
+++ b/acceptance/bundle/resources/permissions/genie_spaces/out_of_band_deletion/output.txt
@@ -75,4 +75,4 @@ The following resources will be deleted:
 
 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
 
-Destroy: 1 deleted
+Destroy: 0 deleted
diff --git a/acceptance/bundle/validate/immutable_workspace_paths/output.txt b/acceptance/bundle/validate/immutable_workspace_paths/output.txt
index a41638ba774..2da05989620 100644
--- a/acceptance/bundle/validate/immutable_workspace_paths/output.txt
+++ b/acceptance/bundle/validate/immutable_workspace_paths/output.txt
@@ -2,14 +2,14 @@
 >>> [CLI] bundle validate -o json
 {
   "workspace": {
-    "artifact_path": "/Workspace/Users/[USERNAME]/.bundle/my-bundle/default/artifacts",
+    "artifact_path": "${resources.internal_immutable_snapshots.immutable.full_path}/artifacts",
     "current_user": {
       "domain_friendly_name": "[USERNAME]",
       "id": "[USERID]",
       "short_name": "[USERNAME]",
       "userName": "[USERNAME]"
     },
-    "file_path": "/Workspace/Users/[USERNAME]/.bundle/my-bundle/default/files",
+    "file_path": "${resources.internal_immutable_snapshots.immutable.full_path}/files",
     "resource_path": "/Workspace/Users/[USERNAME]/.bundle/my-bundle/default/resources",
     "root_path": "/Workspace/Users/[USERNAME]/.bundle/my-bundle/default",
     "state_path": "/Workspace/Users/[USERNAME]/.bundle/my-bundle/default/state"
@@ -19,7 +19,7 @@
       "ai_runtime_task": {
         "deployments": [
           {
-            "command_path": "${workspace.snapshot_path}/files/src/main.py",
+            "command_path": "${resources.internal_immutable_snapshots.immutable.full_path}/files/src/main.py",
             "compute": {
               "accelerator_count": 1,
               "accelerator_type": "GPU_1xA10"
@@ -33,7 +33,7 @@
     {
       "existing_cluster_id": "0101-120000-aaaaaaaa",
       "spark_python_task": {
-        "python_file": "${workspace.snapshot_path}/files/src/main.py"
+        "python_file": "${resources.internal_immutable_snapshots.immutable.full_path}/files/src/main.py"
       },
       "task_key": "my_task"
     }
diff --git a/acceptance/bundle/validate/snapshot_resource_not_allowed/databricks.yml b/acceptance/bundle/validate/snapshot_resource_not_allowed/databricks.yml
new file mode 100644
index 00000000000..a715ab3b03e
--- /dev/null
+++ b/acceptance/bundle/validate/snapshot_resource_not_allowed/databricks.yml
@@ -0,0 +1,9 @@
+bundle:
+  name: snapshot_resource_not_allowed
+
+resources:
+  internal_immutable_snapshots:
+    immutable:
+      bundle_id: 123
+      zip_content: |-
+        This is a test zip content.
diff --git a/acceptance/bundle/validate/snapshot_resource_not_allowed/out.test.toml b/acceptance/bundle/validate/snapshot_resource_not_allowed/out.test.toml
new file mode 100644
index 00000000000..98ea5040486
--- /dev/null
+++ b/acceptance/bundle/validate/snapshot_resource_not_allowed/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
diff --git a/acceptance/bundle/validate/snapshot_resource_not_allowed/output.txt b/acceptance/bundle/validate/snapshot_resource_not_allowed/output.txt
new file mode 100644
index 00000000000..b390e82ce64
--- /dev/null
+++ b/acceptance/bundle/validate/snapshot_resource_not_allowed/output.txt
@@ -0,0 +1,11 @@
+
+>>> [CLI] bundle validate
+Error: Internal resources cannot be set in bundle configuration
+  at resources
+
+Name: snapshot_resource_not_allowed
+Target: default
+
+Found 1 error
+
+Exit code: 1
diff --git a/acceptance/bundle/validate/snapshot_resource_not_allowed/script b/acceptance/bundle/validate/snapshot_resource_not_allowed/script
new file mode 100644
index 00000000000..5350876150f
--- /dev/null
+++ b/acceptance/bundle/validate/snapshot_resource_not_allowed/script
@@ -0,0 +1 @@
+trace $CLI bundle validate
diff --git a/acceptance/invariant_test.go b/acceptance/invariant_test.go
index 1c8204055ed..ad8e1d14022 100644
--- a/acceptance/invariant_test.go
+++ b/acceptance/invariant_test.go
@@ -22,7 +22,8 @@ const invariantConfigsDir = "bundle/invariant/configs"
 // for permissions/grants coverage. Add a config and remove the entry to close a gap;
 // the test fails if an entry here is actually covered, so the list only shrinks.
 var LackingInvariantTest = map[string]bool{
-	"quality_monitors": true,
+	"quality_monitors":             true,
+	"internal_immutable_snapshots": true,
 }
 
 // TestInvariantConfigsCoverage ensures that the invariant test configs in
diff --git a/bundle/config/mutator/default_workspace_paths.go b/bundle/config/mutator/default_workspace_paths.go
index 02a1ddb3b11..33e1531fdd6 100644
--- a/bundle/config/mutator/default_workspace_paths.go
+++ b/bundle/config/mutator/default_workspace_paths.go
@@ -26,7 +26,11 @@ func (m *defineDefaultWorkspacePaths) Apply(ctx context.Context, b *bundle.Bundl
 	}
 
 	if b.Config.Workspace.FilePath == "" {
-		b.Config.Workspace.FilePath = path.Join(root, "files")
+		if b.IsImmutableFolder() {
+			b.Config.Workspace.FilePath = "${resources.internal_immutable_snapshots.immutable.full_path}/files"
+		} else {
+			b.Config.Workspace.FilePath = path.Join(root, "files")
+		}
 	}
 
 	if b.Config.Workspace.ResourcePath == "" {
@@ -34,7 +38,11 @@ func (m *defineDefaultWorkspacePaths) Apply(ctx context.Context, b *bundle.Bundl
 	}
 
 	if b.Config.Workspace.ArtifactPath == "" {
-		b.Config.Workspace.ArtifactPath = path.Join(root, "artifacts")
+		if b.IsImmutableFolder() {
+			b.Config.Workspace.ArtifactPath = "${resources.internal_immutable_snapshots.immutable.full_path}/artifacts"
+		} else {
+			b.Config.Workspace.ArtifactPath = path.Join(root, "artifacts")
+		}
 	}
 
 	if b.Config.Workspace.StatePath == "" {
diff --git a/bundle/config/mutator/reject_internal_resources.go b/bundle/config/mutator/reject_internal_resources.go
new file mode 100644
index 00000000000..c377ecb0aaa
--- /dev/null
+++ b/bundle/config/mutator/reject_internal_resources.go
@@ -0,0 +1,34 @@
+package mutator
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+type rejectInternalResources struct{}
+
+// RejectInternalResources returns a mutator that errors when a user has set
+// any internal resource field in their bundle configuration.
+func RejectInternalResources() bundle.Mutator {
+	return &rejectInternalResources{}
+}
+
+func (m *rejectInternalResources) Name() string {
+	return "RejectInternalResources"
+}
+
+func (m *rejectInternalResources) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	var diags diag.Diagnostics
+	if b.Config.Resources.HasInternalResources() {
+		diags = diags.Append(diag.Diagnostic{
+			Severity: diag.Error,
+			Summary:  "Internal resources cannot be set in bundle configuration",
+			Paths:    []dyn.Path{dyn.MustPathFromString("resources")},
+		})
+	}
+
+	return diags
+}
diff --git a/bundle/config/mutator/resolve_variable_references.go b/bundle/config/mutator/resolve_variable_references.go
index 0b1a02680d4..257309c55f8 100644
--- a/bundle/config/mutator/resolve_variable_references.go
+++ b/bundle/config/mutator/resolve_variable_references.go
@@ -81,24 +81,6 @@ func ResolveVariableReferencesOnlyResources(prefixes ...string) bundle.Mutator {
 	}
 }
 
-// ResolveVariableReferencesOnlyResourcesExcluding is like ResolveVariableReferencesOnlyResources
-// but leaves the listed variable reference paths unresolved. Use this when a workspace path will
-// be updated by a later mutator (e.g. snapshot.Upload sets workspace.file_path to the snapshot
-// location) and the final value should be substituted at that later point.
-func ResolveVariableReferencesOnlyResourcesExcluding(excluded []string, prefixes ...string) bundle.Mutator {
-	if len(prefixes) == 0 {
-		prefixes = defaultPrefixes
-	}
-	return &resolveVariableReferences{
-		prefixes:         prefixes,
-		lookupFn:         lookup,
-		extraRounds:      maxResolutionRounds - 1,
-		pattern:          dyn.NewPattern(dyn.Key("resources")),
-		includeResources: true,
-		excludePaths:     excluded,
-	}
-}
-
 func ResolveVariableReferencesWithoutResources(prefixes ...string) bundle.Mutator {
 	if len(prefixes) == 0 {
 		prefixes = defaultPrefixes
diff --git a/bundle/config/mutator/resourcemutator/apply_bundle_permissions_test.go b/bundle/config/mutator/resourcemutator/apply_bundle_permissions_test.go
index 46262ba8dbf..29940dd3821 100644
--- a/bundle/config/mutator/resourcemutator/apply_bundle_permissions_test.go
+++ b/bundle/config/mutator/resourcemutator/apply_bundle_permissions_test.go
@@ -35,6 +35,7 @@ var unsupportedResources = []string{
 	"postgres_synced_tables",
 	"vector_search_indexes",
 	"job_runs",
+	"internal_immutable_snapshots",
 	"secrets",
 }
 
diff --git a/bundle/config/mutator/resourcemutator/apply_target_mode_test.go b/bundle/config/mutator/resourcemutator/apply_target_mode_test.go
index 53fdf89f50e..3a6eaae4b70 100644
--- a/bundle/config/mutator/resourcemutator/apply_target_mode_test.go
+++ b/bundle/config/mutator/resourcemutator/apply_target_mode_test.go
@@ -320,6 +320,12 @@ func mockBundle(mode config.Mode) *bundle.Bundle {
 						},
 					},
 				},
+				Snapshots: map[string]*resources.Snapshot{
+					"snapshot1": {
+						BundleID: "bundle1",
+						ACL:      nil,
+					},
+				},
 			},
 		},
 		SyncRoot: vfs.MustNew("/Users/lennart.kats@databricks.com"),
diff --git a/bundle/config/mutator/resourcemutator/process_static_resources.go b/bundle/config/mutator/resourcemutator/process_static_resources.go
index 28a7cbefe78..b1f689808f6 100644
--- a/bundle/config/mutator/resourcemutator/process_static_resources.go
+++ b/bundle/config/mutator/resourcemutator/process_static_resources.go
@@ -45,7 +45,7 @@ func (p processStaticResources) Apply(ctx context.Context, b *bundle.Bundle) dia
 		// Reads (dynamic): * (strings) (searches for variable references in string values)
 		// Updates (dynamic): resources.* (strings) (resolves variable references to their actual values)
 		// Resolves variable references in 'resources' using bundle, workspace, and variables prefixes
-		immutableExcludingResolver(b),
+		mutator.ResolveVariableReferencesOnlyResources(),
 		mutator.NormalizePaths(),
 
 		// Translate dashboard paths into paths in the workspace file system
diff --git a/bundle/config/mutator/resourcemutator/resource_mutator.go b/bundle/config/mutator/resourcemutator/resource_mutator.go
index e8c33f0c59b..32243687074 100644
--- a/bundle/config/mutator/resourcemutator/resource_mutator.go
+++ b/bundle/config/mutator/resourcemutator/resource_mutator.go
@@ -133,22 +133,6 @@ func applyInitializeMutators(ctx context.Context, b *bundle.Bundle) {
 	)
 }
 
-// immutableExcludingResolver returns a variable reference resolver for the resources
-// section. When experimental.immutable_folder is enabled it excludes
-// workspace.file_path, workspace.artifact_path, and workspace.snapshot_path from
-// resolution: those paths are set by snapshot.Upload() in the Deploy phase, so
-// resolving them here would freeze them to the default bundle path instead.
-// workspace.snapshot_path is also excluded so it stays as a literal ${...} template
-// in the plan output (making the pre-upload intent visible).
-func immutableExcludingResolver(b *bundle.Bundle) bundle.Mutator {
-	if b.IsImmutableFolder() {
-		return mutator.ResolveVariableReferencesOnlyResourcesExcluding(
-			[]string{"workspace.file_path", "workspace.artifact_path", "workspace.snapshot_path"},
-		)
-	}
-	return mutator.ResolveVariableReferencesOnlyResources()
-}
-
 // Normalization is applied multiple times if resource is modified during initialization
 //
 // If bundle is modified outside of 'resources' section, these changes are discarded.
@@ -162,7 +146,7 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) {
 		// Reads (dynamic): * (strings) (searches for variable references in string values)
 		// Updates (dynamic): resources.* (strings) (resolves variable references to their actual values)
 		// Resolves variable references in 'resources' using bundle, workspace, and variables prefixes
-		immutableExcludingResolver(b),
+		mutator.ResolveVariableReferencesOnlyResources(),
 
 		// Reads (dynamic): resources.pipelines.*.libraries (checks for notebook.path and file.path fields)
 		// Updates (dynamic): resources.pipelines.*.libraries (expands glob patterns in path fields to multiple library entries)
diff --git a/bundle/config/mutator/resourcemutator/run_as_test.go b/bundle/config/mutator/resourcemutator/run_as_test.go
index 87312608de8..47cfba799a9 100644
--- a/bundle/config/mutator/resourcemutator/run_as_test.go
+++ b/bundle/config/mutator/resourcemutator/run_as_test.go
@@ -43,6 +43,7 @@ func allResourceTypes(t *testing.T) []string {
 		"external_locations",
 		"genie_spaces",
 		"instance_pools",
+		"internal_immutable_snapshots",
 		"job_runs",
 		"jobs",
 		"model_serving_endpoints",
@@ -194,6 +195,7 @@ var allowList = []string{
 	"genie_spaces",
 	"instance_pools",
 	"job_runs",
+	"internal_immutable_snapshots",
 	"schemas",
 	"secret_scopes",
 	"secrets",
diff --git a/bundle/config/mutator/translate_paths.go b/bundle/config/mutator/translate_paths.go
index c44e91160b9..8e2d65d1af0 100644
--- a/bundle/config/mutator/translate_paths.go
+++ b/bundle/config/mutator/translate_paths.go
@@ -322,21 +322,17 @@ func (t *translateContext) rewriteValue(ctx context.Context, p dyn.Path, v dyn.V
 func applyTranslations(ctx context.Context, b *bundle.Bundle, t *translateContext, translations []func(context.Context, dyn.Value) (dyn.Value, error)) diag.Diagnostics {
 	switch {
 	case b.IsImmutableFolder():
-		// Reject an explicit workspace.file_path: immutable bundles control that path
-		// automatically (it is set to the content-addressed snapshot location after upload).
-		// A user-supplied value would be silently discarded, so we error early instead.
+		// Reject an explicit workspace.file_path: immutable bundles set it
+		// automatically to the content-addressed snapshot location. A user-supplied
+		// value would be silently discarded during path translation, so we error early.
 		if loc := b.Config.GetLocation("workspace.file_path"); loc.File != "" {
 			return diag.Diagnostics{{
 				Severity:  diag.Error,
-				Summary:   "workspace.file_path cannot be configured when experimental.immutable_folder is true",
+				Summary:   "workspace.file_path cannot be configured when experimental.immutable_folder is enabled",
 				Locations: []dyn.Location{loc},
 			}}
 		}
-		// Use a placeholder referencing workspace.snapshot_path so that paths are stored
-		// as ${workspace.snapshot_path}/files/<rel> during validate. After
-		// snapshot.Upload() sets workspace.snapshot_path, a variable-resolution pass
-		// expands these references to the actual content-addressed paths.
-		t.remoteRoot = "${workspace.snapshot_path}/files"
+		t.remoteRoot = "${resources.internal_immutable_snapshots.immutable.full_path}/files"
 	case config.IsExplicitlyEnabled(t.b.Config.Presets.SourceLinkedDeployment):
 		t.remoteRoot = t.b.SyncRootPath
 	default:
diff --git a/bundle/config/resources.go b/bundle/config/resources.go
index ab12ec9f052..891a72333ed 100644
--- a/bundle/config/resources.go
+++ b/bundle/config/resources.go
@@ -45,6 +45,9 @@ type Resources struct {
 	VectorSearchIndexes   map[string]*resources.VectorSearchIndex    `json:"vector_search_indexes,omitempty"`
 	InstancePools         map[string]*resources.InstancePool         `json:"instance_pools,omitempty"`
 	Secrets               map[string]*resources.Secret               `json:"secrets,omitempty"`
+
+	// Internal resources
+	Snapshots map[string]*resources.Snapshot `json:"internal_immutable_snapshots,omitempty" bundle:"internal"`
 }
 
 type ConfigResource interface {
@@ -130,6 +133,7 @@ func (r *Resources) AllResources() []ResourceGroup {
 		collectResourceMap(descriptions["vector_search_endpoints"], r.VectorSearchEndpoints),
 		collectResourceMap(descriptions["vector_search_indexes"], r.VectorSearchIndexes),
 		collectResourceMap(descriptions["instance_pools"], r.InstancePools),
+		collectResourceMap(descriptions["internal_immutable_snapshots"], r.Snapshots),
 		collectResourceMap(descriptions["secrets"], r.Secrets),
 	}
 }
@@ -159,41 +163,46 @@ func (r *Resources) FindResourceByConfigKey(key string) (ConfigResource, error)
 	return found[0], nil
 }
 
+func (r *Resources) HasInternalResources() bool {
+	return len(r.Snapshots) > 0
+}
+
 // SupportedResources returns a map which keys correspond to the resource key in the bundle configuration.
 func SupportedResources() map[string]resources.ResourceDescription {
 	return map[string]resources.ResourceDescription{
-		"jobs":                    (&resources.Job{}).ResourceDescription(),
-		"job_runs":                (&resources.JobRun{}).ResourceDescription(),
-		"pipelines":               (&resources.Pipeline{}).ResourceDescription(),
-		"models":                  (&resources.MlflowModel{}).ResourceDescription(),
-		"experiments":             (&resources.MlflowExperiment{}).ResourceDescription(),
-		"instance_pools":          (&resources.InstancePool{}).ResourceDescription(),
-		"model_serving_endpoints": (&resources.ModelServingEndpoint{}).ResourceDescription(),
-		"registered_models":       (&resources.RegisteredModel{}).ResourceDescription(),
-		"quality_monitors":        (&resources.QualityMonitor{}).ResourceDescription(),
-		"catalogs":                (&resources.Catalog{}).ResourceDescription(),
-		"schemas":                 (&resources.Schema{}).ResourceDescription(),
-		"external_locations":      (&resources.ExternalLocation{}).ResourceDescription(),
-		"clusters":                (&resources.Cluster{}).ResourceDescription(),
-		"dashboards":              (&resources.Dashboard{}).ResourceDescription(),
-		"genie_spaces":            (&resources.GenieSpace{}).ResourceDescription(),
-		"volumes":                 (&resources.Volume{}).ResourceDescription(),
-		"apps":                    (&resources.App{}).ResourceDescription(),
-		"secret_scopes":           (&resources.SecretScope{}).ResourceDescription(),
-		"alerts":                  (&resources.Alert{}).ResourceDescription(),
-		"sql_warehouses":          (&resources.SqlWarehouse{}).ResourceDescription(),
-		"database_instances":      (&resources.DatabaseInstance{}).ResourceDescription(),
-		"database_catalogs":       (&resources.DatabaseCatalog{}).ResourceDescription(),
-		"synced_database_tables":  (&resources.SyncedDatabaseTable{}).ResourceDescription(),
-		"postgres_projects":       (&resources.PostgresProject{}).ResourceDescription(),
-		"postgres_branches":       (&resources.PostgresBranch{}).ResourceDescription(),
-		"postgres_endpoints":      (&resources.PostgresEndpoint{}).ResourceDescription(),
-		"postgres_catalogs":       (&resources.PostgresCatalog{}).ResourceDescription(),
-		"postgres_databases":      (&resources.PostgresDatabase{}).ResourceDescription(),
-		"postgres_roles":          (&resources.PostgresRole{}).ResourceDescription(),
-		"postgres_synced_tables":  (&resources.PostgresSyncedTable{}).ResourceDescription(),
-		"vector_search_endpoints": (&resources.VectorSearchEndpoint{}).ResourceDescription(),
-		"vector_search_indexes":   (&resources.VectorSearchIndex{}).ResourceDescription(),
-		"secrets":                 (&resources.Secret{}).ResourceDescription(),
+		"jobs":                         (&resources.Job{}).ResourceDescription(),
+		"job_runs":                     (&resources.JobRun{}).ResourceDescription(),
+		"pipelines":                    (&resources.Pipeline{}).ResourceDescription(),
+		"models":                       (&resources.MlflowModel{}).ResourceDescription(),
+		"experiments":                  (&resources.MlflowExperiment{}).ResourceDescription(),
+		"instance_pools":               (&resources.InstancePool{}).ResourceDescription(),
+		"model_serving_endpoints":      (&resources.ModelServingEndpoint{}).ResourceDescription(),
+		"registered_models":            (&resources.RegisteredModel{}).ResourceDescription(),
+		"quality_monitors":             (&resources.QualityMonitor{}).ResourceDescription(),
+		"catalogs":                     (&resources.Catalog{}).ResourceDescription(),
+		"schemas":                      (&resources.Schema{}).ResourceDescription(),
+		"external_locations":           (&resources.ExternalLocation{}).ResourceDescription(),
+		"clusters":                     (&resources.Cluster{}).ResourceDescription(),
+		"dashboards":                   (&resources.Dashboard{}).ResourceDescription(),
+		"genie_spaces":                 (&resources.GenieSpace{}).ResourceDescription(),
+		"volumes":                      (&resources.Volume{}).ResourceDescription(),
+		"apps":                         (&resources.App{}).ResourceDescription(),
+		"secret_scopes":                (&resources.SecretScope{}).ResourceDescription(),
+		"alerts":                       (&resources.Alert{}).ResourceDescription(),
+		"sql_warehouses":               (&resources.SqlWarehouse{}).ResourceDescription(),
+		"database_instances":           (&resources.DatabaseInstance{}).ResourceDescription(),
+		"database_catalogs":            (&resources.DatabaseCatalog{}).ResourceDescription(),
+		"synced_database_tables":       (&resources.SyncedDatabaseTable{}).ResourceDescription(),
+		"postgres_projects":            (&resources.PostgresProject{}).ResourceDescription(),
+		"postgres_branches":            (&resources.PostgresBranch{}).ResourceDescription(),
+		"postgres_endpoints":           (&resources.PostgresEndpoint{}).ResourceDescription(),
+		"postgres_catalogs":            (&resources.PostgresCatalog{}).ResourceDescription(),
+		"postgres_databases":           (&resources.PostgresDatabase{}).ResourceDescription(),
+		"postgres_roles":               (&resources.PostgresRole{}).ResourceDescription(),
+		"postgres_synced_tables":       (&resources.PostgresSyncedTable{}).ResourceDescription(),
+		"vector_search_endpoints":      (&resources.VectorSearchEndpoint{}).ResourceDescription(),
+		"vector_search_indexes":        (&resources.VectorSearchIndex{}).ResourceDescription(),
+		"internal_immutable_snapshots": (&resources.Snapshot{}).ResourceDescription(),
+		"secrets":                      (&resources.Secret{}).ResourceDescription(),
 	}
 }
diff --git a/bundle/config/resources/snapshot.go b/bundle/config/resources/snapshot.go
new file mode 100644
index 00000000000..7a53eb575fc
--- /dev/null
+++ b/bundle/config/resources/snapshot.go
@@ -0,0 +1,81 @@
+package resources
+
+import (
+	"context"
+	"net/url"
+	"path"
+
+	"github.com/databricks/cli/libs/snapshot"
+	"github.com/databricks/databricks-sdk-go"
+)
+
+// Snapshot is an internal resource that stores the bundle zip as an immutable
+// workspace object. It is created by the deploy pipeline and is not intended
+// to be declared in user-authored databricks.yml files.
+//
+// JSON tags are present because the direct-deploy engine serialises the in-memory
+// state to a JSON plan file (resources.internal_immutable_snapshots.*). Fields
+// that must not leak into the plan file use json:"-".
+type Snapshot struct {
+	// BundleID is the stable UUID that identifies the bundle deployment, used
+	// as the first path component of the snapshot workspace path.
+	BundleID string `json:"bundle_id"`
+	// ACL is the access control list applied to the uploaded snapshot, granting
+	// CAN_READ to the deploying user and to every principal in bundle.permissions.
+	ACL []snapshot.ACLEntry `json:"acl"`
+	// ZipContent holds the raw zip bytes of the bundle source tree. It is
+	// populated just before upload. The counterpart SnapshotState.ZipContent
+	// carries json:"-" so the zip bytes never reach the plan file; SyncZipContent
+	// re-injects them from here when deploying from a plan.
+	ZipContent string `json:"zip_content"`
+	// RemoteRoot is the workspace root path returned by the snapshot rootpath
+	// API (e.g. /Workspace/Users/<user>/.snapshots).
+	RemoteRoot string `json:"remote_root"`
+
+	Lifecycle Lifecycle `json:"-"`
+}
+
+func (s *Snapshot) RelativePath() string {
+	return path.Join(s.BundleID, snapshot.HashFromContent([]byte(s.ZipContent)))
+}
+
+func (s *Snapshot) FullPath() string {
+	return path.Join(s.RemoteRoot, s.RelativePath())
+}
+
+func (s *Snapshot) Exists(ctx context.Context, w *databricks.WorkspaceClient, name string) (bool, error) {
+	_, err := w.Workspace.GetStatusByPath(ctx, s.FullPath())
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+func (s *Snapshot) ResourceDescription() ResourceDescription {
+	return ResourceDescription{
+		SingularName:  "internal_immutable_snapshot",
+		PluralName:    "internal_immutable_snapshots",
+		SingularTitle: "Internal Immutable Snapshot",
+		PluralTitle:   "Internal Immutable Snapshots",
+	}
+}
+
+func (s *Snapshot) GetName() string {
+	return s.RelativePath()
+}
+
+func (s *Snapshot) GetURL() string {
+	// A snapshot is a workspace folder owned by the project's service principal, so
+	// a browser URL is constructible from its path. We don't surface one yet:
+	// workspaceurls has no folder-path helper, and the path is only known during
+	// deploy (its content hash depends on ZipContent, which is empty otherwise).
+	return ""
+}
+
+func (s *Snapshot) InitializeURL(_ url.URL) {
+	// See GetURL: no browser URL is surfaced for the snapshot folder yet.
+}
+
+func (s *Snapshot) GetLifecycle() LifecycleConfig {
+	return s.Lifecycle
+}
diff --git a/bundle/config/resources_test.go b/bundle/config/resources_test.go
index d83f4da59b8..21322b418fb 100644
--- a/bundle/config/resources_test.go
+++ b/bundle/config/resources_test.go
@@ -15,6 +15,9 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/serving"
 
 	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/libs/structs/structpath"
+	"github.com/databricks/cli/libs/structs/structtag"
+	"github.com/databricks/cli/libs/structs/structwalk"
 	"github.com/databricks/cli/libs/workspaceurls"
 	"github.com/databricks/databricks-sdk-go/experimental/mocks"
 	"github.com/databricks/databricks-sdk-go/service/apps"
@@ -128,13 +131,14 @@ func TestBundleResourcePluralNamesResolveInWorkspaceURLs(t *testing.T) {
 		// A job run does have a workspace URL, but it's addressed by two IDs
 		// (job + run) so it can't be expressed as a single-ID pattern here; it's
 		// built in JobRun.InitializeURL via workspaceurls.JobRunURL instead.
-		"job_runs":           true,
-		"postgres_branches":  true,
-		"postgres_databases": true,
-		"postgres_endpoints": true,
-		"postgres_projects":  true,
-		"postgres_roles":     true,
-		"secret_scopes":      true,
+		"job_runs":                     true,
+		"postgres_branches":            true,
+		"postgres_databases":           true,
+		"postgres_endpoints":           true,
+		"postgres_projects":            true,
+		"postgres_roles":               true,
+		"secret_scopes":                true,
+		"internal_immutable_snapshots": true,
 	}
 
 	supported := SupportedResources()
@@ -351,7 +355,8 @@ func TestResourcesBindSupport(t *testing.T) {
 		},
 	}
 	unbindableResources := map[string]bool{
-		"model": true,
+		"model":                       true,
+		"internal_immutable_snapshot": true,
 	}
 
 	ctx := t.Context()
@@ -409,3 +414,35 @@ func TestResourcesBindSupport(t *testing.T) {
 		}
 	}
 }
+
+func TestAllInteralResourcesAreMarkedAsInternal(t *testing.T) {
+	internalResourceKeys := map[string]reflect.Type{}
+	err := structwalk.WalkType(reflect.TypeFor[Resources](), func(path *structpath.PatternNode, typ reflect.Type, field *reflect.StructField) bool {
+		if path.Len() > 2 {
+			return false
+		}
+		if field == nil {
+			return true
+		}
+		tag := field.Tag.Get("bundle")
+		if structtag.BundleTag(tag).Internal() {
+			internalResourceKeys[field.Name] = typ
+		}
+		return true
+	})
+	assert.NoError(t, err)
+
+	for key, typ := range internalResourceKeys {
+		r := reflect.MakeMap(typ)
+		r.SetMapIndex(reflect.ValueOf("my_resources"), reflect.New(typ.Elem()).Elem())
+
+		resources := &Resources{}
+		res := reflect.ValueOf(resources).Elem()
+		field := res.FieldByName(key)
+		if !field.IsValid() && !field.CanSet() {
+			t.Fatalf("Field %s is not valid", key)
+		}
+		field.Set(r)
+		assert.True(t, resources.HasInternalResources())
+	}
+}
diff --git a/bundle/config/validate/validate_artifact_path.go b/bundle/config/validate/validate_artifact_path.go
index 4ea5c4308ad..e7070b20b02 100644
--- a/bundle/config/validate/validate_artifact_path.go
+++ b/bundle/config/validate/validate_artifact_path.go
@@ -76,7 +76,7 @@ func findVolumeInBundle(r config.Root, catalogName, schemaName, volumeName strin
 
 func (v *validateArtifactPath) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	// We only validate UC Volumes paths right now.
-	if !libraries.IsVolumesPath(b.Config.Workspace.ArtifactPath) {
+	if !libraries.IsVolumesPath(b.Config.Workspace.ArtifactPath) || b.IsImmutableFolder() {
 		return nil
 	}
 
diff --git a/bundle/config/workspace.go b/bundle/config/workspace.go
index f68670b7e86..9e5d2522d0d 100644
--- a/bundle/config/workspace.go
+++ b/bundle/config/workspace.go
@@ -78,12 +78,6 @@ type Workspace struct {
 	// Remote workspace path for deployment state.
 	// This defaults to "${workspace.root}/state".
 	StatePath string `json:"state_path,omitempty"`
-
-	// SnapshotPath is the workspace path of the immutable snapshot uploaded during
-	// deployment. Set by snapshot.Upload() and used by the subsequent variable-resolution
-	// pass to expand ${workspace.snapshot_path} placeholders in resource configs.
-	// Only populated at runtime for bundles with experimental.immutable_folder = true.
-	SnapshotPath string `json:"snapshot_path,omitempty" bundle:"internal"`
 }
 
 type User struct {
diff --git a/bundle/deploy/snapshot/path.go b/bundle/deploy/snapshot/path.go
index 62a4b9f91ab..b82cf42a336 100644
--- a/bundle/deploy/snapshot/path.go
+++ b/bundle/deploy/snapshot/path.go
@@ -4,8 +4,6 @@ import (
 	"archive/zip"
 	"bytes"
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -17,6 +15,7 @@ import (
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/deploy/files"
 	"github.com/databricks/cli/libs/fileset"
+	"github.com/databricks/cli/libs/snapshot"
 	libsync "github.com/databricks/cli/libs/sync"
 )
 
@@ -71,12 +70,12 @@ func BundleZip(ctx context.Context, b *bundle.Bundle) ([]byte, int, error) {
 
 // addMetadataToZip writes the snapshot metadata file into the zip so that
 // any change to the ACL changes the snapshot hash and forces a new snapshot.
-func addMetadataToZip(zw *zip.Writer, acl []ACLEntry) error {
+func addMetadataToZip(zw *zip.Writer, acl []snapshot.ACLEntry) error {
 	aclJSON, err := json.Marshal(acl)
 	if err != nil {
 		return fmt.Errorf("marshal ACL for permissions hash: %w", err)
 	}
-	data, err := json.Marshal(snapshotMetadata{PermissionsHash: IDFromContent(aclJSON)})
+	data, err := json.Marshal(snapshotMetadata{PermissionsHash: snapshot.HashFromContent(aclJSON)})
 	if err != nil {
 		return fmt.Errorf("marshal snapshot metadata: %w", err)
 	}
@@ -93,23 +92,6 @@ func addMetadataToZip(zw *zip.Writer, acl []ACLEntry) error {
 	return err
 }
 
-// IDFromContent returns the SHA-256 hex digest of content.
-func IDFromContent(content []byte) string {
-	h := sha256.Sum256(content)
-	return hex.EncodeToString(h[:])
-}
-
-// SnapshotID builds the bundle zip and returns its SHA-256 hex digest.
-// Called after artifacts are built so that ApplyImmutableWorkspacePaths and
-// snapshot.Upload both hash identical content.
-func SnapshotID(ctx context.Context, b *bundle.Bundle) (string, error) {
-	content, _, err := BundleZip(ctx, b)
-	if err != nil {
-		return "", err
-	}
-	return IDFromContent(content), nil
-}
-
 // addSyncRootToZip returns the number of files added from the sync root.
 func addSyncRootToZip(ctx context.Context, zw *zip.Writer, b *bundle.Bundle) (int, error) {
 	opts, err := files.GetSyncOptions(ctx, b)
@@ -168,8 +150,6 @@ func addArtifactsToZip(zw *zip.Writer, b *bundle.Bundle) error {
 			if af.Patched != "" {
 				source = af.Patched
 			}
-			// ".internal" matches libraries.InternalDirName so that ReplaceWithRemotePath
-			// produces library paths that resolve correctly inside the snapshot.
 			if err := addLocalFileToZip(zw, source, "artifacts/.internal"); err != nil {
 				return err
 			}
diff --git a/bundle/deploy/snapshot/path_test.go b/bundle/deploy/snapshot/path_test.go
index 5f567d9f377..54407a98e20 100644
--- a/bundle/deploy/snapshot/path_test.go
+++ b/bundle/deploy/snapshot/path_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deploy/snapshot"
+	libsnapshot "github.com/databricks/cli/libs/snapshot"
 	"github.com/databricks/cli/libs/vfs"
 	"github.com/databricks/databricks-sdk-go/service/iam"
 	"github.com/stretchr/testify/assert"
@@ -93,26 +94,6 @@ func TestBundleZipRespectsExcludes(t *testing.T) {
 	assert.Less(t, len(zipExcl), len(zipAll))
 }
 
-func TestIDFromContent(t *testing.T) {
-	id := snapshot.IDFromContent([]byte("hello"))
-	// SHA-256 of "hello"
-	assert.Equal(t, "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824", id)
-	assert.Len(t, id, 64, "SHA-256 hex must be 64 characters")
-}
-
-func TestSnapshotIDMatchesBundleZipHash(t *testing.T) {
-	b := makeBundleWithFiles(t, map[string]string{"task.py": "x = 1"})
-
-	zipContent, _, err := snapshot.BundleZip(t.Context(), b)
-	require.NoError(t, err)
-	expectedID := snapshot.IDFromContent(zipContent)
-
-	id, err := snapshot.SnapshotID(t.Context(), b)
-	require.NoError(t, err)
-
-	assert.Equal(t, expectedID, id)
-}
-
 func zipEntryNames(t *testing.T, zipContent []byte) []string {
 	t.Helper()
 	r, err := zip.NewReader(bytes.NewReader(zipContent), int64(len(zipContent)))
@@ -166,6 +147,6 @@ func TestBundleZipChangesWithPermissions(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.NotEqual(t, zipNoPerms, zipWithPerms, "adding top-level permissions must produce a different snapshot zip")
-	assert.NotEqual(t, snapshot.IDFromContent(zipNoPerms), snapshot.IDFromContent(zipWithPerms),
+	assert.NotEqual(t, libsnapshot.HashFromContent(zipNoPerms), libsnapshot.HashFromContent(zipWithPerms),
 		"snapshot IDs must differ when top-level permissions change")
 }
diff --git a/bundle/deploy/snapshot/upload.go b/bundle/deploy/snapshot/upload.go
index 4a73e63b35a..8b20474b4c5 100644
--- a/bundle/deploy/snapshot/upload.go
+++ b/bundle/deploy/snapshot/upload.go
@@ -3,27 +3,27 @@ package snapshot
 import (
 	"context"
 	"fmt"
-	"path"
 
 	"github.com/databricks/cli/bundle"
-	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/bundle/direct/dresources"
 	"github.com/databricks/cli/libs/diag"
-	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/snapshot"
+	"github.com/google/uuid"
 )
 
 // fileLimitWarning is the file count above which immutable folder deployments may fail.
 const fileLimitWarning = 1000
 
 type snapshotUpload struct {
-	// uploader allows test injection of a custom SnapshotUploader.
-	uploader SnapshotUploader
+	skipZip bool
 }
 
-// Upload returns a mutator that builds the bundle zip, uploads it via
-// /api/2.0/repos/snapshots, and updates workspace.file_path and
-// workspace.artifact_path to the content-addressed location returned by the API.
-func Upload() bundle.Mutator {
-	return &snapshotUpload{}
+// PlanUpload returns a mutator that registers the immutable snapshot as an internal
+// resource. Unless skipZip is set, it also builds the bundle zip and stages it in
+// memory on the resource; the zip is uploaded when the resource is created on apply.
+func PlanUpload(skipZip bool) bundle.Mutator {
+	return &snapshotUpload{skipZip: skipZip}
 }
 
 func (m *snapshotUpload) Name() string {
@@ -31,63 +31,88 @@ func (m *snapshotUpload) Name() string {
 }
 
 func (m *snapshotUpload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
-	uploader := m.uploader
-	if uploader == nil {
-		var err error
-		uploader, err = NewSnapshotUploader(b.WorkspaceClient(ctx))
-		if err != nil {
-			return diag.FromErr(err)
-		}
+	uploader, err := snapshot.NewSnapshotClient(b.WorkspaceClient(ctx))
+	if err != nil {
+		return diag.FromErr(err)
 	}
 
-	cmdio.LogProgress(ctx, "Uploading immutable bundle snapshot...")
-
-	zipContent, fileCount, err := BundleZip(ctx, b)
+	remoteRoot, err := uploader.GetSnapshotRootPath(ctx)
 	if err != nil {
-		return diag.FromErr(fmt.Errorf("failed to build snapshot zip: %w", err))
+		return diag.FromErr(err)
 	}
-	var diags diag.Diagnostics
-	if fileCount > fileLimitWarning {
-		diags = append(diags, diag.Warningf(
-			"immutable folder deployment may not work correctly: bundle contains %d files (limit is %d)",
-			fileCount, fileLimitWarning,
-		)...)
+
+	if b.Config.Resources.Snapshots == nil {
+		b.Config.Resources.Snapshots = make(map[string]*resources.Snapshot)
 	}
-	snapshotID := IDFromContent(zipContent)
-	log.Debugf(ctx, "snapshot.Upload: snapshotID=%s zip=%d bytes", snapshotID, len(zipContent))
-
-	acl := BuildACL(b)
-	// Use the deployment lineage UUID as bundle_id so the snapshot directory is
-	// keyed to this specific deployment (not to the bundle name, which can be
-	// reused across unrelated deployments).
-	bundleID := b.DeploymentBundle.StateDB.GetOrInitLineage()
-	info, err := uploader.Upload(ctx, bundleID, snapshotID, acl, zipContent)
-	if err != nil {
-		return diag.FromErr(err)
+	if _, ok := b.Config.Resources.Snapshots["immutable"]; !ok {
+		b.Config.Resources.Snapshots["immutable"] = &resources.Snapshot{
+			BundleID:   BundleID(b),
+			ACL:        BuildACL(b),
+			RemoteRoot: remoteRoot,
+		}
 	}
 
-	log.Infof(ctx, "Snapshot uploaded to %s", info.Path)
+	var diags diag.Diagnostics
+	if !m.skipZip {
+		zipContent, fileCount, err := BundleZip(ctx, b)
+		if err != nil {
+			return diag.FromErr(fmt.Errorf("failed to build snapshot zip: %w", err))
+		}
+
+		if fileCount > fileLimitWarning {
+			diags = append(diags, diag.Warningf(
+				"immutable folder deployment may not work correctly: bundle contains %d files (limit is %d)",
+				fileCount, fileLimitWarning,
+			)...)
+		}
 
-	b.Config.Workspace.SnapshotPath = info.Path
-	b.Config.Workspace.FilePath = path.Join(info.Path, "files")
-	// Only set artifact_path when artifacts are present; with no artifacts the
-	// zip has no "artifacts" directory and a get-status on it would 404.
-	if len(b.Config.Artifacts) > 0 {
-		b.Config.Workspace.ArtifactPath = path.Join(info.Path, "artifacts")
+		b.Config.Resources.Snapshots["immutable"].ZipContent = string(zipContent)
 	}
 
 	return diags
 }
 
+// SyncZipContent copies the zip content from b.Config.Resources.Snapshots["immutable"]
+// into the in-memory state cache entry for the snapshot resource. This is needed when
+// deploying from a plan file: the plan JSON omits ZipContent (json:"-"), so InitForApply
+// leaves it empty, causing DoCreate to upload an empty zip and derive a wrong snapshot ID.
+func SyncZipContent(b *bundle.Bundle) {
+	snap := b.Config.Resources.Snapshots["immutable"]
+	if snap == nil || snap.ZipContent == "" {
+		return
+	}
+	sv, ok := b.DeploymentBundle.StateCache.Load("resources.internal_immutable_snapshots.immutable")
+	if !ok {
+		return
+	}
+	state, ok := sv.Value.(*dresources.SnapshotState)
+	if !ok {
+		return
+	}
+	state.ZipContent = snap.ZipContent
+}
+
+// bundleIDNamespace is the UUID namespace used to derive the bundle ID.
+var bundleIDNamespace = uuid.MustParse("4b4e4b5a-3c3d-4e4f-8b8c-9d9e9f0a0b0c")
+
+// BundleID returns a stable UUID that identifies the bundle deployment.
+// It is derived deterministically from workspace.state_path, which is the
+// canonical unique identifier for a deployment (name, target, and workspace root
+// are all encoded in it). Two bundles with the same name and target but different
+// workspace.state_path values get distinct IDs.
+func BundleID(b *bundle.Bundle) string {
+	return uuid.NewSHA1(bundleIDNamespace, []byte(b.Config.Workspace.StatePath)).String()
+}
+
 // BuildACL constructs the access_control_list for the snapshot upload.
 // It grants CAN_READ to the current user and to every principal listed in the
 // top-level permissions section of the bundle config.
-func BuildACL(b *bundle.Bundle) []ACLEntry {
-	acl := []ACLEntry{
+func BuildACL(b *bundle.Bundle) []snapshot.ACLEntry {
+	acl := []snapshot.ACLEntry{
 		{UserName: b.Config.Workspace.CurrentUser.UserName, PermissionLevel: "CAN_READ"},
 	}
 	for _, p := range b.Config.Permissions {
-		acl = append(acl, ACLEntry{
+		acl = append(acl, snapshot.ACLEntry{
 			UserName:             p.UserName,
 			GroupName:            p.GroupName,
 			ServicePrincipalName: p.ServicePrincipalName,
diff --git a/bundle/deploy/snapshot/upload_test.go b/bundle/deploy/snapshot/upload_test.go
index 3b53f31ae3c..5bace57e2be 100644
--- a/bundle/deploy/snapshot/upload_test.go
+++ b/bundle/deploy/snapshot/upload_test.go
@@ -7,6 +7,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deploy/snapshot"
+	libsnapshot "github.com/databricks/cli/libs/snapshot"
 	"github.com/databricks/databricks-sdk-go/service/iam"
 	"github.com/stretchr/testify/assert"
 )
@@ -29,7 +30,7 @@ func TestBuildACLCurrentUserOnly(t *testing.T) {
 
 	acl := snapshot.BuildACL(b)
 
-	assert.Equal(t, []snapshot.ACLEntry{
+	assert.Equal(t, []libsnapshot.ACLEntry{
 		{UserName: "alice@example.com", PermissionLevel: "CAN_READ"},
 	}, acl)
 }
@@ -44,7 +45,7 @@ func TestBuildACLWithTopLevelPermissions(t *testing.T) {
 
 	acl := snapshot.BuildACL(b)
 
-	assert.Equal(t, []snapshot.ACLEntry{
+	assert.Equal(t, []libsnapshot.ACLEntry{
 		{UserName: "alice@example.com", PermissionLevel: "CAN_READ"},
 		{UserName: "bob@example.com", PermissionLevel: "CAN_READ"},
 		{GroupName: "devs", PermissionLevel: "CAN_READ"},
diff --git a/bundle/deploy/snapshot/upload_warning_test.go b/bundle/deploy/snapshot/upload_warning_test.go
index ae45c45a87d..778c6b4795c 100644
--- a/bundle/deploy/snapshot/upload_warning_test.go
+++ b/bundle/deploy/snapshot/upload_warning_test.go
@@ -3,6 +3,7 @@ package snapshot
 import (
 	"context"
 	"fmt"
+	"math"
 	"os"
 	"path/filepath"
 	"testing"
@@ -12,16 +13,25 @@ import (
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/cli/libs/testserver"
 	"github.com/databricks/cli/libs/vfs"
+	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/service/iam"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-type mockUploader struct{ path string }
-
-func (m *mockUploader) Upload(_ context.Context, _, _ string, _ []ACLEntry, _ []byte) (*SnapshotInfo, error) {
-	return &SnapshotInfo{Path: m.path}, nil
+func setupTestClient(t *testing.T) *databricks.WorkspaceClient {
+	t.Helper()
+	server := testserver.New(t)
+	testserver.AddDefaultHandlers(server)
+	client, err := databricks.NewWorkspaceClient(&databricks.Config{
+		Host:               server.URL,
+		Token:              "testtoken",
+		RateLimitPerSecond: math.MaxInt,
+	})
+	require.NoError(t, err)
+	return client
 }
 
 func makeBundle(t *testing.T, nFiles int) *bundle.Bundle {
@@ -59,19 +69,20 @@ func testContext(t *testing.T) context.Context {
 
 func TestUploadWarnsAboveFileLimit(t *testing.T) {
 	b := makeBundle(t, fileLimitWarning+1)
-	m := &snapshotUpload{uploader: &mockUploader{path: "/snapshots/test"}}
+	b.SetWorkpaceClient(setupTestClient(t))
+	m := &snapshotUpload{}
 
 	diags := m.Apply(testContext(t), b)
 
 	require.Len(t, diags, 1)
 	assert.Equal(t, diag.Warning, diags[0].Severity)
 	assert.Contains(t, diags[0].Summary, fmt.Sprintf("%d files", fileLimitWarning+1))
-	assert.Equal(t, "/snapshots/test", b.Config.Workspace.SnapshotPath)
 }
 
 func TestUploadNoWarningBelowFileLimit(t *testing.T) {
 	b := makeBundle(t, 5)
-	m := &snapshotUpload{uploader: &mockUploader{path: "/snapshots/test"}}
+	b.SetWorkpaceClient(setupTestClient(t))
+	m := &snapshotUpload{}
 
 	diags := m.Apply(testContext(t), b)
 
diff --git a/bundle/deploy/terraform/lifecycle_test.go b/bundle/deploy/terraform/lifecycle_test.go
index f8a5140576f..a0109848b16 100644
--- a/bundle/deploy/terraform/lifecycle_test.go
+++ b/bundle/deploy/terraform/lifecycle_test.go
@@ -23,6 +23,7 @@ func TestConvertLifecycleForAllResources(t *testing.T) {
 		"secrets",
 		"vector_search_endpoints",
 		"vector_search_indexes",
+		"internal_immutable_snapshots",
 	}
 
 	for resourceType := range supportedResources {
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 5b8829e3f59..adec3882c06 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -721,12 +721,6 @@ func splitResourcePath(path *structpath.PathNode) (string, *structpath.PathNode)
 }
 
 func (b *DeploymentBundle) LookupReferencePreDeploy(ctx context.Context, path *structpath.PathNode) (any, error) {
-	// ${workspace.snapshot_path} is resolved by the mutator pipeline after
-	// snapshot.Upload() — not by the direct engine. Return errDelayed so the
-	// template string is preserved in the plan output rather than causing an error.
-	if path.String() == "workspace.snapshot_path" {
-		return nil, errDelayed
-	}
 	targetResourceKey, fieldPath := splitResourcePath(path)
 	targetGroup := config.GetResourceTypeFromKey(targetResourceKey)
 
@@ -784,9 +778,9 @@ func (b *DeploymentBundle) LookupReferencePreDeploy(ctx context.Context, path *s
 
 	localConfig := sv.Value
 
-	adapter := b.Adapters[targetGroup]
-	if adapter == nil {
-		return nil, fmt.Errorf("internal error: %s: unknown resource type %q", targetResourceKey, targetGroup)
+	adapter, err := b.getAdapterForKey(targetResourceKey)
+	if err != nil {
+		return nil, fmt.Errorf("internal error: %s: %w", targetResourceKey, err)
 	}
 
 	configValidErr := structaccess.ValidatePath(reflect.TypeOf(localConfig), fieldPath)
@@ -960,7 +954,6 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root
 	}
 
 	slices.Sort(nodes)
-
 	for _, node := range nodes {
 		delete(existingKeys, node)
 
@@ -1023,9 +1016,7 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root
 
 				targetNodeDP, _ := config.GetNodeAndType(targetPathParsed)
 				targetNode := targetNodeDP.String()
-				// ${workspace.snapshot_path} is resolved by the mutator pipeline after
-				// snapshot.Upload(), not by the direct engine — skip it here.
-				if targetPath == "workspace.snapshot_path" {
+				if targetNode == "" {
 					continue
 				}
 
diff --git a/bundle/direct/dresources/all.go b/bundle/direct/dresources/all.go
index ad310468da0..ef285de031f 100644
--- a/bundle/direct/dresources/all.go
+++ b/bundle/direct/dresources/all.go
@@ -67,6 +67,9 @@ var SupportedResources = map[string]any{
 	"registered_models.grants":     (*ResourceGrants)(nil),
 	"vector_search_indexes.grants": (*ResourceGrants)(nil),
 	"secrets.grants":               (*ResourceGrants)(nil),
+
+	// Internal resources
+	"internal_immutable_snapshots": (*ResourceSnapshot)(nil),
 }
 
 func InitAll(client *databricks.WorkspaceClient) (map[string]*Adapter, error) {
diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go
index 75a94ffc88c..db41e575c15 100644
--- a/bundle/direct/dresources/all_test.go
+++ b/bundle/direct/dresources/all_test.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/libs/snapshot"
 	"github.com/databricks/cli/libs/structs/structaccess"
 	"github.com/databricks/cli/libs/structs/structdiff"
 	"github.com/databricks/cli/libs/structs/structpath"
@@ -307,6 +308,12 @@ var testConfig map[string]any = map[string]any{
 			Privileges: []catalog.Privilege{catalog.PrivilegeSelect},
 		}},
 	},
+
+	"internal_immutable_snapshots": &resources.Snapshot{
+		RemoteRoot: "/Workspace/Users/" + testserver.TestUserSP.UserName + "/.snapshots",
+		BundleID:   "test-bundle-id",
+		ACL:        []snapshot.ACLEntry{{UserName: "user@example.com", PermissionLevel: "CAN_READ"}},
+	},
 }
 
 type prepareWorkspace func(ctx context.Context, client *databricks.WorkspaceClient) (any, error)
@@ -911,6 +918,7 @@ func TestAll(t *testing.T) {
 // testIgnoreFilter encapsulates the logic for filtering fields based on ignore_remote_changes config.
 type testIgnoreFilter struct {
 	ignoreFields map[string]bool
+	adapter      *Adapter
 }
 
 // newTestIgnoreFilter creates a filter from the adapter's resource configs.
@@ -938,7 +946,7 @@ func newTestIgnoreFilter(adapter *Adapter) *testIgnoreFilter {
 		}
 		return true
 	})
-	return &testIgnoreFilter{ignoreFields: ignoreFields}
+	return &testIgnoreFilter{ignoreFields: ignoreFields, adapter: adapter}
 }
 
 // shouldIgnore returns true if the field at the given path should be ignored.
@@ -951,7 +959,17 @@ func (f *testIgnoreFilter) shouldIgnore(path string) bool {
 	if prefix, _, ok := strings.Cut(path, "."); ok {
 		topLevelField = prefix
 	}
-	return f.ignoreFields[topLevelField]
+
+	parts := strings.Split(topLevelField, "[")
+	if len(parts) > 1 {
+		topLevelField = parts[0]
+	}
+
+	if f.ignoreFields[topLevelField] {
+		return true
+	}
+
+	return structaccess.ValidatePath(f.adapter.RemoteType(), structpath.MustParsePath(path)) != nil
 }
 
 // filterChanges returns only the changes that should not be ignored.
@@ -1079,6 +1097,10 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	}
 
 	require.NoError(t, structwalk.Walk(newState, func(path *structpath.PathNode, val any, field *reflect.StructField) {
+		// Skip fields configured in ignore_remote_changes.
+		if ignoreFilter.shouldIgnore(path.String()) {
+			return
+		}
 		remoteValue, err := structaccess.Get(remappedState, path)
 		if err != nil {
 			t.Errorf("Failed to read %s from remapped remote state %#v", path.String(), remappedState)
@@ -1093,10 +1115,6 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 			// testserver can set field to backend-generated value
 			return
 		}
-		// Skip fields configured in ignore_remote_changes.
-		if ignoreFilter.shouldIgnore(path.String()) {
-			return
-		}
 		// t.Logf("Testing %s v=%#v, remoteValue=%#v", path.String(), val, remoteValue)
 		// We expect fields set explicitly to be preserved by testserver, which is true for all resources as of today.
 		// If not true for your resource, add exception here:
@@ -1118,6 +1136,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	}
 
 	deleteIsNoop := strings.HasSuffix(group, "permissions") || strings.HasSuffix(group, "grants")
+	isImmutable := strings.HasSuffix(group, "internal_immutable_snapshots")
 	// Apps DoDelete is fire-and-forget: the API returns success while the app
 	// sits in DELETING state for up to ~20 minutes before the record is removed.
 	// A GET on the DELETING app returns the app, not 404 -- the testserver
@@ -1128,6 +1147,9 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 
 	remoteAfterDelete, err := adapter.DoRead(ctx, createdID)
 	switch {
+	case isImmutable:
+		require.NoError(t, err)
+		assert.True(t, adapter.IsGone(remoteAfterDelete))
 	case deleteIsNoop:
 		require.NoError(t, err)
 		// The resource genuinely still exists, so it must not report as gone.
diff --git a/bundle/direct/dresources/resources.yml b/bundle/direct/dresources/resources.yml
index 0bc65b0a660..6cf5ea299f5 100644
--- a/bundle/direct/dresources/resources.yml
+++ b/bundle/direct/dresources/resources.yml
@@ -846,3 +846,7 @@ resources:
     backend_defaults:
       # The Vector Search API assigns index_subtype when the config omits it
       - field: index_subtype
+
+  internal_immutable_snapshots:
+    recreate_on_changes:
+      - reason: immutable
diff --git a/bundle/direct/dresources/snapshot.go b/bundle/direct/dresources/snapshot.go
new file mode 100644
index 00000000000..872aa424f62
--- /dev/null
+++ b/bundle/direct/dresources/snapshot.go
@@ -0,0 +1,100 @@
+package dresources
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/libs/snapshot"
+	"github.com/databricks/databricks-sdk-go"
+)
+
+type ResourceSnapshot struct {
+	uploader *snapshot.SnapshotClient
+}
+
+type SnapshotState struct {
+	RemoteRoot   string              `json:"remote_root"`
+	RelativePath string              `json:"relative_path"`
+	FullPath     string              `json:"full_path"`
+	BundleID     string              `json:"bundle_id"`
+	ACL          []snapshot.ACLEntry `json:"acl"`
+	ZipContent   string              `json:"-"`
+}
+
+type SnapshotRemote struct {
+	RelativePath string `json:"relative_path"`
+	FullPath     string `json:"full_path"`
+}
+
+func (s *ResourceSnapshot) New(client *databricks.WorkspaceClient) *ResourceSnapshot {
+	// Return a zero-value instance when client is nil (e.g. refschema introspection).
+	if client == nil {
+		return &ResourceSnapshot{
+			uploader: nil,
+		}
+	}
+
+	uploader, err := snapshot.NewSnapshotClient(client)
+	if err != nil {
+		panic(err)
+	}
+
+	return &ResourceSnapshot{
+		uploader: uploader,
+	}
+}
+
+func (s *ResourceSnapshot) PrepareState(input *resources.Snapshot) *SnapshotState {
+	return &SnapshotState{
+		RemoteRoot:   input.RemoteRoot,
+		RelativePath: input.RelativePath(),
+		FullPath:     input.FullPath(),
+		BundleID:     input.BundleID,
+		ACL:          input.ACL,
+		ZipContent:   input.ZipContent,
+	}
+}
+
+func (s *ResourceSnapshot) RemapState(remote *SnapshotRemote) *SnapshotState {
+	return &SnapshotState{
+		RemoteRoot:   "",
+		RelativePath: remote.RelativePath,
+		FullPath:     remote.FullPath,
+		BundleID:     "",
+		ACL:          nil,
+		ZipContent:   "",
+	}
+}
+
+func (s *ResourceSnapshot) DoRead(ctx context.Context, id string) (*SnapshotRemote, error) {
+	info, err := s.uploader.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	return &SnapshotRemote{
+		RelativePath: id,
+		FullPath:     info.Path,
+	}, nil
+}
+
+func (s *ResourceSnapshot) DoCreate(ctx context.Context, state *SnapshotState) (string, *SnapshotRemote, error) {
+	path := state.RelativePath
+	info, err := s.uploader.Upload(ctx, path, state.BundleID, state.ACL, []byte(state.ZipContent))
+	if err != nil {
+		return "", nil, err
+	}
+	return path, &SnapshotRemote{RelativePath: path, FullPath: info.Path}, nil
+}
+
+func (s *ResourceSnapshot) DoUpdate(ctx context.Context, id string, newState *SnapshotState, entry *PlanEntry) (*SnapshotRemote, error) {
+	return nil, nil
+}
+
+func (s *ResourceSnapshot) DoDelete(ctx context.Context, id string, state *SnapshotState) error {
+	return nil
+}
+
+// IsGone treats a snapshot as already-deleted. The snapshot is immutable, so it can't be deleted.
+func (s *ResourceSnapshot) IsGone(remote *SnapshotRemote) bool {
+	return true
+}
diff --git a/bundle/direct/dresources/type_test.go b/bundle/direct/dresources/type_test.go
index 2d5516d59c7..b41862fb729 100644
--- a/bundle/direct/dresources/type_test.go
+++ b/bundle/direct/dresources/type_test.go
@@ -56,6 +56,11 @@ var knownMissingInRemoteType = map[string][]string{
 	"vector_search_endpoints": {
 		"usage_policy_id",
 	},
+	"internal_immutable_snapshots": {
+		"bundle_id",
+		"acl",
+		"remote_root",
+  },
 	"job_runs": {
 		// Local-only trigger fingerprints under lifecycle.
 		"lifecycle",
@@ -102,6 +107,9 @@ var knownMissingInStateType = map[string][]string{
 		"keyvault_metadata",
 		"name",
 	},
+	"internal_immutable_snapshots": {
+		"zip_content",
+	},
 }
 
 // TestInputSubset validates that all fields in InputType exist in StateType.
diff --git a/bundle/libraries/filer.go b/bundle/libraries/filer.go
index 762732262be..ddd178ae948 100644
--- a/bundle/libraries/filer.go
+++ b/bundle/libraries/filer.go
@@ -26,6 +26,9 @@ func GetFilerForLibraries(ctx context.Context, b *bundle.Bundle) (filer.Filer, s
 
 	uploadPath := path.Join(b.Config.Workspace.ArtifactPath, InternalDirName)
 	uploadPath = ensureWorkspaceOrVolumesPrefix(uploadPath)
+	if b.IsImmutableFolder() {
+		uploadPath = path.Join("${resources.internal_immutable_snapshots.immutable.full_path}", "artifacts", InternalDirName)
+	}
 
 	switch {
 	case IsVolumesPath(artifactPath):
@@ -43,6 +46,9 @@ func GetFilerForLibrariesCleanup(ctx context.Context, b *bundle.Bundle) (filer.F
 	}
 
 	artifactPath = ensureWorkspaceOrVolumesPrefix(artifactPath)
+	if b.IsImmutableFolder() {
+		artifactPath = path.Join("${resources.internal_immutable_snapshots.immutable.full_path}", "artifacts")
+	}
 
 	switch {
 	case IsVolumesPath(artifactPath):
diff --git a/bundle/phases/build.go b/bundle/phases/build.go
index 580a18f7ab6..c5e3d8f28c3 100644
--- a/bundle/phases/build.go
+++ b/bundle/phases/build.go
@@ -56,12 +56,6 @@ func Build(ctx context.Context, b *bundle.Bundle) LibLocationMap {
 		return nil
 	}
 
-	// For immutable bundles, library remote paths are set in the deploy phase
-	// after snapshot.Upload() provides the content-addressed workspace.artifact_path.
-	if b.IsImmutableFolder() {
-		return nil
-	}
-
 	libs, diags := libraries.ReplaceWithRemotePath(ctx, b)
 	for _, d := range diags {
 		logdiag.LogDiag(ctx, d)
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index bc06575baf3..74b1c8f3488 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -11,7 +11,6 @@ import (
 	"github.com/databricks/cli/bundle/artifacts"
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/engine"
-	"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"
@@ -110,7 +109,8 @@ func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, st
 		return
 	}
 
-	bundle.ApplySeqContext(ctx, b,
+	bundle.ApplySeqContext(
+		ctx, b,
 		statemgmt.Load(state),
 		metadata.Compute(),
 		metadata.Upload(),
@@ -177,7 +177,8 @@ func logDeploySummary(ctx context.Context, b *bundle.Bundle, plan *deployplan.Pl
 // 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,
+	bundle.ApplySeqContext(
+		ctx, b,
 		artifacts.CleanUp(),
 		libraries.Upload(libs),
 	)
@@ -193,7 +194,8 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 
 	// Core mutators that CRUD resources and modify deployment state. These
 	// mutators need informed consent if they are potentially destructive.
-	bundle.ApplySeqContext(ctx, b,
+	bundle.ApplySeqContext(
+		ctx, b,
 		scripts.Execute(config.ScriptPreDeploy),
 		lock.Acquire(lock.GoalDeploy),
 	)
@@ -214,29 +216,12 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 		return
 	}
 
-	if immutable {
-		// Upload all source files and built artifacts as a single immutable snapshot.
-		// snapshot.Upload() sets workspace.snapshot_path; the variable-resolution
-		// pass expands ${workspace.snapshot_path} placeholders written by translate_paths.
-		bundle.ApplySeqContext(ctx, b,
-			snapshot.Upload(),
-			mutator.ResolveVariableReferencesOnlyResources("workspace"),
-		)
-		if !logdiag.HasError(ctx) {
-			_, libDiags := libraries.ReplaceWithRemotePath(ctx, b)
-			for _, d := range libDiags {
-				logdiag.LogDiag(ctx, d)
-			}
-		}
-	} else {
+	if !immutable {
 		uploadLibraries(ctx, b, libs)
-	}
-
-	if logdiag.HasError(ctx) {
-		return
-	}
+		if logdiag.HasError(ctx) {
+			return
+		}
 
-	if !immutable {
 		bundle.ApplySeqContext(ctx, b, files.Upload(outputHandler))
 		if logdiag.HasError(ctx) {
 			return
@@ -267,6 +252,13 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 		return
 	}
 
+	if immutable {
+		bundle.ApplyContext(ctx, b, snapshot.PlanUpload(false))
+		if logdiag.HasError(ctx) {
+			return
+		}
+	}
+
 	planFromFile := plan != nil
 	if plan == nil {
 		// State is already open for read by process.go (for direct engine)
@@ -295,6 +287,12 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 			logdiag.LogError(ctx, err)
 			return
 		}
+		if immutable {
+			// The plan JSON omits ZipContent (json:"-"), so InitForApply leaves it
+			// empty in the state cache. Transfer the zip content built by PlanUpload
+			// above so DoCreate uploads the correct content and derives the right ID.
+			snapshot.SyncZipContent(b)
+		}
 	}
 
 	// InitForApply receives ctx and could log a diagnostic without returning an
@@ -348,7 +346,8 @@ func RunPlan(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) *d
 	// b.Select is rejected for the terraform engine in ProcessBundleRet, so it is
 	// never set here.
 
-	bundle.ApplySeqContext(ctx, b,
+	bundle.ApplySeqContext(
+		ctx, b,
 		terraform.Interpolate(),
 		terraform.Write(),
 		terraform.Plan(terraform.PlanGoal("deploy")),
diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go
index e19dd11fb56..5d608744b6b 100644
--- a/bundle/phases/destroy.go
+++ b/bundle/phases/destroy.go
@@ -162,13 +162,14 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, e
 
 	if !logdiag.HasError(ctx) && b.Quiet < bundle.QuietAll {
 		// Count top-level resources only, matching the approval list above (which
-		// skips children); this also keeps the count stable across engines. Gone
-		// resources are included: they are excluded from the approval prompt because
-		// they are not destructive, but destroying them does remove them from state,
-		// and "bundle deploy" likewise counts them as deleted.
+		// skips children and gone resources). A gone delete does not destroy
+		// anything: the resource is already absent remotely, or, like an immutable
+		// snapshot, cannot be deleted at all, so the delete only cleans up state.
+		// Counting it would overstate what was destroyed and disagree with the list
+		// of resources shown above.
 		deleted := 0
 		for _, a := range plan.GetActions() {
-			if a.ActionType == deployplan.Delete && !a.IsChildResource() {
+			if a.ActionType == deployplan.Delete && !a.IsChildResource() && !a.Gone {
 				deleted++
 			}
 		}
diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go
index b15e1c30df6..cf4caa2c6c8 100644
--- a/bundle/phases/initialize.go
+++ b/bundle/phases/initialize.go
@@ -27,6 +27,10 @@ func Initialize(ctx context.Context, b *bundle.Bundle) {
 	log.Info(ctx, "Phase: initialize")
 
 	bundle.ApplySeqContext(ctx, b,
+		// Reads (dynamic): resources.internal_immutable_snapshots (and other internal keys)
+		// Warns and removes any internal resource fields set by the user.
+		mutator.RejectInternalResources(),
+
 		// Reads (dynamic): resource.*.*
 		// Checks that none of resources.<type>.<key> is nil. Raises error otherwise.
 		validate.AllResourcesHaveValues(),
diff --git a/bundle/run/app.go b/bundle/run/app.go
index 8c0f135cc5d..b2d261d65fc 100644
--- a/bundle/run/app.go
+++ b/bundle/run/app.go
@@ -2,6 +2,7 @@ package run
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -142,10 +143,84 @@ func (a *appRunner) deploy(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	deployment := appdeploy.BuildDeployment(a.app.SourceCodePath, config, a.app.GitSource)
+	sourceCodePath, err := a.resolvedSourceCodePath()
+	if err != nil {
+		return err
+	}
+	deployment := appdeploy.BuildDeployment(sourceCodePath, config, a.app.GitSource)
 	return appdeploy.Deploy(ctx, w, a.app.Name, deployment)
 }
 
+// resolvedSourceCodePath returns source_code_path with any ${resources.*} variable
+// references resolved against the current bundle state. This is needed for immutable
+// folder bundles where source_code_path contains a reference to the snapshot's
+// full_path, which is only known after deploy.
+func (a *appRunner) resolvedSourceCodePath() (string, error) {
+	if !dynvar.ContainsVariableReference(a.app.SourceCodePath) {
+		return a.app.SourceCodePath, nil
+	}
+
+	root := a.bundle.Config.Value()
+
+	// Build a lookup for ${resources.*} references. For most fields, the
+	// normalized config value is enough. For compute-only fields like the
+	// snapshot's full_path (a method, not a stored field), we supplement with
+	// the deployed state from the direct engine's state DB.
+	stateOverrides := a.snapshotStateOverrides()
+	normalized, _ := convert.Normalize(a.bundle.Config, root, convert.IncludeMissingFields)
+
+	sourceCodePathKey := dyn.MustPathFromString("resources." + a.Key() + ".source_code_path")
+	pathV, err := dyn.GetByPath(root, sourceCodePathKey)
+	if err != nil || !pathV.IsValid() {
+		return a.app.SourceCodePath, nil //nolint:nilerr
+	}
+
+	resourcesPrefix := dyn.MustPathFromString("resources")
+	resolved, err := dynvar.Resolve(pathV, func(path dyn.Path) (dyn.Value, error) {
+		if !path.HasPrefix(resourcesPrefix) {
+			return dyn.InvalidValue, dynvar.ErrSkipResolution
+		}
+		// Prefer state-derived overrides (e.g. snapshot.full_path) over the
+		// config-computed value, which would be wrong when ZipContent is empty.
+		if v, ok := stateOverrides[path.String()]; ok {
+			return dyn.V(v), nil
+		}
+		return dyn.GetByPath(normalized, path)
+	})
+	if err != nil {
+		return "", err
+	}
+
+	s, ok := resolved.AsString()
+	if !ok {
+		return a.app.SourceCodePath, nil
+	}
+	return s, nil
+}
+
+// snapshotStateOverrides returns a map of resource path → deployed value for
+// fields that are only correct in the persisted state (not computable from the
+// bundle config alone). Currently this covers the snapshot's full_path and
+// relative_path, whose content hash depends on ZipContent (json:"-").
+func (a *appRunner) snapshotStateOverrides() map[string]string {
+	const snapshotKey = "resources.internal_immutable_snapshots.immutable"
+	entry, ok := a.bundle.DeploymentBundle.StateDB.GetResourceEntry(snapshotKey)
+	if !ok {
+		return nil
+	}
+	var s struct {
+		RelativePath string `json:"relative_path"`
+		FullPath     string `json:"full_path"`
+	}
+	if err := json.Unmarshal(entry.State, &s); err != nil || s.FullPath == "" {
+		return nil
+	}
+	return map[string]string{
+		snapshotKey + ".full_path":     s.FullPath,
+		snapshotKey + ".relative_path": s.RelativePath,
+	}
+}
+
 // resolvedConfig returns the app config with any ${resources.*} variable references
 // resolved against the current bundle state. This is needed because the app runtime
 // configuration (env vars, command) can reference other bundle resources whose
diff --git a/bundle/statemgmt/state_load_test.go b/bundle/statemgmt/state_load_test.go
index b706b0770cf..ce1b6b779c3 100644
--- a/bundle/statemgmt/state_load_test.go
+++ b/bundle/statemgmt/state_load_test.go
@@ -1191,6 +1191,11 @@ func AssertFullResourceCoverage(t *testing.T, config *config.Root) {
 	resources := reflect.ValueOf(config.Resources)
 	for i := range resources.NumField() {
 		field := resources.Field(i)
+		// Get Tag field from reflect.Value
+		tag := resources.Type().Field(i).Tag.Get("bundle")
+		if tag == "internal" {
+			continue
+		}
 		if field.Kind() == reflect.Map {
 			assert.True(
 				t,
diff --git a/cmd/bundle/plan.go b/cmd/bundle/plan.go
index cb151856d67..0d5b707100b 100644
--- a/cmd/bundle/plan.go
+++ b/cmd/bundle/plan.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/snapshot"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/phases"
 	"github.com/databricks/cli/cmd/bundle/utils"
@@ -63,6 +64,12 @@ It is useful for previewing changes before running 'bundle deploy'.`,
 		}
 		ctx := cmd.Context()
 
+		if b.IsImmutableFolder() {
+			bundle.ApplyContext(ctx, b, snapshot.PlanUpload(false))
+			if logdiag.HasError(ctx) {
+				return root.ErrAlreadyPrinted
+			}
+		}
 		plan := phases.RunPlan(ctx, b, stateDesc.Engine)
 		if logdiag.HasError(ctx) {
 			return root.ErrAlreadyPrinted
diff --git a/bundle/deploy/snapshot/client.go b/libs/snapshot/client.go
similarity index 60%
rename from bundle/deploy/snapshot/client.go
rename to libs/snapshot/client.go
index 4e5df29e408..8539cf0bda5 100644
--- a/bundle/deploy/snapshot/client.go
+++ b/libs/snapshot/client.go
@@ -8,6 +8,7 @@ import (
 	"mime/multipart"
 	"net/http"
 	"net/textproto"
+	"path"
 
 	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/databricks-sdk-go"
@@ -29,18 +30,10 @@ type ACLEntry struct {
 	PermissionLevel      string `json:"permission_level"`
 }
 
-// SnapshotUploader abstracts the /api/2.0/repos/snapshots endpoint.
-// snapshotID is the content-addressed key supplied by the caller; the API uses
-// it as the final path component so that identical content always resolves to
-// the same workspace location.
-// This interface exists so the implementation can later be replaced with a Go SDK call.
-type SnapshotUploader interface {
-	Upload(ctx context.Context, bundleID, snapshotID string, acl []ACLEntry, zipContent []byte) (*SnapshotInfo, error)
-}
-
-// snapshotAPIClient implements SnapshotUploader against /api/2.0/repos/snapshots.
-type snapshotAPIClient struct {
-	client *databricksclient.DatabricksClient
+// SnapshotClient implements the /api/2.0/repos/snapshots endpoint.
+type SnapshotClient struct {
+	workspaceClient *databricks.WorkspaceClient
+	client          *databricksclient.DatabricksClient
 }
 
 // snapshotUploadResponse mirrors the /api/2.0/repos/snapshots response body.
@@ -50,23 +43,28 @@ type snapshotUploadResponse struct {
 	} `json:"snapshot"`
 }
 
-// NewSnapshotUploader creates a SnapshotUploader backed by /api/2.0/repos/snapshots.
-func NewSnapshotUploader(w *databricks.WorkspaceClient) (SnapshotUploader, error) {
+type snapshotRootPathResponse struct {
+	Path string `json:"path"`
+}
+
+// NewSnapshotClient creates a SnapshotClient backed by /api/2.0/repos/snapshots.
+func NewSnapshotClient(w *databricks.WorkspaceClient) (*SnapshotClient, error) {
 	c, err := databricksclient.New(w.Config)
 	if err != nil {
 		return nil, err
 	}
-	return &snapshotAPIClient{client: c}, nil
+	return &SnapshotClient{workspaceClient: w, client: c}, nil
 }
 
-// Upload uploads zipContent as an immutable snapshot identified by snapshotID.
-// snapshotID is the SHA-256 of the zip and is used by the server as the
-// content-addressed path component. acl grants CAN_READ to each listed principal.
-func (c *snapshotAPIClient) Upload(ctx context.Context, bundleID, snapshotID string, acl []ACLEntry, zipContent []byte) (*SnapshotInfo, error) {
+// Upload uploads zipContent as an immutable snapshot. The server derives the
+// content-addressed path from the SHA-256 of the zip, and acl grants CAN_READ to
+// each listed principal. path is reserved for forthcoming API support for a
+// caller-specified upload location; it is not sent yet.
+func (c *SnapshotClient) Upload(ctx context.Context, path, bundleID string, acl []ACLEntry, zipContent []byte) (*SnapshotInfo, error) {
 	var body bytes.Buffer
 	mw := multipart.NewWriter(&body)
 
-	if err := mw.WriteField("snapshot_id", snapshotID); err != nil {
+	if err := mw.WriteField("snapshot_id", HashFromContent(zipContent)); err != nil {
 		return nil, fmt.Errorf("failed to write snapshot_id: %w", err)
 	}
 	if err := mw.WriteField("bundle_id", bundleID); err != nil {
@@ -112,3 +110,25 @@ func (c *snapshotAPIClient) Upload(ctx context.Context, bundleID, snapshotID str
 
 	return &SnapshotInfo{Path: resp.Snapshot.Path}, nil
 }
+
+func (c *SnapshotClient) Get(ctx context.Context, snapshotRelativePath string) (*SnapshotInfo, error) {
+	rootPath, err := c.GetSnapshotRootPath(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get snapshot root path: %w", err)
+	}
+	snapshotPath := path.Join(rootPath, snapshotRelativePath)
+	resp, err := c.workspaceClient.Workspace.GetStatusByPath(ctx, snapshotPath)
+	if err != nil {
+		return nil, fmt.Errorf("snapshot get: %w", err)
+	}
+	return &SnapshotInfo{Path: resp.Path}, nil
+}
+
+func (c *SnapshotClient) GetSnapshotRootPath(ctx context.Context) (string, error) {
+	var resp snapshotRootPathResponse
+	err := c.client.Do(ctx, http.MethodGet, "/api/2.0/repos/snapshots/rootpath", auth.WorkspaceIDHeaders(c.client.Config), nil, nil, &resp)
+	if err != nil {
+		return "", fmt.Errorf("snapshot root path get: %w", err)
+	}
+	return path.Clean(resp.Path), nil
+}
diff --git a/libs/snapshot/hash.go b/libs/snapshot/hash.go
new file mode 100644
index 00000000000..104858896c3
--- /dev/null
+++ b/libs/snapshot/hash.go
@@ -0,0 +1,12 @@
+package snapshot
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+// HashFromContent returns the SHA-256 hex digest of content.
+func HashFromContent(content []byte) string {
+	h := sha256.Sum256(content)
+	return hex.EncodeToString(h[:])
+}
diff --git a/libs/sync/path.go b/libs/sync/path.go
index 6b60ef682de..760695acfe8 100644
--- a/libs/sync/path.go
+++ b/libs/sync/path.go
@@ -26,6 +26,12 @@ func repoPathForPath(me *iam.User, remotePath string) string {
 // expected base paths and if it is a directory or repository.
 // If dryRun is set, a missing remote directory is not created.
 func EnsureRemotePathIsUsable(ctx context.Context, wsc *databricks.WorkspaceClient, remotePath string, me *iam.User, dryRun bool) error {
+	// If the remote path is a variable, we don't need to check if it's usable.
+	// This happens for immutable folder paths. Their paths are fully resolved at deployment stage and can't be validated at this point.
+	if strings.HasPrefix(remotePath, "${") {
+		return nil
+	}
+
 	var err error
 
 	// TODO: we should cache CurrentUser.Me at the SDK level
diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go
index 099230ad94e..0065e2f5ea4 100644
--- a/libs/testserver/handlers.go
+++ b/libs/testserver/handlers.go
@@ -639,6 +639,12 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.ReposDelete(req)
 	})
 
+	server.Handle("GET", "/api/2.0/repos/snapshots/rootpath", func(req Request) any {
+		return map[string]any{
+			"path": "/Workspace/Users/" + TestUserSP.UserName + "/.snapshots/",
+		}
+	})
+
 	server.Handle("POST", "/api/2.0/repos/snapshots", func(req Request) any {
 		contentType := req.Headers.Get("Content-Type")
 		mediaType, params, err := mime.ParseMediaType(contentType)
@@ -671,6 +677,7 @@ func AddDefaultHandlers(server *Server) {
 		// The real API uses the workspace user UUID (not email) in the snapshot path,
 		// matching service-principal identities used in cloud acceptance tests.
 		snapshotPath := fmt.Sprintf("/Workspace/Users/%s/.snapshots/%s/%s", TestUserSP.UserName, bundleID, snapshotID)
+		req.Workspace.WorkspaceMkdirs(workspace.Mkdirs{Path: snapshotPath})
 		return map[string]any{
 			"snapshot": map[string]any{
 				"path": snapshotPath,