diff --git a/acceptance/bundle/deployment/bind/test.toml b/acceptance/bundle/deployment/bind/test.toml
index b693e4ded26..117379e3907 100644
--- a/acceptance/bundle/deployment/bind/test.toml
+++ b/acceptance/bundle/deployment/bind/test.toml
@@ -1,2 +1,4 @@
-# Bind operations are not yet supported by the Deployment Metadata Service (DMS)
+# These tests cover the file-based bind, which stages a change for the next deploy. Under the
+# deployment metadata service (DMS) bind is a distinct operation - it plans and applies the change
+# immediately - so it cannot share these goldens and is covered by bundle/dms/bind instead.
 EnvMatrix.DMS = [""]
diff --git a/acceptance/bundle/dms/bind-unbind-unsupported/output.txt b/acceptance/bundle/dms/bind-unbind-unsupported/output.txt
deleted file mode 100644
index 14686850b03..00000000000
--- a/acceptance/bundle/dms/bind-unbind-unsupported/output.txt
+++ /dev/null
@@ -1,16 +0,0 @@
-
-=== bind is refused while recording deployment history
->>> musterr [CLI] bundle deployment bind foo [JOB_ID]
-Error: bind is not supported for a bundle target that records deployment history
-
-
-=== Deploy so the bundle records state, then unbind is refused too
->>> [CLI] bundle deploy
-Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-bind-unbind-[UNIQUE_NAME]/default/files...
-Created jobs.foo
-Files: 6 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
-
->>> musterr [CLI] bundle deployment unbind foo
-Error: unbind is not supported for a bundle target that records deployment history
-
diff --git a/acceptance/bundle/dms/bind-unbind-unsupported/script b/acceptance/bundle/dms/bind-unbind-unsupported/script
deleted file mode 100644
index 9af766ca4d3..00000000000
--- a/acceptance/bundle/dms/bind-unbind-unsupported/script
+++ /dev/null
@@ -1,16 +0,0 @@
-envsubst < databricks.yml.tmpl > databricks.yml
-
-# The gate that refuses bind while recording deployment history is on the config, so it fires
-# before any deploy. Create a real workspace job so bind gets past fetching the resource and
-# reaches the gate, instead of failing because the resource does not exist.
-title "bind is refused while recording deployment history"
-job_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
-add_repl "$job_id" JOB_ID
-trace musterr $CLI bundle deployment bind foo "$job_id"
-
-# unbind refuses when it opens an existing recording state, so it needs a deployment first.
-title "Deploy so the bundle records state, then unbind is refused too"
-trace $CLI bundle deploy
-trace musterr $CLI bundle deployment unbind foo
-
-rm -f out.requests.txt
diff --git a/acceptance/bundle/dms/bind-unbind-unsupported/test.toml b/acceptance/bundle/dms/bind-unbind-unsupported/test.toml
deleted file mode 100644
index 8bc2d5e8cf8..00000000000
--- a/acceptance/bundle/dms/bind-unbind-unsupported/test.toml
+++ /dev/null
@@ -1,3 +0,0 @@
-# bind/unbind refusal is a client-side gate on the recording config; it needs no real
-# workspace, so this stays local-only rather than inheriting the suite's cloud run.
-Cloud = false
diff --git a/acceptance/bundle/dms/bind/databricks.yml.tmpl b/acceptance/bundle/dms/bind/databricks.yml.tmpl
new file mode 100644
index 00000000000..e5b40a1ed15
--- /dev/null
+++ b/acceptance/bundle/dms/bind/databricks.yml.tmpl
@@ -0,0 +1,18 @@
+bundle:
+  name: dms-bind-$UNIQUE_NAME
+experimental:
+  deployment_history: true
+
+resources:
+  schemas:
+    sc:
+      name: dms_bind_$UNIQUE_NAME
+      catalog_name: main
+  jobs:
+    foo:
+      name: foo
+      permissions:
+        - group_name: users
+          level: CAN_MANAGE
+    bar:
+      name: bar
diff --git a/acceptance/bundle/dms/bind/out.test.toml b/acceptance/bundle/dms/bind/out.test.toml
new file mode 100644
index 00000000000..23c07f70dca
--- /dev/null
+++ b/acceptance/bundle/dms/bind/out.test.toml
@@ -0,0 +1,3 @@
+Cloud = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+EnvMatrix.DMS = ["true"]
diff --git a/acceptance/bundle/dms/bind/output.txt b/acceptance/bundle/dms/bind/output.txt
new file mode 100644
index 00000000000..1da486db75a
--- /dev/null
+++ b/acceptance/bundle/dms/bind/output.txt
@@ -0,0 +1,110 @@
+
+>>> [CLI] schemas create dms_bind_[UNIQUE_NAME] main
+{
+  "browse_only": false,
+  "catalog_name": "main",
+  "catalog_type": "MANAGED_CATALOG",
+  "created_at": [UNIX_TIME_MILLIS][0],
+  "created_by": "[USERNAME]",
+  "effective_predictive_optimization_flag": {
+    "inherited_from_name": "[METASTORE_NAME]",
+    "inherited_from_type": "METASTORE",
+    "value": "ENABLE"
+  },
+  "enable_predictive_optimization": "INHERIT",
+  "full_name": "main.dms_bind_[UNIQUE_NAME]",
+  "metastore_id": "[UUID]",
+  "name": "dms_bind_[UNIQUE_NAME]",
+  "owner": "[USERNAME]",
+  "schema_id": "[UUID]",
+  "updated_at": [UNIX_TIME_MILLIS][0],
+  "updated_by": "[USERNAME]"
+}
+
+=== Bind sc, the first bind: it creates the deployment. The config already matches, so this is a plain bind that records the resource without touching it
+>>> [CLI] bundle deployment bind sc main.dms_bind_[UNIQUE_NAME] --auto-approve
+Successfully bound schema with an id 'main.dms_bind_[UNIQUE_NAME]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+=== Version 1 records only sc, with the plain bind action type; the jobs are left untouched
+>>> print_requests.py --dms --method POST //versions --nostamp
+{
+  "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": "schemas.sc",
+        "action_type": "OPERATION_ACTION_TYPE_BIND"
+      }
+    ]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+
+=== Bind foo into the existing deployment: its config differs, so this is a bind-and-update that applies the config now
+>>> [CLI] bundle deployment bind foo [FOO_ID] --auto-approve
+Successfully bound job with an id '[FOO_ID]'
+Run 'bundle deploy' to deploy changes to your workspace
+
+=== Version 2 records only foo and its permissions (bar is untouched) and chains to version 1
+>>> print_requests.py --dms --method POST //versions --nostamp
+{
+  "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",
+        "action_type": "OPERATION_ACTION_TYPE_BIND_AND_UPDATE"
+      },
+      {
+        "resource_key": "jobs.foo.permissions",
+        "action_type": "OPERATION_ACTION_TYPE_CREATE"
+      }
+    ]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+
+=== sc and foo are now managed; bar is still an unmanaged create
+>>> [CLI] bundle plan
+create jobs.bar
+
+Plan: 1 to add, 0 to change, 0 to delete, 3 unchanged
+
+=== Test cleanup: the bound resources are now bundle-managed, so destroy removes them
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.foo
+  delete resources.schemas.sc
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.sc
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-bind-[UNIQUE_NAME]/default
+
+Destroy: 2 deleted
diff --git a/acceptance/bundle/dms/bind/script b/acceptance/bundle/dms/bind/script
new file mode 100644
index 00000000000..6858036c7d1
--- /dev/null
+++ b/acceptance/bundle/dms/bind/script
@@ -0,0 +1,29 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+# sc's config matches this existing schema exactly, so binding it changes nothing (a plain bind).
+trace $CLI schemas create "dms_bind_$UNIQUE_NAME" main
+# foo's config ("foo") differs from this existing job, so binding it applies an update.
+foo_id=$($CLI jobs create --json '{"name": "bind-target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
+add_repl "$foo_id" FOO_ID
+
+cleanup() {
+    title "Test cleanup: the bound resources are now bundle-managed, so destroy removes them"
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+title "Bind sc, the first bind: it creates the deployment. The config already matches, so this is a plain bind that records the resource without touching it"
+trace $CLI bundle deployment bind sc "main.dms_bind_$UNIQUE_NAME" --auto-approve
+
+title "Version 1 records only sc, with the plain bind action type; the jobs are left untouched"
+trace print_requests.py --dms --method POST //versions --nostamp
+
+title "Bind foo into the existing deployment: its config differs, so this is a bind-and-update that applies the config now"
+trace $CLI bundle deployment bind foo "$foo_id" --auto-approve
+
+title "Version 2 records only foo and its permissions (bar is untouched) and chains to version 1"
+trace print_requests.py --dms --method POST //versions --nostamp
+
+title "sc and foo are now managed; bar is still an unmanaged create"
+trace $CLI bundle plan
diff --git a/acceptance/bundle/dms/bind-unbind-unsupported/databricks.yml.tmpl b/acceptance/bundle/dms/record-op-fault/databricks.yml.tmpl
similarity index 69%
rename from acceptance/bundle/dms/bind-unbind-unsupported/databricks.yml.tmpl
rename to acceptance/bundle/dms/record-op-fault/databricks.yml.tmpl
index e60f4605610..235760efa24 100644
--- a/acceptance/bundle/dms/bind-unbind-unsupported/databricks.yml.tmpl
+++ b/acceptance/bundle/dms/record-op-fault/databricks.yml.tmpl
@@ -1,5 +1,5 @@
 bundle:
-  name: dms-bind-unbind-$UNIQUE_NAME
+  name: dms-record-op-fault-$UNIQUE_NAME
 experimental:
   deployment_history: true
 
diff --git a/acceptance/bundle/dms/bind-unbind-unsupported/out.test.toml b/acceptance/bundle/dms/record-op-fault/out.test.toml
similarity index 100%
rename from acceptance/bundle/dms/bind-unbind-unsupported/out.test.toml
rename to acceptance/bundle/dms/record-op-fault/out.test.toml
diff --git a/acceptance/bundle/dms/record-op-fault/output.txt b/acceptance/bundle/dms/record-op-fault/output.txt
new file mode 100644
index 00000000000..d65e155ac1f
--- /dev/null
+++ b/acceptance/bundle/dms/record-op-fault/output.txt
@@ -0,0 +1,39 @@
+
+=== The service rejects the operation upload, so recording the bind fails
+>>> fault.py PATCH /api/2.0/bundle/* 500 0 1
+
+>>> musterr [CLI] bundle deployment bind foo [JOB_ID] --auto-approve
+Error: recording operation for resources.jobs.foo: Fault injected by test. (500 INJECTED)
+
+Endpoint: PATCH [DATABRICKS_URL]/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo?update_mask=state%2Cerror_message%2Cresource_id%2Cstatus
+HTTP Status: 500 Internal Server Error
+API error_code: INJECTED
+API message: Fault injected by test.
+
+
+=== The version is still completed - with failure - rather than left holding its lease
+>>> print_requests.py --dms --method POST //versions
+{
+  "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_BIND_AND_UPDATE"
+      }
+    ]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_FAILURE"
+  }
+}
diff --git a/acceptance/bundle/dms/record-op-fault/script b/acceptance/bundle/dms/record-op-fault/script
new file mode 100644
index 00000000000..fdc8afc00bf
--- /dev/null
+++ b/acceptance/bundle/dms/record-op-fault/script
@@ -0,0 +1,11 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+job_id=$($CLI jobs create --json '{"name": "target-'$UNIQUE_NAME'"}' | jq -r '.job_id')
+add_repl "$job_id" JOB_ID
+
+title "The service rejects the operation upload, so recording the bind fails"
+trace fault.py "PATCH /api/2.0/bundle/*" 500 0 1
+trace musterr $CLI bundle deployment bind foo "$job_id" --auto-approve
+
+title "The version is still completed - with failure - rather than left holding its lease"
+trace print_requests.py --dms --method POST //versions
diff --git a/acceptance/bundle/dms/record-op-fault/test.toml b/acceptance/bundle/dms/record-op-fault/test.toml
new file mode 100644
index 00000000000..24f1970a8f2
--- /dev/null
+++ b/acceptance/bundle/dms/record-op-fault/test.toml
@@ -0,0 +1,3 @@
+# Local only: the case needs an injected failure (DMS must reject the operation upload once), and
+# fault.py registers its rule on the fake, which a cloud run does not have.
+Cloud = false
diff --git a/bundle/deployplan/action.go b/bundle/deployplan/action.go
index e855dbb2197..b1ede32887d 100644
--- a/bundle/deployplan/action.go
+++ b/bundle/deployplan/action.go
@@ -31,25 +31,32 @@ type ActionType string
 // If case of several options, action with highest severity wins.
 // Note, Create/Delete are handled explicitly and never compared.
 const (
-	Undefined    ActionType = ""
-	Skip         ActionType = "skip"
-	Resize       ActionType = "resize"
-	Update       ActionType = "update"
-	UpdateWithID ActionType = "update_id"
-	Create       ActionType = "create"
-	Recreate     ActionType = "recreate"
-	Delete       ActionType = "delete"
+	Undefined ActionType = ""
+	Skip      ActionType = "skip"
+	Resize    ActionType = "resize"
+	Update    ActionType = "update"
+	// Bind adopts an existing workspace resource whose config already matches it: record it in
+	// state without touching the resource. BindAndUpdate adopts it and applies the config in one
+	// step. Both are set only for the resource named by `bundle deployment bind`.
+	Bind          ActionType = "bind"
+	BindAndUpdate ActionType = "bind_and_update"
+	UpdateWithID  ActionType = "update_id"
+	Create        ActionType = "create"
+	Recreate      ActionType = "recreate"
+	Delete        ActionType = "delete"
 )
 
 var actionOrder = map[ActionType]int{
-	Undefined:    0,
-	Skip:         1,
-	Resize:       2,
-	Update:       3,
-	UpdateWithID: 4,
-	Create:       5,
-	Recreate:     6,
-	Delete:       7,
+	Undefined:     0,
+	Skip:          1,
+	Bind:          2,
+	Resize:        3,
+	Update:        4,
+	BindAndUpdate: 5,
+	UpdateWithID:  6,
+	Create:        7,
+	Recreate:      8,
+	Delete:        9,
 }
 
 func (a ActionType) KeepsID() bool {
diff --git a/bundle/deployplan/plan.go b/bundle/deployplan/plan.go
index 67236f74766..23d5e7b75b8 100644
--- a/bundle/deployplan/plan.go
+++ b/bundle/deployplan/plan.go
@@ -54,7 +54,7 @@ func (p *Plan) CountActions() ActionCounts {
 		switch entry.Action {
 		case Create:
 			c.Create++
-		case Update, UpdateWithID, Resize:
+		case Update, UpdateWithID, Resize, BindAndUpdate:
 			c.Change++
 		case Delete:
 			c.Delete++
@@ -62,7 +62,8 @@ func (p *Plan) CountActions() ActionCounts {
 			// A recreate counts as both a delete and a create.
 			c.Delete++
 			c.Create++
-		case Skip, Undefined:
+		case Skip, Undefined, Bind:
+			// Bind adopts an existing resource without changing it.
 			c.Unchanged++
 		}
 	}
@@ -116,6 +117,8 @@ func LoadPlanFromFile(path string) (*Plan, error) {
 }
 
 type PlanEntry struct {
+	// ID carries the workspace id for a Bind/BindAndUpdate action, whose resource is not yet in
+	// state; every other action reads the id from state instead, so it is empty for them.
 	ID        string           `json:"id,omitempty"`
 	DependsOn []DependsOnEntry `json:"depends_on,omitempty"`
 	Action    ActionType       `json:"action,omitempty"`
diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go
index 21822cd0fb3..fe9c2c53934 100644
--- a/bundle/direct/apply.go
+++ b/bundle/direct/apply.go
@@ -35,8 +35,18 @@ func (d *DeploymentUnit) Destroy(ctx context.Context, db *dstate.DeploymentState
 func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType, planEntry *deployplan.PlanEntry) error {
 	ctx = log.WithPrefix(ctx, "deploying "+d.ResourceKey)
 	ctx = d.withResourceKey(ctx)
-	if actionType == deployplan.Create {
+
+	// Bind adopts an existing resource, so its id comes from the plan entry rather than state.
+	switch actionType {
+	case deployplan.Create:
 		return d.Create(ctx, db, newState)
+	case deployplan.Bind:
+		return d.Bind(ctx, db, planEntry.ID, newState, planEntry)
+	case deployplan.BindAndUpdate:
+		// Adopt and apply the config in one step; the update is the same as any other.
+		return d.Update(ctx, db, planEntry.ID, newState, planEntry)
+	default:
+		// The remaining actions act on a resource already in state; handled below.
 	}
 
 	oldID := db.GetResourceID(d.ResourceKey)
@@ -58,6 +68,15 @@ func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState,
 	}
 }
 
+// Bind adopts an existing workspace resource: it records the id and config as state without any
+// API write, since the config already matches the resource. The remote etag is copied in so
+// etag-based drift detection (dashboards, genie_spaces) still works on the next plan.
+func (d *DeploymentUnit) Bind(ctx context.Context, db *dstate.DeploymentState, id string, newState any, planEntry *deployplan.PlanEntry) error {
+	copyRemoteEtag(d.ResourceKey, planEntry.RemoteState, newState)
+	log.Infof(ctx, "Bound %s id=%#v", d.ResourceKey, id)
+	return d.saveState(ctx, db, id, newState, d.DependsOn)
+}
+
 // Create creates the resource and records its state.
 func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState, newState any) error {
 	var newID string
diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go
index 115673db731..d2e86005ce7 100644
--- a/bundle/direct/bind.go
+++ b/bundle/direct/bind.go
@@ -35,13 +35,9 @@ func (e ErrResourceAlreadyBound) Error() string {
 		e.ResourceKey, e.ExistingID)
 }
 
-// BindResult contains the result of a bind operation including any detected changes.
+// BindResult contains the result of a bind operation.
 type BindResult struct {
-	// HasChanges is true if deploying after bind would make changes to the resource
-	HasChanges bool
-	// Action is the planned action for the bound resource (e.g., "skip", "update", "recreate")
-	Action deployplan.ActionType
-	// Plan contains the full deployment plan for the bound resource
+	// Plan contains the full deployment plan; confirmBindPlan reads the bound resource's action.
 	Plan *deployplan.Plan
 	// TempStatePath is the path to the temporary state file
 	TempStatePath string
@@ -49,6 +45,25 @@ type BindResult struct {
 	StatePath string
 }
 
+// copyRemoteEtag copies the remote etag into newState for resources that use etag-based drift
+// detection (dashboards, genie_spaces). The etag comes from remote, not the user; without it the
+// next plan reports a bogus update. A no-op for other resources or when there is no remote state.
+func copyRemoteEtag(resourceKey string, remoteState, newState any) {
+	if remoteState == nil {
+		return
+	}
+	if !strings.Contains(resourceKey, ".dashboards.") && !strings.Contains(resourceKey, ".genie_spaces.") {
+		return
+	}
+	etag, err := structaccess.Get(remoteState, structpath.NewStringKey(nil, "etag"))
+	if err != nil || etag == nil {
+		return
+	}
+	if etagStr, ok := etag.(string); ok && etagStr != "" {
+		_ = structaccess.Set(newState, structpath.NewStringKey(nil, "etag"), etagStr)
+	}
+}
+
 // Bind adds an existing workspace resource to a temporary state and calculates
 // if there will be any changes when deploying.
 //
@@ -135,18 +150,11 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 			dependsOn = entry.DependsOn
 		}
 
-		// Copy etag from remote state for resources that use etag-based drift
-		// detection (dashboards and genie spaces). The etag is not provided by the
-		// user; it comes from remote. If we don't store it in state, we won't
-		// detect remote drift correctly and the next plan shows a bogus update.
-		if (strings.Contains(resourceKey, ".dashboards.") || strings.Contains(resourceKey, ".genie_spaces.")) && entry != nil && entry.RemoteState != nil {
-			etag, err := structaccess.Get(entry.RemoteState, structpath.NewStringKey(nil, "etag"))
-			if err == nil && etag != nil {
-				if etagStr, ok := etag.(string); ok && etagStr != "" {
-					_ = structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etagStr)
-				}
-			}
+		var remoteState any
+		if entry != nil {
+			remoteState = entry.RemoteState
 		}
+		copyRemoteEtag(resourceKey, remoteState, sv.Value)
 
 		err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true), dstate.WithDeploymentHistory(false), dstate.OpenDmsArgs{})
 		if err != nil {
@@ -182,22 +190,11 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 		return nil, err
 	}
 
-	// Check if the bound resource has changes
-	result := &BindResult{
-		HasChanges:    false,
-		Action:        deployplan.Skip,
+	return &BindResult{
 		Plan:          plan,
 		TempStatePath: tmpStatePath,
 		StatePath:     statePath,
-	}
-
-	entry = plan.Plan[resourceKey]
-	if entry != nil {
-		result.Action = entry.Action
-		result.HasChanges = result.Action != deployplan.Skip && result.Action != deployplan.Undefined
-	}
-
-	return result, nil
+	}, nil
 }
 
 // Finalize completes the bind operation by renaming the temp state to the final location.
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 48d76abe97f..56228358d80 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -258,32 +258,47 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 		}
 
 		dbentry, hasEntry := b.StateDB.GetResourceEntry(resourceKey)
+
+		// adopt is `bundle deployment bind`: take over the existing workspace resource b.BindID
+		// rather than creating a new one. It has no prior state, so the id and baseline below come
+		// from the bind request and config instead of the state file.
+		adopt := b.BindKey == resourceKey
+
 		// Tolerate empty-ID entries from older partial-recreate failures
 		// (apply.Recreate now deletes state on the way through, but pre-fix
 		// state files may still carry a malformed entry). Treat as missing
 		// and let the resource be re-created on this plan.
-		if !hasEntry || dbentry.ID == "" {
+		if !adopt && (!hasEntry || dbentry.ID == "") {
 			entry.Action = deployplan.Create
 			return true
 		}
 
-		savedState, err := parseState(adapter.StateType(), dbentry.State)
-		if err != nil {
-			logdiag.LogError(ctx, fmt.Errorf("%s: interpreting state: %w", errorPrefix, err))
+		sv, ok := b.StateCache.Load(resourceKey)
+		if !ok {
+			logdiag.LogError(ctx, fmt.Errorf("%s: internal error: no state cache entry found for %q", errorPrefix, resourceKey))
 			return false
 		}
 
+		resourceID := dbentry.ID
+		var savedState any
+		if adopt {
+			resourceID = b.BindID
+			// No prior state, so diff config against itself: only remote drift shows below.
+			savedState = sv.Value
+		} else {
+			savedState, err = parseState(adapter.StateType(), dbentry.State)
+			if err != nil {
+				logdiag.LogError(ctx, fmt.Errorf("%s: interpreting state: %w", errorPrefix, err))
+				return false
+			}
+		}
+
 		// Note, currently we're diffing static structs, not dynamic value.
 		// This means for fields that contain references like ${resources.group.foo.id} we do one of the following:
 		// for strings: comparing unresolved string like "${resoures.group.foo.id}" with actual object id. As long as IDs do not have ${...} format we're good.
 		// for integers: compare 0 with actual object ID. As long as real object IDs are never 0 we're good.
 		// Once we add non-id fields or add per-field details to "bundle plan", we must read dynamic data and deal with references as first class citizen.
 		// This means distinguishing between 0 that are actually object ids and 0 that are there because typed struct integer cannot contain ${...} string.
-		sv, ok := b.StateCache.Load(resourceKey)
-		if !ok {
-			logdiag.LogError(ctx, fmt.Errorf("%s: internal error: no state cache entry found for %q", errorPrefix, resourceKey))
-			return false
-		}
 		localDiff, err := structdiff.GetStructDiff(savedState, sv.Value, adapter.KeyedSlices())
 		if err != nil {
 			logdiag.LogError(ctx, fmt.Errorf("%s: diffing local state: %w", errorPrefix, err))
@@ -291,13 +306,13 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 		}
 
 		remoteState, err := retryOnTransient(ctx, func() (any, error) {
-			return adapter.DoRead(ctx, dbentry.ID)
+			return adapter.DoRead(ctx, resourceID)
 		})
 		if err != nil {
 			if apierr.IsMissing(err) {
 				remoteState = nil
 			} else {
-				logdiag.LogError(ctx, fmt.Errorf("%s: reading id=%q: %w", errorPrefix, dbentry.ID, err))
+				logdiag.LogError(ctx, fmt.Errorf("%s: reading id=%q: %w", errorPrefix, resourceID, err))
 				return false
 			}
 		}
@@ -313,7 +328,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 		if remoteState != nil {
 			remoteStateComparable, err = adapter.RemapState(remoteState)
 			if err != nil {
-				logdiag.LogError(ctx, fmt.Errorf("%s: interpreting remote state id=%q: %w", errorPrefix, dbentry.ID, err))
+				logdiag.LogError(ctx, fmt.Errorf("%s: interpreting remote state id=%q: %w", errorPrefix, resourceID, err))
 				return false
 			}
 
@@ -336,11 +351,32 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 			return false
 		}
 
-		if remoteState == nil {
+		switch {
+		case remoteState == nil && adopt:
+			logdiag.LogError(ctx, fmt.Errorf("%s: cannot bind to id=%q: resource not found", errorPrefix, resourceID))
+			return false
+		case remoteState == nil:
 			// Even if local action is "recreate" which is higher than "create", we should still pick "create" here
 			// because we know remote does not exist.
 			action = deployplan.Create
-		} else {
+		case adopt:
+			// Adopting an existing resource: no change is a plain bind, an in-place update is a
+			// bind-and-update that applies the config in the same step. A heavier change
+			// (recreate/resize) cannot be applied by adopting, so reject it rather than silently
+			// downgrading to an update and skipping the destructive-change confirmation a deploy
+			// would show.
+			switch maxAction := getMaxAction(entry.Changes); maxAction {
+			case deployplan.Skip:
+				action = deployplan.Bind
+			case deployplan.Update:
+				action = deployplan.BindAndUpdate
+			default:
+				logdiag.LogError(ctx, fmt.Errorf("%s: cannot bind id=%q: the config differs from the resource in a field that requires %s, which bind does not apply; align the config with the existing resource first", errorPrefix, resourceID, maxAction))
+				return false
+			}
+			// The id is not in state yet, so carry it to apply on the plan entry.
+			entry.ID = resourceID
+		default:
 			action = getMaxAction(entry.Changes)
 		}
 
@@ -349,7 +385,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 		b.RemoteStateCache.Store(resourceKey, remoteState)
 
 		// Validate that resources without DoUpdate don't have update actions
-		if action == deployplan.Update && !adapter.HasDoUpdate() {
+		if (action == deployplan.Update || action == deployplan.BindAndUpdate) && !adapter.HasDoUpdate() {
 			logdiag.LogError(ctx, fmt.Errorf("%s: resource does not support update action but plan produced update", errorPrefix))
 			return false
 		}
@@ -808,6 +844,11 @@ func (b *DeploymentBundle) LookupReferencePreDeploy(ctx context.Context, path *s
 	if fieldPathS == "id" {
 		if targetAction.KeepsID() {
 			id := b.StateDB.GetResourceID(targetResourceKey)
+			if id == "" {
+				// A resource being adopted (bundle deployment bind) is not in state yet; its id is
+				// carried on the plan entry instead, so a sub-resource can resolve it here.
+				id = targetEntry.ID
+			}
 			if id == "" {
 				return nil, errors.New("internal error: no db entry")
 			}
diff --git a/bundle/direct/pkg.go b/bundle/direct/pkg.go
index ea7ae1e51b6..98bffde2494 100644
--- a/bundle/direct/pkg.go
+++ b/bundle/direct/pkg.go
@@ -50,6 +50,12 @@ type DeploymentBundle struct {
 	Plan             *deployplan.Plan
 	RemoteStateCache sync.Map
 	StateCache       structvar.Cache
+
+	// BindKey and BindID, when BindKey is set, make CalculatePlan adopt the existing workspace
+	// resource BindID as resource BindKey - planned as Bind (config already matches) or
+	// BindAndUpdate (config differs) instead of Create. Set only by `bundle deployment bind`.
+	BindKey string
+	BindID  string
 }
 
 // SetRemoteState updates the remote state with type validation and marks as fresh.
diff --git a/bundle/phases/bind.go b/bundle/phases/bind.go
index 8fb5967196c..21d817a4507 100644
--- a/bundle/phases/bind.go
+++ b/bundle/phases/bind.go
@@ -33,11 +33,6 @@ func Bind(ctx context.Context, b *bundle.Bundle, opts *terraform.BindOptions, en
 	}()
 
 	if engine.IsDirect() {
-		if b.ConfiguresDeploymentHistory(ctx) {
-			logdiag.LogError(ctx, errors.New("bind is not supported for a bundle target that records deployment history"))
-			return
-		}
-
 		// Direct engine: import into temp state, run plan, check for changes
 		// This follows the same pattern as terraform import
 		groupName, ok := terraform.TerraformToGroupName[opts.ResourceType]
@@ -45,57 +40,33 @@ func Bind(ctx context.Context, b *bundle.Bundle, opts *terraform.BindOptions, en
 			groupName = opts.ResourceType
 		}
 		resourceKey := fmt.Sprintf("resources.%s.%s", groupName, opts.ResourceKey)
-		_, statePath := b.StateFilenameDirect(ctx)
 
-		result, err := b.DeploymentBundle.Bind(ctx, b.WorkspaceClient(ctx), &b.Config, statePath, resourceKey, opts.ResourceId)
-		if err != nil {
-			logdiag.LogError(ctx, err)
-			return
-		}
-
-		// If there are changes and auto-approve is not set, show plan and ask for confirmation
-		if result.HasChanges && !opts.AutoApprove {
-			// Display the planned changes for the bound resource
-			cmdio.LogString(ctx, fmt.Sprintf("Plan: %s %s", result.Action, resourceKey))
-
-			// Show details of what will change
-			if result.Plan != nil {
-				if entry, ok := result.Plan.Plan[resourceKey]; ok && entry != nil && len(entry.Changes) > 0 {
-					cmdio.LogString(ctx, "\nChanges detected:")
-					for _, field := range slices.Sorted(maps.Keys(entry.Changes)) {
-						change := entry.Changes[field]
-						if change.Action != deployplan.Skip {
-							cmdio.LogString(ctx, fmt.Sprintf("  ~ %s: %v -> %v", field, jsonDump(ctx, change.Remote, field), jsonDump(ctx, change.New, field)))
-						}
-					}
-					cmdio.LogString(ctx, "")
-				}
-			}
-
-			if !cmdio.IsPromptSupported(ctx) {
-				result.Cancel()
-				logdiag.LogError(ctx, fmt.Errorf("this bind operation requires user confirmation, but the current console does not support prompting.\nTo proceed, use --auto-approve after reviewing the plan above.%s", agent.AgentNotice()))
+		if b.ConfiguresDeploymentHistory(ctx) {
+			// A recorded deployment keeps its resources in the metadata service, so the bind is
+			// recorded there rather than written to the state file.
+			bindWithHistory(ctx, b, resourceKey, opts.ResourceId, opts.AutoApprove)
+			if logdiag.HasError(ctx) {
 				return
 			}
+		} else {
+			_, statePath := b.StateFilenameDirect(ctx)
 
-			ans, err := cmdio.AskYesOrNo(ctx, "Confirm import changes? Changes will be remotely applied only after running 'bundle deploy'.")
+			result, err := b.DeploymentBundle.Bind(ctx, b.WorkspaceClient(ctx), &b.Config, statePath, resourceKey, opts.ResourceId)
 			if err != nil {
-				result.Cancel()
 				logdiag.LogError(ctx, err)
 				return
 			}
-			if !ans {
+
+			if !confirmBindPlan(ctx, resourceKey, result.Plan, opts.AutoApprove, false) {
 				result.Cancel()
-				logdiag.LogError(ctx, errors.New("import aborted"))
 				return
 			}
-		}
 
-		// Finalize: rename temp state to final location
-		err = result.Finalize()
-		if err != nil {
-			logdiag.LogError(ctx, err)
-			return
+			// Finalize: rename temp state to final location
+			if err := result.Finalize(); err != nil {
+				logdiag.LogError(ctx, err)
+				return
+			}
 		}
 	} else {
 		// Terraform engine: use terraform import
@@ -122,6 +93,56 @@ func jsonDump(ctx context.Context, v any, field string) string {
 	return string(b)
 }
 
+// confirmBindPlan shows the bound resource's planned action and, unless autoApprove, asks the user
+// to confirm. It reports whether the bind should proceed; on decline or an unpromptable console it
+// logs the reason and returns false. A plain bind or skip changes nothing, so it proceeds without a
+// prompt. immediate is true when the caller applies the change now (the DMS path) rather than
+// deferring it to the next deploy, so the prompt describes the right timing. The caller owns any
+// cleanup on a false return.
+func confirmBindPlan(ctx context.Context, resourceKey string, plan *deployplan.Plan, autoApprove, immediate bool) bool {
+	var entry *deployplan.PlanEntry
+	if plan != nil {
+		entry = plan.Plan[resourceKey]
+	}
+	changesWorkspace := entry != nil && entry.Action != deployplan.Skip && entry.Action != deployplan.Bind && entry.Action != deployplan.Undefined
+	if !changesWorkspace || autoApprove {
+		return true
+	}
+
+	cmdio.LogString(ctx, fmt.Sprintf("Plan: %s %s", entry.Action, resourceKey))
+	if len(entry.Changes) > 0 {
+		cmdio.LogString(ctx, "\nChanges detected:")
+		for _, field := range slices.Sorted(maps.Keys(entry.Changes)) {
+			change := entry.Changes[field]
+			if change.Action != deployplan.Skip {
+				cmdio.LogString(ctx, fmt.Sprintf("  ~ %s: %v -> %v", field, jsonDump(ctx, change.Remote, field), jsonDump(ctx, change.New, field)))
+			}
+		}
+		cmdio.LogString(ctx, "")
+	}
+
+	if !cmdio.IsPromptSupported(ctx) {
+		logdiag.LogError(ctx, fmt.Errorf("this bind operation requires user confirmation, but the current console does not support prompting.\nTo proceed, use --auto-approve after reviewing the plan above.%s", agent.AgentNotice()))
+		return false
+	}
+
+	prompt := "Confirm import changes? Changes will be remotely applied only after running 'bundle deploy'."
+	if immediate {
+		prompt = "Confirm bind? The change will be applied to the workspace now."
+	}
+	ans, err := cmdio.AskYesOrNo(ctx, prompt)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return false
+	}
+	if !ans {
+		logdiag.LogError(ctx, errors.New("import aborted"))
+		return false
+	}
+
+	return true
+}
+
 func Unbind(ctx context.Context, b *bundle.Bundle, bundleType, tfResourceType, resourceKey string, engine engine.EngineType) {
 	log.Info(ctx, "Phase: unbind")
 
@@ -140,9 +161,10 @@ func Unbind(ctx context.Context, b *bundle.Bundle, bundleType, tfResourceType, r
 			groupName = tfResourceType
 		}
 		fullResourceKey := fmt.Sprintf("resources.%s.%s", groupName, resourceKey)
+		// Unbind under the deployment metadata service is not supported yet (no unbind operation
+		// action type exists); DeploymentBundle.Unbind errors for a recorded deployment.
 		_, statePath := b.StateFilenameDirect(ctx)
-		err := b.DeploymentBundle.Unbind(ctx, statePath, fullResourceKey)
-		if err != nil {
+		if err := b.DeploymentBundle.Unbind(ctx, statePath, fullResourceKey); err != nil {
 			logdiag.LogError(ctx, err)
 			return
 		}
diff --git a/bundle/phases/bind_dms.go b/bundle/phases/bind_dms.go
new file mode 100644
index 00000000000..40fd05057ab
--- /dev/null
+++ b/bundle/phases/bind_dms.go
@@ -0,0 +1,133 @@
+package phases
+
+import (
+	"context"
+	"strings"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/metadata"
+	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/databricks/cli/libs/dms"
+	"github.com/databricks/cli/libs/logdiag"
+)
+
+// bindWithHistory adopts an existing workspace resource for a deployment that records history with
+// the metadata service. The service is the source of truth, so the bind is planned and applied now
+// (as a Bind or BindAndUpdate operation in its own version), unlike the file-based bind which
+// defers the change to the next deploy.
+func bindWithHistory(ctx context.Context, b *bundle.Bundle, resourceKey, resourceID string, autoApprove bool) {
+	wsc := b.WorkspaceClient(ctx)
+
+	deploymentID, deployment, lastVersionID, err := dms.FetchDeployment(ctx, wsc, b.Config.Workspace.StatePath)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
+	ctx = withWorkspaceClient(ctx, b)
+	db := &b.DeploymentBundle.StateDB
+	if err := openRecordedState(ctx, db, localStatePath(ctx, b), deploymentID, lastVersionID); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
+	if existingID := db.GetResourceID(resourceKey); existingID != "" {
+		finalizeState(ctx, db)
+		logdiag.LogError(ctx, direct.ErrResourceAlreadyBound{ResourceKey: resourceKey, ExistingID: existingID, NewID: resourceID})
+		return
+	}
+
+	// Stamp deployment metadata into config (as a deploy does) so the recorded state matches and a
+	// later plan sees no drift. A first bind's deployment_id is stamped after it is created below.
+	firstBind := deploymentID == ""
+	muts := []bundle.Mutator{metadata.AnnotateDeploymentVersion(lastVersionID + 1)}
+	if !firstBind {
+		muts = append(muts, metadata.AnnotateDeployment(deploymentID))
+	}
+	bundle.ApplySeqContext(ctx, b, muts...)
+	if logdiag.HasError(ctx) {
+		finalizeState(ctx, db)
+		return
+	}
+
+	// Plan the resource as an adoption of the existing id, then narrow the plan to it (and its own
+	// grants/permissions) so the bind leaves the rest of the deployment untouched.
+	b.DeploymentBundle.BindKey = resourceKey
+	b.DeploymentBundle.BindID = resourceID
+	plan, err := b.DeploymentBundle.CalculatePlan(ctx, wsc, &b.Config)
+	if err != nil {
+		finalizeState(ctx, db)
+		logdiag.LogError(ctx, err)
+		return
+	}
+	plan.FilterToSelected([]string{strings.TrimPrefix(resourceKey, "resources.")})
+
+	if !confirmBindPlan(ctx, resourceKey, plan, autoApprove, true) {
+		finalizeState(ctx, db)
+		return
+	}
+
+	// Commit now: claim a version, apply the adoption, and complete it.
+	if err := db.UpgradeToWrite(); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	// From here the state is open for write and may hold a version; drain and complete it (with
+	// failure on error) on every path, so nothing is left open or a version left dangling.
+	defer completeRecordedVersion(ctx, b)
+
+	if !createDeploymentAndStamp(ctx, b, deployment, firstBind) {
+		return
+	}
+	staged, err := stagedOperations(plan)
+	if err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+	if err := startVersion(ctx, b, dms.VersionTypeDeploy, staged); err != nil {
+		logdiag.LogError(ctx, err)
+		return
+	}
+
+	b.DeploymentBundle.Apply(ctx, wsc, plan)
+}
+
+// completeRecordedVersion drains the buffered operations and closes the version out, completing
+// with failure if anything went wrong. Deferred once a version exists so every path completes it.
+func completeRecordedVersion(ctx context.Context, b *bundle.Bundle) {
+	db := &b.DeploymentBundle.StateDB
+	if _, err := db.Finalize(ctx); err != nil {
+		logdiag.LogError(ctx, err)
+	}
+	if _, err := db.CompleteVersion(ctx, !logdiag.HasError(ctx)); err != nil {
+		logdiag.LogError(ctx, err)
+	}
+}
+
+// openRecordedState opens the deployment's recorded state for read, reading its resources from the
+// metadata service.
+func openRecordedState(ctx context.Context, db *dstate.DeploymentState, path, deploymentID string, lastVersionID int) error {
+	return db.Open(ctx, path, dstate.WithRecovery(false), dstate.WithWrite(false), dstate.WithDeploymentHistory(true), dstate.OpenDmsArgs{DeploymentID: deploymentID, LastVersionID: lastVersionID})
+}
+
+// finalizeState drains and closes the state without recording a version, for the paths that open it
+// but do not commit (a no-op or a declined bind).
+func finalizeState(ctx context.Context, db *dstate.DeploymentState) {
+	if _, err := db.Finalize(ctx); err != nil {
+		logdiag.LogError(ctx, err)
+	}
+}
+
+func localStatePath(ctx context.Context, b *bundle.Bundle) string {
+	_, localPath := b.StateFilenameDirect(ctx)
+	return localPath
+}
+
+func withWorkspaceClient(ctx context.Context, b *bundle.Bundle) context.Context {
+	if !cmdctx.HasWorkspaceClient(ctx) {
+		return cmdctx.SetWorkspaceClient(ctx, b.WorkspaceClient(ctx))
+	}
+	return ctx
+}
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index 425c4874283..0d8c57db3be 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -321,16 +321,9 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 	// IsDirect first: the state must be open to read its features, and only the direct engine opens it.
 	if stateEngine.IsDirect() && b.DeploymentBundle.StateDB.IsDeploymentMetadataService() {
 		firstDeploy := b.DeploymentBundle.StateDB.DeploymentID == ""
-		createOrUpdateDeployment(ctx, b, dmsDeployment)
-		if logdiag.HasError(ctx) {
+		if !createDeploymentAndStamp(ctx, b, dmsDeployment, firstDeploy) {
 			return
 		}
-		if firstDeploy {
-			if err := b.DeploymentBundle.StampDeploymentIdForFirstVersion(b.DeploymentBundle.StateDB.DeploymentID); err != nil {
-				logdiag.LogError(ctx, err)
-				return
-			}
-		}
 
 		// Only create a version when the plan has at least one operation to record, so it moves
 		// with serial. Done after the prompt, so a declined deploy never claims a number.
diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go
index 9ceea29509d..7f162c2864c 100644
--- a/bundle/phases/dms.go
+++ b/bundle/phases/dms.go
@@ -48,6 +48,10 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType
 	switch a {
 	case deployplan.Create:
 		return bundledeployments.OperationActionTypeOperationActionTypeCreate, nil
+	case deployplan.Bind:
+		return bundledeployments.OperationActionTypeOperationActionTypeBind, nil
+	case deployplan.BindAndUpdate:
+		return bundledeployments.OperationActionTypeOperationActionTypeBindAndUpdate, nil
 	case deployplan.Update:
 		return bundledeployments.OperationActionTypeOperationActionTypeUpdate, nil
 	case deployplan.UpdateWithID:
@@ -93,6 +97,24 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu
 	})
 }
 
+// createDeploymentAndStamp runs createOrUpdateDeployment, then on a first deploy stamps the created
+// id into the plan the apply reads (the id did not exist at plan time). It reports whether it
+// succeeded; on failure it has already logged. firstDeploy is captured before the create, which
+// assigns the id. Shared by the deploy and bind phases.
+func createDeploymentAndStamp(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment, firstDeploy bool) bool {
+	createOrUpdateDeployment(ctx, b, current)
+	if logdiag.HasError(ctx) {
+		return false
+	}
+	if firstDeploy {
+		if err := b.DeploymentBundle.StampDeploymentIdForFirstVersion(b.DeploymentBundle.StateDB.DeploymentID); err != nil {
+			logdiag.LogError(ctx, err)
+			return false
+		}
+	}
+	return true
+}
+
 // 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.
diff --git a/bundle/phases/dms_test.go b/bundle/phases/dms_test.go
index f6854999db9..0528db54dce 100644
--- a/bundle/phases/dms_test.go
+++ b/bundle/phases/dms_test.go
@@ -14,10 +14,12 @@ func TestStagedOperationsCoversEveryTouchedResource(t *testing.T) {
 	// The version fixes its operation set, so anything the apply will write has to appear
 	// here. Keys go out in the service's form, without the CLI's "resources." prefix.
 	plan := &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{
-		"resources.jobs.foo":       {Action: deployplan.Create},
-		"resources.pipelines.bar":  {Action: deployplan.Recreate},
-		"resources.schemas.baz":    {Action: deployplan.Delete},
-		"resources.clusters.small": {Action: deployplan.Resize},
+		"resources.jobs.foo":         {Action: deployplan.Create},
+		"resources.pipelines.bar":    {Action: deployplan.Recreate},
+		"resources.schemas.baz":      {Action: deployplan.Delete},
+		"resources.clusters.small":   {Action: deployplan.Resize},
+		"resources.jobs.adopted":     {Action: deployplan.Bind},
+		"resources.jobs.adopted_upd": {Action: deployplan.BindAndUpdate},
 	}}
 
 	staged, err := stagedOperations(plan)
@@ -28,6 +30,8 @@ func TestStagedOperationsCoversEveryTouchedResource(t *testing.T) {
 		{ResourceKey: "resources.pipelines.bar", ActionType: bundledeployments.OperationActionTypeOperationActionTypeRecreate},
 		{ResourceKey: "resources.schemas.baz", ActionType: bundledeployments.OperationActionTypeOperationActionTypeDelete},
 		{ResourceKey: "resources.clusters.small", ActionType: bundledeployments.OperationActionTypeOperationActionTypeResize},
+		{ResourceKey: "resources.jobs.adopted", ActionType: bundledeployments.OperationActionTypeOperationActionTypeBind},
+		{ResourceKey: "resources.jobs.adopted_upd", ActionType: bundledeployments.OperationActionTypeOperationActionTypeBindAndUpdate},
 	}, staged)
 }
 
@@ -60,6 +64,8 @@ func TestActionToSDK(t *testing.T) {
 		want   bundledeployments.OperationActionType
 	}{
 		{deployplan.Create, bundledeployments.OperationActionTypeOperationActionTypeCreate},
+		{deployplan.Bind, bundledeployments.OperationActionTypeOperationActionTypeBind},
+		{deployplan.BindAndUpdate, bundledeployments.OperationActionTypeOperationActionTypeBindAndUpdate},
 		{deployplan.Update, bundledeployments.OperationActionTypeOperationActionTypeUpdate},
 		{deployplan.UpdateWithID, bundledeployments.OperationActionTypeOperationActionTypeUpdateWithId},
 		{deployplan.Recreate, bundledeployments.OperationActionTypeOperationActionTypeRecreate},
diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go
index ea29079f580..39a2ed24641 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"
@@ -31,8 +29,6 @@ import (
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/cli/libs/sync"
 	"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/spf13/cobra"
 	"golang.org/x/mod/semver"
@@ -248,7 +244,8 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 
 			if b.ConfiguresDeploymentHistory(ctx) {
 				var err error
-				dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
+				var lastVersionID int
+				dmsDeploymentID, dmsDeployment, lastVersionID, err = dms.FetchDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
 				if err != nil {
 					logdiag.LogError(ctx, err)
 					return b, stateDesc, root.ErrAlreadyPrinted
@@ -258,16 +255,6 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 				// the plan carries them. version_id is always known (last recorded + 1); deployment_id
 				// does not exist until a first deploy creates it, so it is left off here and the deploy
 				// phase stamps the created id.
-				// The service reports the version as a string; parse it here so everything below
-				// carries a number.
-				lastVersionID := 0
-				if dmsDeployment != nil && dmsDeployment.LastVersionId != "" {
-					lastVersionID, err = strconv.Atoi(dmsDeployment.LastVersionId)
-					if err != nil {
-						logdiag.LogError(ctx, fmt.Errorf("failed to parse last_version_id %q: %w", dmsDeployment.LastVersionId, err))
-						return b, stateDesc, root.ErrAlreadyPrinted
-					}
-				}
 				nextVersion := lastVersionID + 1
 				muts := []bundle.Mutator{metadata.AnnotateDeploymentVersion(nextVersion)}
 				if dmsDeploymentID != "" {
@@ -484,36 +471,6 @@ func ResolveEngineSetting(ctx context.Context, b *bundle.Bundle) (engine.EngineS
 	return engine.EngineSetting{Type: engine.Default, Source: engine.SourceDefault, IsDefault: true}, 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
-}
-
 // isNewerVersion reports whether the state's recorded CLI version is strictly
 // newer than the running build. Both are bare versions without a leading "v".
 // An empty stateVersion (state not written by any CLI yet) or an unparseable
diff --git a/libs/dms/client.go b/libs/dms/client.go
index 605b4bbd4db..57e033fb9b4 100644
--- a/libs/dms/client.go
+++ b/libs/dms/client.go
@@ -2,13 +2,16 @@ package dms
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
+	"path"
 	"strconv"
 	"strings"
 
 	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/client"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
@@ -67,6 +70,47 @@ func (c *Client) CreateDeployment(ctx context.Context, parentPath string, metada
 	return deploymentIDFromName(created.Name)
 }
 
+// FetchDeployment looks up the deployment recorded under statePath, returning its id, record, and
+// last recorded version number. It returns "", nil, 0, nil when none exists yet: the deployment
+// node is the workspace node the service creates on the first recorded deploy, and its id is that
+// node's object id.
+//
+// TODO: a deployment is usable only when both the node and the service's record exist; a
+// half-created one (node present, record missing) errors here even though CreateDeployment can
+// recover from it. A dms.ReadDeployment returning empty unless both halves are present would let
+// the caller create or finalize it instead.
+//
+// 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 FetchDeployment(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, int, 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, 0, nil
+	}
+	if err != nil {
+		return "", nil, 0, 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, 0, err
+	}
+
+	lastVersionID := 0
+	if deployment.LastVersionId != "" {
+		lastVersionID, err = strconv.Atoi(deployment.LastVersionId)
+		if err != nil {
+			return "", nil, 0, fmt.Errorf("failed to parse last_version_id %q: %w", deployment.LastVersionId, err)
+		}
+	}
+	return deploymentID, deployment, lastVersionID, nil
+}
+
 // 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 {