diff --git a/acceptance/bundle/dms/declined-deploy/output.txt b/acceptance/bundle/dms/declined-deploy/output.txt
index 870af8c885..83f2eb3dc4 100644
--- a/acceptance/bundle/dms/declined-deploy/output.txt
+++ b/acceptance/bundle/dms/declined-deploy/output.txt
@@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "main.dms_declined_deploy_schema_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_declined_deploy_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -174,14 +174,14 @@ false
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DESTROY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_DELETE"
+        "action_type": "OPERATION_ACTION_TYPE_DELETE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DESTROY"
   }
 }
 {
@@ -192,7 +192,7 @@ false
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
 }
diff --git a/acceptance/bundle/dms/depends-on/output.txt b/acceptance/bundle/dms/depends-on/output.txt
index 4237ad47e7..15f4e6ae5d 100644
--- a/acceptance/bundle/dms/depends-on/output.txt
+++ b/acceptance/bundle/dms/depends-on/output.txt
@@ -33,7 +33,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "[NUMID]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"parent\",\"queue\":{\"enabled\":true}}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -47,7 +47,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "[NUMID]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"description\":\"depends on [NUMID]\",\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"child\",\"queue\":{\"enabled\":true}},\"depends_on\":[{\"node\":\"resources.jobs.parent\",\"label\":\"${resources.jobs.parent.id}\"}]}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
diff --git a/acceptance/bundle/dms/deployment-metadata-change/output.txt b/acceptance/bundle/dms/deployment-metadata-change/output.txt
index 29a3fd6366..eabfc61f07 100644
--- a/acceptance/bundle/dms/deployment-metadata-change/output.txt
+++ b/acceptance/bundle/dms/deployment-metadata-change/output.txt
@@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "main.dms_metadata_schema_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_metadata_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
diff --git a/acceptance/bundle/dms/empty-grants/output.txt b/acceptance/bundle/dms/empty-grants/output.txt
index 81164aaa23..1713a8e83d 100644
--- a/acceptance/bundle/dms/empty-grants/output.txt
+++ b/acceptance/bundle/dms/empty-grants/output.txt
@@ -29,17 +29,17 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo"
       },
       {
-        "resource_key": "schemas.foo.grants",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo.grants"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -51,7 +51,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "main.dms_empty_grants_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_empty_grants_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -65,7 +65,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "schema/main.dms_empty_grants_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"securable_type\":\"schema\",\"full_name\":\"main.dms_empty_grants_[UNIQUE_NAME]\",\"__embed__\":[{\"principal\":\"deco-test-user@databricks.com\",\"privileges\":[\"USE_SCHEMA\"]}]},\"depends_on\":[{\"node\":\"resources.schemas.foo\",\"label\":\"${resources.schemas.foo.id}\"}]}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -96,14 +96,14 @@ Resources: 0 created, 1 changed, 0 deleted, 1 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "schemas.foo.grants",
-        "action_type": "OPERATION_ACTION_TYPE_UPDATE"
+        "action_type": "OPERATION_ACTION_TYPE_UPDATE",
+        "resource_key": "schemas.foo.grants"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -114,7 +114,7 @@ Resources: 0 created, 1 changed, 0 deleted, 1 unchanged
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
 }
diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt
index 5fe12f71a8..0ab3b90fbd 100644
--- a/acceptance/bundle/dms/existing-state/output.txt
+++ b/acceptance/bundle/dms/existing-state/output.txt
@@ -69,8 +69,8 @@ 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/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"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/[NUMID]/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/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"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/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
 
 >>> find.py resources[.]json[.]wal --expect 0
diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt
index 6d58016385..008a503468 100644
--- a/acceptance/bundle/dms/failed-delete/output.txt
+++ b/acceptance/bundle/dms/failed-delete/output.txt
@@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "jobs.stuck",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "jobs.stuck"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "[NUMID]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"stuck\",\"queue\":{\"enabled\":true}}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -122,14 +122,14 @@ API message: Fault injected by test.
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DESTROY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "jobs.stuck",
-        "action_type": "OPERATION_ACTION_TYPE_DELETE"
+        "action_type": "OPERATION_ACTION_TYPE_DELETE",
+        "resource_key": "jobs.stuck"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DESTROY"
   }
 }
 {
@@ -141,7 +141,7 @@ API message: Fault injected by test.
   "body": {
     "error_message": "deleting id=[NUMID]: Fault injected by test. (400 INVALID_PARAMETER_VALUE)",
     "resource_id": "[NUMID]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"stuck\",\"queue\":{\"enabled\":true}}}",
     "status": "OPERATION_STATUS_FAILED"
   }
diff --git a/acceptance/bundle/dms/failed-recreate/output.txt b/acceptance/bundle/dms/failed-recreate/output.txt
index c487df25ac..0c57f181f9 100644
--- a/acceptance/bundle/dms/failed-recreate/output.txt
+++ b/acceptance/bundle/dms/failed-recreate/output.txt
@@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "main.dms_failed_recreate_schema_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_failed_recreate_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -124,14 +124,14 @@ Files: 4 uploaded, 0 deleted
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_RECREATE"
+        "action_type": "OPERATION_ACTION_TYPE_RECREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -142,7 +142,7 @@ Files: 4 uploaded, 0 deleted
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_PENDING"
   }
 }
@@ -154,7 +154,7 @@ Files: 4 uploaded, 0 deleted
   },
   "body": {
     "error_message": "Fault injected by test. (400 INVALID_PARAMETER_VALUE)",
-    "sequence_id": "1",
+    "sequence_id": 1,
     "status": "OPERATION_STATUS_FAILED"
   }
 }
