diff --git a/.nextchanges/bundles/cluster-policy-no-drift.md b/.nextchanges/bundles/cluster-policy-no-drift.md
new file mode 100644
index 00000000000..0caa1d030c1
--- /dev/null
+++ b/.nextchanges/bundles/cluster-policy-no-drift.md
@@ -0,0 +1 @@
+* `bundle plan` no longer reports a permanent update on cluster fields supplied by a cluster policy; a field the config never declares is no longer drift when the cluster spec has a `policy_id`. ([#6531](https://github.com/databricks/cli/pull/6531))
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/databricks.yml.tmpl b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/databricks.yml.tmpl
new file mode 100644
index 00000000000..c80fed9f81c
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/databricks.yml.tmpl
@@ -0,0 +1,36 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    pol:
+      name: test-policy-$UNIQUE_NAME
+      definition: |-
+        {
+          "custom_tags.CostCenter": {
+            "type": "fixed",
+            "value": "from-fixed"
+          }
+        }
+
+  jobs:
+    # The config sets the fixed attribute to a value the policy forbids.
+    j:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            policy_id: ${resources.cluster_policies.pol.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: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/hello_world.py b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/out.test.toml b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/output.txt b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/output.txt
new file mode 100644
index 00000000000..097e5a7e98e
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/output.txt
@@ -0,0 +1,23 @@
+
+=== Deploy a cluster whose config contradicts a fixed policy value
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Error: cannot create resources.jobs.j: Cluster validation error: Validation failed for custom_tags, CostCenter must be from-fixed (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-fixed (is "not-what-the-policy-says")
+
+Files: 6 uploaded, 0 deleted
+
+Exit code: 1
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.pol
+
+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/cluster_policies/fixed_value_conflict/script b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/script
new file mode 100644
index 00000000000..bf454016241
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/script
@@ -0,0 +1,9 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+title "Deploy a cluster whose config contradicts a fixed policy value\n"
+errcode trace $CLI bundle deploy
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/test.toml b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/test.toml
new file mode 100644
index 00000000000..ae35e046923
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_value_conflict/test.toml
@@ -0,0 +1,5 @@
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_values_applied/databricks.yml.tmpl b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/databricks.yml.tmpl
new file mode 100644
index 00000000000..8fb232538eb
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/databricks.yml.tmpl
@@ -0,0 +1,38 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    pol:
+      name: test-policy-$UNIQUE_NAME
+      definition: |-
+        {
+          "spark_version": {
+            "type": "fixed",
+            "value": "$DEFAULT_SPARK_VERSION"
+          },
+          "custom_tags.CostCenter": {
+            "type": "fixed",
+            "value": "policy-supplied"
+          }
+        }
+
+  jobs:
+    # new_cluster deliberately omits spark_version and custom_tags: the policy is
+    # expected to supply both. apply_policy_default_values is NOT set.
+    j:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            policy_id: ${resources.cluster_policies.pol.id}
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_values_applied/hello_world.py b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_values_applied/out.test.toml b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_values_applied/output.txt b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/output.txt
new file mode 100644
index 00000000000..92a3a9b4e23
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/output.txt
@@ -0,0 +1,50 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.pol
+Created jobs.j
+Files: 6 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Did the fixed policy SUPPLY spark_version and custom_tags, or only validate them?
+The bundle declares neither, and does not set apply_policy_default_values.
+{
+  "spark_version": "13.3.x-snapshot-scala2.12",
+  "custom_tags": {
+    "CostCenter": "policy-supplied"
+  }
+}
+
+=== Is the policy-supplied value reported as drift?
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+=== spark_version is not covered by any backend_defaults rule, so show its verdict
+
+>>> [CLI] bundle plan -o json
+[
+  {
+    "job_clusters[job_cluster_key='small'].new_cluster.custom_tags": {
+      "action": "skip",
+      "reason": "policy_managed",
+      "remote": {
+        "CostCenter": "policy-supplied"
+      }
+    },
+    "job_clusters[job_cluster_key='small'].new_cluster.spark_version": {
+      "action": "skip",
+      "reason": "policy_managed",
+      "remote": "13.3.x-snapshot-scala2.12"
+    }
+  }
+]
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.pol
+  delete resources.jobs.j
+
+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/cluster_policies/fixed_values_applied/script b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/script
new file mode 100644
index 00000000000..a20c901153d
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/script
@@ -0,0 +1,26 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+# Mask the server-assigned policy id (real backend returns a hex id the built-in
+# UUID replacement misses).
+read_id.py pol > /dev/null
+job_id=$(read_id.py j)
+
+title "Did the fixed policy SUPPLY spark_version and custom_tags, or only validate them?\n"
+echo "The bundle declares neither, and does not set apply_policy_default_values."
+$CLI jobs get "$job_id" | jq '.settings.job_clusters[0].new_cluster | {spark_version, custom_tags}'
+
+title "Is the policy-supplied value reported as drift?\n"
+trace $CLI bundle plan
+
+title "spark_version is not covered by any backend_defaults rule, so show its verdict\n"
+trace $CLI bundle plan -o json | jq -S '
+  [.plan | to_entries[] | (.value.changes // {})
+   | with_entries(select(.key | endswith("spark_version") or endswith("custom_tags")))]
+  | map(select(length > 0))'
diff --git a/acceptance/bundle/resources/cluster_policies/fixed_values_applied/test.toml b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/test.toml
new file mode 100644
index 00000000000..ae35e046923
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/fixed_values_applied/test.toml
@@ -0,0 +1,5 @@
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/databricks.yml.tmpl b/acceptance/bundle/resources/cluster_policies/policy_drift/databricks.yml.tmpl
new file mode 100644
index 00000000000..1c988bcdb8f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/databricks.yml.tmpl
@@ -0,0 +1,68 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    pol:
+      name: test-policy-$UNIQUE_NAME
+      definition: |-
+        {
+          "custom_tags.CostCenter": {
+            "type": "fixed",
+            "value": "from-policy"
+          }
+        }
+
+  jobs:
+    # policy attached: the tag the policy adds must not be drift.
+    with_policy:
+      name: test-with-policy-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            policy_id: ${resources.cluster_policies.pol.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py
+
+    # policy attached, and the config owns a tag of its own. An out-of-band change to that
+    # tag must still be reported: a policy suppresses additions, never disagreements.
+    owned_tag:
+      name: test-owned-tag-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            policy_id: ${resources.cluster_policies.pol.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+            custom_tags:
+              Mine: mine
+      tasks:
+        - task_key: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py
+
+    # no policy attached: a remote-only tag is still drift. edit_resource.py injects it below.
+    no_policy:
+      name: test-no-policy-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/hello_world.py b/acceptance/bundle/resources/cluster_policies/policy_drift/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/out.test.toml b/acceptance/bundle/resources/cluster_policies/policy_drift/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/output.txt b/acceptance/bundle/resources/cluster_policies/policy_drift/output.txt
new file mode 100644
index 00000000000..2242def2f4c
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/output.txt
@@ -0,0 +1,64 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.pol
+Created jobs.no_policy
+Created jobs.owned_tag
+Created jobs.with_policy
+Files: 6 uploaded, 0 deleted
+Resources: 4 created, 0 changed, 0 deleted, 0 unchanged
+
+=== A tag the cluster policy added is not drift
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 4 unchanged
+
+=== Without a policy_id the same remote-only tag IS drift
+
+>>> [CLI] bundle plan
+update jobs.no_policy
+
+Plan: 0 to add, 1 to change, 0 to delete, 3 unchanged
+
+=== A policy does not mask a change to a tag the config owns
+
+>>> [CLI] bundle plan
+update jobs.no_policy
+update jobs.owned_tag
+
+Plan: 0 to add, 2 to change, 0 to delete, 2 unchanged
+
+=== Removing a tag from config is a change the user asked for, not an addition
+
+>>> update_file.py databricks.yml Mine: mine Other: other
+
+>>> [CLI] bundle plan -o json
+[
+  {
+    "job_clusters[job_cluster_key='small'].new_cluster.custom_tags['CostCenter']": {
+      "action": "skip",
+      "reason": "policy_managed",
+      "remote": "from-policy"
+    },
+    "job_clusters[job_cluster_key='small'].new_cluster.custom_tags['Mine']": {
+      "action": "update",
+      "old": "mine",
+      "remote": "changed-out-of-band"
+    },
+    "job_clusters[job_cluster_key='small'].new_cluster.custom_tags['Other']": {
+      "action": "update",
+      "new": "other"
+    }
+  }
+]
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.pol
+  delete resources.jobs.no_policy
+  delete resources.jobs.owned_tag
+  delete resources.jobs.with_policy
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 4 deleted
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/script b/acceptance/bundle/resources/cluster_policies/policy_drift/script
new file mode 100644
index 00000000000..5d927fd49ab
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/script
@@ -0,0 +1,35 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+read_id.py pol > /dev/null
+no_policy_id="$(read_id.py no_policy)"
+owned_tag_id="$(read_id.py owned_tag)"
+
+title "A tag the cluster policy added is not drift\n"
+trace $CLI bundle plan | contains.py "0 to change"
+
+title "Without a policy_id the same remote-only tag IS drift\n"
+edit_resource.py jobs "$no_policy_id" <<'EOF'
+for jc in r["job_clusters"]:
+    jc["new_cluster"]["custom_tags"] = {"CostCenter": "added-out-of-band"}
+EOF
+trace $CLI bundle plan | contains.py "1 to change"
+
+title "A policy does not mask a change to a tag the config owns\n"
+edit_resource.py jobs "$owned_tag_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 "2 to change"
+
+title "Removing a tag from config is a change the user asked for, not an addition\n"
+trace update_file.py databricks.yml "Mine: mine" "Other: other"
+trace $CLI bundle plan -o json | jq -S '
+  [.plan | to_entries[]
+   | select(.key | endswith("owned_tag"))
+   | .value.changes | with_entries(select(.key | contains("custom_tags")))]'
diff --git a/acceptance/bundle/resources/cluster_policies/policy_drift/test.toml b/acceptance/bundle/resources/cluster_policies/policy_drift/test.toml
new file mode 100644
index 00000000000..ae35e046923
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_drift/test.toml
@@ -0,0 +1,5 @@
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/databricks.yml.tmpl b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/databricks.yml.tmpl
new file mode 100644
index 00000000000..9c7520a8f5f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/databricks.yml.tmpl
@@ -0,0 +1,74 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    pol:
+      name: test-policy-$UNIQUE_NAME
+      definition: |-
+        {
+          "custom_tags.CostCenter": {
+            "type": "fixed",
+            "value": "from-policy"
+          }
+        }
+
+  # Every place a cluster spec can appear must be covered by a rule; a gap here is how
+  # https://github.com/databricks/cli/issues/5179 stayed open after a partial fix.
+  clusters:
+    standalone:
+      cluster_name: test-cluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      # Use the shared instance pool: a cold cluster boot made this the slowest test in the
+      # cloud suite (6-8 minutes per env), and the policy tag under test is unaffected.
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      policy_id: ${resources.cluster_policies.pol.id}
+
+  jobs:
+    task_cluster:
+      name: test-task-cluster-$UNIQUE_NAME
+      tasks:
+        - task_key: t
+          new_cluster:
+            policy_id: ${resources.cluster_policies.pol.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+          spark_python_task:
+            python_file: ./hello_world.py
+
+    for_each_cluster:
+      name: test-for-each-cluster-$UNIQUE_NAME
+      tasks:
+        - task_key: outer
+          for_each_task:
+            inputs: "[1,2]"
+            task:
+              task_key: inner
+              new_cluster:
+                policy_id: ${resources.cluster_policies.pol.id}
+                spark_version: $DEFAULT_SPARK_VERSION
+                node_type_id: $NODE_TYPE_ID
+                num_workers: 1
+              spark_python_task:
+                python_file: ./hello_world.py
+
+  # 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. Kept here to record that.
+  pipelines:
+    pipe:
+      name: test-pipeline-$UNIQUE_NAME
+      clusters:
+        - label: default
+          policy_id: ${resources.cluster_policies.pol.id}
+          node_type_id: $NODE_TYPE_ID
+          num_workers: 1
+      libraries:
+        - file:
+            path: ./hello_world.py
diff --git a/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/hello_world.py b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/out.test.toml b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/out.test.toml
new file mode 100644
index 00000000000..f9f4880725d
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/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/cluster_policies/policy_no_drift_variants/output.txt b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/output.txt
new file mode 100644
index 00000000000..585eeb57771
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/output.txt
@@ -0,0 +1,73 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.pol
+Created clusters.standalone
+Created jobs.for_each_cluster
+Created jobs.task_cluster
+Created pipelines.pipe
+Files: 6 uploaded, 0 deleted
+Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Every cluster-spec location converges: the policy tag is not drift anywhere
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged
+
+=== Confirm the tag really is present remotely and was classified, not just absent
+
+>>> [CLI] bundle plan -o json
+[
+  {
+    "node": "resources.clusters.standalone",
+    "tags": {
+      "custom_tags": {
+        "action": "skip",
+        "reason": "policy_managed",
+        "remote": {
+          "CostCenter": "from-policy"
+        }
+      }
+    }
+  },
+  {
+    "node": "resources.jobs.for_each_cluster",
+    "tags": {
+      "tasks[task_key='outer'].for_each_task.task.new_cluster.custom_tags": {
+        "action": "skip",
+        "reason": "policy_managed",
+        "remote": {
+          "CostCenter": "from-policy"
+        }
+      }
+    }
+  },
+  {
+    "node": "resources.jobs.task_cluster",
+    "tags": {
+      "tasks[task_key='t'].new_cluster.custom_tags": {
+        "action": "skip",
+        "reason": "policy_managed",
+        "remote": {
+          "CostCenter": "from-policy"
+        }
+      }
+    }
+  }
+]
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.pol
+  delete resources.clusters.standalone
+  delete resources.jobs.for_each_cluster
+  delete resources.jobs.task_cluster
+  delete resources.pipelines.pipe
+
+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.pipe
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 5 deleted
diff --git a/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/script b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/script
new file mode 100644
index 00000000000..c12e8caaf7f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/script
@@ -0,0 +1,17 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+title "Every cluster-spec location converges: the policy tag is not drift anywhere\n"
+trace $CLI bundle plan
+
+title "Confirm the tag really is present remotely and was classified, not just absent\n"
+trace $CLI bundle plan -o json | jq -S '
+  [.plan | to_entries[]
+   | {node: .key, tags: ((.value.changes // {}) | with_entries(select(.key | contains("custom_tags"))))}
+   | select(.tags | length > 0)]'
diff --git a/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/test.toml b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/test.toml
new file mode 100644
index 00000000000..1002b7e9e54
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/test.toml
@@ -0,0 +1,10 @@
+# Creating a real cluster made this the slowest test in the cloud suite (~8 min per env), and
+# the per-location classification it checks is fully covered locally. The cheap job-based
+# tests in this directory keep the core behaviour in every cloud run.
+CloudSlow = true
+
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/databricks.yml.tmpl b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/databricks.yml.tmpl
new file mode 100644
index 00000000000..747138e1661
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/databricks.yml.tmpl
@@ -0,0 +1,76 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    # A "fixed" element. Docs say it "limits the attribute to the specified value",
+    # which is validation language; whether it also SUPPLIES the value to a request
+    # that omits the attribute is what this test pins down.
+    fixed:
+      name: test-policy-fixed-$UNIQUE_NAME
+      definition: |-
+        {
+          "custom_tags.FixedTag": {
+            "type": "fixed",
+            "value": "from-fixed"
+          }
+        }
+
+    # A "defaultValue" element. Documented to apply only when the request sets
+    # apply_policy_default_values=true.
+    defaulted:
+      name: test-policy-default-$UNIQUE_NAME
+      definition: |-
+        {
+          "custom_tags.DefaultTag": {
+            "type": "unlimited",
+            "defaultValue": "from-default",
+            "isOptional": true
+          }
+        }
+
+  jobs:
+    j:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        # A: fixed element, apply_policy_default_values NOT set
+        - job_cluster_key: a_fixed_noflag
+          new_cluster:
+            policy_id: ${resources.cluster_policies.fixed.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+
+        # B: defaultValue element, apply_policy_default_values NOT set
+        - job_cluster_key: b_default_noflag
+          new_cluster:
+            policy_id: ${resources.cluster_policies.defaulted.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+
+        # C: defaultValue element, apply_policy_default_values = true
+        - job_cluster_key: c_default_flag
+          new_cluster:
+            policy_id: ${resources.cluster_policies.defaulted.id}
+            apply_policy_default_values: true
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+
+        # D: fixed element, apply_policy_default_values = true
+        - job_cluster_key: d_fixed_flag
+          new_cluster:
+            policy_id: ${resources.cluster_policies.fixed.id}
+            apply_policy_default_values: true
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+      tasks:
+        - task_key: t
+          job_cluster_key: a_fixed_noflag
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/hello_world.py b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/hello_world.py
new file mode 100644
index 00000000000..11b15b1a458
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/hello_world.py
@@ -0,0 +1 @@
+print("hello")
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/out.test.toml b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/out.test.toml
new file mode 100644
index 00000000000..ae5c7bd798f
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/output.txt b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/output.txt
new file mode 100644
index 00000000000..30a4ff46b97
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/output.txt
@@ -0,0 +1,53 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.defaulted
+Created cluster_policies.fixed
+Created jobs.j
+Files: 6 uploaded, 0 deleted
+Resources: 3 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Which policy elements did the backend SUPPLY into each job_cluster?
+[
+  {
+    "apply_policy_default_values": null,
+    "custom_tags": {
+      "FixedTag": "from-fixed"
+    },
+    "job_cluster_key": "a_fixed_noflag"
+  },
+  {
+    "apply_policy_default_values": null,
+    "custom_tags": null,
+    "job_cluster_key": "b_default_noflag"
+  },
+  {
+    "apply_policy_default_values": null,
+    "custom_tags": {
+      "DefaultTag": "from-default"
+    },
+    "job_cluster_key": "c_default_flag"
+  },
+  {
+    "apply_policy_default_values": null,
+    "custom_tags": {
+      "FixedTag": "from-fixed"
+    },
+    "job_cluster_key": "d_fixed_flag"
+  }
+]
+
+=== Are the supplied values reported as drift?
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 3 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.defaulted
+  delete resources.cluster_policies.fixed
+  delete resources.jobs.j
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 3 deleted
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/script b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/script
new file mode 100644
index 00000000000..300ee0780c6
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/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
+
+read_id.py fixed > /dev/null
+read_id.py defaulted > /dev/null
+job_id=$(read_id.py j)
+
+title "Which policy elements did the backend SUPPLY into each job_cluster?\n"
+$CLI jobs get "$job_id" | jq -S '[.settings.job_clusters[]
+  | {job_cluster_key,
+     apply_policy_default_values: .new_cluster.apply_policy_default_values,
+     custom_tags: .new_cluster.custom_tags}]'
+
+title "Are the supplied values reported as drift?\n"
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/cluster_policies/policy_value_semantics/test.toml b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/test.toml
new file mode 100644
index 00000000000..ae35e046923
--- /dev/null
+++ b/acceptance/bundle/resources/cluster_policies/policy_value_semantics/test.toml
@@ -0,0 +1,5 @@
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]
diff --git a/bundle/deployplan/plan.go b/bundle/deployplan/plan.go
index c9f5bec3516..3ffbb50ae27 100644
--- a/bundle/deployplan/plan.go
+++ b/bundle/deployplan/plan.go
@@ -147,6 +147,10 @@ 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"
+	// ReasonPolicyManaged: the field is inside a cluster spec that has a policy_id, and the
+	// config never declared it. A cluster policy supplies values server-side, so the backend
+	// legitimately extends the spec beyond what the bundle asks for; that is not drift.
+	ReasonPolicyManaged = "policy_managed"
 
 	// 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 739ca3bd8e7..0ab971dfc1e 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -281,7 +281,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
@@ -382,7 +382,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()
 
@@ -418,6 +418,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
@@ -571,6 +574,47 @@ 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/cluster_policies/policy_value_semantics
+// 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
+		}
+		gate := structpath.NewDotString(path.Prefix(rule.Field.Len()), rule.WhenSet)
+		value, err := structaccess.Get(newState, gate)
+		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
+		}
+		if allEmpty(value) {
+			continue
+		}
+		return deployplan.ReasonPolicyManaged, 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 bf875680c2b..ba75e1953ed 100644
--- a/bundle/direct/bundle_plan_test.go
+++ b/bundle/direct/bundle_plan_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/databricks/cli/libs/dyn/yamlloader"
 	"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"
@@ -256,7 +257,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 != "" {
@@ -323,7 +324,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"]
 }
 
@@ -401,3 +402,108 @@ func bundleWithSkippedJobRun(t *testing.T, remote *dresources.JobRunRemote) *Dep
 	b.RemoteStateCache.Store(jobRunKey, remote)
 	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: "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{},
+	}}}
+
+	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,
+		},
+		{
+			// 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.ReasonPolicyManaged, reason)
+			}
+		})
+	}
+}
diff --git a/bundle/direct/dresources/config.go b/bundle/direct/dresources/config.go
index 33a00e0a2f7..0fab4e24e8b 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
+// the name of a field within that object which gates the rule.
+type RemoteAdditionRule struct {
+	Field   *structpath.PatternNode `yaml:"field"`
+	WhenSet string                  `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..1bcae5f2e27 100644
--- a/bundle/direct/dresources/config_test.go
+++ b/bundle/direct/dresources/config_test.go
@@ -148,3 +148,34 @@ 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 {
+			require.NotEmpty(t, rule.WhenSet, "%s: ignore_remote_additions entry %q has no when_set", resourceType, rule.Field.String())
+
+			gate := rule.Field
+			if !gate.IsRoot() {
+				assert.NoError(t, structaccess.ValidatePattern(adapter.StateType(), gate),
+					"%s: ignore_remote_additions field %q does not resolve in the state type", resourceType, gate.String())
+			}
+			gatePattern, err := structpath.ParsePattern(joinPattern(gate, rule.WhenSet))
+			require.NoError(t, err)
+			assert.NoError(t, structaccess.ValidatePattern(adapter.StateType(), gatePattern),
+				"%s: ignore_remote_additions when_set %q does not resolve under %q", resourceType, rule.WhenSet, gate.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 044fbc4fddb..ce9ee2a4dcf 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/cluster_policies/policy_value_semantics.
+      # 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()
@@ -593,6 +608,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/cluster_policies/policy_value_semantics.
+    # 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..f2871870594 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/cluster_policies/policy_value_semantics.
+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/cluster_policies/policy_no_drift_variants. 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 3b16c61b023..29c023e4038 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