diff --git a/.nextchanges/bundles/cluster-policy-no-drift.md b/.nextchanges/bundles/cluster-policy-no-drift.md
new file mode 100644
index 00000000000..92b5c88e573
--- /dev/null
+++ b/.nextchanges/bundles/cluster-policy-no-drift.md
@@ -0,0 +1 @@
+* direct: `bundle plan` no longer reports a permanent update on a cluster that uses a cluster policy: when the cluster spec sets `policy_id`, a field present in the remote but absent from the bundle config is not treated as drift. ([#6531](https://github.com/databricks/cli/pull/6531))
diff --git a/acceptance/bundle/resources/clusters/cluster_policy/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/cluster_policy/databricks.yml.tmpl
new file mode 100644
index 00000000000..a049ea600b4
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/cluster_policy/databricks.yml.tmpl
@@ -0,0 +1,27 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        # Only a map key: the clusters resource force-sends spark_version, so a scalar the
+        # config omits arrives as "" and a fixed policy rejects it. The scalar-supply case is
+        # covered by fixed_addition, whose job-cluster new_cluster fields are omitempty.
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  clusters:
+    my_cluster:
+      cluster_name: test-cluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      # Shared pool avoids a cold boot; the policy tag under test is unaffected.
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      policy_id: ${resources.cluster_policies.my_policy.id}
diff --git a/acceptance/bundle/resources/clusters/cluster_policy/out.test.toml b/acceptance/bundle/resources/clusters/cluster_policy/out.test.toml
new file mode 100644
index 00000000000..f9f4880725d
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/cluster_policy/out.test.toml
@@ -0,0 +1,4 @@
+Cloud = true
+CloudSlow = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/clusters/cluster_policy/output.txt b/acceptance/bundle/resources/clusters/cluster_policy/output.txt
new file mode 100644
index 00000000000..ed2d9a7c5f1
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/cluster_policy/output.txt
@@ -0,0 +1,34 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created clusters.my_cluster
+Files: 4 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== A cluster policy tag on a standalone cluster is not drift
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+=== ...classified remote_addition
+
+>>> [CLI] bundle plan -o json
+{
+  "custom_tags": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": {
+      "CostCenter": "from-policy"
+    }
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.clusters.my_cluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/clusters/cluster_policy/script b/acceptance/bundle/resources/clusters/cluster_policy/script
new file mode 100644
index 00000000000..d32f03284ae
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/cluster_policy/script
@@ -0,0 +1,15 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+title "A cluster policy tag on a standalone cluster is not drift\n"
+trace $CLI bundle plan
+
+title "...classified remote_addition\n"
+trace $CLI bundle plan -o json | jq -S '.plan["resources.clusters.my_cluster"].changes
+  | with_entries(select(.key | test("custom_tags")))'
diff --git a/acceptance/bundle/resources/clusters/cluster_policy/test.toml b/acceptance/bundle/resources/clusters/cluster_policy/test.toml
new file mode 100644
index 00000000000..a3789595417
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/cluster_policy/test.toml
@@ -0,0 +1,12 @@
+# The standalone clusters resource boots a real cluster at deploy, which is slow and has hit
+# capacity limits on cloud; CloudSlow keeps it out of the per-PR run (it still runs locally and
+# in the full cloud sweep). Output is projected to the policy tag so it is cloud-portable.
+Cloud = true
+CloudSlow = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/databricks.yml.tmpl b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/databricks.yml.tmpl
new file mode 100644
index 00000000000..fc16656ff3e
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/databricks.yml.tmpl
@@ -0,0 +1,33 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  jobs:
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        # The config sets the fixed attribute to a value the policy forbids.
+        - job_cluster_key: main
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+            custom_tags:
+              CostCenter: not-what-the-policy-says
+      tasks:
+        - task_key: main
+          job_cluster_key: main
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/hello_world.py b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/out.test.toml b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/output.txt b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/output.txt
new file mode 100644
index 00000000000..697226c2e9c
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/output.txt
@@ -0,0 +1,21 @@
+
+=== A cluster tag contradicting a fixed policy value is rejected at create
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Error: cannot create resources.jobs.my_job: Cluster validation error: Validation failed for custom_tags, CostCenter must be from-policy (is "not-what-the-policy-says") (400 INVALID_PARAMETER_VALUE)
+
+Endpoint: POST [DATABRICKS_URL]/api/2.2/jobs/create
+HTTP Status: 400 Bad Request
+API error_code: INVALID_PARAMETER_VALUE
+API message: Cluster validation error: Validation failed for custom_tags, CostCenter must be from-policy (is "not-what-the-policy-says")
+
+Files: 4 uploaded, 0 deleted
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 1 deleted
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/script b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/script
new file mode 100644
index 00000000000..b8cb4a39cbf
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/config_conflicts_policy/script
@@ -0,0 +1,9 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+title "A cluster tag contradicting a fixed policy value is rejected at create\n"
+musterr trace $CLI bundle deploy
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/default_flag/databricks.yml.tmpl b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/databricks.yml.tmpl
new file mode 100644
index 00000000000..c358664437f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/databricks.yml.tmpl
@@ -0,0 +1,40 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        # A defaultValue element: the backend fills it into an omitted attribute only when
+        # the request sets apply_policy_default_values (unlike a fixed element).
+        custom_tags.DefaultTag:
+          type: unlimited
+          defaultValue: from-default
+          isOptional: true
+
+  jobs:
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: flag_off
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+        - job_cluster_key: flag_on
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            apply_policy_default_values: true
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: main
+          job_cluster_key: flag_off
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/default_flag/hello_world.py b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/default_flag/out.test.toml b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/default_flag/output.txt b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/output.txt
new file mode 100644
index 00000000000..7982c9afe57
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/output.txt
@@ -0,0 +1,35 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created jobs.my_job
+Files: 4 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== defaultValue is supplied only with apply_policy_default_values (flag_on), not flag_off
+[
+  {
+    "custom_tags": null,
+    "job_cluster_key": "flag_off"
+  },
+  {
+    "custom_tags": {
+      "DefaultTag": "from-default"
+    },
+    "job_cluster_key": "flag_on"
+  }
+]
+
+=== Neither is drift on a second plan
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/default_flag/script b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/script
new file mode 100644
index 00000000000..f359b41276d
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/default_flag/script
@@ -0,0 +1,16 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+job_id=$(read_id.py my_job)
+
+title "defaultValue is supplied only with apply_policy_default_values (flag_on), not flag_off\n"
+$CLI jobs get "$job_id" | jq -S '[.settings.job_clusters[]
+  | {job_cluster_key, custom_tags: .new_cluster.custom_tags}]'
+
+title "Neither is drift on a second plan\n"
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/databricks.yml.tmpl b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/databricks.yml.tmpl
new file mode 100644
index 00000000000..63ae19cddc1
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/databricks.yml.tmpl
@@ -0,0 +1,52 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        # A fixed scalar and a fixed map key; every cluster below omits both, so the policy is
+        # the sole source and each supplied value must be classified remote_addition.
+        spark_version:
+          type: fixed
+          value: $DEFAULT_SPARK_VERSION
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  jobs:
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: job_cluster
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: on_job_cluster
+          job_cluster_key: job_cluster
+          spark_python_task:
+            python_file: ./hello_world.py
+        - task_key: task_cluster
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+          spark_python_task:
+            python_file: ./hello_world.py
+        - task_key: for_each
+          for_each_task:
+            inputs: "[1]"
+            task:
+              task_key: for_each_cluster
+              new_cluster:
+                policy_id: ${resources.cluster_policies.my_policy.id}
+                node_type_id: $NODE_TYPE_ID
+                num_workers: 1
+              spark_python_task:
+                python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/hello_world.py b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/out.test.toml b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/output.txt b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/output.txt
new file mode 100644
index 00000000000..52e33b9bf3e
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/output.txt
@@ -0,0 +1,63 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created jobs.my_job
+Files: 4 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Policy-supplied spark_version and custom_tags are not drift at any cluster-spec location
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+=== Each is classified remote_addition
+
+>>> [CLI] bundle plan -o json
+{
+  "job_clusters[job_cluster_key='job_cluster'].new_cluster.custom_tags": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": {
+      "CostCenter": "from-policy"
+    }
+  },
+  "job_clusters[job_cluster_key='job_cluster'].new_cluster.spark_version": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "13.3.x-snapshot-scala2.12"
+  },
+  "tasks[task_key='for_each'].for_each_task.task.new_cluster.custom_tags": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": {
+      "CostCenter": "from-policy"
+    }
+  },
+  "tasks[task_key='for_each'].for_each_task.task.new_cluster.spark_version": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "13.3.x-snapshot-scala2.12"
+  },
+  "tasks[task_key='task_cluster'].new_cluster.custom_tags": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": {
+      "CostCenter": "from-policy"
+    }
+  },
+  "tasks[task_key='task_cluster'].new_cluster.spark_version": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "13.3.x-snapshot-scala2.12"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/script b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/script
new file mode 100644
index 00000000000..3c6c5f02c10
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/fixed_addition/script
@@ -0,0 +1,15 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+title "Policy-supplied spark_version and custom_tags are not drift at any cluster-spec location\n"
+trace $CLI bundle plan
+
+title "Each is classified remote_addition\n"
+trace $CLI bundle plan -o json | jq -S '.plan["resources.jobs.my_job"].changes
+  | with_entries(select(.key | test("custom_tags|spark_version")))'
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/databricks.yml.tmpl b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/databricks.yml.tmpl
new file mode 100644
index 00000000000..0614a23657b
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/databricks.yml.tmpl
@@ -0,0 +1,35 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  jobs:
+    # The config owns custom_tags.Mine. It is deliberately NOT a policy-governed tag: a fixed
+    # policy enforces on edit too, so an out-of-band change to the policy's own tag would be
+    # rejected (see config_conflicts_policy) and there would be no drifted state to observe.
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: main
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+            custom_tags:
+              Mine: mine
+      tasks:
+        - task_key: main
+          job_cluster_key: main
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/hello_world.py b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/out.test.toml b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/output.txt b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/output.txt
new file mode 100644
index 00000000000..f49402a4dd9
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/output.txt
@@ -0,0 +1,45 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created jobs.my_job
+Files: 4 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== An out-of-band change to a config-owned tag is drift, even with a policy attached
+
+>>> [CLI] bundle plan
+update jobs.my_job
+
+Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
+
+=== Removing a config-owned tag is a change the user asked for, not a backend addition
+
+>>> update_file.py databricks.yml Mine: mine Other: other
+
+>>> [CLI] bundle plan -o json
+{
+  "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['CostCenter']": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "from-policy"
+  },
+  "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['Mine']": {
+    "action": "update",
+    "old": "mine",
+    "remote": "changed-out-of-band"
+  },
+  "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['Other']": {
+    "action": "update",
+    "new": "other"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/script b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/script
new file mode 100644
index 00000000000..d86c89445ee
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/owned_tag_drift/script
@@ -0,0 +1,21 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+job_id=$(read_id.py my_job)
+
+title "An out-of-band change to a config-owned tag is drift, even with a policy attached\n"
+edit_resource.py jobs "$job_id" <<'EOF'
+for jc in r["job_clusters"]:
+    jc["new_cluster"]["custom_tags"]["Mine"] = "changed-out-of-band"
+EOF
+trace $CLI bundle plan | contains.py "1 to change"
+
+title "Removing a config-owned tag is a change the user asked for, not a backend addition\n"
+trace update_file.py databricks.yml "Mine: mine" "Other: other"
+trace $CLI bundle plan -o json | jq -S '.plan["resources.jobs.my_job"].changes
+  | with_entries(select(.key | test("custom_tags")))'
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/databricks.yml.tmpl b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/databricks.yml.tmpl
new file mode 100644
index 00000000000..61fc54a7c3b
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/databricks.yml.tmpl
@@ -0,0 +1,41 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  # policy_id gates suppression per cluster spec, not per resource: the gated cluster's
+  # policy tag is skipped, while an out-of-band tag on the ungated sibling still surfaces.
+  jobs:
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: gated
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+        - job_cluster_key: ungated
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: on_gated
+          job_cluster_key: gated
+          spark_python_task:
+            python_file: ./hello_world.py
+        - task_key: on_ungated
+          job_cluster_key: ungated
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/hello_world.py b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/out.test.toml b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/output.txt b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/output.txt
new file mode 100644
index 00000000000..6ed5b94d741
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/output.txt
@@ -0,0 +1,37 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created jobs.my_job
+Files: 4 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Inject a remote-only tag on the ungated cluster (no policy_id) to show it is not gated
+
+=== The gated cluster's policy tag is skipped; the ungated cluster's tag is drift
+
+>>> [CLI] bundle plan -o json
+{
+  "job_clusters[job_cluster_key='gated'].new_cluster.custom_tags": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": {
+      "CostCenter": "from-policy"
+    }
+  },
+  "job_clusters[job_cluster_key='ungated'].new_cluster.custom_tags": {
+    "action": "update",
+    "remote": {
+      "Injected": "out-of-band"
+    }
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/script b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/script
new file mode 100644
index 00000000000..cdbdcaba7c0
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/policy_id_scoping/script
@@ -0,0 +1,20 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+job_id=$(read_id.py my_job)
+
+title "Inject a remote-only tag on the ungated cluster (no policy_id) to show it is not gated\n"
+edit_resource.py jobs "$job_id" <<'EOF'
+for jc in r["job_clusters"]:
+    if jc["job_cluster_key"] == "ungated":
+        jc["new_cluster"]["custom_tags"] = {"Injected": "out-of-band"}
+EOF
+
+title "The gated cluster's policy tag is skipped; the ungated cluster's tag is drift\n"
+trace $CLI bundle plan -o json | jq -S '.plan["resources.jobs.my_job"].changes
+  | with_entries(select(.key | test("custom_tags")))'
diff --git a/acceptance/bundle/resources/jobs/cluster_policy/test.toml b/acceptance/bundle/resources/jobs/cluster_policy/test.toml
new file mode 100644
index 00000000000..1114c31f86d
--- /dev/null
+++ b/acceptance/bundle/resources/jobs/cluster_policy/test.toml
@@ -0,0 +1,11 @@
+# The drift-suppression under test is a direct-engine feature (bundle/direct), so restrict the
+# inherited engine matrix to direct. Cloud = true adds a real-workspace run; every test here
+# also runs locally against the testserver.
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/databricks.yml.tmpl b/acceptance/bundle/resources/pipelines/cluster_policy/databricks.yml.tmpl
new file mode 100644
index 00000000000..7af2175aa84
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/databricks.yml.tmpl
@@ -0,0 +1,29 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  # A pipeline cluster carries policy_id too, but the Pipelines API does not expand the policy
+  # into the stored spec: it reads back exactly as authored, so there is no addition to suppress
+  # and no ignore_remote_additions rule for pipelines. This records that it simply converges.
+  pipelines:
+    my_pipeline:
+      name: test-pipeline-$UNIQUE_NAME
+      clusters:
+        - label: default
+          policy_id: ${resources.cluster_policies.my_policy.id}
+          node_type_id: $NODE_TYPE_ID
+          num_workers: 1
+      libraries:
+        - file:
+            path: ./hello_world.py
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/hello_world.py b/acceptance/bundle/resources/pipelines/cluster_policy/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/out.test.toml b/acceptance/bundle/resources/pipelines/cluster_policy/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/output.txt b/acceptance/bundle/resources/pipelines/cluster_policy/output.txt
new file mode 100644
index 00000000000..26453f9beb4
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/output.txt
@@ -0,0 +1,25 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created pipelines.my_pipeline
+Files: 5 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== A pipeline cluster with policy_id converges: the Pipelines API does not expand the policy
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.pipelines.my_pipeline
+
+This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
+Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion:
+  delete resources.pipelines.my_pipeline
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/script b/acceptance/bundle/resources/pipelines/cluster_policy/script
new file mode 100644
index 00000000000..dfe1ad7caa2
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/script
@@ -0,0 +1,11 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+title "A pipeline cluster with policy_id converges: the Pipelines API does not expand the policy\n"
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/pipelines/cluster_policy/test.toml b/acceptance/bundle/resources/pipelines/cluster_policy/test.toml
new file mode 100644
index 00000000000..6131aeaf3a2
--- /dev/null
+++ b/acceptance/bundle/resources/pipelines/cluster_policy/test.toml
@@ -0,0 +1,8 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/bundle/deployplan/plan.go b/bundle/deployplan/plan.go
index 67236f74766..f4ecfaeea14 100644
--- a/bundle/deployplan/plan.go
+++ b/bundle/deployplan/plan.go
@@ -153,6 +153,11 @@ const (
 	// ReasonMissingInRemote: field is not present in RemoteType (write-only / input-only).
 	// Remote always appears nil, so treat the absence as a no-op when there is no local change.
 	ReasonMissingInRemote = "missing_in_remote"
+	// ReasonRemoteAddition: the field is a remote-only addition (absent from config, present
+	// in the remote) inside an object whose gate is set (e.g. a cluster with a policy_id). The
+	// backend may extend such an object beyond what the bundle declares, so the addition is not
+	// treated as drift. We do not attribute the value to any particular source.
+	ReasonRemoteAddition = "remote_addition"
 
 	// Special reason that results in removing this change from the plan
 	ReasonDrop = "!drop"
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 26582def481..000426b3d38 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -330,7 +330,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 			return false
 		}
 
-		err = addPerFieldActions(ctx, adapter, entry.Changes, remoteState)
+		err = addPerFieldActions(ctx, adapter, entry.Changes, sv.Value, remoteState)
 		if err != nil {
 			logdiag.LogError(ctx, fmt.Errorf("%s: classifying changes: %w", errorPrefix, err))
 			return false
@@ -431,7 +431,7 @@ func prepareChanges(ctx context.Context, adapter *dresources.Adapter, localDiff,
 	return m, nil
 }
 
-func addPerFieldActions(ctx context.Context, adapter *dresources.Adapter, changes deployplan.Changes, remoteState any) error {
+func addPerFieldActions(ctx context.Context, adapter *dresources.Adapter, changes deployplan.Changes, newState, remoteState any) error {
 	cfg := adapter.ResourceConfig()
 	generatedCfg := adapter.GeneratedResourceConfig()
 
@@ -467,6 +467,9 @@ func addPerFieldActions(ctx context.Context, adapter *dresources.Adapter, change
 		} else if action, reason, ok := classifyIDField(generatedCfg, path, ch); ok {
 			ch.Action = action
 			ch.Reason = reason
+		} else if reason, ok := shouldSkipRemoteAddition(cfg, path, ch, newState); ok {
+			ch.Action = deployplan.Skip
+			ch.Reason = reason
 		} else if reason, ok := shouldSkipBackendDefault(cfg, path, ch); ok {
 			ch.Action = deployplan.Skip
 			ch.Reason = reason
@@ -646,6 +649,50 @@ func shouldSkipNormalized(cfg *dresources.ResourceLifecycleConfig, path *structp
 	return "", false
 }
 
+// shouldSkipRemoteAddition skips a field the backend added to an object it co-owns.
+//
+// It fires only on an addition: absent from both old state and new config, present in the
+// remote. A disagreement between config and remote (New != nil) is left alone and still
+// reports an update, and so does a field the user removed from config (Old != nil) — that
+// is a deletion the user asked for, not a backend addition.
+//
+// The rule is gated on a field within the same object (ignore_remote_additions.when_set).
+// For cluster specs that gate is policy_id: an attached cluster policy supplies values
+// server-side — "fixed" elements always, "defaultValue" elements when the request sets
+// apply_policy_default_values — so the remote spec is legitimately a superset of what the
+// bundle declares. See acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag)
+// for the measured backend behavior.
+//
+// Suppressed values are never echoed back on write: an update sends the config spec as-is
+// and the backend re-supplies the policy values.
+func shouldSkipRemoteAddition(cfg *dresources.ResourceLifecycleConfig, path *structpath.PathNode, ch *deployplan.ChangeDesc, newState any) (string, bool) {
+	if cfg == nil || ch.Old != nil || ch.New != nil || ch.Remote == nil {
+		return "", false
+	}
+	for _, rule := range cfg.IgnoreRemoteAdditions {
+		if !path.HasPatternPrefix(rule.Field) {
+			continue
+		}
+		// Resolve the gate in two steps: the concrete object the rule matched, then the gate
+		// path relative to it. The wildcards in Field are filled in from the change path, so
+		// each object is gated on its own value.
+		object, err := structaccess.Get(newState, path.Prefix(rule.Field.Len()))
+		if err != nil {
+			// The gated object is absent from the config entirely (e.g. the remote grew a
+			// whole new_cluster the bundle does not declare), so there is no policy to gate
+			// on and the addition is real drift. Rule typos cannot reach here: the patterns
+			// are validated against the state type by TestResourcesYMLRemoteAdditionGates.
+			continue
+		}
+		value, err := structaccess.Get(object, rule.WhenSet)
+		if err != nil || allEmpty(value) {
+			continue
+		}
+		return deployplan.ReasonRemoteAddition, true
+	}
+	return "", false
+}
+
 // shouldSkipBackendDefault checks if a change should be skipped because the remote value
 // is a known backend default. Applies when old and new are nil but remote is set.
 // If the rule has allowed values, the remote value must match one of them.
diff --git a/bundle/direct/bundle_plan_test.go b/bundle/direct/bundle_plan_test.go
index c178522f7d7..1ebcecede36 100644
--- a/bundle/direct/bundle_plan_test.go
+++ b/bundle/direct/bundle_plan_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/databricks/cli/libs/structs/structdiff"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/structs/structvar"
+	"github.com/databricks/databricks-sdk-go/service/compute"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	"github.com/stretchr/testify/assert"
@@ -258,7 +259,7 @@ func TestRemoteAlreadySetGuards(t *testing.T) {
 			adapter, ok := adapters[tt.resource]
 			require.True(t, ok)
 			changes := deployplan.Changes{tt.field: tt.ch}
-			err := addPerFieldActions(t.Context(), adapter, changes, nil)
+			err := addPerFieldActions(t.Context(), adapter, changes, nil, nil)
 			require.NoError(t, err)
 			assert.Equal(t, tt.expectedAction, tt.ch.Action)
 			if tt.expectedReason != "" {
@@ -325,7 +326,7 @@ func jobRunResultStateAction(t *testing.T, state *jobs.RunState) *deployplan.Cha
 		Remote: state.ResultState,
 	}}
 
-	require.NoError(t, addPerFieldActions(t.Context(), adapters["job_runs"], changes, remote))
+	require.NoError(t, addPerFieldActions(t.Context(), adapters["job_runs"], changes, nil, remote))
 	return changes["result_state"]
 }
 
@@ -404,6 +405,131 @@ func bundleWithSkippedJobRun(t *testing.T, remote *dresources.JobRunRemote) *Dep
 	return b
 }
 
+func TestShouldSkipRemoteAddition(t *testing.T) {
+	// Rules mirror clusters/jobs ignore_remote_additions in resources.yml, but the test is
+	// deliberately self-contained so edits to resources.yml don't break it. The real wiring
+	// is covered by acceptance/bundle/resources/cluster_policies/*.
+	jobCluster, err := structpath.ParsePattern("job_clusters[*].new_cluster")
+	require.NoError(t, err)
+	cfg := &dresources.ResourceLifecycleConfig{
+		IgnoreRemoteAdditions: []dresources.RemoteAdditionRule{
+			{Field: jobCluster, WhenSet: structpath.MustParsePath("policy_id")},
+		},
+	}
+
+	withPolicy := &jobs.JobSettings{JobClusters: []jobs.JobCluster{{
+		JobClusterKey: "small",
+		NewCluster:    &compute.ClusterSpec{PolicyId: "p1"},
+	}}}
+	withoutPolicy := &jobs.JobSettings{JobClusters: []jobs.JobCluster{{
+		JobClusterKey: "small",
+		NewCluster:    &compute.ClusterSpec{},
+	}}}
+	// when_set resolves against the concrete object the rule matched, so two clusters in one
+	// job are gated independently.
+	mixed := &jobs.JobSettings{JobClusters: []jobs.JobCluster{
+		{JobClusterKey: "gated", NewCluster: &compute.ClusterSpec{PolicyId: "p1"}},
+		{JobClusterKey: "plain", NewCluster: &compute.ClusterSpec{}},
+	}}
+
+	const tagPath = "job_clusters[job_cluster_key='small'].new_cluster.custom_tags['CostCenter']"
+
+	tests := []struct {
+		name     string
+		path     string
+		state    *jobs.JobSettings
+		change   deployplan.ChangeDesc
+		expected bool
+	}{
+		{
+			name:     "policy attached, backend added a tag",
+			path:     tagPath,
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: true,
+		},
+		{
+			name:     "policy attached, whole map added by backend",
+			path:     "job_clusters[job_cluster_key='small'].new_cluster.custom_tags",
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{Remote: map[string]string{"CostCenter": "dev-1234"}},
+			expected: true,
+		},
+		{
+			name:     "no policy attached: an addition is still drift",
+			path:     tagPath,
+			state:    withoutPolicy,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: false,
+		},
+		{
+			name:     "config disagrees with remote: still an update",
+			path:     tagPath,
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{New: "mine", Remote: "dev-1234"},
+			expected: false,
+		},
+		{
+			name:     "user removed the field from config: still an update",
+			path:     tagPath,
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{Old: "mine", Remote: "dev-1234"},
+			expected: false,
+		},
+		{
+			name:     "remote has nothing: not an addition",
+			path:     tagPath,
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{},
+			expected: false,
+		},
+		{
+			name:     "outside the gated object: not covered",
+			path:     "tags['CostCenter']",
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: false,
+		},
+		{
+			name:     "sibling cluster with a policy does not gate one without",
+			path:     "job_clusters[job_cluster_key='plain'].new_cluster.custom_tags['CostCenter']",
+			state:    mixed,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: false,
+		},
+		{
+			name:     "the cluster with the policy is gated on its own policy_id",
+			path:     "job_clusters[job_cluster_key='gated'].new_cluster.custom_tags['CostCenter']",
+			state:    mixed,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: true,
+		},
+		{
+			// The remote grew a whole cluster the config does not declare: there is no
+			// policy_id to gate on, so this is real drift rather than a policy addition.
+			name:     "gated object absent from config: still drift",
+			path:     "job_clusters[job_cluster_key='other'].new_cluster.custom_tags['CostCenter']",
+			state:    withPolicy,
+			change:   deployplan.ChangeDesc{Remote: "dev-1234"},
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path, err := structpath.ParsePath(tt.path)
+			require.NoError(t, err)
+
+			change := tt.change
+			reason, ok := shouldSkipRemoteAddition(cfg, path, &change, tt.state)
+			assert.Equal(t, tt.expected, ok)
+			if tt.expected {
+				assert.Equal(t, deployplan.ReasonRemoteAddition, reason)
+			}
+		})
+	}
+}
+
 // Types for TestPrepareChangesWholeBlockOverlap: two levels of nesting under an
 // optional pointer.
 type threeWayInner struct {
diff --git a/bundle/direct/dresources/config.go b/bundle/direct/dresources/config.go
index 33a00e0a2f7..0aa011605c0 100644
--- a/bundle/direct/dresources/config.go
+++ b/bundle/direct/dresources/config.go
@@ -45,6 +45,17 @@ func (b *BackendDefaultRule) UnmarshalYAML(unmarshal func(any) error) error {
 	return nil
 }
 
+// RemoteAdditionRule marks a sub-object whose contents the backend co-owns whenever the
+// object's WhenSet field is set. Inside such an object, a field the config never declared
+// coming back set from the remote is an addition by the backend, not drift.
+//
+// Field is a prefix pattern selecting the object (omitted = the resource root); WhenSet is a
+// path within that object, relative to it, whose value gates the rule.
+type RemoteAdditionRule struct {
+	Field   *structpath.PatternNode `yaml:"field"`
+	WhenSet *structpath.PathNode    `yaml:"when_set"`
+}
+
 // ResourceLifecycleConfig defines lifecycle behavior for a resource type.
 type ResourceLifecycleConfig struct {
 	// IgnoreRemoteChanges: field patterns where remote changes are ignored (output-only, policy-set).
@@ -76,6 +87,11 @@ type ResourceLifecycleConfig struct {
 	// A change is skipped when local and remote differ only by trailing slashes.
 	NormalizeSlash []FieldRule `yaml:"normalize_slash,omitempty"`
 
+	// IgnoreRemoteAdditions: objects whose fields the backend may add to when a gate field
+	// is set. A field that is absent from both old and new state but present in the remote
+	// is skipped; a disagreement between config and remote is still an update.
+	IgnoreRemoteAdditions []RemoteAdditionRule `yaml:"ignore_remote_additions,omitempty"`
+
 	// BackendDefaults: fields where the backend may set defaults.
 	// When old and new are nil but remote is set, and the remote value matches allowed values (if specified), the change is skipped.
 	BackendDefaults []BackendDefaultRule `yaml:"backend_defaults,omitempty"`
@@ -96,14 +112,15 @@ var resourcesYAML []byte
 var resourcesGeneratedYAML []byte
 
 var empty = ResourceLifecycleConfig{
-	IgnoreRemoteChanges: nil,
-	IgnoreLocalChanges:  nil,
-	RecreateOnChanges:   nil,
-	ProvidedIDFields:    nil,
-	UpdatableIDFields:   nil,
-	NormalizeSlash:      nil,
-	BackendDefaults:     nil,
-	SensitiveFields:     nil,
+	IgnoreRemoteChanges:   nil,
+	IgnoreLocalChanges:    nil,
+	RecreateOnChanges:     nil,
+	ProvidedIDFields:      nil,
+	UpdatableIDFields:     nil,
+	NormalizeSlash:        nil,
+	IgnoreRemoteAdditions: nil,
+	BackendDefaults:       nil,
+	SensitiveFields:       nil,
 }
 
 func mustParseConfig(data []byte) func() *Config {
diff --git a/bundle/direct/dresources/config_test.go b/bundle/direct/dresources/config_test.go
index 33345306b0b..b50c633bb72 100644
--- a/bundle/direct/dresources/config_test.go
+++ b/bundle/direct/dresources/config_test.go
@@ -148,3 +148,35 @@ func TestResourcesYMLActionCategoriesExclusive(t *testing.T) {
 		}
 	}
 }
+
+// TestResourcesYMLRemoteAdditionGates validates every ignore_remote_additions rule against
+// the resource's state type: the field pattern must resolve, and so must the when_set gate
+// relative to it. Without this a typo produces a rule that silently never matches.
+func TestResourcesYMLRemoteAdditionGates(t *testing.T) {
+	for resourceType, rc := range MustLoadConfig().Resources {
+		adapter, err := NewAdapter(SupportedResources[resourceType], resourceType, nil)
+		require.NoError(t, err)
+
+		for _, rule := range rc.IgnoreRemoteAdditions {
+			field := rule.Field
+			require.False(t, rule.WhenSet.IsRoot(),
+				"%s: ignore_remote_additions entry %q needs a when_set", resourceType, field.String())
+
+			if !field.IsRoot() {
+				assert.NoError(t, structaccess.ValidatePattern(adapter.StateType(), field),
+					"%s: ignore_remote_additions field %q does not resolve in the state type", resourceType, field.String())
+			}
+			gate, err := structpath.ParsePattern(joinPattern(field, rule.WhenSet.String()))
+			require.NoError(t, err)
+			assert.NoError(t, structaccess.ValidatePattern(adapter.StateType(), gate),
+				"%s: ignore_remote_additions when_set %q does not resolve under %q", resourceType, rule.WhenSet, field.String())
+		}
+	}
+}
+
+func joinPattern(prefix *structpath.PatternNode, field string) string {
+	if prefix.IsRoot() {
+		return field
+	}
+	return prefix.String() + "." + field
+}
diff --git a/bundle/direct/dresources/resources.yml b/bundle/direct/dresources/resources.yml
index 83ec73c421a..04aa91ed45c 100644
--- a/bundle/direct/dresources/resources.yml
+++ b/bundle/direct/dresources/resources.yml
@@ -85,6 +85,21 @@ resources:
       - field: triggers[*].table_update.condition
         reason: input_only
 
+    ignore_remote_additions:
+      # A cluster policy supplies cluster settings server-side: "fixed" elements always,
+      # "defaultValue" elements when the request sets apply_policy_default_values. The bundle
+      # is not the source for those, so the remote spec is legitimately a superset of what the
+      # config declares and an added field is not drift. Measured backend behavior is pinned by
+      # acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag).
+      # https://github.com/databricks/cli/issues/5179
+      # https://github.com/databricks/cli/issues/6512
+      - field: tasks[*].new_cluster
+        when_set: policy_id
+      - field: tasks[*].for_each_task.task.new_cluster
+        when_set: policy_id
+      - field: job_clusters[*].new_cluster
+        when_set: policy_id
+
     backend_defaults:
       # Same as clusters.enable_elastic_disk — see clusters/resource_cluster.go#L331
       # s.SchemaPath("enable_elastic_disk").SetComputed()
@@ -634,6 +649,16 @@ resources:
       - field: effective_value
 
   clusters:
+    # A cluster policy supplies cluster settings server-side: "fixed" elements always,
+    # "defaultValue" elements when the request sets apply_policy_default_values. The bundle
+    # is not the source for those, so the remote spec is legitimately a superset of what the
+    # config declares and an added field is not drift. Measured backend behavior is pinned by
+    # acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag).
+    # https://github.com/databricks/cli/issues/5179
+    # https://github.com/databricks/cli/issues/6512
+    ignore_remote_additions:
+      - when_set: policy_id
+
     ignore_remote_changes:
       # https://github.com/databricks/terraform-provider-databricks/blob/4eba541abe1a9f50993ea7b9dd83874207e224a1/clusters/resource_cluster.go#L361-L363
       # s.SchemaPath("aws_attributes").SetSuppressDiff()
diff --git a/libs/testserver/cluster_policies.go b/libs/testserver/cluster_policies.go
index 13c48f0770e..d18aee74fc3 100644
--- a/libs/testserver/cluster_policies.go
+++ b/libs/testserver/cluster_policies.go
@@ -3,9 +3,13 @@ package testserver
 import (
 	"encoding/json"
 	"fmt"
+	"maps"
+	"reflect"
 	"slices"
+	"strings"
 
 	"github.com/databricks/databricks-sdk-go/service/compute"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
 )
 
 // policyFamilyDefinition mimics the real backend: a policy created from a policy
@@ -123,3 +127,188 @@ func (s *FakeWorkspace) ClusterPoliciesDelete(req Request) any {
 
 	return Response{}
 }
+
+// policyElement is the part of a cluster policy element the fake applies. The backend
+// supports more element types, but only these two fields materialize a value:
+// "fixed" elements set Value, every limiting type ("allowlist", "blocklist", "regex",
+// "range", "unlimited") may carry DefaultValue. "forbidden" only rejects and is ignored here.
+type policyElement struct {
+	Type         string `json:"type"`
+	Value        any    `json:"value"`
+	DefaultValue any    `json:"defaultValue"`
+}
+
+// effectiveValue returns the value this element materializes into a spec that omits the
+// attribute, or nil if it materializes none. Mirrors the backend: "fixed" applies whether
+// or not the request sets apply_policy_default_values, "defaultValue" only when it does.
+// Pinned against a real workspace by
+// acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag).
+func (e policyElement) effectiveValue(applyDefaults bool) any {
+	if e.Type == "fixed" {
+		return e.Value
+	}
+	if applyDefaults {
+		return e.DefaultValue
+	}
+	return nil
+}
+
+// clusterPolicyValues returns the values the policy supplies, keyed by the policy's
+// attribute path (e.g. "spark_version", "custom_tags.CostCenter").
+func (s *FakeWorkspace) clusterPolicyValues(policyID string, applyDefaults bool) map[string]any {
+	policy, ok := s.ClusterPolicies[policyID]
+	if !ok {
+		return nil
+	}
+	var elements map[string]policyElement
+	if err := json.Unmarshal([]byte(policy.Definition), &elements); err != nil {
+		return nil
+	}
+	values := make(map[string]any, len(elements))
+	for path, element := range elements {
+		if value := element.effectiveValue(applyDefaults); value != nil {
+			values[path] = value
+		}
+	}
+	return values
+}
+
+// applyClusterPolicy fills in attributes the request omitted from the policy attached via
+// policyID. It returns the backend's validation message when a supplied value contradicts a
+// "fixed" element, or "" when the spec is acceptable; callers must hold the workspace lock and
+// surface a non-empty message as a 400.
+//
+// spec is a pointer to any struct with cluster-spec JSON tags (compute.ClusterDetails or
+// compute.ClusterSpec); the policy's attribute paths are applied against its JSON shape so
+// one implementation covers both.
+func (s *FakeWorkspace) applyClusterPolicy(spec any, policyID string, applyDefaults bool) string {
+	if policyID == "" {
+		return ""
+	}
+	values := s.clusterPolicyValues(policyID, applyDefaults)
+	if len(values) == 0 {
+		return ""
+	}
+
+	// spec came from a successful decode of the request, so re-encoding it cannot fail.
+	raw, err := json.Marshal(spec)
+	if err != nil {
+		return ""
+	}
+	// doc is only read, to test whether an attribute is already present; the spec itself is
+	// updated from patch below, so decoded numbers are never written back.
+	var doc map[string]any
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		return ""
+	}
+
+	// Collect only the attributes the request omitted, then unmarshal just those back onto
+	// the spec. Unmarshaling the whole document instead would rebuild ForceSendFields from
+	// every key present, which makes fields the backend omits (e.g. the Jobs API dropping
+	// apply_policy_default_values) serialize as explicit zeros.
+	patch := map[string]any{}
+	for _, path := range slices.Sorted(maps.Keys(values)) {
+		value := values[path]
+		segments := strings.Split(path, ".")
+		if existing, ok := lookup(doc, segments); ok {
+			// A "fixed" element enforces its value as well as supplying it; anything else
+			// only supplies a default, which the request is free to override.
+			if s.isFixed(policyID, path) && !reflect.DeepEqual(existing, value) {
+				// Message shape copied from the real backend for a nested attribute
+				// ("custom_tags, CostCenter must be ..."); the wording for a top-level
+				// attribute has not been observed, so it is only approximated here.
+				return fmt.Sprintf("Cluster validation error: Validation failed for %s must be %v (is %q)",
+					strings.Join(segments, ", "), value, existing)
+			}
+			continue
+		}
+		setPatch(patch, segments, value)
+	}
+	if len(patch) == 0 {
+		return ""
+	}
+
+	if raw, err = json.Marshal(patch); err == nil {
+		_ = json.Unmarshal(raw, spec)
+	}
+	return ""
+}
+
+// isFixed reports whether the policy's element at path is a "fixed" element.
+func (s *FakeWorkspace) isFixed(policyID, path string) bool {
+	var elements map[string]policyElement
+	if err := json.Unmarshal([]byte(s.ClusterPolicies[policyID].Definition), &elements); err != nil {
+		return false
+	}
+	return elements[path].Type == "fixed"
+}
+
+// lookup returns the value at the given key path in doc.
+func lookup(doc map[string]any, path []string) (any, bool) {
+	for _, key := range path[:len(path)-1] {
+		child, ok := doc[key].(map[string]any)
+		if !ok {
+			return nil, false
+		}
+		doc = child
+	}
+	value, ok := doc[path[len(path)-1]]
+	return value, ok
+}
+
+// setPatch records value in patch at the given key path, creating intermediate maps.
+func setPatch(patch map[string]any, path []string, value any) {
+	for _, key := range path[:len(path)-1] {
+		next, ok := patch[key].(map[string]any)
+		if !ok {
+			next = map[string]any{}
+			patch[key] = next
+		}
+		patch = next
+	}
+	patch[path[len(path)-1]] = value
+}
+
+// applyPolicyDefaultValues reads apply_policy_default_values from a raw cluster request body.
+// compute.ClusterDetails has no such field, so it cannot be read off the decoded request.
+func applyPolicyDefaultValues(body []byte) bool {
+	var spec compute.ClusterSpec
+	if err := json.Unmarshal(body, &spec); err != nil {
+		return false
+	}
+	return spec.ApplyPolicyDefaultValues
+}
+
+// applyJobClusterPolicies applies attached cluster policies to every cluster spec a job can
+// carry. Callers must hold the workspace lock.
+//
+// The Pipelines API does not expand cluster policies into the stored spec: a pipeline cluster
+// with a policy_id reads back exactly as authored, verified against a real workspace by
+// acceptance/bundle/resources/pipelines/cluster_policy. So there is
+// deliberately no pipeline equivalent.
+func (s *FakeWorkspace) applyJobClusterPolicies(settings *jobs.JobSettings) string {
+	for i := range settings.JobClusters {
+		if msg := s.applyClusterSpecPolicy(settings.JobClusters[i].NewCluster); msg != "" {
+			return msg
+		}
+	}
+	for i := range settings.Tasks {
+		task := &settings.Tasks[i]
+		if msg := s.applyClusterSpecPolicy(task.NewCluster); msg != "" {
+			return msg
+		}
+		if task.ForEachTask != nil {
+			if msg := s.applyClusterSpecPolicy(task.ForEachTask.Task.NewCluster); msg != "" {
+				return msg
+			}
+		}
+	}
+	return ""
+}
+
+func (s *FakeWorkspace) applyClusterSpecPolicy(spec *compute.ClusterSpec) string {
+	if spec == nil {
+		return ""
+	}
+	return s.applyClusterPolicy(spec, spec.PolicyId, spec.ApplyPolicyDefaultValues)
+}
diff --git a/libs/testserver/cluster_policies_test.go b/libs/testserver/cluster_policies_test.go
new file mode 100644
index 00000000000..83ef663d3ed
--- /dev/null
+++ b/libs/testserver/cluster_policies_test.go
@@ -0,0 +1,110 @@
+package testserver
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/service/compute"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A "fixed" element both supplies and enforces its value; a "defaultValue" element only
+// supplies one, and only when the request sets apply_policy_default_values.
+const testPolicyDefinition = `{
+	"spark_version": {"type": "fixed", "value": "policy-version"},
+	"custom_tags.Fixed": {"type": "fixed", "value": "f"},
+	"custom_tags.Default": {"type": "unlimited", "defaultValue": "d"},
+	"node_type_id": {"type": "unlimited", "defaultValue": "policy-node"}
+}`
+
+func policyWorkspace() *FakeWorkspace {
+	return &FakeWorkspace{ClusterPolicies: map[string]compute.Policy{
+		"p1": {PolicyId: "p1", Definition: testPolicyDefinition},
+	}}
+}
+
+func TestApplyClusterPolicy(t *testing.T) {
+	tests := []struct {
+		name          string
+		policyID      string
+		spec          compute.ClusterSpec
+		want          string
+		wantForceSend []string
+		wantErr       string
+	}{
+		{
+			name:          "fixed applies without apply_policy_default_values, defaultValue does not",
+			policyID:      "p1",
+			spec:          compute.ClusterSpec{PolicyId: "p1"},
+			want:          `{"custom_tags":{"Fixed":"f"},"policy_id":"p1","spark_version":"policy-version"}`,
+			wantForceSend: []string{"SparkVersion"},
+		},
+		{
+			name:          "defaultValue applies with apply_policy_default_values",
+			policyID:      "p1",
+			spec:          compute.ClusterSpec{PolicyId: "p1", ApplyPolicyDefaultValues: true},
+			want:          `{"apply_policy_default_values":true,"custom_tags":{"Default":"d","Fixed":"f"},"node_type_id":"policy-node","policy_id":"p1","spark_version":"policy-version"}`,
+			wantForceSend: []string{"NodeTypeId", "SparkVersion"},
+		},
+		{
+			name:          "a defaultValue does not override what the request supplied",
+			policyID:      "p1",
+			spec:          compute.ClusterSpec{PolicyId: "p1", ApplyPolicyDefaultValues: true, NodeTypeId: "user-node"},
+			want:          `{"apply_policy_default_values":true,"custom_tags":{"Default":"d","Fixed":"f"},"node_type_id":"user-node","policy_id":"p1","spark_version":"policy-version"}`,
+			wantForceSend: []string{"SparkVersion"},
+		},
+		{
+			name:          "policy tags merge into tags the request supplied",
+			policyID:      "p1",
+			spec:          compute.ClusterSpec{PolicyId: "p1", CustomTags: map[string]string{"Mine": "yes"}},
+			want:          `{"custom_tags":{"Fixed":"f","Mine":"yes"},"policy_id":"p1","spark_version":"policy-version"}`,
+			wantForceSend: []string{"SparkVersion"},
+		},
+		{
+			name:     "a value contradicting a fixed element is rejected",
+			policyID: "p1",
+			spec:     compute.ClusterSpec{PolicyId: "p1", SparkVersion: "user-version"},
+			wantErr:  `Cluster validation error: Validation failed for spark_version must be policy-version (is "user-version")`,
+		},
+		{
+			name:     "a tag contradicting a fixed element is rejected",
+			policyID: "p1",
+			spec:     compute.ClusterSpec{PolicyId: "p1", CustomTags: map[string]string{"Fixed": "mine"}},
+			wantErr:  `Cluster validation error: Validation failed for custom_tags, Fixed must be f (is "mine")`,
+		},
+		{
+			name:     "no policy attached leaves the spec untouched",
+			policyID: "",
+			spec:     compute.ClusterSpec{SparkVersion: "v"},
+			want:     `{"spark_version":"v"}`,
+		},
+		{
+			name:     "unknown policy leaves the spec untouched",
+			policyID: "missing",
+			spec:     compute.ClusterSpec{PolicyId: "missing", SparkVersion: "v"},
+			want:     `{"policy_id":"missing","spark_version":"v"}`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			spec := tt.spec
+			msg := policyWorkspace().applyClusterPolicy(&spec, tt.policyID, spec.ApplyPolicyDefaultValues)
+
+			if tt.wantErr != "" {
+				assert.Equal(t, tt.wantErr, msg)
+				return
+			}
+			require.Empty(t, msg)
+
+			got, err := json.Marshal(&spec)
+			require.NoError(t, err)
+			assert.JSONEq(t, tt.want, string(got))
+			// Only attributes the policy actually supplied may become force-sent. Anything
+			// else would serialize as an explicit zero and break the fake's ability to model
+			// fields the real API drops (the Jobs API drops apply_policy_default_values).
+			assert.Equal(t, tt.wantForceSend, spec.ForceSendFields)
+		})
+	}
+}
diff --git a/libs/testserver/clusters.go b/libs/testserver/clusters.go
index 80991b0091a..e0a046031e6 100644
--- a/libs/testserver/clusters.go
+++ b/libs/testserver/clusters.go
@@ -32,6 +32,17 @@ func (s *FakeWorkspace) ClustersCreate(req Request) any {
 		request.SingleUserName = s.CurrentUser().UserName
 	}
 
+	// Apply the attached cluster policy before computing defaults, matching the backend:
+	// policy values feed the defaults below (e.g. driver_node_type_id from node_type_id).
+	if msg := s.applyClusterPolicy(&request, request.PolicyId, applyPolicyDefaultValues(req.Body)); msg != "" {
+		return Response{
+			StatusCode: 400,
+			Body: map[string]string{
+				"error_code": "INVALID_PARAMETER_VALUE",
+				"message":    msg,
+			},
+		}
+	}
 	clusterFixUps(&request)
 
 	// The cluster GET API returns apply_policy_default_values only under .spec, not at the top
@@ -113,6 +124,17 @@ func (s *FakeWorkspace) ClustersEdit(req Request) any {
 	// Preserve runtime-only fields that the Edit API request doesn't include.
 	request.State = existing.State
 	request.ClusterId = existing.ClusterId
+	// Apply the attached cluster policy before computing defaults, matching the backend:
+	// policy values feed the defaults below (e.g. driver_node_type_id from node_type_id).
+	if msg := s.applyClusterPolicy(&request, request.PolicyId, applyPolicyDefaultValues(req.Body)); msg != "" {
+		return Response{
+			StatusCode: 400,
+			Body: map[string]string{
+				"error_code": "INVALID_PARAMETER_VALUE",
+				"message":    msg,
+			},
+		}
+	}
 	clusterFixUps(&request)
 	// Refresh the .spec snapshot from the new settings, matching cloud behavior on edit.
 	request.Spec = specSnapshot(req.Body)
diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go
index c3387be3edb..a3d4c61f005 100644
--- a/libs/testserver/jobs.go
+++ b/libs/testserver/jobs.go
@@ -101,6 +101,15 @@ func (s *FakeWorkspace) JobsCreate(req Request) Response {
 		}
 	}
 
+	if msg := s.applyJobClusterPolicies(&jobSettings); msg != "" {
+		return Response{
+			StatusCode: 400,
+			Body: map[string]string{
+				"error_code": "INVALID_PARAMETER_VALUE",
+				"message":    msg,
+			},
+		}
+	}
 	jobFixUps(&jobSettings)
 
 	// CreatorUserName field is used by TF to check if the resource exists or not. CreatorUserName should be non-empty for the resource to be considered as "exists"
@@ -130,6 +139,15 @@ func (s *FakeWorkspace) JobsReset(req Request) Response {
 
 	defer s.LockUnlock()()
 
+	if msg := s.applyJobClusterPolicies(&request.NewSettings); msg != "" {
+		return Response{
+			StatusCode: 400,
+			Body: map[string]string{
+				"error_code": "INVALID_PARAMETER_VALUE",
+				"message":    msg,
+			},
+		}
+	}
 	jobFixUps(&request.NewSettings)
 
 	jobId := request.JobId
@@ -223,23 +241,34 @@ func jobFixUps(jobSettings *jobs.JobSettings) {
 
 			// The real Jobs API consumes apply_policy_default_values but does not
 			// return it in GET responses; clear it so testserver matches cloud.
-			task.NewCluster.ApplyPolicyDefaultValues = false
+			clearApplyPolicyDefaultValues(task.NewCluster)
 		}
 
 		// Handle for_each_task inner cluster.
 		if task.ForEachTask != nil && task.ForEachTask.Task.NewCluster != nil {
 			// Same as above: not returned in GET responses.
-			task.ForEachTask.Task.NewCluster.ApplyPolicyDefaultValues = false
+			clearApplyPolicyDefaultValues(task.ForEachTask.Task.NewCluster)
 		}
 	}
 
 	// Handle job cluster new_clusters.
 	for i := range jobSettings.JobClusters {
 		// Same as above: not returned in GET responses.
-		jobSettings.JobClusters[i].NewCluster.ApplyPolicyDefaultValues = false
+		clearApplyPolicyDefaultValues(jobSettings.JobClusters[i].NewCluster)
 	}
 }
 
+// clearApplyPolicyDefaultValues drops apply_policy_default_values from a job's cluster spec.
+// Zeroing the value alone is not enough: decoding the request populates ForceSendFields from
+// the keys it carried, so a request that set the flag would still serialize it as an explicit
+// false instead of omitting it the way the Jobs API does.
+func clearApplyPolicyDefaultValues(spec *compute.ClusterSpec) {
+	spec.ApplyPolicyDefaultValues = false
+	spec.ForceSendFields = slices.DeleteFunc(spec.ForceSendFields, func(field string) bool {
+		return field == "ApplyPolicyDefaultValues"
+	})
+}
+
 // jobsGetTasksPageSize matches the real Databricks API limit of 100 tasks per jobs.get response.
 // https://docs.databricks.com/api/workspace/jobs/get
 const jobsGetTasksPageSize = 100