diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt
index 4057525509b..a239b1b14c1 100644
--- a/acceptance/bundle/dms/existing-state/output.txt
+++ b/acceptance/bundle/dms/existing-state/output.txt
@@ -1,5 +1,5 @@
 
-=== Deploy without recording: ordinary direct-engine state tracking one job, unknown to DMS
+=== 1. Deploy without history: existing state tracks one job
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files...
 Created jobs.one
@@ -25,29 +25,30 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
   "features": null
 }
 
-=== Enabling recording is refused because the existing deployment is not recorded
+=== 2. Enable history on the existing deployment
+Expect: migration succeeds without recreating the job.
 >>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
-Error: deployment history setting (true) does not match the existing state (false)
-
-Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over
-
-
->>> print_requests.py --dms //api/2.0/bundle --oneline
+Enabling deployment history...
+Deployment history enabled: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=2
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files...
+Updated jobs.one
+Files: 0 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
 
-=== Wiping the local cache still refuses after pulling the non-DMS state from the workspace
+=== 3. Remove local state and deploy again
+Expect: remote DMS state is loaded; no migration or resource creation.
 >>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
-Error: deployment history setting (true) does not match the existing state (false)
-
-Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over
-
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files...
+Files: 2 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
 
-=== A leftover local WAL with the wrong lineage is still refused
+=== 4. A leftover local WAL is refused: DMS deployments do not use WALs
 >>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
-Error: reading state from [TEST_TMP_DIR]/.databricks/bundle/default/resources.json: WAL recovery failed: WAL lineage ("stale") does not match state lineage ("[UUID]")
+Error: unexpected WAL file found at [TEST_TMP_DIR]/.databricks/bundle/default/resources.json.wal: this deployment records deployment history, which does not write one
 
 
-=== Destroy clears the deployment. With nothing recorded remotely the bundle is treated as new, so recording can be enabled once the stale WAL is gone
->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle destroy --auto-approve
+=== 5. Destroy follows recorded state; redeploy starts a new history
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
   delete resources.jobs.one
 
@@ -55,6 +56,7 @@ All files and directories at the following location will be deleted: /Workspace/
 
 Destroy: 1 deleted
 
+=== Fresh history: version 1 creates the job without migration
 >>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files...
 Created jobs.one
@@ -63,9 +65,9 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 
 >>> print_requests.py --dms //api/2.0/bundle --oneline
 {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-existing-state-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default"}}}
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.one"}], "version_type": "VERSION_TYPE_DEPLOY"}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[ONE_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[DEPLOYMENT_ID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID_2]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.one"}], "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID_2]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[ONE_ID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[DEPLOYMENT_ID_2]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID_2]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
 
 >>> find.py resources[.]json[.]wal --expect 0
 
diff --git a/acceptance/bundle/dms/existing-state/script b/acceptance/bundle/dms/existing-state/script
index e64ca6febe6..2a9a48abc90 100644
--- a/acceptance/bundle/dms/existing-state/script
+++ b/acceptance/bundle/dms/existing-state/script
@@ -1,38 +1,39 @@
 envsubst < databricks.yml.tmpl > databricks.yml
 
-title "Deploy without recording: ordinary direct-engine state tracking one job, unknown to DMS"
+title "1. Deploy without history: existing state tracks one job"
 trace $CLI bundle deploy
 dms_plan
 trace print_requests.py --dms //api/2.0/bundle --oneline
 trace jq '{state_version, features}' .databricks/bundle/default/resources.json
 
-title "Enabling recording is refused because the existing deployment is not recorded"
-musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
-trace print_requests.py --dms //api/2.0/bundle --oneline
+title "2. Enable history on the existing deployment\nExpect: migration succeeds without recreating the job."
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-existing-state-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+add_repl "$deployment_id" DEPLOYMENT_ID
 
-title "Wiping the local cache still refuses after pulling the non-DMS state from the workspace"
+title "3. Remove local state and deploy again\nExpect: remote DMS state is loaded; no migration or resource creation."
 rm -rf .databricks
-musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
 
-title "A leftover local WAL with the wrong lineage is still refused"
+title "4. A leftover local WAL is refused: DMS deployments do not use WALs"
 mkdir -p .databricks/bundle/default
 printf '{"lineage":"stale","serial":99,"state_version":2,"features":{"deployment_history":{}}}\n' > .databricks/bundle/default/resources.json.wal
 musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
 
-title "Destroy clears the deployment. With nothing recorded remotely the bundle is treated as new, so recording can be enabled once the stale WAL is gone"
-# Clear the stale WAL first: it was refused, not deleted, and destroy runs without recording, so it
-# would otherwise try to replay it and fail on the lineage mismatch.
+title "5. Destroy follows recorded state; redeploy starts a new history"
+# The rejected WAL remains on disk. Remove it before destroying the recorded deployment.
 rm -f .databricks/bundle/default/resources.json.wal
-# Keep the mismatched setting here to verify that destroy ignores configuration and follows state.
-trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle destroy --auto-approve
+# Turn history off to check that destroy still follows the recorded backend.
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false $CLI bundle destroy --auto-approve
+title "Fresh history: version 1 creates the job without migration"
+rm -f "$OUT_REQUESTS"
 trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
 trace print_requests.py --dms //api/2.0/bundle --oneline
 trace find.py 'resources[.]json[.]wal' --expect 0
 trace jq '{state_version, features}' .databricks/bundle/default/resources.json
 DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true replace_ids.py
-deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-existing-state-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-existing-state-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
 add_repl "$deployment_id" DEPLOYMENT_ID
 
-# The recorded requests are asserted above via print_requests.py; drop whatever the run added after
-# that (telemetry, and the state probes of the non-recording variant) so nothing is left to compare.
+# Drop telemetry and state probes recorded after the fresh-deployment assertion above.
 rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/migration-before-deploy/databricks.yml.tmpl b/acceptance/bundle/dms/migration-before-deploy/databricks.yml.tmpl
