diff --git a/.nextchanges/bundles/record-bind-history.md b/.nextchanges/bundles/record-bind-history.md
new file mode 100644
index 00000000000..f1ae02d88be
--- /dev/null
+++ b/.nextchanges/bundles/record-bind-history.md
@@ -0,0 +1 @@
+* Record deployment history when binding existing resources with `bundle deployment bind` when deployment history is enabled. ([#7400](https://github.com/databricks/cli/pull/7400))
diff --git a/acceptance/bin/nostamp.py b/acceptance/bin/nostamp.py
index 8a0c222630c..8f370edc657 100755
--- a/acceptance/bin/nostamp.py
+++ b/acceptance/bin/nostamp.py
@@ -7,13 +7,14 @@
 is a 2-space indent, keys in input order, <>& left unescaped, and integers at full precision.
 Tests under bundle/dms assert the stamp itself and must not use it.
 
-Removed in three shapes plus the plan header:
+Removed in these shapes:
   1. deployment_id/version_id nested in a deployment block, recognized by the
      neighbouring "kind" and "metadata_file_path", which are kept; version_id
      == "" is kept (a Terraform state dump carries that for an unstamped job).
   2. changes entries keyed "deployment.deployment_id" / "deployment.version_id".
   3. a changes object emptied by (2), or already empty.
   4. the recorded plan header field features.
+  5. deployment_id/latest_version_id in bundle.deployment; other settings are kept.
 """
 
 import argparse
@@ -23,6 +24,7 @@
 _STAMP_KEYS = ("deployment_id", "version_id")
 _CHANGE_KEYS = ("deployment.deployment_id", "deployment.version_id")
 _PLAN_HEADER_KEYS = ("features",)
+_BUNDLE_STAMP_KEYS = ("deployment_id", "latest_version_id")
 
 
 def scrub(node):
@@ -52,6 +54,15 @@ def render(data, indent):
     if isinstance(data, dict):
         for k in _PLAN_HEADER_KEYS:
             data.pop(k, None)  # 4. plan header, present only at the root
+        # 5. Strip the summary's DMS IDs, keeping deployment settings such as locks.
+        bundle = data.get("bundle")
+        if isinstance(bundle, dict):
+            deployment = bundle.get("deployment")
+            if isinstance(deployment, dict) and any(k in deployment for k in _BUNDLE_STAMP_KEYS):
+                for k in _BUNDLE_STAMP_KEYS:
+                    deployment.pop(k, None)
+                if not deployment:
+                    bundle.pop("deployment")
     return json.dumps(data, indent=indent, ensure_ascii=False, separators=(",", ": "))
 
 
diff --git a/acceptance/bundle/deployment/bind/dashboard/recreation/script b/acceptance/bundle/deployment/bind/dashboard/recreation/script
index c9656879b7e..6d8f12f1821 100644
--- a/acceptance/bundle/deployment/bind/dashboard/recreation/script
+++ b/acceptance/bundle/deployment/bind/dashboard/recreation/script
@@ -13,7 +13,7 @@ cleanupRemoveDashboard() {
 trap cleanupRemoveDashboard EXIT
 
 trace $CLI bundle deployment bind dashboard1 "${DASHBOARD_ID}" --auto-approve
-trace print_state.py --no-dms > out.state_after_bind.json
+trace print_state.py --no-dms | nostamp.py --indent 1 > out.state_after_bind.json
 ETAG=$(jq '.state["resources.dashboards.dashboard1"].state.etag // .resources[0].instances[0].attributes.etag // "no-etag"' < out.state_after_bind.json)
 echo $ETAG
 add_repl $ETAG ETAG
diff --git a/acceptance/bundle/deployment/bind/pipelines/recreate/script b/acceptance/bundle/deployment/bind/pipelines/recreate/script
index a9aa3403a35..0796fa0e9e6 100644
--- a/acceptance/bundle/deployment/bind/pipelines/recreate/script
+++ b/acceptance/bundle/deployment/bind/pipelines/recreate/script
@@ -12,7 +12,7 @@ rm -f "$OUT_REQUESTS"
 trace $CLI bundle deployment bind foo $NEW_PIPELINE_ID --auto-approve &> out.bind-success.txt
 trace print_requests.py '^//import-file/' '^//workspace/'
 
-trace $CLI bundle summary -o json | jq .resources > out.summary.json
+trace $CLI bundle summary -o json | jq .resources | nostamp.py > out.summary.json
 trace $CLI bundle plan
 
 trace musterr $CLI bundle deploy
diff --git a/acceptance/bundle/deployment/bind/pipelines/update/script b/acceptance/bundle/deployment/bind/pipelines/update/script
index 6bafa319765..77f95d4d3ed 100644
--- a/acceptance/bundle/deployment/bind/pipelines/update/script
+++ b/acceptance/bundle/deployment/bind/pipelines/update/script
@@ -9,7 +9,7 @@ rm -f "$OUT_REQUESTS"
 trace $CLI bundle deployment bind foo $NEW_PIPELINE_ID --auto-approve &> out.bind-success.txt
 print_requests.py '^//import-file/' '^//workspace/delete'
 
-trace $CLI bundle summary -o json > out.summary.json
+trace $CLI bundle summary -o json | nostamp.py > out.summary.json
 trace $CLI bundle plan
 
 rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/bind-history-failures/databricks.yml.tmpl b/acceptance/bundle/dms/bind-history-failures/databricks.yml.tmpl
new file mode 100644
index 00000000000..6f6a7fba6a3
--- /dev/null
+++ b/acceptance/bundle/dms/bind-history-failures/databricks.yml.tmpl
@@ -0,0 +1,13 @@
+bundle:
+  name: dms-bind-history-failures-$UNIQUE_NAME
+
+resources:
+  jobs:
+    create_failed:
+      name: create_failed
+    record_failed:
+      name: record_failed
+    complete_failed:
+      name: complete_failed
+    upload_failed:
+      name: upload_failed
diff --git a/acceptance/bundle/dms/bind-history-failures/out.test.toml b/acceptance/bundle/dms/bind-history-failures/out.test.toml
new file mode 100644
index 00000000000..88e6fbd847c
--- /dev/null
+++ b/acceptance/bundle/dms/bind-history-failures/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/bind-history-failures/output.txt b/acceptance/bundle/dms/bind-history-failures/output.txt
new file mode 100644
index 00000000000..8a2cf5d20f8
--- /dev/null
+++ b/acceptance/bundle/dms/bind-history-failures/output.txt
@@ -0,0 +1,73 @@
+
+=== A version creation failure does not prevent binding
+>>> fault.py POST /api/2.0/bundle/deployments/*/versions 400 0 1
+
+>>> [CLI] bundle deployment bind create_failed [JOB_ID] --auto-approve
+Warn: Failed to record bind history: failed to create deployment version: Fault injected by test.
+Successfully bound job with an id '[JOB_ID]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+=== An operation recording failure does not prevent binding
+>>> fault.py PATCH /api/2.0/bundle/** 400 0 1
+
+>>> [CLI] bundle deployment bind record_failed [JOB_ID] --auto-approve
+Warn: Failed to record bind history: Fault injected by test.
+Successfully bound job with an id '[JOB_ID]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+=== A version completion failure does not prevent binding
+>>> fault.py POST /api/2.0/bundle/deployments/*/versions/*/complete 400 0 1
+
+>>> [CLI] bundle deployment bind complete_failed [JOB_ID] --auto-approve
+Warn: Failed to complete deployment history version: Fault injected by test.
+Successfully bound job with an id '[JOB_ID]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+=== All bindings survived the remote state round trips
+>>> jq .state | map_values(.__id__) .databricks/bundle/default/resources.json
+{
+  "resources.jobs.complete_failed": "[JOB_ID]",
+  "resources.jobs.create_failed": "[JOB_ID]",
+  "resources.jobs.record_failed": "[JOB_ID]"
+}
+
+>>> [CLI] jobs get [JOB_ID]
+{
+  "job_id": [JOB_ID],
+  "name": "bind-target-[UNIQUE_NAME]"
+}
+
+>>> print_requests.py --dms //versions --oneline
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "3"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.create_failed"}], "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "5"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.record_failed"}], "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/5/operations/jobs.record_failed", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[JOB_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-bind-history-failures-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"record_failed\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/5/complete", "body": {"completion_reason": "VERSION_COMPLETE_FAILURE"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "7"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.complete_failed"}], "previous_version_id": "5", "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/7/operations/jobs.complete_failed", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[JOB_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-bind-history-failures-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"complete_failed\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/7/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
+
+=== A remote state upload failure fails the bind version
+>>> fault.py POST /api/2.0/workspace-files/import-file/**resources.json 400 0 1
+
+>>> musterr [CLI] bundle deployment bind upload_failed [JOB_ID] --auto-approve
+Error: Fault injected by test. (400 INJECTED)
+
+Endpoint: POST [DATABRICKS_URL]/api/2.0/workspace-files/import-file/Workspace%2FUsers%2F[USERNAME]%2F.bundle%2Fdms-bind-history-failures-[UNIQUE_NAME]%2Fdefault%2Fstate%2Fresources.json?overwrite=true
+HTTP Status: 400 Bad Request
+API error_code: INJECTED
+API message: Fault injected by test.
+
+
+>>> print_requests.py --dms //versions --oneline
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "9"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.upload_failed"}], "previous_version_id": "7", "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/9/operations/jobs.upload_failed", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[JOB_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-bind-history-failures-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"upload_failed\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/9/complete", "body": {"completion_reason": "VERSION_COMPLETE_FAILURE"}}
+
+=== Remote state still contains only the successful bindings
+>>> [CLI] bundle summary -o json
+{
+  "complete_failed": "[JOB_ID]",
+  "create_failed": "[JOB_ID]",
+  "record_failed": "[JOB_ID]",
+  "upload_failed": null
+}
diff --git a/acceptance/bundle/dms/bind-history-failures/script b/acceptance/bundle/dms/bind-history-failures/script
new file mode 100644
index 00000000000..eecbabcd3e2
--- /dev/null
+++ b/acceptance/bundle/dms/bind-history-failures/script
@@ -0,0 +1,33 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+job_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
+add_repl "$job_id" JOB_ID
+
+title "A version creation failure does not prevent binding"
+trace fault.py "POST /api/2.0/bundle/deployments/*/versions" 400 0 1
+trace $CLI bundle deployment bind create_failed "$job_id" --auto-approve
+rm -rf .databricks
+
+title "An operation recording failure does not prevent binding"
+trace fault.py "PATCH /api/2.0/bundle/**" 400 0 1
+trace $CLI bundle deployment bind record_failed "$job_id" --auto-approve
+rm -rf .databricks
+
+title "A version completion failure does not prevent binding"
+trace fault.py "POST /api/2.0/bundle/deployments/*/versions/*/complete" 400 0 1
+trace $CLI bundle deployment bind complete_failed "$job_id" --auto-approve
+
+title "All bindings survived the remote state round trips"
+trace jq '.state | map_values(.__id__)' .databricks/bundle/default/resources.json
+trace $CLI jobs get "$job_id" | jq '{job_id, name: .settings.name}'
+trace print_requests.py --dms //versions --oneline
+
+
+title "A remote state upload failure fails the bind version"
+trace fault.py "POST /api/2.0/workspace-files/import-file/**resources.json" 400 0 1
+trace musterr $CLI bundle deployment bind upload_failed "$job_id" --auto-approve
+trace print_requests.py --dms //versions --oneline
+
+rm -rf .databricks
+title "Remote state still contains only the successful bindings"
+trace $CLI bundle summary -o json | jq '.resources.jobs | map_values(.id // null)'
+rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/bind-history-failures/test.toml b/acceptance/bundle/dms/bind-history-failures/test.toml
new file mode 100644
index 00000000000..0e4c9337799
--- /dev/null
+++ b/acceptance/bundle/dms/bind-history-failures/test.toml
@@ -0,0 +1 @@
+Cloud = false # Inject failures into the deployment history service.
diff --git a/acceptance/bundle/dms/bind-immutable/databricks.yml.tmpl b/acceptance/bundle/dms/bind-immutable/databricks.yml.tmpl
new file mode 100644
index 00000000000..4d6fc71467e
--- /dev/null
+++ b/acceptance/bundle/dms/bind-immutable/databricks.yml.tmpl
@@ -0,0 +1,8 @@
+bundle:
+  name: dms-bind-immutable-$UNIQUE_NAME
+workspace:
+  sync_mode: immutable
+resources:
+  jobs:
+    foo:
+      name: foo
diff --git a/acceptance/bundle/dms/bind-immutable/out.test.toml b/acceptance/bundle/dms/bind-immutable/out.test.toml
new file mode 100644
index 00000000000..88e6fbd847c
--- /dev/null
+++ b/acceptance/bundle/dms/bind-immutable/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/bind-immutable/output.txt b/acceptance/bundle/dms/bind-immutable/output.txt
new file mode 100644
index 00000000000..99e62a0e129
--- /dev/null
+++ b/acceptance/bundle/dms/bind-immutable/output.txt
@@ -0,0 +1,13 @@
+
+>>> [CLI] bundle deployment bind foo [JOB_ID] --auto-approve
+Successfully bound job with an id '[JOB_ID]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_BIND",
+    "resource_id": "[JOB_ID]",
+    "resource_key": "jobs.foo"
+  }
+]
diff --git a/acceptance/bundle/dms/bind-immutable/script b/acceptance/bundle/dms/bind-immutable/script
new file mode 100644
index 00000000000..929a499ddd6
--- /dev/null
+++ b/acceptance/bundle/dms/bind-immutable/script
@@ -0,0 +1,8 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+job_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
+add_repl "$job_id" JOB_ID
+
+trace $CLI bundle deployment bind foo "$job_id" --auto-approve
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-bind-immutable-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | jq -r '.object_id')
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${deployment_id}" -o json | jq -S 'map({resource_key, resource_id, last_action_type})'
+rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/bind-immutable/test.toml b/acceptance/bundle/dms/bind-immutable/test.toml
new file mode 100644
index 00000000000..ee99e628c60
--- /dev/null
+++ b/acceptance/bundle/dms/bind-immutable/test.toml
@@ -0,0 +1,4 @@
+# This test ensures that deployment history works with bind and immutable folders. This case is interesting
+# because by default we record the snapshot path in deploy, but that is not available during bind. So it's important
+# to ensure that we don't do that during deployment bind, and recording history continues to work.
+Cloud = false
diff --git a/acceptance/bundle/dms/bind-unbind/out.test.toml b/acceptance/bundle/dms/bind-unbind/out.test.toml
index 88e6fbd847c..84b22ba6735 100644
--- a/acceptance/bundle/dms/bind-unbind/out.test.toml
+++ b/acceptance/bundle/dms/bind-unbind/out.test.toml
@@ -1,2 +1,2 @@
-Cloud = false
+Cloud = true
 EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/bind-unbind/output.txt b/acceptance/bundle/dms/bind-unbind/output.txt
index d6d3fcd36e6..922b2985955 100644
--- a/acceptance/bundle/dms/bind-unbind/output.txt
+++ b/acceptance/bundle/dms/bind-unbind/output.txt
@@ -1,10 +1,44 @@
 
 === Bind an existing job with deployment history enabled
+>>> musterr [CLI] bundle deployment bind foo [JOB_ID]
+Plan: update resources.jobs.foo
+
+Changes detected:
+  ~ deployment: null -> {"kind":"BUNDLE","metadata_file_path":"/Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/state/metadata.json"}
+  ~ edit_mode: null -> "UI_LOCKED"
+  ~ name: "bind-target-[UNIQUE_NAME]" -> "foo"
+
+Error: this bind operation requires user confirmation, but the current console does not support prompting.
+To proceed, use --auto-approve after reviewing the plan above.
+
+
+>>> print_requests.py --dms //api/2.0/bundle
+
 >>> [CLI] bundle deployment bind foo [JOB_ID] --auto-approve
 Successfully bound job with an id '[JOB_ID]'
 Run 'bundle deploy' to deploy changes to your workspace
 
-=== Deploy keeps the bound job ID and enables shadow history
+>>> print_requests.py --dms //versions --oneline
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "3"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/3/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[JOB_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/3/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
+
+>>> [CLI] jobs get [JOB_ID]
+{
+  "job_id": [JOB_ID],
+  "name": "bind-target-[UNIQUE_NAME]"
+}
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_BIND",
+    "resource_id": "[JOB_ID]",
+    "resource_key": "jobs.foo"
+  }
+]
+
+=== Deploy keeps the bound job ID with recorded history
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/files...
 Updated jobs.foo
@@ -48,6 +82,11 @@ Run 'bundle deploy' to deploy changes to your workspace
   "resource_id": "[JOB_ID]"
 }
 
+>>> print_requests.py --dms //versions --oneline
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "7"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_BIND", "resource_key": "jobs.foo"}], "previous_version_id": "4", "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/7/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[JOB_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/7/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
+
 === A subsequent deploy continues managing the same job
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/files...
@@ -61,3 +100,12 @@ Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
   },
   "resource_id": "[JOB_ID]"
 }
+
+=== A rejected duplicate bind creates no history version
+>>> musterr [CLI] bundle deployment bind foo [JOB_ID] --auto-approve
+Error: Resource already managed
+
+The bundle is already managing resource for resources.jobs.foo and it is bound to the requested ID [JOB_ID].
+
+
+>>> print_requests.py --dms //versions
diff --git a/acceptance/bundle/dms/bind-unbind/script b/acceptance/bundle/dms/bind-unbind/script
index 32fab5aa0e8..8c7d75c3d43 100644
--- a/acceptance/bundle/dms/bind-unbind/script
+++ b/acceptance/bundle/dms/bind-unbind/script
@@ -1,11 +1,25 @@
 envsubst < databricks.yml.tmpl > databricks.yml
 
 title "Bind an existing job with deployment history enabled"
-job_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
+# Match Jobs API defaults so the bind preview is identical locally and on cloud.
+job_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'", "format": "MULTI_TASK", "max_concurrent_runs": 1, "queue": {"enabled": true}}' | jq -r '.job_id')
 add_repl "$job_id" JOB_ID
+cleanup_bind_target() {
+    errcode $CLI bundle destroy --auto-approve >> LOG.cleanup 2>&1
+    errcode $CLI jobs delete "$job_id" >> LOG.cleanup 2>&1
+}
+trap cleanup_bind_target EXIT
+
+trace musterr $CLI bundle deployment bind foo "$job_id"
+trace print_requests.py --dms //api/2.0/bundle
 trace $CLI bundle deployment bind foo "$job_id" --auto-approve
+trace print_requests.py --dms //versions --oneline
+trace $CLI jobs get "$job_id" | jq '{job_id, name: .settings.name}'
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-bind-unbind-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | jq -r '.object_id')
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${deployment_id}" -o json | jq -S 'map({resource_key, resource_id, last_action_type})'
+
 
-title "Deploy keeps the bound job ID and enables shadow history"
+title "Deploy keeps the bound job ID with recorded history"
 trace $CLI bundle deploy
 trace jq '{features, resource_id: .state["resources.jobs.foo"].__id__}' .databricks/bundle/default/resources.json
 
@@ -14,10 +28,20 @@ trace $CLI bundle deployment unbind foo
 trace jq '{features, state}' .databricks/bundle/default/resources.json
 trace $CLI jobs get "$job_id" | jq '{job_id}'
 
+rm -f "$OUT_REQUESTS"
+
 title "Rebind preserves the existing shadow history marker"
 trace $CLI bundle deployment bind foo "$job_id" --auto-approve
 trace jq '{features, resource_id: .state["resources.jobs.foo"].__id__}' .databricks/bundle/default/resources.json
 
+trace print_requests.py --dms //versions --oneline
+
 title "A subsequent deploy continues managing the same job"
 trace $CLI bundle deploy
 trace jq '{features, resource_id: .state["resources.jobs.foo"].__id__}' .databricks/bundle/default/resources.json
+
+rm -f "$OUT_REQUESTS"
+
+title "A rejected duplicate bind creates no history version"
+trace musterr $CLI bundle deployment bind foo "$job_id" --auto-approve
+trace print_requests.py --dms //versions
diff --git a/acceptance/bundle/dms/bind-unbind/test.toml b/acceptance/bundle/dms/bind-unbind/test.toml
index f620acda97e..b7cf7fdd9a0 100644
--- a/acceptance/bundle/dms/bind-unbind/test.toml
+++ b/acceptance/bundle/dms/bind-unbind/test.toml
@@ -1,3 +1,2 @@
-RecordRequests = false
 # Verify shadow history markers and resource IDs across bind, deploy, unbind, and rebind.
-Cloud = false
+Cloud = true
diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go
index 7fbf7ede662..558d02e1ece 100644
--- a/bundle/direct/bind.go
+++ b/bundle/direct/bind.go
@@ -10,6 +10,7 @@ import (
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct/dresources"
 	"github.com/databricks/cli/bundle/direct/dstate"
+	bundleenv "github.com/databricks/cli/bundle/env"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/structs/structaccess"
 	"github.com/databricks/cli/libs/structs/structpath"
@@ -35,8 +36,16 @@ func (e ErrResourceAlreadyBound) Error() string {
 		e.ResourceKey, e.ExistingID)
 }
 
+// DMSBindMetadata holds the bound resource state and serial to publish after bind is committed.
+type DMSBindMetadata struct {
+	Serial        int
+	ResourceState dstate.ResourceEntry
+}
+
 // BindResult contains the result of a bind operation including any detected changes.
 type BindResult struct {
+	// DMSMetadata is nil when deployment history is disabled.
+	DMSMetadata *DMSBindMetadata
 	// HasChanges is true if deploying after bind would make changes to the resource
 	HasChanges bool
 	// Action is the planned action for the bound resource (e.g., "skip", "update", "recreate")
@@ -197,6 +206,15 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 		os.Remove(tmpStatePath)
 		return nil, err
 	}
+	var configured *bool
+	if configRoot.Experimental != nil {
+		configured = configRoot.Experimental.DeploymentHistory
+	}
+	var dmsMetadata *DMSBindMetadata
+	if b.StateDB.RecordsHistory() || bundleenv.RecordsDeploymentHistory(ctx, configured) {
+		resourceState, _ := b.StateDB.GetResourceEntry(resourceKey)
+		dmsMetadata = &DMSBindMetadata{Serial: b.StateDB.Data.Serial, ResourceState: resourceState}
+	}
 	plan, err = b.CalculatePlan(ctx, client, configRoot)
 	if _, ferr := b.StateDB.Finalize(ctx); ferr != nil {
 		log.Warnf(ctx, "failed to finalize state: %v", ferr)
@@ -208,6 +226,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 
 	// Check if the bound resource has changes
 	result := &BindResult{
+		DMSMetadata:   dmsMetadata,
 		HasChanges:    false,
 		Action:        deployplan.Skip,
 		Plan:          plan,
diff --git a/bundle/phases/bind.go b/bundle/phases/bind.go
index 36c4fcf56b4..2e935e06be2 100644
--- a/bundle/phases/bind.go
+++ b/bundle/phases/bind.go
@@ -7,16 +7,25 @@ import (
 	"fmt"
 	"maps"
 	"slices"
+	"strconv"
+	"strings"
+	"time"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/bundle/deploy/lock"
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dstate"
 	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/internal/build"
 	"github.com/databricks/cli/libs/agent"
 	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/dms"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/cli/libs/retry"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
 type BindOptions struct {
@@ -27,6 +36,28 @@ type BindOptions struct {
 	ResourceId  string
 }
 
+// createBindingDeployment registers a first binding without changing metadata from an existing deploy.
+func createBindingDeployment(ctx context.Context, b *bundle.Bundle) (string, error) {
+	// Bind does not upload files or stage an immutable snapshot, so it cannot report a deployed file path.
+	created, err := b.WorkspaceClient(ctx).BundleDeployments.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{
+		Deployment: bundledeployments.Deployment{
+			DisplayName:       b.Config.Bundle.Name,
+			TargetName:        b.Config.Bundle.Target,
+			DeploymentMode:    deploymentModeToSDK(b.Config.Bundle.Mode),
+			WorkspaceInfo:     &bundledeployments.WorkspaceInfo{RootPath: b.Config.Workspace.RootPath},
+			InitialParentPath: b.Config.Workspace.StatePath,
+		},
+	})
+	if err != nil {
+		return "", fmt.Errorf("failed to create deployment: %w", err)
+	}
+	deploymentID, err := dms.DeploymentIDFromName(created.Name)
+	if err != nil {
+		return "", fmt.Errorf("failed to create deployment: %w", err)
+	}
+	return deploymentID, nil
+}
+
 func Bind(ctx context.Context, b *bundle.Bundle, opts *BindOptions, stateDesc *statemgmt.StateDesc) {
 	log.Info(ctx, "Phase: bind")
 	engine := stateDesc.Engine
@@ -36,7 +67,10 @@ func Bind(ctx context.Context, b *bundle.Bundle, opts *BindOptions, stateDesc *s
 		return
 	}
 
+	var historyVersion string
+	var historyErr error
 	defer func() {
+		completeBindingHistory(ctx, b, historyVersion, historyErr == nil && !logdiag.HasError(ctx))
 		bundle.ApplyContext(ctx, b, lock.Release(lock.GoalBind))
 	}()
 
@@ -102,9 +136,32 @@ func Bind(ctx context.Context, b *bundle.Bundle, opts *BindOptions, stateDesc *s
 		return
 	}
 
+	historyVersion, historyErr = recordBindHistory(ctx, b, resourceKey, result.DMSMetadata)
+	if historyErr != nil {
+		log.Warnf(ctx, "Failed to record bind history: %s", historyErr)
+	}
+
 	statemgmt.PushResourcesState(ctx, b)
 }
 
+// completeBindingHistory completes the recorded version after the remote state upload.
+func completeBindingHistory(ctx context.Context, b *bundle.Bundle, versionName string, success bool) {
+	if versionName == "" {
+		return
+	}
+	reason := bundledeployments.VersionCompleteVersionCompleteFailure
+	if success {
+		reason = bundledeployments.VersionCompleteVersionCompleteSuccess
+	}
+	_, err := b.WorkspaceClient(ctx).BundleDeployments.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{
+		Name:             versionName,
+		CompletionReason: reason,
+	})
+	if err != nil {
+		log.Warnf(ctx, "Failed to complete deployment history version: %s", err)
+	}
+}
+
 func jsonDump(ctx context.Context, v any, field string) string {
 	b, err := json.Marshal(v)
 	if err != nil {
@@ -143,3 +200,94 @@ func Unbind(ctx context.Context, b *bundle.Bundle, group, resourceKey string, en
 
 	statemgmt.PushResourcesState(ctx, b)
 }
+
+// recordBindHistory publishes a committed binding when DMS metadata was requested.
+func recordBindHistory(ctx context.Context, b *bundle.Bundle, resourceKey string, metadata *direct.DMSBindMetadata) (string, error) {
+	if metadata == nil {
+		return "", nil
+	}
+	versionName, err := createBindingVersion(ctx, b, metadata.Serial, []bundledeployments.StagedOperation{{
+		ResourceKey: strings.TrimPrefix(resourceKey, dms.StatePrefix),
+		ActionType:  bundledeployments.OperationActionTypeOperationActionTypeBind,
+	}})
+	if err != nil {
+		return "", err
+	}
+	return versionName, recordBindingOperation(ctx, b, versionName, resourceKey, &metadata.ResourceState)
+}
+
+// createBindingVersion claims a history version for the staged binding operations.
+func createBindingVersion(ctx context.Context, b *bundle.Bundle, serial int, staged []bundledeployments.StagedOperation) (string, error) {
+	deploymentID, current, err := dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+	if err != nil {
+		return "", err
+	}
+	previous, err := dms.ParseLastVersionID(current)
+	if err != nil {
+		return "", err
+	}
+	if current == nil {
+		deploymentID, err = createBindingDeployment(ctx, b)
+		if err != nil {
+			return "", err
+		}
+	}
+	versionID := max(previous, serial) + 1
+	previousVersionID := ""
+	if previous > 0 {
+		previousVersionID = strconv.Itoa(previous)
+	}
+	var gitInfo *bundledeployments.GitInfo
+	if git := b.Config.Bundle.Git; git.Branch != "" || git.Commit != "" || git.OriginURL != "" {
+		gitInfo = &bundledeployments.GitInfo{Branch: git.Branch, Commit: git.Commit, OriginUrl: git.OriginURL}
+	}
+	version, err := b.WorkspaceClient(ctx).BundleDeployments.CreateVersion(ctx, bundledeployments.CreateVersionRequest{
+		Parent:    dms.DeploymentName(deploymentID),
+		VersionId: strconv.Itoa(versionID),
+		Version: bundledeployments.Version{
+			CliVersion:        build.GetInfo().Version,
+			VersionType:       dms.VersionTypeDeploy,
+			PreviousVersionId: previousVersionID,
+			GitInfo:           gitInfo,
+			Operations:        staged,
+		},
+	})
+	if err != nil {
+		return "", fmt.Errorf("failed to create deployment version: %w", err)
+	}
+	return version.Name, nil
+}
+
+// recordBindingOperation sends an operation's saved state, or its removal, directly to DMS.
+func recordBindingOperation(ctx context.Context, b *bundle.Bundle, versionName, resourceKey string, entry *dstate.ResourceEntry) error {
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	var state json.RawMessage
+	var resourceID string
+	if entry != nil {
+		var err error
+		state, err = json.Marshal(dstate.RecordedState{State: entry.State, DependsOn: entry.DependsOn})
+		if err != nil {
+			return err
+		}
+		resourceID = entry.ID
+	}
+	update, err := dms.NewStateUpdate(resourceID, state, false)
+	if err != nil {
+		return err
+	}
+	// Newly staged operations start at sequence 0; DMS requires it even though Go omits zero values.
+	_, err = retry.With(ctx, retry.IsTransient, func() (*bundledeployments.Operation, error) {
+		return b.WorkspaceClient(ctx).BundleDeployments.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{
+			Name: versionName + "/operations/" + strings.TrimPrefix(resourceKey, dms.StatePrefix),
+			Operation: bundledeployments.Operation{
+				State:           string(update.State),
+				ResourceId:      update.ResourceID,
+				Status:          update.Status,
+				ForceSendFields: []string{"SequenceId", "ErrorMessage"},
+			},
+			UpdateMask: update.Fields.Mask(),
+		})
+	})
+	return err
+}
diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go
index 68798a4bb01..0f866174802 100644
--- a/cmd/bundle/utils/process.go
+++ b/cmd/bundle/utils/process.go
@@ -4,10 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"path"
 	"path/filepath"
 	"slices"
-	"strconv"
 	"time"
 
 	"github.com/databricks/cli/bundle"
@@ -32,14 +30,10 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
-	"github.com/databricks/cli/libs/retry"
 	"github.com/databricks/cli/libs/sync"
 	"github.com/databricks/cli/libs/telemetry"
 	"github.com/databricks/cli/libs/telemetry/protos"
-	"github.com/databricks/databricks-sdk-go"
-	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
-	"github.com/databricks/databricks-sdk-go/service/workspace"
 	"github.com/databricks/databricks-sdk-go/useragent"
 	"github.com/spf13/cobra"
 	"golang.org/x/mod/semver"
@@ -411,7 +405,7 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 
 		if needDirectState && stateMode == dstate.WritingDeploymentHistoryOnly {
 			var err error
-			dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+			dmsDeploymentID, dmsDeployment, err = dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
 			if err != nil {
 				logdiag.LogError(ctx, err)
 				return b, stateDesc, root.ErrAlreadyPrinted
@@ -421,7 +415,7 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 			// the plan carries them. version_id is always known (last recorded + 1); deployment_id
 			// does not exist until a first deploy creates it, so it is left off here and the deploy
 			// phase stamps the created id.
-			lastVersionID, err := parseLastVersionID(dmsDeployment)
+			lastVersionID, err := dms.ParseLastVersionID(dmsDeployment)
 			if err != nil {
 				logdiag.LogError(ctx, err)
 				return b, stateDesc, root.ErrAlreadyPrinted
@@ -637,53 +631,6 @@ func ValidateEngineSetting(ctx context.Context, b *bundle.Bundle) error {
 	return nil
 }
 
-// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json
-//
-// TODO: a deployment is only usable when both the node and the service's record exist, and a
-// half-created one - node present, record missing - blocks the bundle here even though
-// CreateDeployment already recovers from it. Move this behind a dms.ReadDeployment(ctx, statePath)
-// that returns an empty id and version unless both halves are there, leaving the caller to call
-// CreateDeployment to create or finalize it.
-//
-// TODO: ask the service for a lookup by state path, so this is one round trip rather than two - a
-// workspace lookup to turn the node into an id, then a get by that id.
-func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) {
-	nodePath := path.Join(statePath, dms.DeploymentNodeName)
-
-	obj, err := retry.With(ctx, retry.IsTransientRead, func() (*workspace.ObjectInfo, error) {
-		return w.Workspace.GetStatusByPath(ctx, nodePath)
-	})
-	if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) {
-		return "", nil, nil
-	}
-	if err != nil {
-		return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err)
-	}
-	deploymentID := strconv.FormatInt(obj.ObjectId, 10)
-	deployment, err := retry.With(ctx, retry.IsTransientRead, func() (*bundledeployments.Deployment, error) {
-		return w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{
-			Name: dms.DeploymentName(deploymentID),
-		})
-	})
-	if err != nil {
-		return "", nil, err
-	}
-	return deploymentID, deployment, nil
-}
-
-// parseLastVersionID parses the deployment's last recorded version, which the service reports
-// as a string. It returns 0 when the deployment does not exist yet or has no recorded version.
-func parseLastVersionID(dmsDeployment *bundledeployments.Deployment) (int, error) {
-	if dmsDeployment == nil || dmsDeployment.LastVersionId == "" {
-		return 0, nil
-	}
-	v, err := strconv.Atoi(dmsDeployment.LastVersionId)
-	if err != nil {
-		return 0, fmt.Errorf("failed to parse last_version_id %q: %w", dmsDeployment.LastVersionId, err)
-	}
-	return v, nil
-}
-
 // OpenDirectStateForRead opens the direct-engine state database read-only. When the bundle
 // records deployment history the local state file is only a tombstone, so the resources are
 // read from the deployment metadata service instead.
@@ -696,11 +643,11 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st
 		return nil
 	}
 
-	dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+	dmsDeploymentID, dmsDeployment, err := dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
 	if err != nil {
 		return err
 	}
-	lastVersionID, err := parseLastVersionID(dmsDeployment)
+	lastVersionID, err := dms.ParseLastVersionID(dmsDeployment)
 	if err != nil {
 		return err
 	}
@@ -716,11 +663,11 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st
 
 // prepareDeploymentHistoryShadow reads version metadata, never resource state, from DMS.
 func prepareDeploymentHistoryShadow(ctx context.Context, b *bundle.Bundle) (*bundledeployments.Deployment, error) {
-	id, deployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+	id, deployment, err := dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
 	if err != nil {
 		return nil, err
 	}
-	lastVersion, err := parseLastVersionID(deployment)
+	lastVersion, err := dms.ParseLastVersionID(deployment)
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/bundle/utils/process_test.go b/cmd/bundle/utils/process_test.go
index 21910039344..b05fb9b0a84 100644
--- a/cmd/bundle/utils/process_test.go
+++ b/cmd/bundle/utils/process_test.go
@@ -3,6 +3,7 @@ package utils
 import (
 	"testing"
 
+	"github.com/databricks/cli/libs/dms"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -61,7 +62,7 @@ func TestParseLastVersionID(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := parseLastVersionID(tt.deployment)
+			got, err := dms.ParseLastVersionID(tt.deployment)
 			if tt.wantErr {
 				require.Error(t, err)
 				return
diff --git a/libs/dms/metadata.go b/libs/dms/metadata.go
index f9e23a67db0..98010ac7b6f 100644
--- a/libs/dms/metadata.go
+++ b/libs/dms/metadata.go
@@ -1,12 +1,19 @@
 package dms
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"path"
 	"strconv"
 	"strings"
 
+	"github.com/databricks/cli/libs/retry"
 	"github.com/databricks/cli/libs/structs/structdiff"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
+	"github.com/databricks/databricks-sdk-go/service/workspace"
 )
 
 // VersionType identifies the kind of deployment a version records.
@@ -68,6 +75,19 @@ func (m Metadata) StaleFields(current *bundledeployments.Deployment) string {
 	return strings.Join(stale, ",")
 }
 
+// ParseLastVersionID parses the deployment's last recorded version, which the service reports
+// as a string. It returns 0 when the deployment does not exist yet or has no recorded version.
+func ParseLastVersionID(dmsDeployment *bundledeployments.Deployment) (int, error) {
+	if dmsDeployment == nil || dmsDeployment.LastVersionId == "" {
+		return 0, nil
+	}
+	v, err := strconv.Atoi(dmsDeployment.LastVersionId)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse last_version_id %q: %w", dmsDeployment.LastVersionId, err)
+	}
+	return v, nil
+}
+
 // NextVersion is the version number this run will create, given the deployment's most recent
 // one. lastVersionID is empty before the deployment has any version.
 func NextVersion(lastVersionID string) (int, error) {
@@ -132,3 +152,40 @@ func DeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployme
 	}
 	return dep
 }
+
+// LookupDeployment resolves the deployment node under statePath and reads its DMS record.
+// A missing workspace node returns no deployment.
+//
+// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json
+//
+// TODO: a deployment is only usable when both the node and the service's record exist, and a
+// half-created one - node present, record missing - blocks the bundle here even though
+// CreateDeployment already recovers from it. Move this behind a dms.ReadDeployment(ctx, statePath)
+// that returns an empty id and version unless both halves are there, leaving the caller to call
+// CreateDeployment to create or finalize it.
+//
+// TODO: ask the service for a lookup by state path, so this is one round trip rather than two - a
+// workspace lookup to turn the node into an id, then a get by that id.
+func LookupDeployment(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) {
+	nodePath := path.Join(statePath, DeploymentNodeName)
+
+	obj, err := retry.With(ctx, retry.IsTransientRead, func() (*workspace.ObjectInfo, error) {
+		return w.Workspace.GetStatusByPath(ctx, nodePath)
+	})
+	if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) {
+		return "", nil, nil
+	}
+	if err != nil {
+		return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err)
+	}
+	deploymentID := strconv.FormatInt(obj.ObjectId, 10)
+	deployment, err := retry.With(ctx, retry.IsTransientRead, func() (*bundledeployments.Deployment, error) {
+		return w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{
+			Name: DeploymentName(deploymentID),
+		})
+	})
+	if err != nil {
+		return "", nil, err
+	}
+	return deploymentID, deployment, nil
+}
diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go
index 0eec1a4e463..e341789fb2a 100644
--- a/libs/testserver/bundledeployments.go
+++ b/libs/testserver/bundledeployments.go
@@ -50,6 +50,7 @@ var dmsResourceTypes = []string{
 	"database_instances",
 	"experiments",
 	"external_locations",
+	"features",
 	"genie_spaces",
 	"instance_pools",
 	"internal_immutable_folders",