diff --git a/.nextchanges/bundles/cluster-libraries.md b/.nextchanges/bundles/cluster-libraries.md
new file mode 100644
index 00000000000..1ff0e422fa5
--- /dev/null
+++ b/.nextchanges/bundles/cluster-libraries.md
@@ -0,0 +1 @@
+* Add libraries field to clusters. ([#6365](https://github.com/databricks/cli/pull/6365))
diff --git a/acceptance/bundle/invariant/configs/cluster_libraries.yml.tmpl b/acceptance/bundle/invariant/configs/cluster_libraries.yml.tmpl
new file mode 100644
index 00000000000..507730ec287
--- /dev/null
+++ b/acceptance/bundle/invariant/configs/cluster_libraries.yml.tmpl
@@ -0,0 +1,14 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+resources:
+  clusters:
+    foo:
+      cluster_name: test-cluster-$UNIQUE_NAME
+      spark_version: 13.3.x-scala2.12
+      node_type_id: $NODE_TYPE_ID
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      libraries:
+        - pypi:
+            package: requests
diff --git a/acceptance/bundle/invariant/continue_293/test.toml b/acceptance/bundle/invariant/continue_293/test.toml
index a02577b3504..15890459519 100644
--- a/acceptance/bundle/invariant/continue_293/test.toml
+++ b/acceptance/bundle/invariant/continue_293/test.toml
@@ -23,6 +23,9 @@ EnvMatrixExclude.no_instance_pool = ["INPUT_CONFIG=instance_pool.yml.tmpl"]
 # cluster_policies resource is not supported on v0.293.0
 EnvMatrixExclude.no_cluster_policy = ["INPUT_CONFIG=cluster_policy.yml.tmpl"]
 
+# cluster libraries (clusters.libraries) are not supported on v0.293.0
+EnvMatrixExclude.no_cluster_libraries = ["INPUT_CONFIG=cluster_libraries.yml.tmpl"]
+
 # job_runs resource is not supported on v0.293.0
 EnvMatrixExclude.no_job_run = ["INPUT_CONFIG=job_run.yml.tmpl"]
 
diff --git a/acceptance/bundle/invariant/delete_idempotent/out.test.toml b/acceptance/bundle/invariant/delete_idempotent/out.test.toml
index 59018deb1ce..158eb038ad8 100644
--- a/acceptance/bundle/invariant/delete_idempotent/out.test.toml
+++ b/acceptance/bundle/invariant/delete_idempotent/out.test.toml
@@ -8,6 +8,7 @@ EnvMatrix.INPUT_CONFIG = [
   "catalog_optional_fields.yml.tmpl",
   "cluster.yml.tmpl",
   "cluster_apply_policy_default_values.yml.tmpl",
+  "cluster_libraries.yml.tmpl",
   "cluster_policy.yml.tmpl",
   "dashboard.yml.tmpl",
   "job_apply_policy_default_values_job_cluster.yml.tmpl",
diff --git a/acceptance/bundle/invariant/destroy_idempotent/out.test.toml b/acceptance/bundle/invariant/destroy_idempotent/out.test.toml
index 59018deb1ce..158eb038ad8 100644
--- a/acceptance/bundle/invariant/destroy_idempotent/out.test.toml
+++ b/acceptance/bundle/invariant/destroy_idempotent/out.test.toml
@@ -8,6 +8,7 @@ EnvMatrix.INPUT_CONFIG = [
   "catalog_optional_fields.yml.tmpl",
   "cluster.yml.tmpl",
   "cluster_apply_policy_default_values.yml.tmpl",
+  "cluster_libraries.yml.tmpl",
   "cluster_policy.yml.tmpl",
   "dashboard.yml.tmpl",
   "job_apply_policy_default_values_job_cluster.yml.tmpl",
diff --git a/acceptance/bundle/invariant/migrate/test.toml b/acceptance/bundle/invariant/migrate/test.toml
index 2924a3dec1c..cbba4ab177f 100644
--- a/acceptance/bundle/invariant/migrate/test.toml
+++ b/acceptance/bundle/invariant/migrate/test.toml
@@ -21,6 +21,8 @@ EnvMatrixExclude.no_genie_space = ["INPUT_CONFIG=genie_space.yml.tmpl"]
 EnvMatrixExclude.no_instance_pool = ["INPUT_CONFIG=instance_pool.yml.tmpl"]
 # Cluster policies are direct-only; the terraform deploy that seeds the migration fails for them.
 EnvMatrixExclude.no_cluster_policy = ["INPUT_CONFIG=cluster_policy.yml.tmpl"]
+# Cluster libraries are direct-only; the terraform deploy that seeds the migration fails for them.
+EnvMatrixExclude.no_cluster_libraries = ["INPUT_CONFIG=cluster_libraries.yml.tmpl"]
 
 # Cross-resource permission references (e.g. ${resources.jobs.job_b.permissions[0].level})
 # don't work in terraform mode: the terraform interpolator converts the path to
diff --git a/acceptance/bundle/invariant/no_drift/out.test.toml b/acceptance/bundle/invariant/no_drift/out.test.toml
index 8439713320e..acbbe99db77 100644
--- a/acceptance/bundle/invariant/no_drift/out.test.toml
+++ b/acceptance/bundle/invariant/no_drift/out.test.toml
@@ -8,6 +8,7 @@ EnvMatrix.INPUT_CONFIG = [
   "catalog_optional_fields.yml.tmpl",
   "cluster.yml.tmpl",
   "cluster_apply_policy_default_values.yml.tmpl",
+  "cluster_libraries.yml.tmpl",
   "cluster_policy.yml.tmpl",
   "dashboard.yml.tmpl",
   "job_apply_policy_default_values_job_cluster.yml.tmpl",
diff --git a/acceptance/bundle/invariant/test.toml b/acceptance/bundle/invariant/test.toml
index 37bbb2c5103..0302d3dcac9 100644
--- a/acceptance/bundle/invariant/test.toml
+++ b/acceptance/bundle/invariant/test.toml
@@ -25,6 +25,7 @@ EnvMatrix.INPUT_CONFIG = [
     "catalog_optional_fields.yml.tmpl",
     "cluster.yml.tmpl",
     "cluster_apply_policy_default_values.yml.tmpl",
+    "cluster_libraries.yml.tmpl",
     "cluster_policy.yml.tmpl",
     "dashboard.yml.tmpl",
     "job_apply_policy_default_values_job_cluster.yml.tmpl",
diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt
index 748939d3c56..3852bd76bef 100644
--- a/acceptance/bundle/refschema/out.fields.txt
+++ b/acceptance/bundle/refschema/out.fields.txt
@@ -461,6 +461,23 @@ resources.clusters.*.jdbc_port	int	REMOTE
 resources.clusters.*.kind	compute.Kind	ALL
 resources.clusters.*.last_restarted_time	int64	REMOTE
 resources.clusters.*.last_state_loss_time	int64	REMOTE
+resources.clusters.*.libraries	[]compute.Library	ALL
+resources.clusters.*.libraries[*]	compute.Library	ALL
+resources.clusters.*.libraries[*].cran	*compute.RCranLibrary	ALL
+resources.clusters.*.libraries[*].cran.package	string	ALL
+resources.clusters.*.libraries[*].cran.repo	string	ALL
+resources.clusters.*.libraries[*].egg	string	ALL
+resources.clusters.*.libraries[*].jar	string	ALL
+resources.clusters.*.libraries[*].maven	*compute.MavenLibrary	ALL
+resources.clusters.*.libraries[*].maven.coordinates	string	ALL
+resources.clusters.*.libraries[*].maven.exclusions	[]string	ALL
+resources.clusters.*.libraries[*].maven.exclusions[*]	string	ALL
+resources.clusters.*.libraries[*].maven.repo	string	ALL
+resources.clusters.*.libraries[*].pypi	*compute.PythonPyPiLibrary	ALL
+resources.clusters.*.libraries[*].pypi.package	string	ALL
+resources.clusters.*.libraries[*].pypi.repo	string	ALL
+resources.clusters.*.libraries[*].requirements	string	ALL
+resources.clusters.*.libraries[*].whl	string	ALL
 resources.clusters.*.lifecycle	*dresources.StateLifecycle	REMOTE	STATE
 resources.clusters.*.lifecycle	*resources.LifecycleWithStarted	INPUT
 resources.clusters.*.lifecycle	resources.Lifecycle	INPUT
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/libraries-content-change/databricks.yml.tmpl
new file mode 100644
index 00000000000..f5e3173e888
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/databricks.yml.tmpl
@@ -0,0 +1,23 @@
+bundle:
+  name: cluster-libraries-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+artifacts:
+  my_test_code:
+    type: whl
+    path: .
+    # Use 'python' because 'python3' does not exist in Windows virtualenvs.
+    build: python setup.py bdist_wheel
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      num_workers: 1
+      data_security_mode: $DATA_SECURITY_MODE
+      libraries:
+        - whl: ./dist/*.whl
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__init__.py b/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__init__.py
new file mode 100644
index 00000000000..909f1f3220d
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__init__.py
@@ -0,0 +1,2 @@
+__version__ = "0.0.1"
+__author__ = "Databricks"
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__main__.py b/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__main__.py
new file mode 100644
index 00000000000..ea918ce2d53
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/my_test_code/__main__.py
@@ -0,0 +1,16 @@
+"""
+The entry point of the Python Wheel
+"""
+
+import sys
+
+
+def main():
+    # This method will print the provided arguments
+    print("Hello from my func")
+    print("Got arguments:")
+    print(sys.argv)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/out.test.toml b/acceptance/bundle/resources/clusters/libraries-content-change/out.test.toml
new file mode 100644
index 00000000000..f9f4880725d
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/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/libraries-content-change/output.txt b/acceptance/bundle/resources/clusters/libraries-content-change/output.txt
new file mode 100644
index 00000000000..8c13e12e371
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/output.txt
@@ -0,0 +1,30 @@
+
+=== Deploy a cluster with a local wheel library
+OK: cluster created with wheel
+
+>>> print_requests.py //libraries/install
+{
+  "method": "POST",
+  "path": "/api/2.0/libraries/install",
+  "body": {
+    "cluster_id": "[UUID]",
+    "libraries": [
+      {
+        "whl": "/Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/artifacts/.internal/my_test_code-0.0.1-py3-none-any.whl"
+      }
+    ]
+  }
+}
+
+=== Badness: wheel code changes but version does not, so redeploy neither reinstalls nor restarts
+OK: no restart on content-only change (known limitation)
+
+>>> print_requests.py //libraries/install
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+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/clusters/libraries-content-change/script b/acceptance/bundle/resources/clusters/libraries-content-change/script
new file mode 100644
index 00000000000..2fa40f5b891
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/script
@@ -0,0 +1,27 @@
+export DATA_SECURITY_MODE=USER_ISOLATION
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# Deploy output goes to LOG (differs fake vs cloud); echo a pass line instead.
+title "Deploy a cluster with a local wheel library"
+$CLI bundle deploy &> LOG.deploy
+cat LOG.deploy | contains.py "Created clusters.mycluster" > /dev/null
+printf '\n%s\n' "OK: cluster created with wheel"
+trace print_requests.py //libraries/install
+
+# Change only the wheel's code, keeping version 0.0.1 so the built filename - and
+# thus the uploaded workspace path - is byte-for-byte identical to the first deploy.
+title "Badness: wheel code changes but version does not, so redeploy neither reinstalls nor restarts"
+update_file.py my_test_code/__main__.py "Hello from my func" "Hello from my func v2"
+$CLI bundle deploy &> LOG.redeploy
+# The libraries path is unchanged, so the plan sees no library change: the cluster is
+# not restarted and no install request is issued, leaving the stale wheel running. The
+# following two assertions encode that known limitation (see the TODO in cluster.go).
+cat LOG.redeploy | contains.py "!Restarting cluster" > /dev/null
+printf '\n%s\n' "OK: no restart on content-only change (known limitation)"
+trace print_requests.py //libraries/install
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/script.prepare b/acceptance/bundle/resources/clusters/libraries-content-change/script.prepare
new file mode 100644
index 00000000000..31e6de1dbf5
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/script.prepare
@@ -0,0 +1,3 @@
+uv venv -q .venv
+venv_activate
+uv pip install -q --no-index setuptools
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/setup.py b/acceptance/bundle/resources/clusters/libraries-content-change/setup.py
new file mode 100644
index 00000000000..8b48a92b4ce
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/setup.py
@@ -0,0 +1,15 @@
+from setuptools import setup, find_packages
+
+import my_test_code
+
+setup(
+    name="my_test_code",
+    version=my_test_code.__version__,
+    author=my_test_code.__author__,
+    url="https://databricks.com",
+    author_email="john.doe@databricks.com",
+    description="my example wheel",
+    packages=find_packages(include=["my_test_code"]),
+    entry_points={"group1": "run=my_test_code.__main__:main"},
+    install_requires=["setuptools"],
+)
diff --git a/acceptance/bundle/resources/clusters/libraries-content-change/test.toml b/acceptance/bundle/resources/clusters/libraries-content-change/test.toml
new file mode 100644
index 00000000000..afbab5783ff
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-content-change/test.toml
@@ -0,0 +1,13 @@
+Badness = "A local wheel whose contents change but whose version (and thus workspace path) does not is not detected in non-dev mode, so redeploy neither reinstalls it nor restarts the cluster; the stale wheel keeps running. See the TODO in bundle/direct/dresources/cluster.go DoUpdate."
+
+Cloud = true
+CloudSlow = true
+RecordRequests = true
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+Ignore = [".databricks", ".venv", "build", "dist", "my_test_code.egg-info", "databricks.yml"]
+
+[[Repls]]
+Old = "[0-9]{4}-[0-9]{6}-[0-9a-z]{8}"
+New = "[UUID]"
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/databricks.yml b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/databricks.yml
new file mode 100644
index 00000000000..db8e632173f
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/databricks.yml
@@ -0,0 +1,13 @@
+bundle:
+  name: libraries-readplan-not-uploaded
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster
+      spark_version: 15.4.x-scala2.12
+      node_type_id: Standard_DS3_v2
+      num_workers: 1
+      libraries:
+        # Prebuilt local wheel (no build: step) to keep the test hermetic.
+        - whl: ./dist/*.whl
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/dist/my_test_code-0.0.1-py3-none-any.whl b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/dist/my_test_code-0.0.1-py3-none-any.whl
new file mode 100644
index 00000000000..4bb80477caf
Binary files /dev/null and b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/dist/my_test_code-0.0.1-py3-none-any.whl differ
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/out.test.toml b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/out.test.toml
new file mode 100644
index 00000000000..59b56a2037c
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/output.txt b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/output.txt
new file mode 100644
index 00000000000..bbf64e49796
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/output.txt
@@ -0,0 +1,16 @@
+
+>>> [CLI] bundle plan -o json
+
+>>> [CLI] bundle deploy --plan tmp.plan.json
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/libraries-readplan-not-uploaded/default/files...
+Created clusters.mycluster
+Files: 6 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/libraries-readplan-not-uploaded/default
+
+Destroy: 1 deleted
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/script b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/script
new file mode 100644
index 00000000000..8fff35358c4
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/script
@@ -0,0 +1,7 @@
+# A cluster's local wheel library is uploaded by `bundle deploy` but NOT by
+# `bundle deploy --plan`. This tripwire pins that difference.
+# Can be removed once fixed, and other tests can be moved to READPLAN variants.
+#See acceptance/bundle/deploy/readplan/whl-not-uploaded for the bug.
+trace $CLI bundle plan -o json > tmp.plan.json
+trace $CLI bundle deploy --plan tmp.plan.json
+trace $CLI bundle destroy --auto-approve
diff --git a/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/test.toml b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/test.toml
new file mode 100644
index 00000000000..764f1385977
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-readplan-not-uploaded/test.toml
@@ -0,0 +1,6 @@
+Badness = "bundle deploy --plan skips the build phase, so a cluster's local wheel is never uploaded (proven in deploy/readplan/whl-not-uploaded). Kept as a note: when the bug is fixed, deploy --plan will print 'Uploading dist/...' and this golden changes - then delete this test and restore EnvMatrix.READPLAN = [\"\", \"1\"] on clusters/libraries and clusters/libraries-content-change."
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+# This test is just asserting bad behaviour so that it can be removed once that behaviour is fixed
+RecordRequests = false
+Ignore = [".databricks", "tmp.plan.json"]
diff --git a/acceptance/bundle/resources/clusters/libraries-repo/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/libraries-repo/databricks.yml.tmpl
new file mode 100644
index 00000000000..40f63c9241c
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-repo/databricks.yml.tmpl
@@ -0,0 +1,18 @@
+bundle:
+  name: cluster-libraries-repo-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      libraries:
+        - pypi:
+            package: requests
+            repo: https://pypi.org/simple
diff --git a/acceptance/bundle/resources/clusters/libraries-repo/out.test.toml b/acceptance/bundle/resources/clusters/libraries-repo/out.test.toml
new file mode 100644
index 00000000000..17e040e07dc
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-repo/out.test.toml
@@ -0,0 +1,5 @@
+Cloud = true
+CloudSlow = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
+EnvMatrix.READPLAN = ["", "1"]
diff --git a/acceptance/bundle/resources/clusters/libraries-repo/output.txt b/acceptance/bundle/resources/clusters/libraries-repo/output.txt
new file mode 100644
index 00000000000..03dfa68dc62
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-repo/output.txt
@@ -0,0 +1,21 @@
+
+=== Deploy creates the cluster and installs the pypi library
+OK: cluster created with pypi library
+
+=== Plan is a no-op after deploy: the repo round-trips
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
+
+=== Changing only the repo is planned as an update
+>>> [CLI] bundle plan
+update clusters.mycluster
+
+Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+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/clusters/libraries-repo/script b/acceptance/bundle/resources/clusters/libraries-repo/script
new file mode 100644
index 00000000000..c44c39e8b1f
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-repo/script
@@ -0,0 +1,21 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# Deploy output goes to LOG (differs fake vs cloud); echo a pass line instead.
+title "Deploy creates the cluster and installs the pypi library"
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.deploy
+cat LOG.deploy | contains.py "Created clusters.mycluster" > /dev/null
+printf '\n%s\n' "OK: cluster created with pypi library"
+
+title "Plan is a no-op after deploy: the repo round-trips"
+trace $CLI bundle plan
+
+title "Changing only the repo is planned as an update"
+update_file.py databricks.yml "repo: https://pypi.org/simple" "repo: https://example.invalid/simple"
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/clusters/libraries-repo/test.toml b/acceptance/bundle/resources/clusters/libraries-repo/test.toml
new file mode 100644
index 00000000000..4724ab632f1
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-repo/test.toml
@@ -0,0 +1,12 @@
+Cloud = true
+CloudSlow = true
+RecordRequests = false
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.READPLAN = ["", "1"]
+
+Ignore = [".databricks", "databricks.yml", "tmp.plan.json"]
+
+[[Repls]]
+Old = "[0-9]{4}-[0-9]{6}-[0-9a-z]{8}"
+New = "[UUID]"
diff --git a/acceptance/bundle/resources/clusters/libraries-restart/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/libraries-restart/databricks.yml.tmpl
new file mode 100644
index 00000000000..3c9a4946638
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-restart/databricks.yml.tmpl
@@ -0,0 +1,17 @@
+bundle:
+  name: cluster-libraries-restart-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      libraries:
+        - pypi:
+            package: requests
diff --git a/acceptance/bundle/resources/clusters/libraries-restart/out.test.toml b/acceptance/bundle/resources/clusters/libraries-restart/out.test.toml
new file mode 100644
index 00000000000..17e040e07dc
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-restart/out.test.toml
@@ -0,0 +1,5 @@
+Cloud = true
+CloudSlow = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
+EnvMatrix.READPLAN = ["", "1"]
diff --git a/acceptance/bundle/resources/clusters/libraries-restart/output.txt b/acceptance/bundle/resources/clusters/libraries-restart/output.txt
new file mode 100644
index 00000000000..df7c44549eb
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-restart/output.txt
@@ -0,0 +1,31 @@
+
+=== Install (create): libraries install on the fresh cluster with no restart
+>>> print_requests.py //clusters/restart
+
+=== Update: adding a library restarts the running cluster
+>>> print_requests.py //clusters/restart
+{
+  "method": "POST",
+  "path": "/api/2.1/clusters/restart",
+  "body": {
+    "cluster_id": "[UUID]"
+  }
+}
+
+=== Delete: removing all libraries restarts the running cluster
+>>> print_requests.py //clusters/restart
+{
+  "method": "POST",
+  "path": "/api/2.1/clusters/restart",
+  "body": {
+    "cluster_id": "[UUID]"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+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/clusters/libraries-restart/script b/acceptance/bundle/resources/clusters/libraries-restart/script
new file mode 100644
index 00000000000..21ac92e7312
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-restart/script
@@ -0,0 +1,38 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# Deploy output goes to LOG (differs fake vs cloud); the captured restart request is the proof.
+title "Install (create): libraries install on the fresh cluster with no restart"
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.create
+cat LOG.create | contains.py "Created clusters.mycluster" > /dev/null
+# No restart request on create (empty below).
+trace print_requests.py //clusters/restart
+
+title "Update: adding a library restarts the running cluster"
+update_file.py databricks.yml "        - pypi:
+            package: requests" "        - pypi:
+            package: requests
+        - pypi:
+            package: six"
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.update
+cat LOG.update | contains.py "Updated clusters.mycluster" > /dev/null
+trace print_requests.py //clusters/restart
+
+title "Delete: removing all libraries restarts the running cluster"
+update_file.py databricks.yml "      libraries:
+        - pypi:
+            package: requests
+        - pypi:
+            package: six
+" ""
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.remove
+cat LOG.remove | contains.py "Updated clusters.mycluster" > /dev/null
+trace print_requests.py //clusters/restart
diff --git a/acceptance/bundle/resources/clusters/libraries-restart/test.toml b/acceptance/bundle/resources/clusters/libraries-restart/test.toml
new file mode 100644
index 00000000000..472e3f3459e
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-restart/test.toml
@@ -0,0 +1,14 @@
+Cloud = true
+CloudSlow = true
+# This test runs three full cluster lifecycles (create, add-library restart, remove-library
+# restart); on slower clouds that exceeds the default 25m cloud timeout, so give it 50m.
+TimeoutCloud = '50m'
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.READPLAN = ["", "1"]
+
+Ignore = [".databricks", "databricks.yml", "tmp.plan.json"]
+
+[[Repls]]
+Old = "[0-9]{4}-[0-9]{6}-[0-9a-z]{8}"
+New = "[UUID]"
diff --git a/acceptance/bundle/resources/clusters/libraries-stopped-cluster/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/databricks.yml.tmpl
new file mode 100644
index 00000000000..705177abe13
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/databricks.yml.tmpl
@@ -0,0 +1,19 @@
+bundle:
+  name: cluster-libraries-stopped-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      lifecycle:
+        started: false
+      libraries:
+        - pypi:
+            package: requests
diff --git a/acceptance/bundle/resources/clusters/libraries-stopped-cluster/out.test.toml b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/out.test.toml
new file mode 100644
index 00000000000..17e040e07dc
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/out.test.toml
@@ -0,0 +1,5 @@
+Cloud = true
+CloudSlow = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["", "true"]
+EnvMatrix.READPLAN = ["", "1"]
diff --git a/acceptance/bundle/resources/clusters/libraries-stopped-cluster/output.txt b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/output.txt
new file mode 100644
index 00000000000..6319906394e
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/output.txt
@@ -0,0 +1,14 @@
+
+=== Deploy libraries on a started=false cluster: no restart
+OK: created on stopped cluster, no restart
+
+=== Change the libraries and redeploy: cluster is stopped, so no restart is issued
+OK: libraries changed on stopped cluster, no restart
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+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/clusters/libraries-stopped-cluster/script b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/script
new file mode 100644
index 00000000000..02f0b8eeed6
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/script
@@ -0,0 +1,25 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# Deploy output goes to LOG (differs fake vs cloud); echo a pass line instead.
+title "Deploy libraries on a started=false cluster: no restart"
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.deploy
+cat LOG.deploy | contains.py "Created clusters.mycluster" "!Restarting cluster" > /dev/null
+printf '\n%s\n' "OK: created on stopped cluster, no restart"
+
+title "Change the libraries and redeploy: cluster is stopped, so no restart is issued"
+update_file.py databricks.yml "        - pypi:
+            package: requests" "        - pypi:
+            package: requests
+        - pypi:
+            package: numpy"
+$CLI bundle plan -o json > tmp.plan.json 2>>LOG.plan
+$CLI bundle deploy $(readplanarg tmp.plan.json) &> LOG.redeploy
+cat LOG.redeploy | contains.py "Updated clusters.mycluster" "!Restarting cluster" > /dev/null
+printf '\n%s\n' "OK: libraries changed on stopped cluster, no restart"
diff --git a/acceptance/bundle/resources/clusters/libraries-stopped-cluster/test.toml b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/test.toml
new file mode 100644
index 00000000000..4724ab632f1
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries-stopped-cluster/test.toml
@@ -0,0 +1,12 @@
+Cloud = true
+CloudSlow = true
+RecordRequests = false
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.READPLAN = ["", "1"]
+
+Ignore = [".databricks", "databricks.yml", "tmp.plan.json"]
+
+[[Repls]]
+Old = "[0-9]{4}-[0-9]{6}-[0-9a-z]{8}"
+New = "[UUID]"
diff --git a/acceptance/bundle/resources/clusters/libraries/databricks.yml.tmpl b/acceptance/bundle/resources/clusters/libraries/databricks.yml.tmpl
new file mode 100644
index 00000000000..2364bafe42a
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/databricks.yml.tmpl
@@ -0,0 +1,25 @@
+bundle:
+  name: cluster-libraries-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+artifacts:
+  my_test_code:
+    type: whl
+    path: .
+    # Use 'python' because 'python3' does not exist in Windows virtualenvs.
+    build: python setup.py bdist_wheel
+
+resources:
+  clusters:
+    mycluster:
+      cluster_name: mycluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      num_workers: 1
+      data_security_mode: $DATA_SECURITY_MODE
+      libraries:
+        - pypi:
+            package: requests
+        - whl: ./dist/*.whl
diff --git a/acceptance/bundle/resources/clusters/libraries/my_test_code/__init__.py b/acceptance/bundle/resources/clusters/libraries/my_test_code/__init__.py
new file mode 100644
index 00000000000..909f1f3220d
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/my_test_code/__init__.py
@@ -0,0 +1,2 @@
+__version__ = "0.0.1"
+__author__ = "Databricks"
diff --git a/acceptance/bundle/resources/clusters/libraries/my_test_code/__main__.py b/acceptance/bundle/resources/clusters/libraries/my_test_code/__main__.py
new file mode 100644
index 00000000000..ea918ce2d53
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/my_test_code/__main__.py
@@ -0,0 +1,16 @@
+"""
+The entry point of the Python Wheel
+"""
+
+import sys
+
+
+def main():
+    # This method will print the provided arguments
+    print("Hello from my func")
+    print("Got arguments:")
+    print(sys.argv)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/acceptance/bundle/resources/clusters/libraries/out.test.toml b/acceptance/bundle/resources/clusters/libraries/out.test.toml
new file mode 100644
index 00000000000..f9f4880725d
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/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/libraries/output.txt b/acceptance/bundle/resources/clusters/libraries/output.txt
new file mode 100644
index 00000000000..ea0567a0dca
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/output.txt
@@ -0,0 +1,53 @@
+
+=== Deploy a cluster with a pypi and a local wheel library
+OK: cluster created with libraries
+
+=== Libraries installed via the Libraries API (wheel rewritten to its uploaded path)
+>>> print_requests.py //libraries/install
+{
+  "method": "POST",
+  "path": "/api/2.0/libraries/install",
+  "body": {
+    "cluster_id": "[UUID]",
+    "libraries": [
+      {
+        "pypi": {
+          "package": "requests"
+        }
+      },
+      {
+        "whl": "/Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/artifacts/.internal/my_test_code-0.0.1-py3-none-any.whl"
+      }
+    ]
+  }
+}
+
+=== Removing the wheel and redeploying uninstalls it and restarts the running cluster
+>>> print_requests.py //libraries/uninstall //clusters/restart
+{
+  "method": "POST",
+  "path": "/api/2.0/libraries/uninstall",
+  "body": {
+    "cluster_id": "[UUID]",
+    "libraries": [
+      {
+        "whl": "/Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/artifacts/.internal/my_test_code-0.0.1-py3-none-any.whl"
+      }
+    ]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.1/clusters/restart",
+  "body": {
+    "cluster_id": "[UUID]"
+  }
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.clusters.mycluster
+
+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/clusters/libraries/script b/acceptance/bundle/resources/clusters/libraries/script
new file mode 100644
index 00000000000..576d5af03b0
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/script
@@ -0,0 +1,23 @@
+export DATA_SECURITY_MODE=USER_ISOLATION
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# Deploy output goes to LOG (differs fake vs cloud); echo a pass line instead.
+title "Deploy a cluster with a pypi and a local wheel library"
+$CLI bundle deploy &> LOG.deploy
+cat LOG.deploy | contains.py "Created clusters.mycluster" > /dev/null
+printf '\n%s\n' "OK: cluster created with libraries"
+
+title "Libraries installed via the Libraries API (wheel rewritten to its uploaded path)"
+trace print_requests.py //libraries/install
+
+title "Removing the wheel and redeploying uninstalls it and restarts the running cluster"
+update_file.py databricks.yml "        - whl: ./dist/*.whl" ""
+$CLI bundle deploy &> LOG.redeploy
+cat LOG.redeploy | contains.py "Updated clusters.mycluster" > /dev/null
+trace print_requests.py //libraries/uninstall //clusters/restart
diff --git a/acceptance/bundle/resources/clusters/libraries/script.prepare b/acceptance/bundle/resources/clusters/libraries/script.prepare
new file mode 100644
index 00000000000..31e6de1dbf5
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/script.prepare
@@ -0,0 +1,3 @@
+uv venv -q .venv
+venv_activate
+uv pip install -q --no-index setuptools
diff --git a/acceptance/bundle/resources/clusters/libraries/setup.py b/acceptance/bundle/resources/clusters/libraries/setup.py
new file mode 100644
index 00000000000..8b48a92b4ce
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/setup.py
@@ -0,0 +1,15 @@
+from setuptools import setup, find_packages
+
+import my_test_code
+
+setup(
+    name="my_test_code",
+    version=my_test_code.__version__,
+    author=my_test_code.__author__,
+    url="https://databricks.com",
+    author_email="john.doe@databricks.com",
+    description="my example wheel",
+    packages=find_packages(include=["my_test_code"]),
+    entry_points={"group1": "run=my_test_code.__main__:main"},
+    install_requires=["setuptools"],
+)
diff --git a/acceptance/bundle/resources/clusters/libraries/test.toml b/acceptance/bundle/resources/clusters/libraries/test.toml
new file mode 100644
index 00000000000..3101985b060
--- /dev/null
+++ b/acceptance/bundle/resources/clusters/libraries/test.toml
@@ -0,0 +1,11 @@
+Cloud = true
+CloudSlow = true
+RecordRequests = true
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+Ignore = [".databricks", ".venv", "build", "dist", "my_test_code.egg-info", "databricks.yml"]
+
+[[Repls]]
+Old = "[0-9]{4}-[0-9]{6}-[0-9a-z]{8}"
+New = "[UUID]"
diff --git a/bundle/config/mutator/validate_cluster_libraries.go b/bundle/config/mutator/validate_cluster_libraries.go
new file mode 100644
index 00000000000..f13cd3be373
--- /dev/null
+++ b/bundle/config/mutator/validate_cluster_libraries.go
@@ -0,0 +1,43 @@
+package mutator
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/libs/diag"
+)
+
+type validateClusterLibraries struct {
+	engine engine.EngineType
+}
+
+// ValidateClusterLibraries returns a mutator that errors when cluster libraries are used with
+// the terraform deployment engine. Cluster libraries are only supported in direct deployment mode.
+func ValidateClusterLibraries(e engine.EngineType) bundle.Mutator {
+	return &validateClusterLibraries{engine: e}
+}
+
+func (m *validateClusterLibraries) Name() string {
+	return "ValidateClusterLibraries"
+}
+
+func (m *validateClusterLibraries) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+	if m.engine.IsDirect() {
+		return nil
+	}
+
+	var diags diag.Diagnostics
+	for key, cluster := range b.Config.Resources.Clusters {
+		if cluster == nil || len(cluster.Libraries) == 0 {
+			continue
+		}
+		path := "resources.clusters." + key + ".libraries"
+		diags = diags.Append(diag.Diagnostic{
+			Severity:  diag.Error,
+			Summary:   "cluster libraries are only supported in direct deployment mode",
+			Locations: b.Config.GetLocations(path),
+		})
+	}
+	return diags
+}
diff --git a/bundle/config/resources/clusters.go b/bundle/config/resources/clusters.go
index 235ea6eee1a..cb69fc6b752 100644
--- a/bundle/config/resources/clusters.go
+++ b/bundle/config/resources/clusters.go
@@ -18,6 +18,9 @@ type Cluster struct {
 	// Lifecycle shadows BaseResource.Lifecycle to add support for lifecycle.started.
 	Lifecycle *LifecycleWithStarted `json:"lifecycle,omitempty"`
 
+	// Libraries are installed via the Libraries API, not the cluster spec.
+	Libraries []compute.Library `json:"libraries,omitempty"`
+
 	Permissions []ClusterPermission `json:"permissions,omitempty"`
 }
 
diff --git a/bundle/direct/dresources/cluster.go b/bundle/direct/dresources/cluster.go
index 0605d50b86c..fd5b20034ed 100644
--- a/bundle/direct/dresources/cluster.go
+++ b/bundle/direct/dresources/cluster.go
@@ -4,10 +4,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/utils"
@@ -16,8 +18,12 @@ import (
 	"github.com/databricks/databricks-sdk-go/marshal"
 	"github.com/databricks/databricks-sdk-go/retries"
 	"github.com/databricks/databricks-sdk-go/service/compute"
+	"golang.org/x/sync/errgroup"
 )
 
+// librariesWaitTimeout bounds how long we poll for libraries to finish installing.
+const librariesWaitTimeout = 15 * time.Minute
+
 // clusterWaitTimeout bounds how long we poll for a cluster to reach its target
 // state (RUNNING/TERMINATED) after create, edit, or start/stop. Provisioning can
 // legitimately take longer than 15 minutes on capacity-constrained workspaces
@@ -33,6 +39,10 @@ type ClusterState struct {
 	compute.ClusterSpec
 
 	Lifecycle *StateLifecycle `json:"lifecycle,omitempty"`
+
+	// Libraries are installed via the Libraries API, not the cluster spec, and managed as
+	// part of the cluster (see reconcileLibraries and the install in WaitAfterCreate).
+	Libraries []compute.Library `json:"libraries,omitempty"`
 }
 
 // Custom marshalers needed because embedded compute.ClusterSpec has its own MarshalJSON
@@ -57,6 +67,10 @@ type ClusterRemote struct {
 	compute.ClusterDetails
 	ApplyPolicyDefaultValues bool            `json:"apply_policy_default_values,omitempty"`
 	Lifecycle                *StateLifecycle `json:"lifecycle,omitempty"`
+
+	// Libraries is populated by DoRead from the Libraries cluster-status API (the cluster
+	// GET does not return installed libraries), so it participates in drift detection.
+	Libraries []compute.Library `json:"libraries,omitempty"`
 }
 
 func (r *ClusterRemote) UnmarshalJSON(b []byte) error {
@@ -81,6 +95,7 @@ func (r *ResourceCluster) PrepareState(input *resources.Cluster) *ClusterState {
 	s := &ClusterState{
 		ClusterSpec: input.ClusterSpec,
 		Lifecycle:   nil,
+		Libraries:   input.Libraries,
 	}
 	if input.Lifecycle != nil && input.Lifecycle.Started != nil {
 		s.Lifecycle = &StateLifecycle{Started: input.Lifecycle.Started}
@@ -132,19 +147,36 @@ func (r *ResourceCluster) RemapState(input *ClusterRemote) *ClusterState {
 			ForceSendFields:            utils.FilterFields[compute.ClusterSpec](input.ForceSendFields),
 		},
 		Lifecycle: &StateLifecycle{Started: &started},
+		Libraries: input.Libraries,
 	}
 	return spec
 }
 
 func (r *ResourceCluster) DoRead(ctx context.Context, id string) (*ClusterRemote, error) {
-	details, err := r.client.Clusters.GetByClusterId(ctx, id)
-	if err != nil {
+	var details *compute.ClusterDetails
+	var libraries []compute.Library
+
+	// The cluster GET and the library-status GET are independent, so run them concurrently.
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		var err error
+		details, err = r.client.Clusters.GetByClusterId(ctx, id)
+		return err
+	})
+	g.Go(func() error {
+		var err error
+		libraries, err = r.readLibraries(ctx, id)
+		return err
+	})
+	if err := g.Wait(); err != nil {
 		return nil, err
 	}
+
 	remote := &ClusterRemote{
 		ClusterDetails:           *details,
 		ApplyPolicyDefaultValues: false,
 		Lifecycle:                nil,
+		Libraries:                libraries,
 	}
 	// The GET response carries apply_policy_default_values only under .spec (a snapshot of the
 	// create/edit settings), not at the top level. Promote it so RemapState is a dumb copy.
@@ -162,9 +194,29 @@ func (r *ResourceCluster) DoRead(ctx context.Context, id string) (*ClusterRemote
 	default:
 		remote.Lifecycle = nil
 	}
+
 	return remote, nil
 }
 
+// readLibraries returns the bundle-managed libraries installed on the cluster.
+// https://docs.databricks.com/api/workspace/libraries/clusterstatus
+func (r *ResourceCluster) readLibraries(ctx context.Context, id string) ([]compute.Library, error) {
+	statuses, err := r.client.Libraries.ClusterStatusByClusterId(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	var libraries []compute.Library
+	for _, s := range statuses.LibraryStatuses {
+		// Libraries set for all clusters via the UI are not managed by the bundle; a library
+		// pending uninstall on restart is on its way out. Skip both.
+		if s.Library == nil || s.IsLibraryForAllClusters || s.Status == compute.LibraryInstallStatusUninstallOnRestart {
+			continue
+		}
+		libraries = append(libraries, *s.Library)
+	}
+	return libraries, nil
+}
+
 func (r *ResourceCluster) DoCreate(ctx context.Context, config *ClusterState) (string, *ClusterRemote, error) {
 	wait, err := r.client.Clusters.Create(ctx, makeCreateCluster(&config.ClusterSpec))
 	if err != nil {
@@ -173,14 +225,28 @@ func (r *ResourceCluster) DoCreate(ctx context.Context, config *ClusterState) (s
 	return wait.ClusterId, nil, nil
 }
 
-// hasClusterChanges reports whether the plan entry contains any Update changes
-// to fields that belong to the Cluster Edit API (i.e., not lifecycle-only fields).
-func hasClusterChanges(entry *PlanEntry) bool {
-	return entry.Changes.HasChangeExcept("lifecycle", "lifecycle.started")
+// hasClusterSpecChanges reports whether the plan entry changes a Cluster Edit API field —
+// anything other than lifecycle (start/stop) and libraries (handled via the Libraries API).
+func hasClusterSpecChanges(entry *PlanEntry) bool {
+	for field, change := range entry.Changes {
+		if change.Action == deployplan.Skip {
+			continue
+		}
+		node, err := structpath.ParsePath(field)
+		if err != nil {
+			continue
+		}
+		top, _ := node.Prefix(1).StringKey()
+		if top != "lifecycle" && top != "libraries" {
+			return true
+		}
+	}
+	return false
 }
 
 func (r *ResourceCluster) DoUpdate(ctx context.Context, id string, config *ClusterState, entry *PlanEntry) (*ClusterRemote, error) {
-	if hasClusterChanges(entry) {
+	edited := hasClusterSpecChanges(entry)
+	if edited {
 		// Same retry as in TF provider logic
 		// https://github.com/databricks/terraform-provider-databricks/blob/3eecd0f90cf99d7777e79a3d03c41f9b2aafb004/clusters/resource_cluster.go#L624
 		_, err := retries.Poll(ctx, clusterWaitTimeout, func() (*compute.WaitGetClusterRunning[struct{}], *retries.Err) {
@@ -201,6 +267,25 @@ func (r *ResourceCluster) DoUpdate(ctx context.Context, id string, config *Clust
 		}
 	}
 
+	// TODO: a local whl/jar whose workspace path is unchanged but whose contents
+	// changed (same name+version, non-dev mode) is not detected here, so no restart fires.
+	// Dev mode handles this via patchwheel (a source-derived version bump).
+	if entry.Changes.HasChange(librariesPath) {
+		if err := r.reconcileLibraries(ctx, id, config.Libraries, entry); err != nil {
+			return nil, err
+		}
+		// A cluster edit restarts the cluster on its own, which applies the library change.
+		// Without an edit we restart so the change takes effect on a running cluster.
+		if !edited {
+			if err := r.restartIfRunning(ctx, id); err != nil {
+				return nil, err
+			}
+			if err := r.waitForInstall(ctx, id, config.Libraries); err != nil {
+				return nil, err
+			}
+		}
+	}
+
 	if config.Lifecycle == nil || config.Lifecycle.Started == nil {
 		return nil, nil
 	}
@@ -245,6 +330,21 @@ func (r *ResourceCluster) WaitAfterCreate(ctx context.Context, id string, config
 		return nil, err
 	}
 
+	// Install libraries once the cluster is running. A freshly-created cluster has no
+	// attached sessions, so the install applies live without a restart.
+	// TODO: Wait is supposed to be side effect free, but in this case moving it to
+	// the create will cause a wait for libraries to be installed befor the cluster is installed.
+	// this increases the risk of losing the cluster. This is a limitation
+	if len(config.Libraries) > 0 {
+		err = r.client.Libraries.Install(ctx, compute.InstallLibraries{ClusterId: id, Libraries: config.Libraries})
+		if err != nil {
+			return nil, err
+		}
+		if err := r.waitForInstall(ctx, id, config.Libraries); err != nil {
+			return nil, err
+		}
+	}
+
 	if config.Lifecycle != nil && config.Lifecycle.Started != nil && !*config.Lifecycle.Started {
 		// started=false: terminate the cluster after it reaches RUNNING.
 		// Note: Delete terminates the cluster; permanent removal is a separate API (permanent-delete).
@@ -427,3 +527,171 @@ func makeEditCluster(id string, config *compute.ClusterSpec) compute.EditCluster
 
 	return edit
 }
+
+// librariesPath is the ClusterState path of the libraries slice, used to detect library changes.
+var librariesPath = structpath.MustParsePath("libraries")
+
+// KeyedSlices compares libraries by identity rather than index so reordering (or the
+// read-order the status API returns) does not produce phantom diffs.
+func (*ResourceCluster) KeyedSlices() map[string]any {
+	return map[string]any{"libraries": libraryKey}
+}
+
+// reconcileLibraries uninstalls libraries dropped from config and installs the desired set.
+// The Libraries API exposes install and uninstall as separate endpoints, so this is two calls.
+func (r *ResourceCluster) reconcileLibraries(ctx context.Context, id string, desired []compute.Library, entry *PlanEntry) error {
+	removed := removedLibraries(desired, entry)
+	if len(removed) > 0 {
+		err := r.client.Libraries.Uninstall(ctx, compute.UninstallLibraries{ClusterId: id, Libraries: removed})
+		if err != nil {
+			return err
+		}
+	}
+	if len(desired) > 0 {
+		return r.client.Libraries.Install(ctx, compute.InstallLibraries{ClusterId: id, Libraries: desired})
+	}
+	return nil
+}
+
+// removedLibraries returns libraries present in the remote state but absent from the desired set.
+func removedLibraries(desired []compute.Library, entry *PlanEntry) []compute.Library {
+	remote, ok := entry.RemoteState.(*ClusterRemote)
+	if !ok || remote == nil {
+		return nil
+	}
+	desiredKeys := make(map[string]struct{}, len(desired))
+	for _, l := range desired {
+		desiredKeys[libraryMapKey(l)] = struct{}{}
+	}
+	var result []compute.Library
+	for _, l := range remote.Libraries {
+		if _, ok := desiredKeys[libraryMapKey(l)]; !ok {
+			result = append(result, l)
+		}
+	}
+	return result
+}
+
+// restartIfRunning restarts the cluster so a library change takes effect, but only when it is
+// running: a stopped cluster applies pending install/uninstall on its next start. It waits for
+// the cluster to return to RUNNING before returning.
+func (r *ResourceCluster) restartIfRunning(ctx context.Context, id string) error {
+	details, err := r.client.Clusters.GetByClusterId(ctx, id)
+	if err != nil {
+		return err
+	}
+	if details.State != compute.StateRunning {
+		log.Debugf(ctx, "cluster %s is not running (%s); skipping restart for library change", id, details.State)
+		return nil
+	}
+	cmdio.LogString(ctx, fmt.Sprintf("Restarting cluster %s because its libraries changed", id))
+	wait, err := r.client.Clusters.Restart(ctx, compute.RestartCluster{ClusterId: id, RestartUser: "", ForceSendFields: nil})
+	if err != nil {
+		return err
+	}
+	_, err = wait.GetWithTimeout(clusterWaitTimeout)
+	return err
+}
+
+// waitForInstall polls until every desired library reaches a terminal installed state. It returns
+// early without waiting when the cluster is not running: installs only progress on a running
+// cluster and are queued until it next starts.
+func (r *ResourceCluster) waitForInstall(ctx context.Context, id string, desired []compute.Library) error {
+	if len(desired) == 0 {
+		return nil
+	}
+	details, err := r.client.Clusters.GetByClusterId(ctx, id)
+	if err != nil {
+		return err
+	}
+	if details.State != compute.StateRunning {
+		log.Debugf(ctx, "cluster %s is not running (%s); skipping wait for library installation", id, details.State)
+		return nil
+	}
+
+	desiredKeys := make(map[string]struct{}, len(desired))
+	for _, l := range desired {
+		typ, val := libraryWaitKey(l)
+		if typ == "" {
+			// Unknown library type: it can't be matched in cluster-status, so don't wait for it.
+			continue
+		}
+		desiredKeys[typ+"="+val] = struct{}{}
+	}
+
+	_, err = retries.Poll(ctx, librariesWaitTimeout, func() (*struct{}, *retries.Err) {
+		statuses, err := r.client.Libraries.ClusterStatusByClusterId(ctx, id)
+		if err != nil {
+			return nil, retries.Halt(err)
+		}
+		pending := len(desiredKeys)
+		for _, s := range statuses.LibraryStatuses {
+			if s.Library == nil {
+				continue
+			}
+			if _, ok := desiredKeys[libraryWaitMapKey(*s.Library)]; !ok {
+				continue
+			}
+			switch s.Status {
+			case compute.LibraryInstallStatusFailed:
+				return nil, retries.Halt(fmt.Errorf("library %s failed to install: %s", libraryWaitMapKey(*s.Library), strings.Join(s.Messages, "; ")))
+			case compute.LibraryInstallStatusInstalled, compute.LibraryInstallStatusSkipped, compute.LibraryInstallStatusRestored:
+				pending--
+			case compute.LibraryInstallStatusPending, compute.LibraryInstallStatusResolving, compute.LibraryInstallStatusInstalling, compute.LibraryInstallStatusUninstallOnRestart:
+				// Still in progress (or being removed); keep polling.
+			}
+		}
+		if pending > 0 {
+			return nil, retries.Continues(fmt.Sprintf("waiting for %d librar(ies) to install on cluster %s", pending, id))
+		}
+		return &struct{}{}, nil
+	})
+	return err
+}
+
+// libraryWaitKey identifies a library by its primary field only, used to match install-status reports.
+func libraryWaitKey(l compute.Library) (string, string) {
+	switch {
+	case l.Whl != "":
+		return "whl", l.Whl
+	case l.Jar != "":
+		return "jar", l.Jar
+	case l.Egg != "":
+		return "egg", l.Egg
+	case l.Requirements != "":
+		return "requirements", l.Requirements
+	case l.Pypi != nil:
+		return "pypi", l.Pypi.Package
+	case l.Maven != nil:
+		return "maven", l.Maven.Coordinates
+	case l.Cran != nil:
+		return "cran", l.Cran.Package
+	}
+	return "", ""
+}
+
+// libraryKey extends libraryWaitKey with repo/exclusions so a repo-only change is detected.
+func libraryKey(l compute.Library) (string, string) {
+	typ, id := libraryWaitKey(l)
+	switch {
+	case l.Pypi != nil:
+		return typ, id + ";" + l.Pypi.Repo
+	case l.Maven != nil:
+		return typ, id + ";" + l.Maven.Repo + ";" + strings.Join(l.Maven.Exclusions, ",")
+	case l.Cran != nil:
+		return typ, id + ";" + l.Cran.Repo
+	}
+	return typ, id
+}
+
+// libraryMapKey flattens libraryKey into a single string for map lookups.
+func libraryMapKey(l compute.Library) string {
+	f, v := libraryKey(l)
+	return f + "=" + v
+}
+
+// libraryWaitMapKey flattens libraryWaitKey into a single string for map lookups.
+func libraryWaitMapKey(l compute.Library) string {
+	f, v := libraryWaitKey(l)
+	return f + "=" + v
+}
diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml
index e0b712c8ceb..ad7caca86d9 100644
--- a/bundle/internal/schema/annotations.yml
+++ b/bundle/internal/schema/annotations.yml
@@ -618,6 +618,9 @@ resources:
                       notebook_path: "./src/my_notebook.py"
           ```
       "$fields":
+        "libraries":
+          "description": |-
+            A list of libraries to install on the cluster. Installed via the Libraries API after the cluster is created. Only supported in direct deployment mode.
         "lifecycle":
           "description": |-
             Settings that control the deployment lifecycle of the resource, such as preventing it from being destroyed.
diff --git a/bundle/internal/validation/generated/required_fields.go b/bundle/internal/validation/generated/required_fields.go
index fcdd9b11ab9..f3e30b9c49a 100644
--- a/bundle/internal/validation/generated/required_fields.go
+++ b/bundle/internal/validation/generated/required_fields.go
@@ -56,6 +56,9 @@ var RequiredFields = map[string][]string{
 	"resources.clusters.*.init_scripts[*].s3":        {"destination"},
 	"resources.clusters.*.init_scripts[*].volumes":   {"destination"},
 	"resources.clusters.*.init_scripts[*].workspace": {"destination"},
+	"resources.clusters.*.libraries[*].cran":         {"package"},
+	"resources.clusters.*.libraries[*].maven":        {"coordinates"},
+	"resources.clusters.*.libraries[*].pypi":         {"package"},
 	"resources.clusters.*.permissions[*]":            {"level"},
 	"resources.clusters.*.workload_type":             {"clients"},
 
diff --git a/bundle/libraries/expand_glob_references.go b/bundle/libraries/expand_glob_references.go
index 720142fe6d7..ab1da3df68d 100644
--- a/bundle/libraries/expand_glob_references.go
+++ b/bundle/libraries/expand_glob_references.go
@@ -198,6 +198,13 @@ var pipelineEnvDepsPattern = dyn.NewPattern(
 	dyn.Key("dependencies"),
 )
 
+var clusterLibrariesPattern = dyn.NewPattern(
+	dyn.Key("resources"),
+	dyn.Key("clusters"),
+	dyn.AnyKey(),
+	dyn.Key("libraries"),
+)
+
 func (e *expand) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	expanders := []expandPattern{
 		{
@@ -216,6 +223,10 @@ func (e *expand) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 			pattern: pipelineEnvDepsPattern,
 			fn:      expandEnvironmentDeps,
 		},
+		{
+			pattern: clusterLibrariesPattern,
+			fn:      expandLibraries,
+		},
 	}
 
 	var diags diag.Diagnostics
diff --git a/bundle/libraries/remote_path.go b/bundle/libraries/remote_path.go
index 02a1172f36d..e40de7460df 100644
--- a/bundle/libraries/remote_path.go
+++ b/bundle/libraries/remote_path.go
@@ -68,6 +68,8 @@ func collectLocalLibraries(b *bundle.Bundle) (map[string][]LocationToUpdate, err
 		taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
 		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
+		clusterLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
+		clusterLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 		envDepsPattern.Append(dyn.AnyIndex()),
 		pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
 		// The AI Runtime task's code_source_path is a local archive (typically an
diff --git a/bundle/libraries/same_name_libraries.go b/bundle/libraries/same_name_libraries.go
index 49776fbd8c9..8fb140d7aa4 100644
--- a/bundle/libraries/same_name_libraries.go
+++ b/bundle/libraries/same_name_libraries.go
@@ -17,6 +17,8 @@ var patterns = []dyn.Pattern{
 	taskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 	forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
 	forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
+	clusterLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
+	clusterLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 	envDepsPattern.Append(dyn.AnyIndex()),
 	pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
 }
diff --git a/bundle/libraries/switch_to_patched_wheels.go b/bundle/libraries/switch_to_patched_wheels.go
index 56250d713a7..3c5ff184a97 100644
--- a/bundle/libraries/switch_to_patched_wheels.go
+++ b/bundle/libraries/switch_to_patched_wheels.go
@@ -79,6 +79,22 @@ func (c switchToPatchedWheels) Apply(ctx context.Context, b *bundle.Bundle) diag
 		}
 	}
 
+	// Update resources.clusters.*.libraries[*].whl
+	for clusterName, clusterRef := range b.Config.Resources.Clusters {
+		if clusterRef == nil {
+			continue
+		}
+		for libInd, lib := range clusterRef.Libraries {
+			repl := replacements[lib.Whl]
+			if repl != "" {
+				log.Debugf(ctx, "Updating resources.clusters.%s.libraries[%d].whl from %s to %s", clusterName, libInd, lib.Whl, repl)
+				clusterRef.Libraries[libInd].Whl = repl
+			} else {
+				log.Debugf(ctx, "Not updating resources.clusters.%s.libraries[%d].whl from %s. Available replacements: %v", clusterName, libInd, lib.Whl, slices.Sorted(maps.Keys(replacements)))
+			}
+		}
+	}
+
 	return nil
 }
 
diff --git a/bundle/phases/plan.go b/bundle/phases/plan.go
index aee481b2b14..f937e2e91c9 100644
--- a/bundle/phases/plan.go
+++ b/bundle/phases/plan.go
@@ -28,6 +28,7 @@ func PreDeployChecks(ctx context.Context, b *bundle.Bundle, isPlan bool, engine
 		mutator.ValidateGitDetails(),
 		mutator.ValidateDirectOnlyResources(engine),
 		mutator.ValidateLifecycleStarted(engine),
+		mutator.ValidateClusterLibraries(engine),
 		mutator.ValidateCascadeOnDestroy(engine),
 		statemgmt.CheckRunningResource(engine),
 	)
diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json
index 2da7431503d..babaaf07731 100644
--- a/bundle/schema/jsonschema.json
+++ b/bundle/schema/jsonschema.json
@@ -540,6 +540,10 @@
                         "$ref": "#/$defs/github.com/databricks/databricks-sdk-go/service/compute.Kind",
                         "x-databricks-launch-stage": "GA"
                       },
+                      "libraries": {
+                        "description": "A list of libraries to install on the cluster. Installed via the Libraries API after the cluster is created. Only supported in direct deployment mode.",
+                        "$ref": "#/$defs/slice/github.com/databricks/databricks-sdk-go/service/compute.Library"
+                      },
                       "lifecycle": {
                         "description": "Settings that control the deployment lifecycle of the resource, such as preventing it from being destroyed.",
                         "$ref": "#/$defs/github.com/databricks/cli/bundle/config/resources.LifecycleWithStarted"
diff --git a/bundle/terraform_dabs_map/generated.go b/bundle/terraform_dabs_map/generated.go
index fc245721c8a..c1d2430d36e 100644
--- a/bundle/terraform_dabs_map/generated.go
+++ b/bundle/terraform_dabs_map/generated.go
@@ -6,7 +6,8 @@ package terraform_dabs_map
 // alerts / databricks_alert_v2: 3 tf-only
 // apps / databricks_app: 6 dabs-only
 // apps / databricks_app: 1 tf-only
-// clusters / databricks_cluster: 26 tf-only
+// clusters / databricks_cluster: 1 renames
+// clusters / databricks_cluster: 11 tf-only
 // dashboards / databricks_dashboard: 2 tf-only
 // database_instances / databricks_database_instance: 1 tf-only
 // experiments / databricks_mlflow_experiment: 1 tf-only
@@ -34,6 +35,9 @@ package terraform_dabs_map
 // TerraformToDABsFieldMap maps DABs group name → nested TF segments → DABs segment name.
 // Navigate using TF field name segments; DABs is the corresponding DABs name when it differs.
 var TerraformToDABsFieldMap = map[string]RenameTree{
+	"clusters": {
+		"library": {NewName: "libraries"},
+	},
 	"jobs": {
 		"environment": {NewName: "environments"},
 		"git_source": {Children: RenameTree{
@@ -168,27 +172,8 @@ var TerraformOnlyFields = map[string]FieldSet{
 		},
 		"idempotency_token": {},
 		"is_pinned":         {},
-		"library": {
-			"cran": {
-				"package": {}, // databricks_cluster.*.library.cran.package
-				"repo":    {}, // databricks_cluster.*.library.cran.repo
-			},
-			"egg": {}, // databricks_cluster.*.library.egg
-			"jar": {}, // databricks_cluster.*.library.jar
-			"maven": {
-				"coordinates": {}, // databricks_cluster.*.library.maven.coordinates
-				"exclusions":  {}, // databricks_cluster.*.library.maven.exclusions
-				"repo":        {}, // databricks_cluster.*.library.maven.repo
-			},
-			"pypi": {
-				"package": {}, // databricks_cluster.*.library.pypi.package
-				"repo":    {}, // databricks_cluster.*.library.pypi.repo
-			},
-			"requirements": {}, // databricks_cluster.*.library.requirements
-			"whl":          {}, // databricks_cluster.*.library.whl
-		},
-		"no_wait": {},
-		"url":     {},
+		"no_wait":           {},
+		"url":               {},
 	},
 	"dashboards": {
 		"dashboard_change_detected": {},
@@ -571,6 +556,9 @@ var TerraformOnlyFields = map[string]FieldSet{
 // DABsToTerraformRenameMap maps DABs group name → nested DABs segments → TF segment name.
 // Navigate using DABs field name segments; NewName is the TF name when it differs.
 var DABsToTerraformRenameMap = map[string]RenameTree{
+	"clusters": {
+		"libraries": {NewName: "library"},
+	},
 	"jobs": {
 		"environments": {NewName: "environment"},
 		"git_source": {Children: RenameTree{
diff --git a/libs/testserver/clusters.go b/libs/testserver/clusters.go
index 80991b0091a..14a6d45eef7 100644
--- a/libs/testserver/clusters.go
+++ b/libs/testserver/clusters.go
@@ -222,6 +222,35 @@ func (s *FakeWorkspace) ClustersStart(req Request) any {
 	return Response{}
 }
 
+// ClustersRestart restarts a running cluster. It moves the cluster back to PENDING so the next
+// ClustersGet transitions it to RUNNING, mirroring how a real restart cycles through states, and
+// clears the venv cache like ClustersEdit (a restart rebuilds the library environment).
+func (s *FakeWorkspace) ClustersRestart(req Request) any {
+	var request compute.RestartCluster
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return Response{
+			StatusCode: 400,
+			Body:       fmt.Sprintf("request parsing error: %s", err),
+		}
+	}
+	defer s.LockUnlock()()
+
+	cluster, ok := s.Clusters[request.ClusterId]
+	if !ok {
+		return Response{StatusCode: 404}
+	}
+
+	cluster.State = compute.StatePending
+	s.Clusters[request.ClusterId] = cluster
+
+	if env, ok := s.clusterVenvs[request.ClusterId]; ok {
+		os.RemoveAll(env.dir)
+		delete(s.clusterVenvs, request.ClusterId)
+	}
+
+	return Response{}
+}
+
 func (s *FakeWorkspace) ClustersPermanentDelete(req Request) any {
 	var request compute.PermanentDeleteCluster
 	if err := json.Unmarshal(req.Body, &request); err != nil {
diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go
index 2f077b1b631..41d015e771d 100644
--- a/libs/testserver/fake_workspace.go
+++ b/libs/testserver/fake_workspace.go
@@ -199,6 +199,7 @@ type FakeWorkspace struct {
 	ModelRegistryModels   map[string]ml.Model
 	ModelRegistryModelIDs map[string]string // model name -> numeric ID
 	Clusters              map[string]compute.ClusterDetails
+	ClusterLibraries      map[string][]compute.Library // cluster id -> installed libraries
 	InstancePools         map[string]compute.GetInstancePool
 	ClusterPolicies       map[string]compute.Policy
 	Catalogs              map[string]catalog.CatalogInfo
@@ -512,7 +513,8 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace {
 				SingleUserName:   TestUser.UserName,
 			},
 		},
-		InstancePools: map[string]compute.GetInstancePool{},
+		InstancePools:    map[string]compute.GetInstancePool{},
+		ClusterLibraries: map[string][]compute.Library{},
 		ClusterPolicies: map[string]compute.Policy{
 			// Seeded so the stateful list keeps backing the variable-lookup tests
 			// (e.g. acceptance/bundle/variables/env_overrides resolves these by name).
diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go
index 8c44d97eae7..7fa5776a604 100644
--- a/libs/testserver/handlers.go
+++ b/libs/testserver/handlers.go
@@ -923,10 +923,27 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.ClustersStart(req)
 	})
 
+	server.Handle("POST", "/api/2.1/clusters/restart", func(req Request) any {
+		return req.Workspace.ClustersRestart(req)
+	})
+
 	server.Handle("POST", "/api/2.1/clusters/permanent-delete", func(req Request) any {
 		return req.Workspace.ClustersPermanentDelete(req)
 	})
 
+	// Cluster libraries:
+	server.Handle("POST", "/api/2.0/libraries/install", func(req Request) any {
+		return req.Workspace.LibrariesInstall(req)
+	})
+
+	server.Handle("POST", "/api/2.0/libraries/uninstall", func(req Request) any {
+		return req.Workspace.LibrariesUninstall(req)
+	})
+
+	server.Handle("GET", "/api/2.0/libraries/cluster-status", func(req Request) any {
+		return req.Workspace.LibrariesClusterStatus(req, req.URL.Query().Get("cluster_id"))
+	})
+
 	// MLflow Experiments:
 	server.Handle("GET", "/api/2.0/mlflow/experiments/get", func(req Request) any {
 		experimentId := req.URL.Query().Get("experiment_id")
diff --git a/libs/testserver/libraries.go b/libs/testserver/libraries.go
new file mode 100644
index 00000000000..63042ca8e66
--- /dev/null
+++ b/libs/testserver/libraries.go
@@ -0,0 +1,87 @@
+package testserver
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"reflect"
+
+	"github.com/databricks/databricks-sdk-go/service/compute"
+)
+
+func (s *FakeWorkspace) LibrariesInstall(req Request) any {
+	var request compute.InstallLibraries
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return Response{StatusCode: http.StatusBadRequest, Body: fmt.Sprintf("request parsing error: %s", err)}
+	}
+
+	defer s.LockUnlock()()
+
+	if _, ok := s.Clusters[request.ClusterId]; !ok {
+		return Response{StatusCode: http.StatusNotFound}
+	}
+
+	// Install is additive and idempotent: installing an already-present library is a no-op.
+	installed := s.ClusterLibraries[request.ClusterId]
+	for _, lib := range request.Libraries {
+		if !containsLibrary(installed, lib) {
+			installed = append(installed, lib)
+		}
+	}
+	s.ClusterLibraries[request.ClusterId] = installed
+
+	return Response{}
+}
+
+func (s *FakeWorkspace) LibrariesUninstall(req Request) any {
+	var request compute.UninstallLibraries
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return Response{StatusCode: http.StatusBadRequest, Body: fmt.Sprintf("request parsing error: %s", err)}
+	}
+
+	defer s.LockUnlock()()
+
+	installed := s.ClusterLibraries[request.ClusterId]
+	remaining := make([]compute.Library, 0, len(installed))
+	for _, lib := range installed {
+		if !containsLibrary(request.Libraries, lib) {
+			remaining = append(remaining, lib)
+		}
+	}
+	s.ClusterLibraries[request.ClusterId] = remaining
+
+	return Response{}
+}
+
+func (s *FakeWorkspace) LibrariesClusterStatus(req Request, clusterId string) any {
+	defer s.LockUnlock()()
+
+	if _, ok := s.Clusters[clusterId]; !ok {
+		return Response{StatusCode: http.StatusNotFound}
+	}
+
+	installed := s.ClusterLibraries[clusterId]
+	statuses := make([]compute.LibraryFullStatus, 0, len(installed))
+	for i := range installed {
+		statuses = append(statuses, compute.LibraryFullStatus{
+			Library: &installed[i],
+			Status:  compute.LibraryInstallStatusInstalled,
+		})
+	}
+
+	return Response{
+		Body: compute.ClusterLibraryStatuses{
+			ClusterId:       clusterId,
+			LibraryStatuses: statuses,
+		},
+	}
+}
+
+func containsLibrary(libs []compute.Library, target compute.Library) bool {
+	for _, l := range libs {
+		if reflect.DeepEqual(l, target) {
+			return true
+		}
+	}
+	return false
+}