new file mode 100644
index 00000000000..a0af2a05d62
--- /dev/null
+++ b/acceptance/bundle/dms/migration-before-deploy/databricks.yml.tmpl
@@ -0,0 +1,18 @@
+bundle:
+  name: dms-migration-before-deploy-$UNIQUE_NAME
+
+experimental:
+  deployment_history: false
+
+sync:
+  exclude:
+    - LOG.*
+
+resources:
+  jobs:
+    job:
+      name: migration-before-deploy
+  schemas:
+    schema:
+      name: dms_migration_before_deploy_$UNIQUE_NAME
+      catalog_name: main
diff --git a/acceptance/bundle/dms/migration-before-deploy/out.test.toml b/acceptance/bundle/dms/migration-before-deploy/out.test.toml
new file mode 100644
index 00000000000..88e6fbd847c
--- /dev/null
+++ b/acceptance/bundle/dms/migration-before-deploy/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/migration-before-deploy/output.txt b/acceptance/bundle/dms/migration-before-deploy/output.txt
new file mode 100644
index 00000000000..2e8d58a6f37
--- /dev/null
+++ b/acceptance/bundle/dms/migration-before-deploy/output.txt
@@ -0,0 +1,80 @@
+
+=== Setup (LOG.setup): create a job and schema without history
+Next deploy enables history and recreates the schema in another catalog, requiring approval.
+=== 1. Fail deployment planning
+Expect: migration version 2 stays complete; no deployment version is created.
+>>> fault.py GET /api/2.2/jobs/get* 403 0 1
+
+>>> musterr [CLI] bundle deploy -qq
+Enabling deployment history...
+Deployment history enabled: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=2
+Error: cannot plan resources.jobs.job: reading id="[JOB_ID]": Fault injected by test. (403 INJECTED)
+
+Endpoint: GET [DATABRICKS_URL]/api/2.2/jobs/get?job_id=[JOB_ID]
+HTTP Status: 403 Forbidden
+API error_code: INJECTED
+API message: Fault injected by test.
+
+Error: planning failed
+
+
+=== Request order: complete migration -> upload marker -> read DMS -> plan jobs
+>>> print_requests.py --dms --get //api/2.0/bundle //jobs/get //workspace/import //workspace-files/import-file --oneline
+{"method":"POST","path":"/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/2/complete"}
+{"method":"POST","path":"/Workspace/Users/[USERNAME]/.bundle/dms-migration-before-deploy-[UNIQUE_NAME]/default/state/resources.json"}
+{"method":"GET","path":"/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/resources"}
+{"method":"GET","path":"/api/2.2/jobs/get"}
+
+=== Remote state: version 2 marker; original resources preserved in the backup
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-migration-before-deploy-[UNIQUE_NAME]/default/state/resources.json --format AUTO --file .databricks/resources.marker.json
+
+>>> jq -c {serial, features, state} .databricks/resources.marker.json
+{"serial":2,"features":{"deployment_history":{}},"state":{}}
+
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-migration-before-deploy-[UNIQUE_NAME]/default/state/resources.json.backup --format AUTO --file .databricks/resources.backup.json
+
+>>> cmp .databricks/resources.before-migration.json .databricks/resources.backup.json
+
+=== Migration result: only completed version 2, with INITIAL_REGISTER operations
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-versions deployments/[DEPLOYMENT_ID] -o json
+[{"version_id":"2","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"}]
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-operations deployments/[DEPLOYMENT_ID]/versions/2 -o json
+[{"resource_key":"jobs.job","action_type":"OPERATION_ACTION_TYPE_INITIAL_REGISTER","status":"OPERATION_STATUS_SUCCEEDED"},{"resource_key":"schemas.schema","action_type":"OPERATION_ACTION_TYPE_INITIAL_REGISTER","status":"OPERATION_STATUS_SUCCEEDED"}]
+
+=== 2. Retry without approval
+Expect: deployment declined; migration stays at version 2 and is not repeated.
+>>> musterr [CLI] bundle deploy -qq
+
+This action will result in the deletion or recreation of the following UC schemas. Any underlying data may be lost:
+  recreate resources.schemas.schema
+Error: the deployment requires destructive actions, but the current console does not support prompting.
+Deleting data assets such as schemas, pipelines, or volumes may cause permanent data loss and should be carefully reviewed.
+To proceed, use --auto-approve after reviewing the plan above.
+
+
+=== Local and remote markers must stay byte-for-byte unchanged
+>>> cmp .databricks/resources.marker.json .databricks/bundle/default/resources.json
+
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-migration-before-deploy-[UNIQUE_NAME]/default/state/resources.json --format AUTO --file .databricks/resources.after-decline.json
+
+>>> cmp .databricks/resources.marker.json .databricks/resources.after-decline.json
+
+=== No DMS writes and no extra version on the declined retry
+>>> print_requests.py --dms //api/2.0/bundle
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-versions deployments/[DEPLOYMENT_ID] -o json
+[{"version_id":"2","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"}]
+
+=== 3. Approve deployment
+Expect: ordinary deployment version 3; version 2 remains the only migration.
+>>> [CLI] bundle deploy --auto-approve -qq
+
+This action will result in the deletion or recreation of the following UC schemas. Any underlying data may be lost:
+  recreate resources.schemas.schema
+
+>>> [CLI] bundle summary -o json
+{"latest_version_id":3}
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-versions deployments/[DEPLOYMENT_ID] -o json
+[{"version_id":"2","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"},{"version_id":"3","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"}]
diff --git a/acceptance/bundle/dms/migration-before-deploy/script b/acceptance/bundle/dms/migration-before-deploy/script
new file mode 100644
index 00000000000..18b106a7c48
--- /dev/null
+++ b/acceptance/bundle/dms/migration-before-deploy/script
@@ -0,0 +1,50 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+state_path="/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-migration-before-deploy-${UNIQUE_NAME}/default/state"
+other_catalog="dms_other_${UNIQUE_NAME}"
+
+title "Setup (LOG.setup): create a job and schema without history\nNext deploy enables history and recreates the schema in another catalog, requiring approval."
+{
+    $CLI bundle deploy -qq
+    DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false replace_ids.py
+    cp .databricks/bundle/default/resources.json .databricks/resources.before-migration.json
+    $CLI catalogs create "$other_catalog"
+    update_file.py databricks.yml 'deployment_history: false' 'deployment_history: true'
+    update_file.py databricks.yml 'catalog_name: main' "catalog_name: ${other_catalog}"
+} &> LOG.setup
+trap '$CLI bundle destroy --auto-approve >> LOG.cleanup 2>&1; $CLI catalogs delete "$other_catalog" --force >> LOG.cleanup 2>&1; rm -f "$OUT_REQUESTS"' EXIT
+rm -f "$OUT_REQUESTS"
+
+title "1. Fail deployment planning\nExpect: migration version 2 stays complete; no deployment version is created."
+trace fault.py 'GET /api/2.2/jobs/get*' 403 0 1
+trace musterr $CLI bundle deploy -qq
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "$state_path/resources.deployment.json" -o json | jq -r '.object_id')
+add_repl "$deployment_id" DEPLOYMENT_ID
+title "Request order: complete migration -> upload marker -> read DMS -> plan jobs"
+trace print_requests.py --dms --get //api/2.0/bundle //jobs/get //workspace/import //workspace-files/import-file --oneline |
+    MSYS_NO_PATHCONV=1 jq -c --arg marker "$state_path/resources.json" '
+        (.body.multipart_form.path // .body.path // (.path | ltrimstr("/api/2.0/workspace-files/import-file"))) as $destination |
+        select((.path | endswith("/complete")) or (.path | endswith("/resources")) or .path == "/api/2.2/jobs/get" or $destination == $marker) |
+        {method, path: (if $destination == $marker then $destination else .path end)}'
+title "Remote state: version 2 marker; original resources preserved in the backup"
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "$state_path/resources.json" --format AUTO --file .databricks/resources.marker.json
+trace jq -c '{serial, features, state}' .databricks/resources.marker.json
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "$state_path/resources.json.backup" --format AUTO --file .databricks/resources.backup.json
+trace cmp .databricks/resources.before-migration.json .databricks/resources.backup.json
+title "Migration result: only completed version 2, with INITIAL_REGISTER operations"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-versions "deployments/${deployment_id}" -o json | jq -c 'map({version_id, status, completion_reason})'
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/2" -o json | jq -c 'sort_by(.resource_key) | map({resource_key, action_type, status})'
+
+title "2. Retry without approval\nExpect: deployment declined; migration stays at version 2 and is not repeated."
+trace musterr $CLI bundle deploy -qq
+title "Local and remote markers must stay byte-for-byte unchanged"
+trace cmp .databricks/resources.marker.json .databricks/bundle/default/resources.json
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "$state_path/resources.json" --format AUTO --file .databricks/resources.after-decline.json
+trace cmp .databricks/resources.marker.json .databricks/resources.after-decline.json
+title "No DMS writes and no extra version on the declined retry"
+trace print_requests.py --dms //api/2.0/bundle
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-versions "deployments/${deployment_id}" -o json | jq -c 'map({version_id, status, completion_reason})'
+
+title "3. Approve deployment\nExpect: ordinary deployment version 3; version 2 remains the only migration."
+trace $CLI bundle deploy --auto-approve -qq
+trace $CLI bundle summary -o json | jq -c '.bundle.deployment | {latest_version_id}'
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-versions "deployments/${deployment_id}" -o json | jq -c 'sort_by(.version_id | tonumber) | map({version_id, status, completion_reason})'
diff --git a/acceptance/bundle/dms/migration-before-deploy/test.toml b/acceptance/bundle/dms/migration-before-deploy/test.toml
new file mode 100644
index 00000000000..7123b8a9421
--- /dev/null
+++ b/acceptance/bundle/dms/migration-before-deploy/test.toml
@@ -0,0 +1,2 @@
+# Fault injection exercises a real planner failure against the local testserver.
+Cloud = false
diff --git a/acceptance/bundle/dms/migration/databricks.yml.tmpl b/acceptance/bundle/dms/migration/databricks.yml.tmpl
new file mode 100644
index 00000000000..bb879a94254
--- /dev/null
+++ b/acceptance/bundle/dms/migration/databricks.yml.tmpl
@@ -0,0 +1,16 @@
+bundle:
+  name: dms-migration-$UNIQUE_NAME
+
+experimental:
+  deployment_history: false
+
+resources:
+  jobs:
+    one:
+      name: one
+    two:
+      name: two
+      tasks:
+        - task_key: run_one
+          run_job_task:
+            job_id: ${resources.jobs.one.id}
diff --git a/acceptance/bundle/dms/migration/out.test.toml b/acceptance/bundle/dms/migration/out.test.toml
new file mode 100644
index 00000000000..f68b4a89c36
--- /dev/null
+++ b/acceptance/bundle/dms/migration/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = true
+EnvMatrix.DMS = [""]
diff --git a/acceptance/bundle/dms/migration/output.txt b/acceptance/bundle/dms/migration/output.txt
new file mode 100644
index 00000000000..808dea82789
--- /dev/null
+++ b/acceptance/bundle/dms/migration/output.txt
@@ -0,0 +1,120 @@
+
+=== 1. Create two jobs without history (direct state serial 1)
+Config disables history even though the environment enables it.
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-migration-[UNIQUE_NAME]/default/files...
+Created jobs.one
+Created jobs.two
+Files: 2 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Save the original resources.json for the backup check
+>>> cp .databricks/bundle/default/resources.json .databricks/resources.before-dms.json
+
+=== 2. Enable history and rename job one
+Expect: version 2 registers old state; version 3 applies the new name; no new jobs.
+>>> update_file.py databricks.yml deployment_history: false deployment_history: true
+
+>>> update_file.py databricks.yml name: one name: updated
+
+>>> [CLI] bundle deploy
+Enabling deployment history...
+Deployment history enabled: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=2
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-migration-[UNIQUE_NAME]/default/files...
+Current Deployment Version: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=3
+Updated jobs.one
+Updated jobs.two
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 2 changed, 0 deleted, 0 unchanged
+
+=== 3. Check local state
+Expect: serial=3; history enabled; no resource state or WAL.
+>>> find.py resources[.]json[.]wal --expect 0
+
+>>> jq {serial, features, state} .databricks/bundle/default/resources.json
+{
+  "serial": 3,
+  "features": {
+    "deployment_history": {}
+  },
+  "state": {}
+}
+
+=== 4. Check backup
+Expect: original IDs, names and dependencies; exact pre-migration bytes.
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-migration-[UNIQUE_NAME]/default/state/resources.json.backup --format AUTO --file .databricks/resources.backup.first.json
+
+>>> cat .databricks/resources.backup.first.json
+{
+ "state_version": 3,
+ "cli_version": "[CLI_VERSION]",
+ "lineage": "[UUID]",
+ "serial": 1,
+ "state": {
+  "resources.jobs.one": {
+   "__id__": "[ONE_ID]",
+   "state": {
+    "deployment": {
+     "kind": "BUNDLE",
+     "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-[UNIQUE_NAME]/default/state/metadata.json"
+    },
+    "edit_mode": "UI_LOCKED",
+    "format": "MULTI_TASK",
+    "max_concurrent_runs": 1,
+    "name": "one",
+    "queue": {
+     "enabled": true
+    }
+   }
+  },
+  "resources.jobs.two": {
+   "__id__": "[TWO_ID]",
+   "state": {
+    "deployment": {
+     "kind": "BUNDLE",
+     "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-[UNIQUE_NAME]/default/state/metadata.json"
+    },
+    "edit_mode": "UI_LOCKED",
+    "format": "MULTI_TASK",
+    "max_concurrent_runs": 1,
+    "name": "two",
+    "queue": {
+     "enabled": true
+    },
+    "tasks": [
+     {
+      "run_job_task": {
+       "job_id": [ONE_ID]
+      },
+      "task_key": "run_one"
+     }
+    ]
+   },
+   "depends_on": [
+    {
+     "node": "resources.jobs.one",
+     "label": "${resources.jobs.one.id}"
+    }
+   ]
+  }
+ }
+}
+>>> cmp .databricks/resources.before-dms.json .databricks/resources.backup.first.json
+
+=== 5. Check DMS records
+Expect: completed versions 2 and 3; original job IDs and dependency preserved.
+=== Version history: migration 2 completes before deployment 3
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-versions deployments/[DEPLOYMENT_ID] -o json
+[{"version_id":"2","previous_version_id":null,"version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"},{"version_id":"3","previous_version_id":"2","version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"}]
+
+=== Migration 2: register the original job IDs (INITIAL_REGISTER)
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-operations deployments/[DEPLOYMENT_ID]/versions/2 -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","action_type":"OPERATION_ACTION_TYPE_INITIAL_REGISTER","status":"OPERATION_STATUS_SUCCEEDED"},{"resource_key":"jobs.two","resource_id":"[TWO_ID]","action_type":"OPERATION_ACTION_TYPE_INITIAL_REGISTER","status":"OPERATION_STATUS_SUCCEEDED"}]
+
+=== Deployment 3: record updates separately
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-operations deployments/[DEPLOYMENT_ID]/versions/3 -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","action_type":"OPERATION_ACTION_TYPE_UPDATE","status":"OPERATION_STATUS_SUCCEEDED"},{"resource_key":"jobs.two","resource_id":"[TWO_ID]","action_type":"OPERATION_ACTION_TYPE_UPDATE","status":"OPERATION_STATUS_SUCCEEDED"}]
+
+=== Current resources: new name, original IDs, and job two still depends on job one
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","last_action_type":"OPERATION_ACTION_TYPE_UPDATE","last_version_id":"3","state":{"name":"updated","tasks":[],"depends_on":[]}},{"resource_key":"jobs.two","resource_id":"[TWO_ID]","last_action_type":"OPERATION_ACTION_TYPE_UPDATE","last_version_id":"3","state":{"name":"two","tasks":[{"task_key":"run_one","job_id":[ONE_ID]}],"depends_on":[{"node":"resources.jobs.one","label":"${resources.jobs.one.id}"}]}}]
diff --git a/acceptance/bundle/dms/migration/script b/acceptance/bundle/dms/migration/script
new file mode 100644
index 00000000000..9e1b6010261
--- /dev/null
+++ b/acceptance/bundle/dms/migration/script
@@ -0,0 +1,40 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+title "1. Create two jobs without history (direct state serial 1)\nConfig disables history even though the environment enables it."
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy 2>&1 | contains.py '!Current Deployment Version:' '!Enabling deployment history'
+replace_ids.py
+title "Save the original resources.json for the backup check"
+trace cp .databricks/bundle/default/resources.json .databricks/resources.before-dms.json
+
+title "2. Enable history and rename job one\nExpect: version 2 registers old state; version 3 applies the new name; no new jobs."
+export DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false
+trace update_file.py databricks.yml 'deployment_history: false' 'deployment_history: true'
+trace update_file.py databricks.yml 'name: one' 'name: updated'
+trace $CLI bundle deploy 2>&1 | contains.py 'Enabling deployment history...' 'Deployment history enabled: ' 'Current Deployment Version: ' '?version=2' '?version=3' 'Resources: 0 created'
+
+title "3. Check local state\nExpect: serial=3; history enabled; no resource state or WAL."
+trace find.py 'resources[.]json[.]wal' --expect 0
+trace jq '{serial, features, state}' .databricks/bundle/default/resources.json
+
+title "4. Check backup\nExpect: original IDs, names and dependencies; exact pre-migration bytes."
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-migration-${UNIQUE_NAME}/default/state/resources.json.backup" --format AUTO --file .databricks/resources.backup.first.json
+trace cat .databricks/resources.backup.first.json
+trace cmp .databricks/resources.before-dms.json .databricks/resources.backup.first.json
+
+title "5. Check DMS records\nExpect: completed versions 2 and 3; original job IDs and dependency preserved."
+deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-migration-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+add_repl "$deployment_id" DEPLOYMENT_ID
+title "Version history: migration 2 completes before deployment 3"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-versions "deployments/${deployment_id}" -o json | jq -c 'sort_by(.version_id | tonumber) | map({version_id, previous_version_id, version_type, status, completion_reason})'
+title "Migration 2: register the original job IDs (INITIAL_REGISTER)"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/2" -o json | jq -c 'sort_by(.resource_key) | map({resource_key, resource_id, action_type, status})'
+title "Deployment 3: record updates separately"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/3" -o json | jq -c 'sort_by(.resource_key) | map({resource_key, resource_id, action_type, status})'
+title "Current resources: new name, original IDs, and job two still depends on job one"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${deployment_id}" -o json | jq -c '
+    sort_by(.resource_key) | map({resource_key, resource_id, last_action_type, last_version_id,
+        state: (.state | fromjson | {name: .state.name,
+            tasks: [.state.tasks[]? | {task_key, job_id: .run_job_task.job_id}],
+            depends_on: (.depends_on // [])})})'
+
+rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/migration/test.toml b/acceptance/bundle/dms/migration/test.toml
new file mode 100644
index 00000000000..4da5c0d15a3
--- /dev/null
+++ b/acceptance/bundle/dms/migration/test.toml
@@ -0,0 +1,17 @@
+# This test checks config precedence with conflicting environment settings,
+# so it drives DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY instead of the parent's matrix.
+EnvMatrix.DMS = [""]
+
+# acceptance/bundle/test.toml removes Current Deployment Version for shared DMS/non-DMS goldens.
+# Inherited Repls append and cannot be disabled here.
+# Mask at Order -1 before the parent's default 0, then restore at +1.
+# This lets these tests assert both migration and deployment links.
+[[Repls]]
+Old = '(?m)^Current Deployment Version: '
+New = '__CURRENT_DEPLOYMENT_VERSION__: '
+Order = -1
+
+[[Repls]]
+Old = '(?m)^__CURRENT_DEPLOYMENT_VERSION__: '
+New = 'Current Deployment Version: '
+Order = 1
diff --git a/acceptance/bundle/dms/repeated-migration/databricks.yml.tmpl b/acceptance/bundle/dms/repeated-migration/databricks.yml.tmpl
new file mode 100644
index 00000000000..6c29e76684e
--- /dev/null
+++ b/acceptance/bundle/dms/repeated-migration/databricks.yml.tmpl
@@ -0,0 +1,14 @@
+bundle:
+  name: dms-repeated-migration-$UNIQUE_NAME
+
+experimental:
+  deployment_history: false
+
+sync:
+  exclude:
+    - LOG.*
+
+resources:
+  jobs:
+    one:
+      name: one
diff --git a/acceptance/bundle/dms/repeated-migration/out.test.toml b/acceptance/bundle/dms/repeated-migration/out.test.toml
new file mode 100644
index 00000000000..84b22ba6735
--- /dev/null
+++ b/acceptance/bundle/dms/repeated-migration/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = true
+EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/repeated-migration/output.txt b/acceptance/bundle/dms/repeated-migration/output.txt
new file mode 100644
index 00000000000..420a829d71b
--- /dev/null
+++ b/acceptance/bundle/dms/repeated-migration/output.txt
@@ -0,0 +1,106 @@
+
+=== Setup (LOG.setup)
+Current: job=updated; DMS version=3; old backup=one.
+
+=== 1. Compare the old backup with the latest saved state
+Expect: old backup name=one; latest state name=updated.
+
+>>> gron.py .databricks/resources.backup.first.json
+json.state.resources.jobs.one.state.name = "one";
+
+>>> gron.py .databricks/resources.before-repeat.json
+json.state.resources.jobs.one.state.name = "updated";
+
+=== 2. Restore saved state without its history flag and deploy again
+Expect: migration=4; deployment=5; name=repeated; no new jobs.
+
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace import /Workspace/Users/[USERNAME]/.bundle/dms-repeated-migration-[UNIQUE_NAME]/default/state/resources.json --file .databricks/resources.before-repeat.json --format AUTO --overwrite
+
+>>> cp .databricks/resources.before-repeat.json .databricks/bundle/default/resources.json
+
+>>> update_file.py databricks.yml name: updated name: repeated
+
+>>> [CLI] bundle deploy
+Enabling deployment history...
+Deployment history enabled: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=4
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-repeated-migration-[UNIQUE_NAME]/default/files...
+Current Deployment Version: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=5
+Updated jobs.one
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
+
+=== 3. Check DMS records
+Expect: completed migration 4 and deployment 5; original ID preserved; name=repeated.
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-versions deployments/[DEPLOYMENT_ID] -o json
+[{"version_id":"2","previous_version_id":null,"version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"},{"version_id":"3","previous_version_id":"2","version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"},{"version_id":"4","previous_version_id":"3","version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"},{"version_id":"5","previous_version_id":"4","version_type":"VERSION_TYPE_DEPLOY","status":"VERSION_STATUS_COMPLETED","completion_reason":"VERSION_COMPLETE_SUCCESS"}]
+
+=== Migration 4: register the existing job ID again (INITIAL_REGISTER)
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-operations deployments/[DEPLOYMENT_ID]/versions/4 -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","action_type":"OPERATION_ACTION_TYPE_INITIAL_REGISTER","status":"OPERATION_STATUS_SUCCEEDED"}]
+
+=== Deployment 5: update the job name without creating a replacement
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-operations deployments/[DEPLOYMENT_ID]/versions/5 -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","action_type":"OPERATION_ACTION_TYPE_UPDATE","status":"OPERATION_STATUS_SUCCEEDED"}]
+
+=== Current resource: original job ID, name=repeated, last version=5
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[{"resource_key":"jobs.one","resource_id":"[ONE_ID]","last_version_id":"5","name":"repeated"}]
+
+=== 4. Check backup was overwritten
+Expect: name=updated (before this deploy), not one or repeated.
+
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-repeated-migration-[UNIQUE_NAME]/default/state/resources.json.backup --format AUTO --file .databricks/resources.backup.repeat.json
+
+>>> cat .databricks/resources.backup.repeat.json
+{
+ "state_version": 3,
+ "cli_version": "[CLI_VERSION]",
+ "lineage": "[UUID]",
+ "serial": 3,
+ "state": {
+  "resources.jobs.one": {
+   "__id__": "[ONE_ID]",
+   "state": {
+    "deployment": {
+     "kind": "BUNDLE",
+     "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-repeated-migration-[UNIQUE_NAME]/default/state/metadata.json"
+    },
+    "edit_mode": "UI_LOCKED",
+    "format": "MULTI_TASK",
+    "max_concurrent_runs": 1,
+    "name": "updated",
+    "queue": {
+     "enabled": true
+    }
+   }
+  }
+ }
+}
+
+=== Backup matches the restored snapshot byte for byte
+>>> cmp .databricks/resources.before-repeat.json .databricks/resources.backup.repeat.json
+
+=== Compare backups: serial 1 -> 3 and job name one -> updated
+>>> diff.py .databricks/resources.backup.first.json .databricks/resources.backup.repeat.json
+--- .databricks/resources.backup.first.json
++++ .databricks/resources.backup.repeat.json
+@@ -3,5 +3,5 @@
+  "cli_version": "[CLI_VERSION]",
+  "lineage": "[UUID]",
+- "serial": 1,
++ "serial": 3,
+  "state": {
+   "resources.jobs.one": {
+@@ -15,5 +15,5 @@
+     "format": "MULTI_TASK",
+     "max_concurrent_runs": 1,
+-    "name": "one",
++    "name": "updated",
+     "queue": {
+      "enabled": true
+@@ -22,3 +22,3 @@
+   }
+  }
+-}
++}
diff --git a/acceptance/bundle/dms/repeated-migration/script b/acceptance/bundle/dms/repeated-migration/script
new file mode 100644
index 00000000000..99d363df7f3
--- /dev/null
+++ b/acceptance/bundle/dms/repeated-migration/script
@@ -0,0 +1,48 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+state_path="/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-repeated-migration-${UNIQUE_NAME}/default/state"
+title "Setup (LOG.setup)\nCurrent: job=updated; DMS version=3; old backup=one.\n"
+{
+    trace $CLI bundle deploy
+    trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false replace_ids.py
+    trace update_file.py databricks.yml 'deployment_history: false' 'deployment_history: true'
+    trace update_file.py databricks.yml 'name: one' 'name: updated'
+    trace $CLI bundle deploy
+    deployment_id=$(trace MSYS_NO_PATHCONV=1 $CLI workspace get-status "$state_path/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+    add_repl "$deployment_id" DEPLOYMENT_ID
+    trace MSYS_NO_PATHCONV=1 $CLI workspace export "$state_path/resources.json.backup" --format AUTO --file .databricks/resources.backup.first.json
+    trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true print_state.py > .databricks/resources.before-repeat.json
+} &> LOG.setup
+
+title "1. Compare the old backup with the latest saved state\nExpect: old backup name=one; latest state name=updated.\n"
+trace gron.py .databricks/resources.backup.first.json | grep '\.state\.name = ' | contains.py '"one"'
+trace gron.py .databricks/resources.before-repeat.json | grep '\.state\.name = ' | contains.py '"updated"'
+
+title "2. Restore saved state without its history flag and deploy again\nExpect: migration=4; deployment=5; name=repeated; no new jobs.\n"
+trace MSYS_NO_PATHCONV=1 $CLI workspace import "$state_path/resources.json" --file .databricks/resources.before-repeat.json --format AUTO --overwrite
+trace cp .databricks/resources.before-repeat.json .databricks/bundle/default/resources.json
+trace update_file.py databricks.yml 'name: updated' 'name: repeated'
+trace $CLI bundle deploy 2>&1 | contains.py 'Deployment history enabled: ' 'Current Deployment Version: ' '?version=4' '?version=5' 'Resources: 0 created' '!Created '
+
+title "3. Check DMS records\nExpect: completed migration 4 and deployment 5; original ID preserved; name=repeated.\n"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-versions "deployments/${deployment_id}" -o json |
+    jq -c 'sort_by(.version_id | tonumber) | map({version_id, previous_version_id, version_type, status, completion_reason})'
+title "Migration 4: register the existing job ID again (INITIAL_REGISTER)"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/4" -o json |
+    jq -c 'sort_by(.resource_key) | map({resource_key, resource_id, action_type, status})'
+title "Deployment 5: update the job name without creating a replacement"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/5" -o json |
+    jq -c 'sort_by(.resource_key) | map({resource_key, resource_id, action_type, status})'
+title "Current resource: original job ID, name=repeated, last version=5"
+trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${deployment_id}" -o json |
+    jq -c 'map({resource_key, resource_id, last_version_id, name: (.state | fromjson | .state.name)})'
+
+title "4. Check backup was overwritten\nExpect: name=updated (before this deploy), not one or repeated.\n"
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "$state_path/resources.json.backup" --format AUTO --file .databricks/resources.backup.repeat.json
+trace cat .databricks/resources.backup.repeat.json | contains.py '"name": "updated"' '!"name": "one"' '!"name": "repeated"'
+title "Backup matches the restored snapshot byte for byte"
+trace cmp .databricks/resources.before-repeat.json .databricks/resources.backup.repeat.json
+title "Compare backups: serial 1 -> 3 and job name one -> updated"
+trace diff.py .databricks/resources.backup.first.json .databricks/resources.backup.repeat.json
+
+rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/repeated-migration/test.toml b/acceptance/bundle/dms/repeated-migration/test.toml
new file mode 100644
index 00000000000..8182a51630b
--- /dev/null
+++ b/acceptance/bundle/dms/repeated-migration/test.toml
@@ -0,0 +1,13 @@
+# acceptance/bundle/test.toml removes Current Deployment Version for shared DMS/non-DMS goldens.
+# Inherited Repls append and cannot be disabled here.
+# Mask at Order -1 before the parent's default 0, then restore at +1.
+# This lets these tests assert both migration and deployment links.
+[[Repls]]
+Old = '(?m)^Current Deployment Version: '
+New = '__CURRENT_DEPLOYMENT_VERSION__: '
+Order = -1
+
+[[Repls]]
+Old = '(?m)^__CURRENT_DEPLOYMENT_VERSION__: '
+New = 'Current Deployment Version: '
+Order = 1
diff --git a/acceptance/bundle/migrate/auto/dms-migration/output.txt b/acceptance/bundle/migrate/auto/dms-migration/output.txt
index ba6f36cc8bf..58cb5d081d3 100644
--- a/acceptance/bundle/migrate/auto/dms-migration/output.txt
+++ b/acceptance/bundle/migrate/auto/dms-migration/output.txt
@@ -2,19 +2,42 @@
 === Replay a prior terraform deployment (captured from a real terraform run)
 >>> replay_tfstate.py terraform-requests.json terraform.tfstate
 
-=== Direct + deployment-history requested against a terraform state is rejected before commit
->>> DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true errcode [CLI] bundle deploy
-Notice: migrating your bundle to direct deployment engine (https://github.com/databricks/cli/issues/6765).
-Error: deployment history setting (true) does not match the existing state (false)
+=== Deployment history rejects Terraform state before conversion, for both plan and deploy
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true musterr [CLI] bundle plan
+Error: deployment history is not supported with Terraform state; first migrate to direct deployment with deployment history disabled
+
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true musterr [CLI] bundle deploy
+Error: deployment history is not supported with Terraform state; first migrate to direct deployment with deployment history disabled
+
+>>> find.py resources[.]json|tfstate[.]backup --expect 0
 
-Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over
+>>> print_requests.py --dms //api/2.0/bundle //jobs
 
+=== First migrate Terraform state to direct with deployment history disabled
+>>> DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false [CLI] bundle deploy
+Notice: migrating your bundle to direct deployment engine (https://github.com/databricks/cli/issues/6765).
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Migrated 1 resource to direct deployment engine.
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
 
-Exit code: 1
+=== Then enable deployment history on the direct state, preserving the job ID
+>>> DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy
+Enabling deployment history...
+Deployment history enabled: [DATABRICKS_URL]/deployments/[DEPLOYMENT_ID]?version=4
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Updated jobs.test_job
+Files: 0 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
 
 >>> find.py \.databricks/.*(\.tfstate|resources.*\.json|\.wal)
-.databricks/bundle/default/terraform/terraform.tfstate
+.databricks/bundle/default/resources.json
+.databricks/bundle/default/terraform/terraform.tfstate.backup
+
+>>> find.py resources[.]json[.]wal --expect 0
+
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle summary -o json
+{"job_id":"[TEST_JOB_ID]","version":5}
 
->>> MSYS_NO_PATHCONV=1 [CLI] workspace list /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/state
-ID                   Type  Language  Path
-[OBJECT_ID]  FILE            /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/state/terraform.tfstate
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/state/resources.json --format AUTO
+{"features":{"deployment_history":{}},"state":{}}
diff --git a/acceptance/bundle/migrate/auto/dms-migration/script b/acceptance/bundle/migrate/auto/dms-migration/script
index 7ef1ca09f94..a5d1a0c53f9 100644
--- a/acceptance/bundle/migrate/auto/dms-migration/script
+++ b/acceptance/bundle/migrate/auto/dms-migration/script
@@ -2,13 +2,22 @@ export DATABRICKS_BUNDLE_ENGINE=
 
 title "Replay a prior terraform deployment (captured from a real terraform run)"
 trace replay_tfstate.py terraform-requests.json terraform.tfstate
+original_job_id=$(read_id.py test_job)
 rm -f "$OUT_REQUESTS"
 
-title "Direct + deployment-history requested against a terraform state is rejected before commit"
-# Migration produces a plain direct state (no deployment_history feature), so requesting
-# deployment history while migrating is a mismatch. It is rejected after the in-memory migration
-# but before CommitMigration writes or pushes anything, so the bundle stays on terraform with no
-# leaked direct state - migrate first (plain deploy), then enable deployment history.
-trace DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true errcode $CLI bundle deploy
+title "Deployment history rejects Terraform state before conversion, for both plan and deploy"
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true musterr $CLI bundle plan
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true musterr $CLI bundle deploy
+trace find.py 'resources[.]json|tfstate[.]backup' --expect 0
+trace print_requests.py --dms //api/2.0/bundle //jobs
+
+title "First migrate Terraform state to direct with deployment history disabled"
+trace DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false $CLI bundle deploy
+
+title "Then enable deployment history on the direct state, preserving the job ID"
+trace DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy
 trace find.py '\.databricks/.*(\.tfstate|resources.*\.json|\.wal)'
-trace MSYS_NO_PATHCONV=1 $CLI workspace list "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/test-bundle/default/state"
+trace find.py 'resources[.]json[.]wal' --expect 0
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle summary -o json |
+    jq -ce --arg id "$original_job_id" '{job_id: .resources.jobs.test_job.id, version: .bundle.deployment.latest_version_id} | select(.job_id == $id)'
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/test-bundle/default/state/resources.json" --format AUTO | jq -c '{features, state}'
diff --git a/bundle/config/experimental.go b/bundle/config/experimental.go
index 6fd5511b79b..498e1b15d47 100644
--- a/bundle/config/experimental.go
+++ b/bundle/config/experimental.go
@@ -54,9 +54,9 @@ type Experimental struct {
 	// service (DMS), which records deployment history and tracks what changed
 	// across deployments.
 	//
-	// Only for bundles with no deployed resources yet: DMS becomes the source of
-	// truth for their state.
-	// An explicit value takes priority over the environment setting.
+	// Existing resources are registered when deployment history is enabled, and
+	// DMS becomes the source of truth for their state. An explicit value takes
+	// priority over the environment setting.
 	DeploymentHistory *bool `json:"deployment_history,omitempty"`
 }
 
diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go
index 57e10b8d4b4..00f5ac29253 100644
--- a/bundle/direct/dstate/dms.go
+++ b/bundle/direct/dstate/dms.go
@@ -16,6 +16,21 @@ type RecordedState struct {
 	DependsOn []deployplan.DependsOnEntry `json:"depends_on,omitempty"`
 }
 
+// MarkDeploymentHistory persists the DMS marker after all existing resources have been recorded.
+func (db *DeploymentState) MarkDeploymentHistory(versionID int) error {
+	db.AssertOpenedForRead()
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	db.recordsHistory = true
+	db.VersionID = versionID
+	if db.Data.Features == nil {
+		db.Data.Features = make(map[string]struct{})
+	}
+	db.Data.Features[FeatureDeploymentHistory] = struct{}{}
+	return db.unlockedSave()
+}
+
 // applyDMSState fills in the resource state from what DMS recorded. For a deployment that uses DMS
 // resources.json does not track state. The service is queried for state for resources and then they
 // are filled in-place in DeploymentState.Data.State:
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index ac8f8226c4b..009f2e2dd84 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -885,6 +885,19 @@ func (db *DeploymentState) Persist() error {
 	return db.unlockedSave()
 }
 
+// Close discards read-only in-memory state so the next Open rereads its backend.
+// It is safe to call on a closed state. Writable state must be finalized instead.
+func (db *DeploymentState) Close() {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	if db.Path == "" {
+		return
+	}
+	db.AssertOpenedForRead()
+	db.reset()
+}
+
 // IsOpen reports whether the state has been opened (for read or write). It lets
 // callers probe the state without risking the panic in AssertOpenedForReadOrWrite,
 // e.g. code paths shared with the terraform engine where the state DB is never opened.
diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index 72ec263f7b7..11f825fbfe7 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -49,6 +49,44 @@ func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) {
 	assert.ErrorIs(t, err, os.ErrNotExist)
 }
 
+func TestCloseReadState(t *testing.T) {
+	// 1. Disk has serial 1; change only the in-memory copy to serial 99.
+	path := filepath.Join(t.TempDir(), "resources.json")
+	var db DeploymentState
+	db.OpenWithData(path, NewDatabase("lineage", 1))
+	require.NoError(t, db.Persist())
+	persistedBeforeClose, err := os.ReadFile(path)
+	require.NoError(t, err)
+	db.Data.Serial = 99
+
+	// 2. Closing twice is safe and leaves the database closed.
+	db.Close()
+	db.Close()
+	assert.False(t, db.IsOpen(), "Close must leave the database closed")
+
+	// 3. Close does not write the in-memory changes to disk.
+	persistedAfterClose, err := os.ReadFile(path)
+	require.NoError(t, err)
+	assert.Equal(t, persistedBeforeClose, persistedAfterClose, "Close must not change resources.json")
+
+	// 4. Reopening reads serial 1 from disk, not the discarded serial 99.
+	require.NoError(t, db.Open(t.Context(), path, WithRecovery(false), WithWrite(false), WithDeploymentHistory(false), OpenDmsArgs{}))
+	assert.Equal(t, 1, db.GetSerial(), "Open must reload the persisted serial")
+	db.Close()
+}
+
+func TestCloseRefusesWritableState(t *testing.T) {
+	// 1. Enter write mode. Writable state must use Finalize, not Close.
+	var db DeploymentState
+	db.OpenWithData(filepath.Join(t.TempDir(), "resources.json"), NewDatabase("lineage", 1))
+	require.NoError(t, db.UpgradeToWrite())
+	defer mustFinalize(t, &db)
+
+	// 2. Reject Close without closing the database, so Finalize can still finish.
+	assert.PanicsWithValue(t, "internal error: DeploymentState must be opened in read mode", db.Close)
+	assert.True(t, db.IsOpen(), "A rejected Close must leave writable state open for Finalize")
+}
+
 func TestDataForFileSerial(t *testing.T) {
 	tests := []struct {
 		name       string
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index 13c5121f569..9de3596aca7 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -321,12 +321,12 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 	// A first deploy's id did not exist at plan time - the version and any existing id were stamped
 	// then - so stamp the one just created into the plan the apply reads.
 	if b.DeploymentBundle.StateDB.IsDeploymentMetadataService() {
-		firstDeploy := b.DeploymentBundle.StateDB.DeploymentID == ""
+		injectDeploymentIdInPlan := b.DeploymentBundle.StateDB.DeploymentID == ""
 		createOrUpdateDeployment(ctx, b, dmsDeployment)
 		if logdiag.HasError(ctx) {
 			return
 		}
-		if firstDeploy {
+		if injectDeploymentIdInPlan {
 			if err := b.DeploymentBundle.StampDeploymentIdForFirstVersion(b.DeploymentBundle.StateDB.DeploymentID); err != nil {
 				logdiag.LogError(ctx, err)
 				return
@@ -345,7 +345,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 				logdiag.LogError(ctx, err)
 				return
 			}
-			logDeploymentVersion(ctx, b)
+			logDeploymentVersion(ctx, b, "Current Deployment Version")
 		}
 	}
 
diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go
index abd89f03f2f..095e4a149b0 100644
--- a/bundle/phases/dms.go
+++ b/bundle/phases/dms.go
@@ -165,7 +165,7 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version
 
 // logDeploymentVersion logs the deployment version URL. Workspace ID is omitted
 // so the page stays clickable in a terminal and redirects correctly without it.
-func logDeploymentVersion(ctx context.Context, b *bundle.Bundle) {
+func logDeploymentVersion(ctx context.Context, b *bundle.Bundle, message string) {
 	deploymentID := b.DeploymentBundle.StateDB.DeploymentID
 	version := b.DeploymentBundle.StateDB.VersionID
 	if version == 0 {
@@ -177,11 +177,11 @@ func logDeploymentVersion(ctx context.Context, b *bundle.Bundle) {
 		// Only the link is lost, so report the version without it rather than failing
 		// a deploy over it.
 		log.Debugf(ctx, "Not linking to the recorded deployment: %s", err)
-		cmdio.LogString(ctx, fmt.Sprintf("Current Deployment Version: %s version %d", deploymentID, version))
+		cmdio.LogString(ctx, fmt.Sprintf("%s: %s version %d", message, deploymentID, version))
 		return
 	}
 
-	cmdio.LogString(ctx, "Current Deployment Version: "+workspaceurls.DeploymentURL(*baseURL, deploymentID, version))
+	cmdio.LogString(ctx, message+": "+workspaceurls.DeploymentURL(*baseURL, deploymentID, version))
 }
 
 // deploymentMetadata describes the bundle this deploy came from and where it
diff --git a/bundle/phases/dms_migration.go b/bundle/phases/dms_migration.go
new file mode 100644
index 00000000000..15bee44d7d5
--- /dev/null
+++ b/bundle/phases/dms_migration.go
@@ -0,0 +1,140 @@
+package phases
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/lock"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/internal/build"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/dms"
+	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
+)
+
+// ErrDMSRequiresDirectState rejects enabling history before Terraform state is migrated to direct.
+var ErrDMSRequiresDirectState = errors.New("deployment history is not supported with Terraform state; first migrate to direct deployment with deployment history disabled")
+
+// MigrateToDMS commits the already-open direct state to DMS and updates its descriptor before deployment planning.
+func MigrateToDMS(ctx context.Context, b *bundle.Bundle, state *statemgmt.StateDesc) {
+	if !state.Engine.IsDirect() {
+		logdiag.LogError(ctx, ErrDMSRequiresDirectState)
+		return
+	}
+	if state.IsDMS() {
+		return
+	}
+
+	bundle.ApplyContext(ctx, b, lock.Acquire(lock.GoalDeploy))
+	if logdiag.HasError(ctx) {
+		return
+	}
+	defer bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDeploy))
+
+	db := &b.DeploymentBundle.StateDB
+	db.AssertOpenedForRead()
+	defer db.Close()
+
+	deploymentID, deployment, err := dms.ReadDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	lastVersion, err := dms.ParseLastVersionID(deployment)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	if err := statemgmt.BackupRemoteResourcesState(ctx, b); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	db.DeploymentID = deploymentID
+	db.VersionID = lastVersion
+	createOrUpdateDeployment(ctx, b, deployment)
+	if logdiag.HasError(ctx) {
+		return
+	}
+	ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx))
+	cmdio.LogString(ctx, "Enabling deployment history...")
+	if err := registerMigrationState(ctx, db); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	logDeploymentVersion(ctx, b, "Deployment history enabled")
+	statemgmt.PushResourcesState(ctx, b)
+	// Remember the migration version and that state now lives in DMS.
+	// The deployment code uses this flag to accept the history setting and read resources from DMS.
+	state.Serial = db.GetSerial()
+	state.Features = db.StateFeatures()
+}
+
+// registerMigrationState records every resource from the open direct state in an initial registration version.
+// The caller must hold the deployment lock and have created the DMS deployment.
+func registerMigrationState(ctx context.Context, db *dstate.DeploymentState) error {
+	keys := slices.Sorted(maps.Keys(db.Data.State))
+	states := make([]json.RawMessage, len(keys))
+	operations := make([]bundledeployments.StagedOperation, len(keys))
+	for i, key := range keys {
+		entry := db.Data.State[key]
+		state, err := json.Marshal(dstate.RecordedState{State: entry.State, DependsOn: entry.DependsOn})
+		if err != nil {
+			return fmt.Errorf("serializing migration state for %s: %w", key, err)
+		}
+		states[i] = state
+		operations[i] = bundledeployments.StagedOperation{
+			ResourceKey: strings.TrimPrefix(key, dms.StatePrefix),
+			ActionType:  bundledeployments.OperationActionTypeOperationActionTypeInitialRegister,
+		}
+	}
+
+	// Migration uses max(state serial, last version)+1; deployment follows at the next version.
+	versionID := max(db.Data.Serial, db.VersionID) + 1
+	previousVersionID := ""
+	if db.VersionID > 0 {
+		previousVersionID = strconv.Itoa(db.VersionID)
+	}
+	w := cmdctx.WorkspaceClient(ctx)
+	_, err := w.BundleDeployments.CreateVersion(ctx, bundledeployments.CreateVersionRequest{
+		Parent:    dms.DeploymentName(db.DeploymentID),
+		VersionId: strconv.Itoa(versionID),
+		Version: bundledeployments.Version{
+			CliVersion:        build.GetInfo().Version,
+			VersionType:       dms.VersionTypeDeploy,
+			PreviousVersionId: previousVersionID,
+			Operations:        operations,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("creating migration version: %w", err)
+	}
+
+	buffer := dms.StartOperationBuffer(ctx, db.DeploymentID, versionID)
+	for i, key := range keys {
+		buffer.RecordOperation(ctx, key, false, db.Data.State[key].ID, states[i])
+	}
+	recordErr := buffer.Drain()
+	reason := bundledeployments.VersionCompleteVersionCompleteSuccess
+	if recordErr != nil {
+		reason = bundledeployments.VersionCompleteVersionCompleteFailure
+	}
+	_, completeErr := w.BundleDeployments.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{
+		Name:             dms.VersionName(db.DeploymentID, versionID),
+		CompletionReason: reason,
+	})
+	if err := errors.Join(recordErr, completeErr); err != nil {
+		return fmt.Errorf("recording migration version: %w", err)
+	}
+
+	return db.MarkDeploymentHistory(versionID)
+}
diff --git a/bundle/phases/dms_migration_test.go b/bundle/phases/dms_migration_test.go
new file mode 100644
index 00000000000..59956a3797c
--- /dev/null
+++ b/bundle/phases/dms_migration_test.go
@@ -0,0 +1,96 @@
+package phases
+
+import (
+	"context"
+	"path/filepath"
+	"strconv"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/databricks-sdk-go/experimental/mocks"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigrateToDMSRejectsTerraformBeforeLock(t *testing.T) {
+	// 1. Try migration with Terraform state and locking enabled by default.
+	b := &bundle.Bundle{}
+	ctx := logdiag.InitContext(t.Context())
+	logdiag.SetCollect(ctx, true)
+
+	MigrateToDMS(ctx, b, &statemgmt.StateDesc{Engine: engine.EngineTerraform})
+
+	// 2. Reject the state before creating a deployment lock.
+	require.True(t, logdiag.HasError(ctx))
+	assert.Nil(t, b.Locker, "Terraform state must be rejected before locking")
+}
+
+func TestMigrateToDMSSkipsAlreadyRecordedBeforeLock(t *testing.T) {
+	// 1. State already uses DMS; no open database or workspace client is needed.
+	b := &bundle.Bundle{}
+	ctx := logdiag.InitContext(t.Context())
+	logdiag.SetCollect(ctx, true)
+	state := &statemgmt.StateDesc{
+		Features: map[string]struct{}{dstate.FeatureDeploymentHistory: {}},
+	}
+
+	MigrateToDMS(ctx, b, state)
+
+	// 2. Return successfully without acquiring a migration lock.
+	require.False(t, logdiag.HasError(ctx), "%v", logdiag.FlushCollected(ctx))
+	assert.Nil(t, b.Locker, "Already-recorded state must not acquire a migration lock")
+}
+
+func TestRegisterMigrationStateVersionNumbers(t *testing.T) {
+	// Acceptance tests cover registration and persistence. These cases cover unequal counters.
+	tests := []struct {
+		name           string
+		stateSerial    int
+		serviceVersion int
+		wantVersion    int
+	}{
+		{
+			name:           "service version is newer",
+			stateSerial:    2,
+			serviceVersion: 5,
+			wantVersion:    6,
+		},
+		{
+			name:           "state serial is newer",
+			stateSerial:    8,
+			serviceVersion: 3,
+			wantVersion:    9,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// 1. Open state with deliberately different file and service counters.
+			statePath := filepath.Join(t.TempDir(), "resources.json")
+			var db dstate.DeploymentState
+			db.OpenWithData(statePath, dstate.NewDatabase("lineage", tt.stateSerial))
+			db.DeploymentID = "123"
+			db.VersionID = tt.serviceVersion
+
+			// 2. Registration uses max(file, service)+1, but its predecessor is the service version.
+			m := mocks.NewMockWorkspaceClient(t)
+			api := m.GetMockBundleDeploymentsAPI()
+			api.EXPECT().CreateVersion(mock.Anything, mock.Anything).
+				Run(func(_ context.Context, request bundledeployments.CreateVersionRequest) {
+					assert.Equal(t, strconv.Itoa(tt.wantVersion), request.VersionId, "Migration must follow the larger counter")
+					assert.Equal(t, strconv.Itoa(tt.serviceVersion), request.Version.PreviousVersionId, "Predecessor must be the actual DMS version")
+				}).Return(&bundledeployments.Version{}, nil).Once()
+			api.EXPECT().CompleteVersion(mock.Anything, mock.Anything).Return(&bundledeployments.Version{}, nil).Once()
+			ctx := cmdctx.SetWorkspaceClient(t.Context(), m.WorkspaceClient)
+
+			// 3. Run registration; the mock checks the version numbers above.
+			require.NoError(t, registerMigrationState(ctx, &db))
+		})
+	}
+}
diff --git a/bundle/statemgmt/state_push.go b/bundle/statemgmt/state_push.go
index 03036ba712d..9df720f6157 100644
--- a/bundle/statemgmt/state_push.go
+++ b/bundle/statemgmt/state_push.go
@@ -3,6 +3,7 @@ package statemgmt
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io/fs"
 	"os"
 
@@ -43,6 +44,27 @@ func PushResourcesState(ctx context.Context, b *bundle.Bundle) {
 	}
 }
 
+// BackupRemoteResourcesState preserves the remote state before deployment history replaces it.
+func BackupRemoteResourcesState(ctx context.Context, b *bundle.Bundle) error {
+	f, err := deploy.StateFiler(ctx, b)
+	if err != nil {
+		return fmt.Errorf("backing up resource state: creating state filer: %w", err)
+	}
+
+	remotePath, _ := b.StateFilenameDirect(ctx)
+	reader, err := f.Read(ctx, remotePath)
+	if err != nil {
+		return fmt.Errorf("backing up resource state: reading %s: %w", remotePath, err)
+	}
+	defer reader.Close()
+
+	backupPath := remotePath + ".backup"
+	if err := f.Write(ctx, backupPath, reader, filer.OverwriteIfExists); err != nil {
+		return fmt.Errorf("backing up resource state: writing %s: %w", backupPath, err)
+	}
+	return nil
+}
+
 func BackupRemoteTerraformState(ctx context.Context, b *bundle.Bundle) {
 	f, err := deploy.StateFiler(ctx, b)
 	if err != nil {
diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go
index 6ccb3367db7..d95d775a7aa 100644
--- a/cmd/bundle/utils/process.go
+++ b/cmd/bundle/utils/process.go
@@ -4,9 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"path"
 	"path/filepath"
-	"strconv"
 	"time"
 
 	"github.com/databricks/cli/bundle"
@@ -34,8 +32,6 @@ import (
 	"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/useragent"
 	"github.com/spf13/cobra"
@@ -254,6 +250,11 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 		// Record the engine the resolved state uses now, so deploy telemetry reports
 		// it even when the deploy fails or is cancelled before deployCore runs.
 		b.Metrics.StateEngine = stateDesc.Engine.ThisOrDefault()
+
+		// TODO: Support the full Terraform -> direct -> DMS migration path at GA.
+		if (opts.Deploy || opts.PreDeployChecks) && !stateDesc.Engine.IsDirect() && configuresDeploymentHistory(ctx, b) {
+			return b, stateDesc, phases.ErrDMSRequiresDirectState
+		}
 	}
 
 	// --plan applies a precomputed plan, so it skips Build and PreDeployChecks; a plain
@@ -349,12 +350,17 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 			cmd.SetContext(ctx)
 		}
 
+		migrateToDMS := opts.Deploy && stateDesc.Engine.IsDirect() && !stateDesc.IsDMS() && configuresDeploymentHistory(ctx, b)
+		// TODO: Support applying plans computed against direct state after migration to DMS.
+		// This restriction blocks automatic migration with --plan once deployment history reaches GA.
+		if migrateToDMS && opts.ReadPlanPath != "" {
+			return b, stateDesc, errors.New("migrating to deployment history does not support --plan; run 'bundle deploy' without --plan")
+		}
 		// Open direct engine state once for all subsequent operations (ExportState, CalculatePlan, Apply, etc.)
 		// A migrated-from-Terraform state is already open (seeded in memory above), so skip the disk open.
 		needDirectState := stateDesc.Engine.IsDirect() && needsState
-		var localPath string
 		if needDirectState && !b.DeploymentBundle.StateDB.IsOpen() {
-			_, localPath = b.StateFilenameDirect(ctx)
+			_, localPath := b.StateFilenameDirect(ctx)
 			if !stateDesc.IsDMS() {
 				if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dstate.WithDeploymentHistory(false), dstate.OpenDmsArgs{}); err != nil {
 					logdiag.LogError(ctx, err)
@@ -362,7 +368,12 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 				}
 			}
 		}
-
+		if migrateToDMS {
+			phases.MigrateToDMS(ctx, b, stateDesc)
+			if logdiag.HasError(ctx) {
+				return b, stateDesc, root.ErrAlreadyPrinted
+			}
+		}
 		if stateDesc.Engine.IsDirect() && !opts.SkipEnforcingDeploymentHistorySetting {
 			if err := enforceDeploymentHistorySetting(ctx, b, stateDesc, opts.Deploy || opts.PreDeployChecks); err != nil {
 				logdiag.LogError(ctx, err)
@@ -372,22 +383,32 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 
 		if needDirectState && stateDesc.IsDMS() {
 			var err error
-			dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+			dmsDeploymentID, dmsDeployment, err = dms.ReadDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
 			if err != nil {
 				logdiag.LogError(ctx, err)
 				return b, stateDesc, root.ErrAlreadyPrinted
 			}
 
-			// Stamp the deployment and the version this run records onto every job and pipeline so
-			// 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
 			}
-			nextVersion := lastVersionID + 1
+			// StateDB.Open builds the DMS client from the workspace client on the context.
+			if !cmdctx.HasWorkspaceClient(ctx) {
+				ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx))
+			}
+			_, localPath := b.StateFilenameDirect(ctx)
+			if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(false), dstate.WithWrite(false), dstate.WithDeploymentHistory(true), dstate.OpenDmsArgs{DeploymentID: dmsDeploymentID, LastVersionID: lastVersionID}); err != nil {
+				logdiag.LogError(ctx, err)
+				return b, stateDesc, root.ErrAlreadyPrinted
+			}
+			nextVersion := b.DeploymentBundle.StateDB.GetSerial() + 1
+			// Stamp jobs and pipelines for the plan:
+			// 1. Existing: deployment ID is known; version = last recorded + 1.
+			// 2. New: AnnotateDeploymentVersion stamps version 1 now; the deploy phase
+			//    stamps the deployment ID after creating the deployment.
+			// 3. Migrated: deployment ID is known; version = migration version + 1.
 			muts := []bundle.Mutator{metadata.AnnotateDeploymentVersion(nextVersion)}
 			if dmsDeploymentID != "" {
 				bundle.ApplyFuncContext(ctx, b, func(_ context.Context, b *bundle.Bundle) {
@@ -400,14 +421,6 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 			if logdiag.HasError(ctx) {
 				return b, stateDesc, root.ErrAlreadyPrinted
 			}
-			// StateDB.Open builds the DMS client from the workspace client on the context.
-			if !cmdctx.HasWorkspaceClient(ctx) {
-				ctx = cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx))
-			}
-			if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(false), dstate.WithWrite(false), dstate.WithDeploymentHistory(true), dstate.OpenDmsArgs{DeploymentID: dmsDeploymentID, LastVersionID: lastVersionID}); err != nil {
-				logdiag.LogError(ctx, err)
-				return b, stateDesc, root.ErrAlreadyPrinted
-			}
 		}
 
 		if needDirectState {
@@ -584,49 +597,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 := 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 := 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.
@@ -639,11 +609,11 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st
 		return enforceDeploymentHistorySetting(ctx, b, stateDesc, false)
 	}
 
-	dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+	dmsDeploymentID, dmsDeployment, err := dms.ReadDeployment(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
 	}
diff --git a/cmd/bundle/utils/process_test.go b/cmd/bundle/utils/process_test.go
index b216f1102d2..e52fc4ab6f4 100644
--- a/cmd/bundle/utils/process_test.go
+++ b/cmd/bundle/utils/process_test.go
@@ -7,6 +7,7 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/direct/dstate"
 	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/libs/dms"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -81,7 +82,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..c91ac4d5be5 100644
--- a/libs/dms/metadata.go
+++ b/libs/dms/metadata.go
@@ -1,11 +1,16 @@
 package dms
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"path"
 	"strconv"
 	"strings"
 
 	"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"
 )
 
@@ -85,6 +90,49 @@ func NextVersion(lastVersionID string) (int, error) {
 // Keep it equal to the service's DEPLOYMENT_NODE_NAME.
 const DeploymentNodeName = "resources.deployment.json"
 
+// ReadDeployment looks up the deployment node and its service record.
+//
+// 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. Recover here with 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 ReadDeployment(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) {
+	nodePath := path.Join(statePath, DeploymentNodeName)
+
+	obj, err := 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 := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{
+		Name: 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
+}
+
 // DeploymentName, VersionName and OperationName build the resource names the
 // service uses, so callers pass only ids.
 func DeploymentName(deploymentID string) string {
diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go
index 86aef24d83c..bc52c392cf7 100644
--- a/libs/testserver/bundledeployments.go
+++ b/libs/testserver/bundledeployments.go
@@ -2,6 +2,7 @@ package testserver
 
 import (
 	"bytes"
+	"cmp"
 	"encoding/json"
 	"fmt"
 	"path"
@@ -576,6 +577,27 @@ func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, re
 	return Response{Body: body}
 }
 
+// ListVersions returns recorded versions in descending numeric order.
+func (s *FakeWorkspace) ListVersions(deploymentID string) Response {
+	defer s.LockUnlock()()
+
+	d, ok := s.DmsDeployments[deploymentID]
+	if !ok {
+		return dmsNotFound("deployment " + deploymentID)
+	}
+
+	versions := make([]bundledeployments.Version, 0, len(d.Versions))
+	for _, version := range d.Versions {
+		versions = append(versions, *version)
+	}
+	slices.SortFunc(versions, func(a, b bundledeployments.Version) int {
+		aID, _ := strconv.ParseInt(a.VersionId, 10, 64)
+		bID, _ := strconv.ParseInt(b.VersionId, 10, 64)
+		return cmp.Compare(bID, aID)
+	})
+	return Response{Body: bundledeployments.ListVersionsResponse{Versions: versions}}
+}
+
 func (s *FakeWorkspace) ListResources(deploymentID string) Response {
 	defer s.LockUnlock()()
 
diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go
index f2d9cff5109..c8d4ef74655 100644
--- a/libs/testserver/handlers.go
+++ b/libs/testserver/handlers.go
@@ -468,6 +468,9 @@ func AddDefaultHandlers(server *Server) {
 	server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any {
 		return req.Workspace.CreateVersion(req, req.Vars["deployment_id"])
 	})
+	server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any {
+		return req.Workspace.ListVersions(req.Vars["deployment_id"])
+	})
 	server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/complete", func(req Request) any {
 		return req.Workspace.CompleteVersion(req, req.Vars["deployment_id"], req.Vars["version_id"])
 	})