diff --git a/acceptance/bundle/dms/failed-update-permissions/output.txt b/acceptance/bundle/dms/failed-update-permissions/output.txt
index 58e558971b..2523e5f95c 100644
--- a/acceptance/bundle/dms/failed-update-permissions/output.txt
+++ b/acceptance/bundle/dms/failed-update-permissions/output.txt
@@ -104,6 +104,6 @@ Files: 3 uploaded, 0 deleted
 }
 
 >>> print_requests.py --dms //api/2.0/bundle --oneline
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "previous_version_id": "1", "operations": [{"resource_key": "jobs.foo.permissions", "action_type": "OPERATION_ACTION_TYPE_UPDATE"}]}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo.permissions", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "updating id=/jobs/[NUMID]: Fault injected by test. (403 PERMISSION_DENIED)", "resource_id": "/jobs/[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"object_id\":\"/jobs/[NUMID]\",\"__embed__\":[{\"level\":\"CAN_VIEW\",\"user_name\":\"viewer@example.com\"},{\"level\":\"IS_OWNER\",\"user_name\":\"[USERNAME]\"}]},\"depends_on\":[{\"node\":\"resources.jobs.foo\",\"label\":\"${resources.jobs.foo.id}\"}]}", "status": "OPERATION_STATUS_FAILED"}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_UPDATE", "resource_key": "jobs.foo.permissions"}], "previous_version_id": "1", "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo.permissions", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "updating id=/jobs/[NUMID]: Fault injected by test. (403 PERMISSION_DENIED)", "resource_id": "/jobs/[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"object_id\":\"/jobs/[NUMID]\",\"__embed__\":[{\"level\":\"CAN_VIEW\",\"user_name\":\"viewer@example.com\"},{\"level\":\"IS_OWNER\",\"user_name\":\"[USERNAME]\"}]},\"depends_on\":[{\"node\":\"resources.jobs.foo\",\"label\":\"${resources.jobs.foo.id}\"}]}", "status": "OPERATION_STATUS_FAILED"}}
 {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_FAILURE"}}
diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index ec51a77d2c..1ec44fec71 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -11,11 +11,11 @@ Files: 5 uploaded, 0 deleted
 Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 
 >>> print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
 
 === The plan records the deployment_history feature
 >>> [CLI] bundle plan -o json
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index f930196be6..e424db40da 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -93,14 +93,14 @@ Files: 4 uploaded, 0 deleted
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_RECREATE"
+        "action_type": "OPERATION_ACTION_TYPE_RECREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -111,7 +111,7 @@ Files: 4 uploaded, 0 deleted
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_PENDING"
   }
 }
@@ -124,7 +124,7 @@ Files: 4 uploaded, 0 deleted
   "body": {
     "error_message": "",
     "resource_id": "dms_other_[UNIQUE_NAME].dms_upload_fails_once_schema_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"dms_other_[UNIQUE_NAME]\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
diff --git a/acceptance/bundle/dms/provenance/output.txt b/acceptance/bundle/dms/provenance/output.txt
index 03ff7c406c..770e2f2ea1 100644
--- a/acceptance/bundle/dms/provenance/output.txt
+++ b/acceptance/bundle/dms/provenance/output.txt
@@ -15,7 +15,6 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "git_info": {
       "branch": "main",
       "commit": "[COMMIT]",
@@ -23,10 +22,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
     },
     "operations": [
       {
-        "resource_key": "jobs.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "jobs.foo"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -38,7 +38,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "[NUMID]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-provenance-[UNIQUE_NAME]/dev/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":4,\"name\":\"[dev [USERNAME]] foo\",\"queue\":{\"enabled\":true},\"tags\":{\"dev\":\"[USERNAME]\"}}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
diff --git a/acceptance/bundle/dms/record-failure/output.txt b/acceptance/bundle/dms/record-failure/output.txt
index fbcd3135ff..6a2d8ffe13 100644
--- a/acceptance/bundle/dms/record-failure/output.txt
+++ b/acceptance/bundle/dms/record-failure/output.txt
@@ -49,13 +49,13 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "jobs.doomed",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "jobs.doomed"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -66,7 +66,7 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
   },
   "body": {
     "error_message": "Node type nonexistent-node-type is not supported. Supported node types: [NODE_TYPES] (400 INVALID_PARAMETER_VALUE)",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_FAILED"
   }
 }
diff --git a/acceptance/bundle/dms/stale-plan/output.txt b/acceptance/bundle/dms/stale-plan/output.txt
index 2ce611c624..4e65d40b3d 100644
--- a/acceptance/bundle/dms/stale-plan/output.txt
+++ b/acceptance/bundle/dms/stale-plan/output.txt
@@ -53,8 +53,8 @@ Error: plan lineage "" does not match state lineage "[UUID]"; the state may have
 === Only the first deploy recorded a version; the rejected replay created none
 >>> print_requests.py --dms //api/2.0/bundle --oneline
 {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-stale-plan-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default"}}}
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}}
-{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"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/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}}
+{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"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/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}}
 
 === Save a plan against the recorded deployment, then move the deployment on
diff --git a/acceptance/bundle/dms/successful-recreate/output.txt b/acceptance/bundle/dms/successful-recreate/output.txt
index 320369badc..69e7b02643 100644
--- a/acceptance/bundle/dms/successful-recreate/output.txt
+++ b/acceptance/bundle/dms/successful-recreate/output.txt
@@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "main.dms_successful_recreate_schema_[UNIQUE_NAME]",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_successful_recreate_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -117,14 +117,14 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DEPLOY",
-    "previous_version_id": "1",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_RECREATE"
+        "action_type": "OPERATION_ACTION_TYPE_RECREATE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "previous_version_id": "1",
+    "version_type": "VERSION_TYPE_DEPLOY"
   }
 }
 {
@@ -135,7 +135,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_PENDING"
   }
 }
@@ -148,7 +148,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
   "body": {
     "error_message": "",
     "resource_id": "dms_other_[UNIQUE_NAME].dms_successful_recreate_schema_[UNIQUE_NAME]",
-    "sequence_id": "1",
+    "sequence_id": 1,
     "state": "{\"state\":{\"catalog_name\":\"dms_other_[UNIQUE_NAME]\",\"comment\":\"v1\",\"name\":\"dms_successful_recreate_schema_[UNIQUE_NAME]\"}}",
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
@@ -219,14 +219,14 @@ Destroy: 1 deleted
   },
   "body": {
     "cli_version": "[CLI_VERSION]",
-    "version_type": "VERSION_TYPE_DESTROY",
-    "previous_version_id": "2",
     "operations": [
       {
-        "resource_key": "schemas.foo",
-        "action_type": "OPERATION_ACTION_TYPE_DELETE"
+        "action_type": "OPERATION_ACTION_TYPE_DELETE",
+        "resource_key": "schemas.foo"
       }
-    ]
+    ],
+    "previous_version_id": "2",
+    "version_type": "VERSION_TYPE_DESTROY"
   }
 }
 {
@@ -237,7 +237,7 @@ Destroy: 1 deleted
   },
   "body": {
     "error_message": "",
-    "sequence_id": "0",
+    "sequence_id": 0,
     "status": "OPERATION_STATUS_SUCCEEDED"
   }
 }
diff --git a/acceptance/bundle/dms/version-never-created/output.txt b/acceptance/bundle/dms/version-never-created/output.txt
index 3436a9bf09..529bcf4808 100644
--- a/acceptance/bundle/dms/version-never-created/output.txt
+++ b/acceptance/bundle/dms/version-never-created/output.txt
@@ -32,7 +32,7 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
 
 >>> print_requests.py --dms //api/2.0/bundle --get --oneline
 {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-version-never-created", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default"}}}
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}}
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]"}
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]/resources"}
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]"}
@@ -57,4 +57,4 @@ Files: 2 uploaded, 0 deleted
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]/resources"}
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]"}
 {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]/resources"}
-{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}}
+{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}}
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index f4fb00c6b9..bbb97ac4ab 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -122,9 +122,9 @@ type DeploymentState struct {
 	// history, in which case InitializeOperationBuffer installs it once the version exists.
 	operationBuffer *dms.OperationBuffer
 
-	// dmsClient talks to the deployment metadata service. Open builds it from the workspace
+	// dmsService talks to the deployment metadata service. Open sets it from the workspace
 	// client when the deployment records history; nil otherwise.
-	dmsClient *dms.Client
+	dmsService bundledeployments.BundleDeploymentsInterface
 
 	// versionCompleted makes CompleteVersion a no-op after the first call, so a deferred safety-net
 	// completion after an explicit one does nothing.
@@ -204,7 +204,7 @@ type WALEntry struct {
 // (after approval) - which is why it is not an Open option. It also records the id a first deploy
 // just created, which Open could not know, so CompleteVersion later has it.
 func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploymentID string, versionID int) {
-	buf := dms.StartOperationBuffer(ctx, db.dmsClient, deploymentID, versionID)
+	buf := dms.StartOperationBuffer(ctx, db.dmsService, deploymentID, versionID)
 
 	db.mu.Lock()
 	defer db.mu.Unlock()
@@ -248,7 +248,7 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b
 		return false, nil
 	}
 	db.versionCompleted = true
-	deploymentID, client, versionID := db.DeploymentID, db.dmsClient, db.VersionID
+	deploymentID, service, versionID := db.DeploymentID, db.dmsService, db.VersionID
 	db.mu.Unlock()
 
 	// A recording failure fails the version even when the caller counted the deploy a success: the
@@ -262,7 +262,11 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b
 	if !success {
 		reason = bundledeployments.VersionCompleteVersionCompleteFailure
 	}
-	if err := client.CompleteVersion(ctx, deploymentID, versionID, reason); err != nil {
+	_, err := service.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{
+		Name:             dms.VersionName(deploymentID, versionID),
+		CompletionReason: reason,
+	})
+	if err != nil {
 		return false, err
 	}
 	log.Infof(ctx, "Completed deployment version: deployment=%s version=%d reason=%s", deploymentID, versionID, reason)
@@ -450,10 +454,10 @@ func (db *DeploymentState) IsDeploymentMetadataService() bool {
 	return db.isDeploymentMetadataService()
 }
 
-// DmsClient returns the deployment metadata service client Open built from the workspace client,
+// DmsService returns the deployment metadata service client Open set from the workspace client,
 // or nil when the deployment does not record history.
-func (db *DeploymentState) DmsClient() *dms.Client {
-	return db.dmsClient
+func (db *DeploymentState) DmsService() bundledeployments.BundleDeploymentsInterface {
+	return db.dmsService
 }
 
 // GetOrInitLineage returns the deployment lineage, generating and storing a new
@@ -617,15 +621,11 @@ To record this bundle's history, start it over as a new deployment:
 	if recorded {
 		// The service is the source of truth for a recorded deployment; the file is a tombstone
 		// carrying only the marker, and applyDMSState loads the resources the service holds.
-		client, err := dms.NewClient(cmdctx.WorkspaceClient(ctx))
-		if err != nil {
-			return err
-		}
-		db.dmsClient = client
+		db.dmsService = cmdctx.WorkspaceClient(ctx).BundleDeployments
 		db.DeploymentID = dmsDeployment.DeploymentID
 
 		if dmsDeployment.DeploymentID != "" {
-			resources, err := db.dmsClient.ListResources(ctx, dmsDeployment.DeploymentID)
+			resources, err := dms.ListResources(ctx, db.dmsService, dmsDeployment.DeploymentID)
 			if err != nil {
 				return err
 			}
diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go
index 2ac8c8482d..081f59c3d8 100644
--- a/bundle/phases/destroy.go
+++ b/bundle/phases/destroy.go
@@ -20,6 +20,7 @@ import (
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
 func assertRootPathExists(ctx context.Context, b *bundle.Bundle) (bool, error) {
@@ -169,7 +170,7 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, e
 		// A completed destroy's resources are gone, so its deployment record is deleted too.
 		if completed {
 			deploymentID := b.DeploymentBundle.StateDB.DeploymentID
-			if err := b.DeploymentBundle.StateDB.DmsClient().DeleteDeployment(ctx, deploymentID); err != nil {
+			if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err))
 				return
 			}
@@ -224,7 +225,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) {
 				logdiag.LogError(ctx, err)
 			} else if completed {
 				deploymentID := b.DeploymentBundle.StateDB.DeploymentID
-				if err := b.DeploymentBundle.StateDB.DmsClient().DeleteDeployment(ctx, deploymentID); err != nil {
+				if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil {
 					logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err))
 				}
 			}
diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go
index 9ceea29509..eda1abc7d7 100644
--- a/bundle/phases/dms.go
+++ b/bundle/phases/dms.go
@@ -16,15 +16,16 @@ import (
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/cli/libs/workspaceurls"
+	"github.com/databricks/databricks-sdk-go/common/types/fieldmask"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
 // stagedOperations lists the resources the plan will touch, for CreateVersion to stage an
 // operation each. Skipped and undefined actions are left out: nothing is applied for them, so
 // their operations would stay pending and the service would hold no state for them.
-func stagedOperations(plan *deployplan.Plan) ([]dms.StagedOperation, error) {
+func stagedOperations(plan *deployplan.Plan) ([]bundledeployments.StagedOperation, error) {
 	actions := plan.GetActions()
-	staged := make([]dms.StagedOperation, 0, len(actions))
+	staged := make([]bundledeployments.StagedOperation, 0, len(actions))
 	for _, action := range actions {
 		if action.ActionType == deployplan.Skip || action.ActionType == deployplan.Undefined {
 			continue
@@ -33,8 +34,8 @@ func stagedOperations(plan *deployplan.Plan) ([]dms.StagedOperation, error) {
 		if err != nil {
 			return nil, fmt.Errorf("%s: %w", action.ResourceKey, err)
 		}
-		staged = append(staged, dms.StagedOperation{
-			ResourceKey: action.ResourceKey,
+		staged = append(staged, bundledeployments.StagedOperation{
+			ResourceKey: strings.TrimPrefix(action.ResourceKey, dms.StatePrefix),
 			ActionType:  actionType,
 		})
 	}
@@ -69,19 +70,31 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType
 // no deployment behind; a first deploy's new id is then stamped into the plan (StampDeploymentID).
 func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment) {
 	db := &b.DeploymentBundle
-	dmsClient := db.StateDB.DmsClient()
+	dmsService := db.StateDB.DmsService()
 	metadata := deploymentMetadata(b)
 	deploymentID := db.StateDB.DeploymentID
 	if deploymentID == "" {
-		id, err := dmsClient.CreateDeployment(ctx, b.Config.Workspace.StatePath, metadata)
+		dep := metadata.Deployment()
+		dep.InitialParentPath = b.Config.Workspace.StatePath
+		created, err := dmsService.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep})
+		if err != nil {
+			logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err))
+			return
+		}
+		// The server assigns the id as the workspace node it creates under the parent path.
+		deploymentID, err = dms.DeploymentIDFromName(created.Name)
 		if err != nil {
 			logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err))
 			return
 		}
-		deploymentID = id
 		db.StateDB.DeploymentID = deploymentID
 	} else if mask := metadata.StaleFields(current); mask != "" {
-		if err := dmsClient.UpdateDeployment(ctx, deploymentID, metadata, mask); err != nil {
+		_, err := dmsService.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{
+			Name:       dms.DeploymentName(deploymentID),
+			Deployment: dms.DeploymentUpdate(metadata, mask),
+			UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")},
+		})
+		if err != nil {
 			logdiag.LogError(ctx, fmt.Errorf("failed to update deployment: %w", err))
 			return
 		}
@@ -96,10 +109,10 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu
 // startVersion claims the version the run settled on and opens the buffer that records
 // each state write under it. Called after approval, so a declined deploy never claims a number.
 // A no-op when the bundle does not record deployment history.
-func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.VersionType, staged []dms.StagedOperation) error {
+func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.VersionType, staged []bundledeployments.StagedOperation) error {
 	db := &b.DeploymentBundle
-	dmsClient := db.StateDB.DmsClient()
-	if dmsClient == nil {
+	dmsService := db.StateDB.DmsService()
+	if dmsService == nil {
 		return nil
 	}
 	deploymentID := db.StateDB.DeploymentID
@@ -124,12 +137,16 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version
 			OriginUrl: git.OriginURL,
 		}
 	}
-	version, err := dmsClient.CreateVersion(ctx, deploymentID, versionID, dms.CreateVersionRequest{
-		CliVersion:        build.GetInfo().Version,
-		VersionType:       versionType,
-		PreviousVersionId: previousVersionID,
-		Operations:        staged,
-		GitInfo:           gitInfo,
+	version, err := dmsService.CreateVersion(ctx, bundledeployments.CreateVersionRequest{
+		Parent:    dms.DeploymentName(deploymentID),
+		VersionId: strconv.Itoa(versionID),
+		Version: bundledeployments.Version{
+			CliVersion:        build.GetInfo().Version,
+			VersionType:       versionType,
+			PreviousVersionId: previousVersionID,
+			GitInfo:           gitInfo,
+			Operations:        staged,
+		},
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create deployment version: %w", err)
diff --git a/bundle/phases/dms_test.go b/bundle/phases/dms_test.go
index f6854999db..6b045f863a 100644
--- a/bundle/phases/dms_test.go
+++ b/bundle/phases/dms_test.go
@@ -4,7 +4,6 @@ import (
 	"testing"
 
 	"github.com/databricks/cli/bundle/deployplan"
-	"github.com/databricks/cli/libs/dms"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -23,11 +22,11 @@ func TestStagedOperationsCoversEveryTouchedResource(t *testing.T) {
 	staged, err := stagedOperations(plan)
 	require.NoError(t, err)
 
-	assert.ElementsMatch(t, []dms.StagedOperation{
-		{ResourceKey: "resources.jobs.foo", ActionType: bundledeployments.OperationActionTypeOperationActionTypeCreate},
-		{ResourceKey: "resources.pipelines.bar", ActionType: bundledeployments.OperationActionTypeOperationActionTypeRecreate},
-		{ResourceKey: "resources.schemas.baz", ActionType: bundledeployments.OperationActionTypeOperationActionTypeDelete},
-		{ResourceKey: "resources.clusters.small", ActionType: bundledeployments.OperationActionTypeOperationActionTypeResize},
+	assert.ElementsMatch(t, []bundledeployments.StagedOperation{
+		{ResourceKey: "jobs.foo", ActionType: bundledeployments.OperationActionTypeOperationActionTypeCreate},
+		{ResourceKey: "pipelines.bar", ActionType: bundledeployments.OperationActionTypeOperationActionTypeRecreate},
+		{ResourceKey: "schemas.baz", ActionType: bundledeployments.OperationActionTypeOperationActionTypeDelete},
+		{ResourceKey: "clusters.small", ActionType: bundledeployments.OperationActionTypeOperationActionTypeResize},
 	}, staged)
 }
 
@@ -43,8 +42,8 @@ func TestStagedOperationsLeavesOutUntouchedResources(t *testing.T) {
 	staged, err := stagedOperations(plan)
 	require.NoError(t, err)
 
-	assert.Equal(t, []dms.StagedOperation{
-		{ResourceKey: "resources.jobs.touched", ActionType: bundledeployments.OperationActionTypeOperationActionTypeUpdate},
+	assert.Equal(t, []bundledeployments.StagedOperation{
+		{ResourceKey: "jobs.touched", ActionType: bundledeployments.OperationActionTypeOperationActionTypeUpdate},
 	}, staged)
 }
 
diff --git a/libs/dms/client.go b/libs/dms/client.go
deleted file mode 100644
index 605b4bbd4d..0000000000
--- a/libs/dms/client.go
+++ /dev/null
@@ -1,258 +0,0 @@
-package dms
-
-import (
-	"context"
-	"fmt"
-	"net/http"
-	"strconv"
-	"strings"
-
-	"github.com/databricks/cli/libs/auth"
-	"github.com/databricks/databricks-sdk-go"
-	"github.com/databricks/databricks-sdk-go/client"
-	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
-)
-
-// DeploymentNodeName is the workspace node DMS creates per deployment. Must
-// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side.
-const DeploymentNodeName = "resources.deployment.json"
-
-// statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a
-// job "resources.jobs.foo", DMS calls it "jobs.foo". Every exported name here takes the state
-// form; the prefix comes off where a request is built, and back on where a resource is read.
-const statePrefix = "resources."
-
-// Client carries the calls the CLI makes to DMS, as methods below. Each one goes out through one
-// of two halves: the generated client for the calls it can express, and hand-written requests
-// for the two it cannot.
-//
-// TODO: Remove this and replace with the SDK.
-type Client struct {
-	// Service is the generated client.
-	Service bundledeployments.BundleDeploymentsInterface
-
-	// raw sends what the generated client cannot; see requester.
-	raw requester
-}
-
-// NewClient returns a Client for the workspace w.
-func NewClient(w *databricks.WorkspaceClient) (*Client, error) {
-	api, err := client.New(w.Config)
-	if err != nil {
-		return nil, err
-	}
-	return &Client{Service: w.BundleDeployments, raw: &rawClient{client: api}}, nil
-}
-
-// DeploymentName and versionName are the two resource-name formats the service uses. Every
-// call builds its name here, so a caller only ever passes ids.
-func DeploymentName(deploymentID string) string {
-	return "deployments/" + deploymentID
-}
-
-func versionName(deploymentID string, version int) string {
-	return fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version)
-}
-
-// CreateDeployment registers a deployment under parentPath and returns the id the server
-// assigned it, which is the id of the workspace node it creates there.
-func (c *Client) CreateDeployment(ctx context.Context, parentPath string, metadata Metadata) (string, error) {
-	dep := metadata.deployment()
-	dep.InitialParentPath = parentPath
-
-	created, err := c.Service.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep})
-	if err != nil {
-		return "", err
-	}
-	return deploymentIDFromName(created.Name)
-}
-
-// UpdateDeployment writes the fields mask names onto the deployment. The service ignores every
-// other field, so the mask is what decides the write.
-func (c *Client) UpdateDeployment(ctx context.Context, deploymentID string, metadata Metadata, mask string) error {
-	return c.raw.UpdateDeployment(ctx, deploymentID, metadata.deployment(), mask)
-}
-
-// DeleteDeployment removes the deployment record, which a completed destroy does.
-func (c *Client) DeleteDeployment(ctx context.Context, deploymentID string) error {
-	return c.Service.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{
-		Name: DeploymentName(deploymentID),
-	})
-}
-
-// CreateVersion claims the version and stages the operations body carries.
-func (c *Client) CreateVersion(ctx context.Context, deploymentID string, version int, body CreateVersionRequest) (*bundledeployments.Version, error) {
-	return c.raw.CreateVersion(ctx, deploymentID, strconv.Itoa(version), body)
-}
-
-// CompleteVersion closes the version out, which is what stops the service expiring its lease.
-func (c *Client) CompleteVersion(ctx context.Context, deploymentID string, version int, reason bundledeployments.VersionComplete) error {
-	_, err := c.Service.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{
-		Name:             versionName(deploymentID, version),
-		CompletionReason: reason,
-	})
-	return err
-}
-
-// UpdateOperation fills in one operation the version staged, and returns the sequence id the
-// next update for that resource must send.
-func (c *Client) UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (string, error) {
-	return c.raw.UpdateOperation(ctx, deploymentID, version, stateKey, sequenceID, update)
-}
-
-// deploymentIDFromName extracts the deployment ID from a DMS resource name of
-// the form "deployments/{deployment_id}".
-func deploymentIDFromName(name string) (string, error) {
-	id, ok := strings.CutPrefix(name, DeploymentName(""))
-	if !ok || id == "" {
-		return "", fmt.Errorf("unexpected deployment name %q from the deployment history service", name)
-	}
-	return id, nil
-}
-
-// requester sends the two requests the generated client cannot express, so a test can capture
-// what the CLI puts on the wire. Both are TODO(DMS): drop them once the spec catches up.
-type requester interface {
-	// CreateVersion is hand-written because the generated struct has no operations: the field
-	// is at DEVELOPMENT stage, which keeps it out of the SDK until it is promoted.
-	CreateVersion(ctx context.Context, deploymentID, versionID string, body CreateVersionRequest) (*bundledeployments.Version, error)
-
-	// UpdateDeployment is hand-written because the generated client has no such call yet.
-	UpdateDeployment(ctx context.Context, deploymentID string, deployment bundledeployments.Deployment, mask string) error
-
-	// UpdateOperation is hand-written because the SDK types sequence_id as an int64 while
-	// the service sends a JSON string, so it cannot read the response. sequenceID is the
-	// token the previous update for this resource returned, or 0 for the first, which is
-	// what staging leaves.
-	UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (next string, err error)
-}
-
-// CreateVersionRequest is the CreateVersion request body.
-type CreateVersionRequest struct {
-	CliVersion  string      `json:"cli_version"`
-	VersionType VersionType `json:"version_type"`
-	// PreviousVersionId is the deployment's most recent version, unset for a
-	// deployment's first version.
-	PreviousVersionId string `json:"previous_version_id,omitempty"`
-	// GitInfo records where this version's source came from. The rest of the provenance -
-	// display name, target, mode, workspace paths - belongs to the deployment.
-	GitInfo *bundledeployments.GitInfo `json:"git_info,omitempty"`
-	// Operations is every resource this version will touch; see StagedOperation. It sits in this
-	// body with the version's own fields because the request binds body: "version", and is input
-	// only - the response never carries it back.
-	Operations []StagedOperation `json:"operations,omitempty"`
-}
-
-// StagedOperation is one resource the version will record an operation for. The service
-// creates it in OPERATION_STATUS_PENDING at sequence id 0, and the CLI fills in the outcome
-// with UpdateOperation as the resource is applied.
-type StagedOperation struct {
-	// ResourceKey is the bundle state key; the request carries the form the service uses.
-	ResourceKey string                                `json:"resource_key"`
-	ActionType  bundledeployments.OperationActionType `json:"action_type"`
-}
-
-// operationResponse is the part of an operation response the CLI reads back.
-type operationResponse struct {
-	// SequenceId is the concurrency token for the next update, typed as the service sends it.
-	SequenceId string `json:"sequence_id,omitempty"`
-}
-
-// rawClient sends the three DMS requests the generated SDK client cannot express yet, while the
-// endpoints are at DEVELOPMENT stage. TODO(DMS): drop each once the SDK catches up.
-//   - UpdateDeployment: the SDK exposes no method for it.
-//   - CreateVersion: the SDK's Version type has no operations field, so a version cannot stage
-//     its operations through it.
-//   - UpdateOperation: the SDK types sequence_id as int64, but the service sends and expects it
-//     as a JSON string.
-//
-// CreateDeployment, DeleteDeployment, CompleteVersion and ListResources go through the SDK (see Client).
-type rawClient struct {
-	client *client.DatabricksClient
-}
-
-func (r *rawClient) CreateVersion(ctx context.Context, deploymentID, versionID string, body CreateVersionRequest) (*bundledeployments.Version, error) {
-	staged := make([]StagedOperation, len(body.Operations))
-	for i, op := range body.Operations {
-		op.ResourceKey = strings.TrimPrefix(op.ResourceKey, statePrefix)
-		staged[i] = op
-	}
-	body.Operations = staged
-
-	var version bundledeployments.Version
-	path := "/api/2.0/bundle/" + DeploymentName(deploymentID) + "/versions"
-	err := r.client.Do(ctx, http.MethodPost, path,
-		auth.WorkspaceIDHeaders(r.client.Config),
-		map[string]any{"version_id": versionID},
-		body, &version)
-	if err != nil {
-		return nil, err
-	}
-	return &version, nil
-}
-
-// newDeploymentUpdate builds the request body for update, holding exactly the masked fields for
-// the same reason newUpdateRequest does. A map, not the SDK struct, whose omitempty tags would
-// drop a masked field that is empty - which is how deployment_mode is cleared when a target stops
-// setting mode.
-func newDeploymentUpdate(deployment bundledeployments.Deployment, mask string) map[string]any {
-	values := map[string]any{
-		"display_name":    deployment.DisplayName,
-		"target_name":     deployment.TargetName,
-		"deployment_mode": deployment.DeploymentMode,
-		"workspace_info":  deployment.WorkspaceInfo,
-	}
-
-	body := map[string]any{}
-	for field := range strings.SplitSeq(mask, ",") {
-		body[field] = values[field]
-	}
-	return body
-}
-
-func (r *rawClient) UpdateDeployment(ctx context.Context, deploymentID string, deployment bundledeployments.Deployment, mask string) error {
-	path := "/api/2.0/bundle/" + DeploymentName(deploymentID)
-	return r.client.Do(ctx, http.MethodPatch, path,
-		auth.WorkspaceIDHeaders(r.client.Config),
-		map[string]any{"update_mask": mask},
-		newDeploymentUpdate(deployment, mask), nil)
-}
-
-// newUpdateRequest builds the request body for update. A field is in the body when the mask
-// names it and absent otherwise, which is what the service requires: it rejects an update
-// whose mask names a field the body leaves out, and an empty value is how a field is cleared -
-// no state means the resource is gone, no error_message means an earlier failure is resolved.
-// A map, not a struct, so presence cannot drift from the mask through an omitempty tag.
-func newUpdateRequest(update OperationUpdate, sequenceID string) map[string]any {
-	body := map[string]any{"sequence_id": sequenceID}
-	// A masked state with no value is how the service is told the resource is gone, so a nil
-	// state is left out rather than sent empty.
-	if update.Fields.Has(FieldState) && update.State != nil {
-		body["state"] = string(update.State)
-	}
-	if update.Fields.Has(FieldResourceID) {
-		body["resource_id"] = update.ResourceID
-	}
-	if update.Fields.Has(FieldErrorMessage) {
-		body["error_message"] = update.ErrorMessage
-	}
-	if update.Fields.Has(FieldStatus) {
-		body["status"] = update.Status
-	}
-	return body
-}
-
-func (r *rawClient) UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (string, error) {
-	body := newUpdateRequest(update, sequenceID)
-
-	var result operationResponse
-	path := "/api/2.0/bundle/" + versionName(deploymentID, version) + "/operations/" + strings.TrimPrefix(stateKey, statePrefix)
-	err := r.client.Do(ctx, http.MethodPatch, path,
-		auth.WorkspaceIDHeaders(r.client.Config),
-		map[string]any{"update_mask": update.Fields.Mask()},
-		body, &result)
-	if err != nil {
-		return "", err
-	}
-	return result.SequenceId, nil
-}
diff --git a/libs/dms/client_test.go b/libs/dms/client_test.go
deleted file mode 100644
index 867655bca4..0000000000
--- a/libs/dms/client_test.go
+++ /dev/null
@@ -1,67 +0,0 @@
-package dms
-
-import (
-	"encoding/json"
-	"errors"
-	"testing"
-
-	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestClientNamesEveryResourceTheSameWay(t *testing.T) {
-	// One format each, so a call only ever passes ids.
-	assert.Equal(t, "deployments/dep-1", DeploymentName("dep-1"))
-	assert.Equal(t, "deployments/dep-1/versions/2", versionName("dep-1", 2))
-}
-
-func TestDeploymentIDFromName(t *testing.T) {
-	id, err := deploymentIDFromName("deployments/abc-123")
-	require.NoError(t, err)
-	assert.Equal(t, "abc-123", id)
-
-	_, err = deploymentIDFromName("abc-123")
-	assert.Error(t, err)
-
-	_, err = deploymentIDFromName("deployments/")
-	assert.Error(t, err)
-}
-
-func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) {
-	// The service rejects an update whose mask names a field the body leaves out, and treats a
-	// field it does carry as written - so the body has to hold every masked field and nothing
-	// else, empty values included. Both masks the CLI builds are asserted on the wire by
-	// acceptance/bundle/dms; this pins the rule they both rely on.
-	update := OperationUpdate{
-		Fields:     DescribesResource,
-		State:      json.RawMessage(`{"state":{"name":"foo"}}`),
-		ResourceID: "job-1",
-		Status:     bundledeployments.OperationStatusOperationStatusSucceeded,
-	}
-
-	// A successful write reports no error, and the empty value is what clears an earlier one.
-	assert.Equal(t, map[string]any{
-		"state":         `{"state":{"name":"foo"}}`,
-		"resource_id":   "job-1",
-		"error_message": "",
-		"status":        bundledeployments.OperationStatusOperationStatusSucceeded,
-		"sequence_id":   "3",
-	}, newUpdateRequest(update, "3"))
-
-	// A failure keeps the recorded state, so state is absent rather than empty: naming it
-	// would clear what the resource last recorded.
-	assert.Equal(t, map[string]any{
-		"error_message": "boom",
-		"status":        bundledeployments.OperationStatusOperationStatusFailed,
-		"sequence_id":   "3",
-	}, newUpdateRequest(NewFailureUpdate("job-1", nil, errors.New("boom")), "3"))
-
-	// The deployment's own fields follow the same rule. Clearing deployment_mode - a target that
-	// stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped.
-	deployment := Metadata{DisplayName: "b", TargetName: "t"}.deployment()
-	assert.Equal(t, map[string]any{
-		"target_name":     "t",
-		"deployment_mode": bundledeployments.DeploymentMode(""),
-	}, newDeploymentUpdate(deployment, "target_name,deployment_mode"))
-}
diff --git a/libs/dms/metadata.go b/libs/dms/metadata.go
index 33b7d49f3c..80170b7e56 100644
--- a/libs/dms/metadata.go
+++ b/libs/dms/metadata.go
@@ -34,8 +34,8 @@ type Metadata struct {
 // among them: the service derives the deployment's from the version that carried it.
 var deploymentFields = []string{"display_name", "target_name", "deployment_mode", "workspace_info"}
 
-// deployment renders the metadata the deployment owns.
-func (m Metadata) deployment() bundledeployments.Deployment {
+// Deployment renders the metadata the deployment owns.
+func (m Metadata) Deployment() bundledeployments.Deployment {
 	return bundledeployments.Deployment{
 		DisplayName:    m.DisplayName,
 		TargetName:     m.TargetName,
@@ -51,7 +51,7 @@ func (m Metadata) StaleFields(current *bundledeployments.Deployment) string {
 		return strings.Join(deploymentFields, ",")
 	}
 
-	want := m.deployment()
+	want := m.Deployment()
 	var stale []string
 	if want.DisplayName != current.DisplayName {
 		stale = append(stale, "display_name")
@@ -80,3 +80,53 @@ func NextVersion(lastVersionID string) (int, error) {
 	}
 	return last + 1, nil
 }
+
+// DeploymentNodeName is the workspace node DMS creates per deployment. Must
+// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side.
+const DeploymentNodeName = "resources.deployment.json"
+
+// DeploymentName and VersionName are the two resource-name formats the service uses, so a
+// caller only ever passes ids.
+func DeploymentName(deploymentID string) string {
+	return "deployments/" + deploymentID
+}
+
+func VersionName(deploymentID string, version int) string {
+	return fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version)
+}
+
+// DeploymentIDFromName extracts the deployment ID from a DMS resource name of
+// the form "deployments/{deployment_id}".
+func DeploymentIDFromName(name string) (string, error) {
+	id, ok := strings.CutPrefix(name, DeploymentName(""))
+	if !ok || id == "" {
+		return "", fmt.Errorf("unexpected deployment name %q from the deployment history service", name)
+	}
+	return id, nil
+}
+
+// DeploymentUpdate builds the deployment carrying exactly the masked fields, empty ones
+// included: the service requires every masked field to be present in the body and reads an empty
+// value as a clear (a target that stops setting mode clears deployment_mode). ForceSendFields
+// keeps those empty values on the wire, which omitempty would drop.
+func DeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployment {
+	full := metadata.Deployment()
+	var dep bundledeployments.Deployment
+	for path := range strings.SplitSeq(mask, ",") {
+		switch path {
+		case "display_name":
+			dep.DisplayName = full.DisplayName
+			dep.ForceSendFields = append(dep.ForceSendFields, "DisplayName")
+		case "target_name":
+			dep.TargetName = full.TargetName
+			dep.ForceSendFields = append(dep.ForceSendFields, "TargetName")
+		case "deployment_mode":
+			dep.DeploymentMode = full.DeploymentMode
+			dep.ForceSendFields = append(dep.ForceSendFields, "DeploymentMode")
+		case "workspace_info":
+			dep.WorkspaceInfo = full.WorkspaceInfo
+			dep.ForceSendFields = append(dep.ForceSendFields, "WorkspaceInfo")
+		}
+	}
+	return dep
+}
diff --git a/libs/dms/metadata_test.go b/libs/dms/metadata_test.go
index cddf4205cf..5122533daf 100644
--- a/libs/dms/metadata_test.go
+++ b/libs/dms/metadata_test.go
@@ -23,3 +23,21 @@ func TestNextVersion(t *testing.T) {
 	_, err := NextVersion("not-a-number")
 	assert.ErrorContains(t, err, "last_version_id")
 }
+
+func TestResourceNames(t *testing.T) {
+	// One format each, so a call only ever passes ids.
+	assert.Equal(t, "deployments/dep-1", DeploymentName("dep-1"))
+	assert.Equal(t, "deployments/dep-1/versions/2", VersionName("dep-1", 2))
+}
+
+func TestDeploymentIDFromName(t *testing.T) {
+	id, err := DeploymentIDFromName("deployments/abc-123")
+	require.NoError(t, err)
+	assert.Equal(t, "abc-123", id)
+
+	_, err = DeploymentIDFromName("abc-123")
+	assert.Error(t, err)
+
+	_, err = DeploymentIDFromName("deployments/")
+	assert.Error(t, err)
+}
diff --git a/libs/dms/operation.go b/libs/dms/operation.go
index 57577f5dc2..0bf4452052 100644
--- a/libs/dms/operation.go
+++ b/libs/dms/operation.go
@@ -3,12 +3,18 @@ package dms
 import (
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"unicode/utf8"
 
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
+// statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a
+// job "resources.jobs.foo", DMS calls it "jobs.foo". Exported names take the state form; the
+// prefix comes off where a request is built, and back on where a resource is read.
+const StatePrefix = "resources."
+
 // maxStateSize is the largest serialized state DMS accepts per operation. More than this
 // and the resource cannot be recorded at all, so the deploy fails rather than leaving the
 // service holding a resource with no state.
@@ -140,3 +146,37 @@ func (u OperationUpdate) Merge(newer OperationUpdate) OperationUpdate {
 
 	return merged
 }
+
+// newOperationUpdate builds the operation carrying exactly the fields update.Fields masks, plus
+// the sequence_id precondition. The service requires a masked field to be present and reads an
+// empty value as a write (error_message="" clears it), so masked fields that can be empty are
+// forced onto the wire; sequence_id is always sent and a freshly staged operation sits at 0,
+// which omitempty would drop. State is the exception: an absent value is how the service is told
+// the resource is gone, so a nil state is left off (never forced) while still named in the mask.
+func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) {
+	sequence, err := strconv.ParseInt(sequenceID, 10, 64)
+	if err != nil {
+		return bundledeployments.Operation{}, fmt.Errorf("invalid sequence id %q: %w", sequenceID, err)
+	}
+
+	operation := bundledeployments.Operation{
+		SequenceId:      sequence,
+		ForceSendFields: []string{"SequenceId"},
+	}
+	if update.Fields.Has(FieldState) && update.State != nil {
+		operation.State = string(update.State)
+	}
+	if update.Fields.Has(FieldErrorMessage) {
+		operation.ErrorMessage = update.ErrorMessage
+		operation.ForceSendFields = append(operation.ForceSendFields, "ErrorMessage")
+	}
+	if update.Fields.Has(FieldResourceID) {
+		operation.ResourceId = update.ResourceID
+		operation.ForceSendFields = append(operation.ForceSendFields, "ResourceId")
+	}
+	if update.Fields.Has(FieldStatus) {
+		operation.Status = update.Status
+		operation.ForceSendFields = append(operation.ForceSendFields, "Status")
+	}
+	return operation, nil
+}
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index c4fbb8e379..e17f424f13 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -4,7 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
+	"strings"
 	"sync"
+
+	"github.com/databricks/databricks-sdk-go/common/types/fieldmask"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
 // bufferedOperations caps how far ahead of the service a deploy may get; DMS is what the next
@@ -19,7 +24,7 @@ const stagedSequenceID = "0"
 // path: writes are queued and sent on one background goroutine. It exists only while a bundle
 // records deployment history; callers hold a nil buffer otherwise and must not call it.
 type OperationBuffer struct {
-	client       *Client
+	service      bundledeployments.BundleDeploymentsInterface
 	deploymentID string
 	versionNum   int
 
@@ -56,9 +61,9 @@ type OperationBuffer struct {
 
 // StartOperationBuffer opens the buffer for the version the caller just created. The version
 // must already exist: operations record under it, and nothing here creates it.
-func StartOperationBuffer(ctx context.Context, client *Client, deploymentID string, versionNum int) *OperationBuffer {
+func StartOperationBuffer(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string, versionNum int) *OperationBuffer {
 	b := &OperationBuffer{
-		client:       client,
+		service:      service,
 		deploymentID: deploymentID,
 		versionNum:   versionNum,
 		queue:        make(chan string, bufferedOperations),
@@ -155,13 +160,21 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio
 		sequenceID = stagedSequenceID
 	}
 
-	next, err := b.client.UpdateOperation(ctx, b.deploymentID, b.versionNum, key, sequenceID, update)
+	operation, err := newOperationUpdate(update, sequenceID)
+	if err != nil {
+		return err
+	}
+	result, err := b.service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{
+		Name:       VersionName(b.deploymentID, b.versionNum) + "/operations/" + strings.TrimPrefix(key, StatePrefix),
+		Operation:  operation,
+		UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")},
+	})
 	if err != nil {
 		return err
 	}
 
 	// The next write for this resource echoes the sequence id this one earned.
-	b.sequenceIDs[key] = next
+	b.sequenceIDs[key] = strconv.FormatInt(result.SequenceId, 10)
 
 	return nil
 }
diff --git a/libs/dms/operation_test.go b/libs/dms/operation_test.go
index 2bae9a8114..ef5928a0ef 100644
--- a/libs/dms/operation_test.go
+++ b/libs/dms/operation_test.go
@@ -81,3 +81,54 @@ func TestMergeLetsADeleteClearTheState(t *testing.T) {
 
 	assert.Nil(t, merged.State)
 }
+
+func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) {
+	// The service rejects an update whose mask names a field the body leaves out, and treats a
+	// field it does carry as written - so the body has to hold every masked field, empty values
+	// included, which ForceSendFields is what secures. Both bodies the CLI builds are asserted on
+	// the wire by acceptance/bundle/dms; this pins the rule they both rely on.
+	update := OperationUpdate{
+		Fields:     DescribesResource,
+		State:      json.RawMessage(`{"state":{"name":"foo"}}`),
+		ResourceID: "job-1",
+		Status:     bundledeployments.OperationStatusOperationStatusSucceeded,
+	}
+
+	// A successful write reports no error, and the empty value is what clears an earlier one.
+	operation, err := newOperationUpdate(update, "3")
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"sequence_id": 3,
+		"state": "{\"state\":{\"name\":\"foo\"}}",
+		"resource_id": "job-1",
+		"error_message": "",
+		"status": "OPERATION_STATUS_SUCCEEDED"
+	}`, marshalBody(t, operation))
+
+	// A failure keeps the recorded state, so state is absent rather than empty: naming it
+	// would clear what the resource last recorded.
+	operation, err = newOperationUpdate(NewFailureUpdate("job-1", nil, errors.New("boom")), "3")
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"sequence_id": 3,
+		"error_message": "boom",
+		"status": "OPERATION_STATUS_FAILED"
+	}`, marshalBody(t, operation))
+
+	// The deployment's own fields follow the same rule. Clearing deployment_mode - a target that
+	// stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped.
+	deployment := DeploymentUpdate(Metadata{DisplayName: "b", TargetName: "t"}, "target_name,deployment_mode")
+	assert.JSONEq(t, `{
+		"target_name": "t",
+		"deployment_mode": ""
+	}`, marshalBody(t, deployment))
+}
+
+// marshalBody renders the request body the SDK sends for v, so a test sees the wire the
+// ForceSendFields produce.
+func marshalBody(t *testing.T, v any) string {
+	t.Helper()
+	body, err := json.Marshal(v)
+	require.NoError(t, err)
+	return string(body)
+}
diff --git a/libs/dms/resources.go b/libs/dms/resources.go
index 11321161c1..93024eda3c 100644
--- a/libs/dms/resources.go
+++ b/libs/dms/resources.go
@@ -20,8 +20,8 @@ type Resource struct {
 }
 
 // ListResources returns every resource DMS holds for the deployment.
-func (c *Client) ListResources(ctx context.Context, deploymentID string) ([]Resource, error) {
-	it := c.Service.ListResources(ctx, bundledeployments.ListResourcesRequest{
+func ListResources(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string) ([]Resource, error) {
+	it := service.ListResources(ctx, bundledeployments.ListResourcesRequest{
 		Parent: DeploymentName(deploymentID),
 	})
 
@@ -32,7 +32,7 @@ func (c *Client) ListResources(ctx context.Context, deploymentID string) ([]Reso
 			return nil, fmt.Errorf("listing resources from the deployment history service: %w", err)
 		}
 		out = append(out, Resource{
-			Key:   statePrefix + res.ResourceKey,
+			Key:   StatePrefix + res.ResourceKey,
 			ID:    res.ResourceId,
 			State: res.State,
 		})
diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go
index 4fccd96c70..e77c754962 100644
--- a/libs/testserver/bundledeployments.go
+++ b/libs/testserver/bundledeployments.go
@@ -355,8 +355,9 @@ func (s *FakeWorkspace) Heartbeat() Response {
 	return Response{Body: bundledeployments.HeartbeatResponse{}}
 }
 
-// operationBody renders an operation the way the service does: sequence_id as a
-// JSON string, which the SDK struct cannot express (it types the field int64).
+// operationBody renders an operation the way the service does: sequence_id as a JSON string. The
+// SDK reads it back into its int64 field through its number-or-string normalizer, so the fake
+// keeps emitting a string to mirror the real service's proto3 JSON.
 func operationBody(op *bundledeployments.Operation) (map[string]any, error) {
 	raw, err := json.Marshal(op)
 	if err != nil {
@@ -377,26 +378,16 @@ func operationBody(op *bundledeployments.Operation) (map[string]any, error) {
 // UpdateOperation applies a later write for a resource already recorded in this
 // version. sequence_id is the concurrency precondition and increments on success.
 func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, resourceKey string) Response {
-	// sequence_id arrives as a string, which the SDK struct cannot hold (it types the
-	// field int64), so read the body twice: once for the typed fields with that key
-	// removed, and once for the precondition alone.
+	// raw records which fields the body carried, which the update_mask rules turn on: a masked
+	// path with no value is how a field is cleared.
 	var raw map[string]json.RawMessage
 	if err := json.Unmarshal(req.Body, &raw); err != nil {
 		return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}}
 	}
-	var precondition struct {
-		SequenceId string `json:"sequence_id"`
-	}
-	if err := json.Unmarshal(req.Body, &precondition); err != nil {
-		return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}}
-	}
-	delete(raw, "sequence_id")
-	typedBody, err := json.Marshal(raw)
-	if err != nil {
-		return Response{StatusCode: 500, Body: map[string]string{"message": err.Error()}}
-	}
+	// The SDK sends sequence_id as a JSON number and the real service as a string; Operation reads
+	// either into its int64 field, so the whole body decodes straight into it.
 	var op bundledeployments.Operation
-	if err := json.Unmarshal(typedBody, &op); err != nil {
+	if err := json.Unmarshal(req.Body, &op); err != nil {
 		return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}}
 	}
 
@@ -461,7 +452,7 @@ func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, re
 	if !ok {
 		return dmsNotFound("operation " + opName)
 	}
-	if precondition.SequenceId != strconv.FormatInt(existing.SequenceId, 10) {
+	if op.SequenceId != existing.SequenceId {
 		return dmsAborted("sequence_id is outdated; the operation is at " + strconv.FormatInt(existing.SequenceId, 10))
 	}