From 713af9eba21499c6f1725c740c27d5dd6ce30776 Mon Sep 17 00:00:00 2001
From: Jan Rose <jan.rose@databricks.com>
Date: Wed, 6 May 2026 17:59:35 +0200
Subject: [PATCH 1/5] acceptance: add drift test for model_serving_endpoints
 recreated with same name
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

After deleting and recreating a model serving endpoint remotely with the
same name but a different endpoint_id, V1 permissions API behavior
results in permanent drift on permissions: bundle plan keeps showing an
update for permissions even after a successful deploy. The V1 endpoints
do not delete ACLs immediately when the parent is gone, so DoRead from
the old object_id keeps returning ACL data, while plan computes the new
object_id from the recreated endpoint.

This is the V1 counterpart to vector_search_endpoints/drift/recreated_same_name,
which exercises V2 behavior (404 → create plan, no drift after deploy).

Co-authored-by: Isaac
---
 .../recreated_same_name/databricks.yml.tmpl   | 13 ++++
 .../drift/recreated_same_name/out.test.toml   |  4 ++
 .../drift/recreated_same_name/output.txt      | 63 +++++++++++++++++++
 .../drift/recreated_same_name/script          | 41 ++++++++++++
 .../drift/recreated_same_name/test.toml       |  8 +++
 5 files changed, 129 insertions(+)
 create mode 100644 acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/databricks.yml.tmpl
 create mode 100644 acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/out.test.toml
 create mode 100644 acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
 create mode 100644 acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
 create mode 100644 acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml

diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/databricks.yml.tmpl b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/databricks.yml.tmpl
new file mode 100644
index 00000000000..8dcc1302127
--- /dev/null
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/databricks.yml.tmpl
@@ -0,0 +1,13 @@
+bundle:
+  name: drift-mse-recreated-same-name-$UNIQUE_NAME
+
+sync:
+  paths: []
+
+resources:
+  model_serving_endpoints:
+    my_endpoint:
+      name: mse-endpoint-$UNIQUE_NAME
+      permissions:
+        - level: CAN_VIEW
+          user_name: deco-test-user@databricks.com
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/out.test.toml b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/out.test.toml
new file mode 100644
index 00000000000..fe4076cdf9b
--- /dev/null
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/out.test.toml
@@ -0,0 +1,4 @@
+Local = true
+Cloud = true
+RequiresUnityCatalog = true
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
new file mode 100644
index 00000000000..fea354c1f8a
--- /dev/null
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
@@ -0,0 +1,63 @@
+
+=== Initial deployment
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/drift-mse-recreated-same-name-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] serving-endpoints get mse-endpoint-[UNIQUE_NAME]
+{
+  "name": "mse-endpoint-[UNIQUE_NAME]",
+  "creator": "[USERNAME]"
+}
+
+=== Delete and recreate remotely with the same name
+>>> [CLI] serving-endpoints delete mse-endpoint-[UNIQUE_NAME]
+
+>>> [CLI] serving-endpoints create mse-endpoint-[UNIQUE_NAME] --no-wait
+{
+  "name": "mse-endpoint-[UNIQUE_NAME]",
+  "creator": "[USERNAME]"
+}
+
+>>> [CLI] serving-endpoints get mse-endpoint-[UNIQUE_NAME]
+{
+  "name": "mse-endpoint-[UNIQUE_NAME]",
+  "creator": "[USERNAME]"
+}
+Original endpoint id: [ORIGINAL_ENDPOINT_ID]
+Remote recreated endpoint id: [REMOTE_RECREATED_ENDPOINT_ID]
+
+=== Plan after out-of-band recreate
+>>> [CLI] bundle plan
+update model_serving_endpoints.my_endpoint.permissions
+
+Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/drift-mse-recreated-same-name-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] serving-endpoints get mse-endpoint-[UNIQUE_NAME]
+{
+  "name": "mse-endpoint-[UNIQUE_NAME]",
+  "creator": "[USERNAME]"
+}
+
+=== Persistent drift after deploy: V1 permissions API leaves an ACL on the deleted endpoint id, so plan keeps showing an update.
+>>> [CLI] bundle plan
+update model_serving_endpoints.my_endpoint.permissions
+
+Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.model_serving_endpoints.my_endpoint
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/drift-mse-recreated-same-name-[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
new file mode 100644
index 00000000000..9f764d7914e
--- /dev/null
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
@@ -0,0 +1,41 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+endpoint_name="mse-endpoint-${UNIQUE_NAME}"
+
+title "Initial deployment"
+trace $CLI bundle deploy
+
+original_endpoint_id=$($CLI serving-endpoints get "${endpoint_name}" | jq -r '.id')
+add_repl.py "$original_endpoint_id" "ORIGINAL_ENDPOINT_ID"
+trace $CLI serving-endpoints get "${endpoint_name}" | jq '{name, creator}'
+
+title "Delete and recreate remotely with the same name"
+trace $CLI serving-endpoints delete "${endpoint_name}"
+trace $CLI serving-endpoints create "${endpoint_name}" --no-wait | jq '{name, creator}'
+
+remote_recreated_endpoint_id=$($CLI serving-endpoints get "${endpoint_name}" | jq -r '.id')
+add_repl.py "$remote_recreated_endpoint_id" "REMOTE_RECREATED_ENDPOINT_ID"
+trace $CLI serving-endpoints get "${endpoint_name}" | jq '{name, creator}'
+
+printf "Original endpoint id: %s\n" "$original_endpoint_id"
+printf "Remote recreated endpoint id: %s\n" "$remote_recreated_endpoint_id"
+
+if [ "$original_endpoint_id" = "$remote_recreated_endpoint_id" ]; then
+    echo "Expected remote recreation to assign a different endpoint id" >&2
+    exit 1
+fi
+
+title "Plan after out-of-band recreate"
+trace $CLI bundle plan
+
+trace $CLI bundle deploy
+trace $CLI serving-endpoints get "${endpoint_name}" | jq '{name, creator}'
+
+title "Persistent drift after deploy: V1 permissions API leaves an ACL on the deleted endpoint id, so plan keeps showing an update."
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml
new file mode 100644
index 00000000000..83e36142b53
--- /dev/null
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml
@@ -0,0 +1,8 @@
+Badness = "After deleting and recreating a model serving endpoint remotely with the same name but a different endpoint_id, bundle plan/deploy ends up with a permanent update on permissions because the V1 permissions API does not delete ACLs immediately when the parent is gone."
+
+Local = true
+Cloud = true
+RequiresUnityCatalog = true
+RecordRequests = false
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

From b7ddd8f222ce189746825207ab48f62ec096b01d Mon Sep 17 00:00:00 2001
From: Jan Rose <jan.rose@databricks.com>
Date: Wed, 6 May 2026 18:10:33 +0200
Subject: [PATCH 2/5] direct/permissions: implement DoUpdateWithID to avoid
 permanent drift
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When the parent resource is recreated remotely with a different identifier
(e.g. a model serving endpoint deleted and recreated under the same name),
PermissionsState.ObjectID changes between deploys. Previously the framework
called DoUpdate, which kept the deployment state ID pointing at the old,
gone object_id. On V1 permissions APIs (jobs, pipelines, model serving),
DoRead with the old object_id keeps returning ACL data for the deleted
parent due to eventual consistency, so plan saw the same ObjectID drift
on every subsequent run — a permanent update on permissions.

Add DoUpdateWithID that returns newState.ObjectID as the new resource ID
so the framework persists the new ID in deployment state, and wire up
update_id_on_changes for object_id on every *.permissions resource that
uses ResourcePermissions. Subsequent plans then compare against the new
ObjectID and see no drift.

Update the model_serving_endpoints drift acceptance test to assert no
permanent drift after deploy (same shape as the vector_search V2 test).

Co-authored-by: Isaac
---
 .../drift/recreated_same_name/output.txt      |  6 +-
 .../drift/recreated_same_name/script          |  4 +-
 .../drift/recreated_same_name/test.toml       |  2 +-
 bundle/direct/dresources/permissions.go       | 14 +++++
 bundle/direct/dresources/resources.yml        | 58 +++++++++++++++++++
 5 files changed, 77 insertions(+), 7 deletions(-)

diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
index fea354c1f8a..89140651806 100644
--- a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/output.txt
@@ -47,11 +47,9 @@ Deployment complete!
   "creator": "[USERNAME]"
 }
 
-=== Persistent drift after deploy: V1 permissions API leaves an ACL on the deleted endpoint id, so plan keeps showing an update.
+=== Verify no permanent drift after deploy
 >>> [CLI] bundle plan
-update model_serving_endpoints.my_endpoint.permissions
-
-Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
index 9f764d7914e..e8fe54ca4d9 100644
--- a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/script
@@ -37,5 +37,5 @@ trace $CLI bundle plan
 trace $CLI bundle deploy
 trace $CLI serving-endpoints get "${endpoint_name}" | jq '{name, creator}'
 
-title "Persistent drift after deploy: V1 permissions API leaves an ACL on the deleted endpoint id, so plan keeps showing an update."
-trace $CLI bundle plan
+title "Verify no permanent drift after deploy"
+trace $CLI bundle plan | contains.py "Plan: 0 to add, 0 to change, 0 to delete"
diff --git a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml
index 83e36142b53..7c8ba9edfa1 100644
--- a/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml
+++ b/acceptance/bundle/resources/model_serving_endpoints/drift/recreated_same_name/test.toml
@@ -1,4 +1,4 @@
-Badness = "After deleting and recreating a model serving endpoint remotely with the same name but a different endpoint_id, bundle plan/deploy ends up with a permanent update on permissions because the V1 permissions API does not delete ACLs immediately when the parent is gone."
+Badness = "After deleting and recreating a model serving endpoint remotely with the same name but a different endpoint_id, bundle plan shows an update on permissions (instead of create as in V2) because the V1 permissions API does not delete ACLs immediately when the parent is gone. UpdateWithID persists the new object_id in deployment state so subsequent plans do not show permanent drift."
 
 Local = true
 Cloud = true
diff --git a/bundle/direct/dresources/permissions.go b/bundle/direct/dresources/permissions.go
index 91ca9000aaf..900d74b9609 100644
--- a/bundle/direct/dresources/permissions.go
+++ b/bundle/direct/dresources/permissions.go
@@ -253,6 +253,20 @@ func (r *ResourcePermissions) DoUpdate(ctx context.Context, _ string, newState *
 	return nil, err
 }
 
+// DoUpdateWithID is identical to DoUpdate but reports newState.ObjectID as the new
+// resource ID so the framework persists it in deployment state. Without this, an
+// out-of-band recreate of the parent resource leaves the deployment state pointing
+// at the gone object_id; on V1 permissions APIs that still return ACL data for the
+// old parent (eventual consistency), this manifests as a permanent update on the
+// permissions resource.
+func (r *ResourcePermissions) DoUpdateWithID(ctx context.Context, _ string, newState *PermissionsState) (string, *PermissionsState, error) {
+	_, err := r.DoUpdate(ctx, newState.ObjectID, newState, nil)
+	if err != nil {
+		return "", nil, err
+	}
+	return newState.ObjectID, nil, nil
+}
+
 // DoDelete is activated in 2 distinct cases:
 // 1) 'permissions' field is deleted in DABs config. In that case terraform would restore the default permissions (IS_OWNER for current user).
 // 2) the parent resource is deleted; in that case there is no need to do anything; parent resource deletion is enough.
diff --git a/bundle/direct/dresources/resources.yml b/bundle/direct/dresources/resources.yml
index c283e6f0b68..a3b6c1ebe83 100644
--- a/bundle/direct/dresources/resources.yml
+++ b/bundle/direct/dresources/resources.yml
@@ -404,6 +404,64 @@ resources:
       - field: scope_name
         reason: id_changes
 
+  # Permissions for the resources below use ResourcePermissions, with the deployment state ID
+  # equal to PermissionsState.ObjectID. When the parent resource is recreated remotely with a
+  # different identifier, ObjectID changes, so we trigger UpdateWithID to persist the new ID
+  # in state and avoid permanent drift on V1 permissions APIs (which still return ACL data
+  # for the old parent due to eventual consistency).
+  alerts.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  apps.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  clusters.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  dashboards.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  database_instances.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  experiments.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  jobs.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  model_serving_endpoints.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  models.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  pipelines.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  postgres_projects.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  sql_warehouses.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+  vector_search_endpoints.permissions:
+    update_id_on_changes:
+      - field: object_id
+        reason: id_changes
+
   clusters:
     ignore_remote_changes:
       # https://github.com/databricks/terraform-provider-databricks/blob/4eba541abe1a9f50993ea7b9dd83874207e224a1/clusters/resource_cluster.go#L361-L363

From 8dfaee1d7a6aeb02c227a2c05b4abea485d8ece4 Mon Sep 17 00:00:00 2001
From: Jan Rose <jan.rose@databricks.com>
Date: Thu, 7 May 2026 10:24:09 +0200
Subject: [PATCH 3/5] direct: extend DoRead to receive planned newState
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

DoRead previously took only the deployment-state id, which is stale after
an out-of-band recreate of the parent resource: the id points at the
gone object_id while the new plan has already resolved newState.ObjectID
to the freshly-created identifier. The permissions resource needs to read
against the new identifier, otherwise on V1 permissions APIs (jobs,
pipelines, model serving) DoRead keeps returning ACL data for the deleted
parent and produces a permanent drift; on V2 (vector search) it 404s
when it should be reading the new endpoint's empty ACLs.

Add a newState parameter to DoRead across all resources. Most resources
ignore it and continue to read by id. ResourcePermissions uses
newState.ObjectID when available so subsequent plans see no drift after
a parent recreate, and the test-server's new-endpoint ACL state is
correctly observed.

For the delete path in plan, where there is no planned new state, pass
nil so the adapter falls back to id-based read. For refreshRemoteState
post-deploy, thread newState (sv.Value) through from bundle_apply.

Update vector_search_endpoints/drift/recreated_same_name acceptance to
match the new behaviour: the plan now reads the freshly-created endpoint
and shows an "update" on permissions instead of "create" — both end with
no permanent drift after deploy.

Co-authored-by: Isaac
---
 .../drift/recreated_same_name/output.txt            |  4 ++--
 bundle/direct/apply.go                              |  4 ++--
 bundle/direct/bundle_apply.go                       |  5 ++++-
 bundle/direct/bundle_plan.go                        |  6 ++++--
 bundle/direct/dresources/adapter.go                 | 13 +++++++++----
 bundle/direct/dresources/alert.go                   |  2 +-
 bundle/direct/dresources/all_test.go                | 11 +++++++----
 bundle/direct/dresources/app.go                     |  2 +-
 bundle/direct/dresources/catalog.go                 |  2 +-
 bundle/direct/dresources/cluster.go                 |  2 +-
 bundle/direct/dresources/dashboard.go               |  2 +-
 bundle/direct/dresources/database_catalog.go        |  2 +-
 bundle/direct/dresources/database_instance.go       |  2 +-
 bundle/direct/dresources/experiment.go              |  2 +-
 bundle/direct/dresources/external_location.go       |  2 +-
 bundle/direct/dresources/grants.go                  |  2 +-
 bundle/direct/dresources/job.go                     |  2 +-
 bundle/direct/dresources/model.go                   |  2 +-
 bundle/direct/dresources/model_serving_endpoint.go  |  2 +-
 bundle/direct/dresources/permissions.go             | 12 +++++++++++-
 bundle/direct/dresources/pipeline.go                |  2 +-
 bundle/direct/dresources/postgres_branch.go         |  2 +-
 bundle/direct/dresources/postgres_catalog.go        |  2 +-
 bundle/direct/dresources/postgres_endpoint.go       |  2 +-
 bundle/direct/dresources/postgres_project.go        |  2 +-
 bundle/direct/dresources/postgres_synced_table.go   |  2 +-
 bundle/direct/dresources/quality_monitor.go         |  2 +-
 bundle/direct/dresources/registered_model.go        |  2 +-
 bundle/direct/dresources/schema.go                  |  2 +-
 bundle/direct/dresources/schema_test.go             |  2 +-
 bundle/direct/dresources/secret_scope.go            |  2 +-
 bundle/direct/dresources/secret_scope_acls.go       |  2 +-
 bundle/direct/dresources/sql_warehouse.go           |  2 +-
 bundle/direct/dresources/synced_database_table.go   |  2 +-
 bundle/direct/dresources/vector_search_endpoint.go  |  2 +-
 bundle/direct/dresources/vector_search_index.go     |  2 +-
 bundle/direct/dresources/volume.go                  |  2 +-
 37 files changed, 69 insertions(+), 46 deletions(-)

diff --git a/acceptance/bundle/resources/vector_search_endpoints/drift/recreated_same_name/output.txt b/acceptance/bundle/resources/vector_search_endpoints/drift/recreated_same_name/output.txt
index dece842119f..4567e3f0b4b 100644
--- a/acceptance/bundle/resources/vector_search_endpoints/drift/recreated_same_name/output.txt
+++ b/acceptance/bundle/resources/vector_search_endpoints/drift/recreated_same_name/output.txt
@@ -34,9 +34,9 @@ Remote recreated endpoint UUID: [REMOTE_RECREATED_ENDPOINT_UUID]
 
 === Plan after out-of-band recreate
 >>> [CLI] bundle plan
-create vector_search_endpoints.my_endpoint.permissions
+update vector_search_endpoints.my_endpoint.permissions
 
-Plan: 1 to add, 0 to change, 0 to delete, 1 unchanged
+Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
 
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/drift-vs-endpoint-recreated-same-name-[UNIQUE_NAME]/default/files...
diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go
index a4a61f727f2..1781674fc3c 100644
--- a/bundle/direct/apply.go
+++ b/bundle/direct/apply.go
@@ -287,12 +287,12 @@ func (d *DeploymentUnit) loadPersistedState(db *dstate.DeploymentState) (any, er
 	return state, nil
 }
 
-func (d *DeploymentUnit) refreshRemoteState(ctx context.Context, id string) error {
+func (d *DeploymentUnit) refreshRemoteState(ctx context.Context, id string, newState any) error {
 	if d.RemoteState != nil {
 		return nil
 	}
 	remoteState, err := retryOnTransient(ctx, func() (any, error) {
-		return d.Adapter.DoRead(ctx, id)
+		return d.Adapter.DoRead(ctx, id, newState)
 	})
 	if err != nil {
 		return fmt.Errorf("failed to refresh remote state id=%s: %w", id, err)
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index 16b145f7af8..98c2ff0a69c 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -96,6 +96,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 
 		// We don't keep NewState around for 'skip' nodes
 
+		var newState any
 		if action != deployplan.Skip {
 			if !b.resolveReferences(ctx, resourceKey, entry, errorPrefix, false) {
 				return false
@@ -113,6 +114,8 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				return false
 			}
 
+			newState = sv.Value
+
 			if migrateMode {
 				// In migration mode we're reading resources in DAG order so that we have fully resolved config snapshots stored
 				id := b.StateDB.GetResourceID(resourceKey)
@@ -143,7 +146,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				return false
 			}
 
-			err = d.refreshRemoteState(ctx, id)
+			err = d.refreshRemoteState(ctx, id, newState)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: failed to read remote state: %w", errorPrefix, err))
 				return false
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 5591626cd75..5ac142724c5 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -157,8 +157,10 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 				return false
 			}
 
+			// Delete branch: there is no planned new state, pass typed nil so the
+			// adapter falls back to using the deployment state id.
 			remoteState, err := retryOnTransient(ctx, func() (any, error) {
-				return adapter.DoRead(ctx, id)
+				return adapter.DoRead(ctx, id, nil)
 			})
 			if err != nil {
 				if isResourceGone(err) {
@@ -215,7 +217,7 @@ 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, dbentry.ID, sv.Value)
 		})
 		if err != nil {
 			if isResourceGone(err) {
diff --git a/bundle/direct/dresources/adapter.go b/bundle/direct/dresources/adapter.go
index 6891632b626..667326463d0 100644
--- a/bundle/direct/dresources/adapter.go
+++ b/bundle/direct/dresources/adapter.go
@@ -39,8 +39,12 @@ type IResource interface {
 	RemapState(input any) any
 
 	// DoRead reads and returns remote state from the backend. The return type defines schema for remote field resolution.
-	// Example: func (r *ResourceJob) DoRead(ctx context.Context, id string) (*jobs.Job, error)
-	DoRead(ctx context.Context, id string) (remoteState any, e error)
+	// newState is the resource's planned state for the current deploy with all references resolved;
+	// resources may use it to read against an identifier computed from the new config (e.g. the
+	// permissions resource derives object_id from the parent's freshly-resolved id) instead of the
+	// possibly-stale id stored in deployment state.
+	// Example: func (r *ResourceJob) DoRead(ctx context.Context, id string, newState *jobs.JobSettings) (*jobs.Job, error)
+	DoRead(ctx context.Context, id string, newState any) (remoteState any, e error)
 
 	// DoDelete deletes the resource. The state argument is the last-persisted
 	// state for the resource; resources that don't need it should accept it as
@@ -279,6 +283,7 @@ func (a *Adapter) validate() error {
 	validations := []any{
 		"PrepareState return", a.prepareState.OutTypes[0], stateType,
 		"DoCreate newState", a.doCreate.InTypes[1], stateType,
+		"DoRead newState", a.doRefresh.InTypes[2], stateType,
 		"DoDelete state", a.doDelete.InTypes[2], stateType,
 	}
 
@@ -407,8 +412,8 @@ func (a *Adapter) RemapState(remoteState any) (any, error) {
 	return outs[0], nil
 }
 
-func (a *Adapter) DoRead(ctx context.Context, id string) (any, error) {
-	outs, err := a.doRefresh.Call(ctx, id)
+func (a *Adapter) DoRead(ctx context.Context, id string, newState any) (any, error) {
+	outs, err := a.doRefresh.Call(ctx, id, newState)
 	if err != nil {
 		return nil, err
 	}
diff --git a/bundle/direct/dresources/alert.go b/bundle/direct/dresources/alert.go
index a18641e810a..9a0c48472e3 100644
--- a/bundle/direct/dresources/alert.go
+++ b/bundle/direct/dresources/alert.go
@@ -23,7 +23,7 @@ func (*ResourceAlert) PrepareState(input *resources.Alert) *sql.AlertV2 {
 }
 
 // DoRead reads the alert by id.
-func (r *ResourceAlert) DoRead(ctx context.Context, id string) (*sql.AlertV2, error) {
+func (r *ResourceAlert) DoRead(ctx context.Context, id string, _ *sql.AlertV2) (*sql.AlertV2, error) {
 	alert, err := r.client.AlertsV2.GetAlertById(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go
index f18a84d0efc..b0d8dd2c47a 100644
--- a/bundle/direct/dresources/all_test.go
+++ b/bundle/direct/dresources/all_test.go
@@ -821,8 +821,11 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 
 	ctx := t.Context()
 
-	// initial DoRead() cannot find the resource
-	remote, err := adapter.DoRead(ctx, "1234")
+	// initial DoRead() cannot find the resource. Pass nil newState so resources that
+	// use newState (e.g. permissions) fall back to the id-based path and surface the
+	// expected "not found" error rather than reading against the test fixture's
+	// already-created parent.
+	remote, err := adapter.DoRead(ctx, "1234", nil)
 	require.Nil(t, remote)
 	require.Error(t, err)
 	// TODO: if errors.Is(err, databricks.ErrResourceDoesNotExist) {... }
@@ -831,7 +834,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	require.NoError(t, err, "DoCreate failed state=%v", newState)
 	require.NotEmpty(t, createdID, "ID returned from DoCreate was empty")
 
-	remote, err = adapter.DoRead(ctx, createdID)
+	remote, err = adapter.DoRead(ctx, createdID, newState)
 	require.NoError(t, err)
 	require.NotNil(t, remote)
 
@@ -916,7 +919,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 
 	deleteIsNoop := strings.HasSuffix(group, "permissions") || strings.HasSuffix(group, "grants")
 
-	remoteAfterDelete, err := adapter.DoRead(ctx, createdID)
+	remoteAfterDelete, err := adapter.DoRead(ctx, createdID, newState)
 	if deleteIsNoop {
 		require.NoError(t, err)
 	} else {
diff --git a/bundle/direct/dresources/app.go b/bundle/direct/dresources/app.go
index e629027b24b..616673ac96b 100644
--- a/bundle/direct/dresources/app.go
+++ b/bundle/direct/dresources/app.go
@@ -93,7 +93,7 @@ func (*ResourceApp) RemapState(remote *AppRemote) *AppState {
 	}
 }
 
-func (r *ResourceApp) DoRead(ctx context.Context, id string) (*AppRemote, error) {
+func (r *ResourceApp) DoRead(ctx context.Context, id string, _ *AppState) (*AppRemote, error) {
 	app, err := r.client.Apps.GetByName(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/catalog.go b/bundle/direct/dresources/catalog.go
index 1ce28b54123..ddad6f62ccc 100644
--- a/bundle/direct/dresources/catalog.go
+++ b/bundle/direct/dresources/catalog.go
@@ -36,7 +36,7 @@ func (*ResourceCatalog) RemapState(info *catalog.CatalogInfo) *catalog.CreateCat
 	}
 }
 
-func (r *ResourceCatalog) DoRead(ctx context.Context, id string) (*catalog.CatalogInfo, error) {
+func (r *ResourceCatalog) DoRead(ctx context.Context, id string, _ *catalog.CreateCatalog) (*catalog.CatalogInfo, error) {
 	return r.client.Catalogs.GetByName(ctx, id)
 }
 
diff --git a/bundle/direct/dresources/cluster.go b/bundle/direct/dresources/cluster.go
index b41de453017..f28366364ad 100644
--- a/bundle/direct/dresources/cluster.go
+++ b/bundle/direct/dresources/cluster.go
@@ -124,7 +124,7 @@ func (r *ResourceCluster) RemapState(input *ClusterRemote) *ClusterState {
 	return spec
 }
 
-func (r *ResourceCluster) DoRead(ctx context.Context, id string) (*ClusterRemote, error) {
+func (r *ResourceCluster) DoRead(ctx context.Context, id string, _ *ClusterState) (*ClusterRemote, error) {
 	details, err := r.client.Clusters.GetByClusterId(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/dashboard.go b/bundle/direct/dresources/dashboard.go
index dbf492a0bee..e98ca49f6b6 100644
--- a/bundle/direct/dresources/dashboard.go
+++ b/bundle/direct/dresources/dashboard.go
@@ -106,7 +106,7 @@ func (r *ResourceDashboard) RemapState(state *DashboardState) *DashboardState {
 	}
 }
 
-func (r *ResourceDashboard) DoRead(ctx context.Context, id string) (*DashboardState, error) {
+func (r *ResourceDashboard) DoRead(ctx context.Context, id string, _ *DashboardState) (*DashboardState, error) {
 	var dashboard *dashboards.Dashboard
 	var publishedDashboard *dashboards.PublishedDashboard
 	var publishedErr error
diff --git a/bundle/direct/dresources/database_catalog.go b/bundle/direct/dresources/database_catalog.go
index 9bffa708d73..2197c656892 100644
--- a/bundle/direct/dresources/database_catalog.go
+++ b/bundle/direct/dresources/database_catalog.go
@@ -20,7 +20,7 @@ func (*ResourceDatabaseCatalog) PrepareState(input *resources.DatabaseCatalog) *
 	return &input.DatabaseCatalog
 }
 
-func (r *ResourceDatabaseCatalog) DoRead(ctx context.Context, id string) (*database.DatabaseCatalog, error) {
+func (r *ResourceDatabaseCatalog) DoRead(ctx context.Context, id string, _ *database.DatabaseCatalog) (*database.DatabaseCatalog, error) {
 	return r.client.Database.GetDatabaseCatalogByName(ctx, id)
 }
 
diff --git a/bundle/direct/dresources/database_instance.go b/bundle/direct/dresources/database_instance.go
index 2169a61fc8e..2cf0f0bcc8e 100644
--- a/bundle/direct/dresources/database_instance.go
+++ b/bundle/direct/dresources/database_instance.go
@@ -21,7 +21,7 @@ func (*ResourceDatabaseInstance) PrepareState(input *resources.DatabaseInstance)
 	return &input.DatabaseInstance
 }
 
-func (d *ResourceDatabaseInstance) DoRead(ctx context.Context, id string) (*database.DatabaseInstance, error) {
+func (d *ResourceDatabaseInstance) DoRead(ctx context.Context, id string, _ *database.DatabaseInstance) (*database.DatabaseInstance, error) {
 	return d.client.Database.GetDatabaseInstanceByName(ctx, id)
 }
 
diff --git a/bundle/direct/dresources/experiment.go b/bundle/direct/dresources/experiment.go
index bedabc81365..ee036e8590f 100644
--- a/bundle/direct/dresources/experiment.go
+++ b/bundle/direct/dresources/experiment.go
@@ -40,7 +40,7 @@ func (*ResourceExperiment) RemapState(experiment *ml.Experiment) *ml.CreateExper
 	}
 }
 
-func (r *ResourceExperiment) DoRead(ctx context.Context, id string) (*ml.Experiment, error) {
+func (r *ResourceExperiment) DoRead(ctx context.Context, id string, _ *ml.CreateExperiment) (*ml.Experiment, error) {
 	result, err := r.client.Experiments.GetExperiment(ctx, ml.GetExperimentRequest{
 		ExperimentId: id,
 	})
diff --git a/bundle/direct/dresources/external_location.go b/bundle/direct/dresources/external_location.go
index 64eace48eb3..8a80aac9dc1 100644
--- a/bundle/direct/dresources/external_location.go
+++ b/bundle/direct/dresources/external_location.go
@@ -40,7 +40,7 @@ func (*ResourceExternalLocation) RemapState(info *catalog.ExternalLocationInfo)
 	}
 }
 
-func (r *ResourceExternalLocation) DoRead(ctx context.Context, id string) (*catalog.ExternalLocationInfo, error) {
+func (r *ResourceExternalLocation) DoRead(ctx context.Context, id string, _ *catalog.CreateExternalLocation) (*catalog.ExternalLocationInfo, error) {
 	return r.client.ExternalLocations.GetByName(ctx, id)
 }
 
diff --git a/bundle/direct/dresources/grants.go b/bundle/direct/dresources/grants.go
index 596346f1614..55b22c9f2fd 100644
--- a/bundle/direct/dresources/grants.go
+++ b/bundle/direct/dresources/grants.go
@@ -89,7 +89,7 @@ func (*ResourceGrants) KeyedSlices() map[string]any {
 	}
 }
 
-func (r *ResourceGrants) DoRead(ctx context.Context, id string) (*GrantsState, error) {
+func (r *ResourceGrants) DoRead(ctx context.Context, id string, _ *GrantsState) (*GrantsState, error) {
 	securableType, fullName, err := parseGrantsID(id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/job.go b/bundle/direct/dresources/job.go
index 49ca54036c4..13f02b653e7 100644
--- a/bundle/direct/dresources/job.go
+++ b/bundle/direct/dresources/job.go
@@ -86,7 +86,7 @@ func (*ResourceJob) KeyedSlices() map[string]any {
 	}
 }
 
-func (r *ResourceJob) DoRead(ctx context.Context, id string) (*JobRemote, error) {
+func (r *ResourceJob) DoRead(ctx context.Context, id string, _ *jobs.JobSettings) (*JobRemote, error) {
 	idInt, err := parseJobID(id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/model.go b/bundle/direct/dresources/model.go
index 9d04231456d..c45f8cec73c 100644
--- a/bundle/direct/dresources/model.go
+++ b/bundle/direct/dresources/model.go
@@ -39,7 +39,7 @@ func (*ResourceMlflowModel) RemapState(output *MlflowModelRemote) *ml.CreateMode
 	}
 }
 
-func (r *ResourceMlflowModel) DoRead(ctx context.Context, id string) (*MlflowModelRemote, error) {
+func (r *ResourceMlflowModel) DoRead(ctx context.Context, id string, _ *ml.CreateModelRequest) (*MlflowModelRemote, error) {
 	response, err := r.client.ModelRegistry.GetModel(ctx, ml.GetModelRequest{
 		Name: id,
 	})
diff --git a/bundle/direct/dresources/model_serving_endpoint.go b/bundle/direct/dresources/model_serving_endpoint.go
index ccab3a13dea..b435bb6068b 100644
--- a/bundle/direct/dresources/model_serving_endpoint.go
+++ b/bundle/direct/dresources/model_serving_endpoint.go
@@ -117,7 +117,7 @@ type ModelServingEndpointRemote struct {
 	EndpointId      string                           `json:"endpoint_id"`
 }
 
-func (r *ResourceModelServingEndpoint) DoRead(ctx context.Context, id string) (*ModelServingEndpointRemote, error) {
+func (r *ResourceModelServingEndpoint) DoRead(ctx context.Context, id string, _ *serving.CreateServingEndpoint) (*ModelServingEndpointRemote, error) {
 	endpoint, err := r.client.ServingEndpoints.GetByName(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/permissions.go b/bundle/direct/dresources/permissions.go
index 900d74b9609..f5a52ff0222 100644
--- a/bundle/direct/dresources/permissions.go
+++ b/bundle/direct/dresources/permissions.go
@@ -177,7 +177,17 @@ func parsePermissionsID(id string) (extractedType, extractedID string, err error
 	return extractedType, extractedID, nil
 }
 
-func (r *ResourcePermissions) DoRead(ctx context.Context, id string) (*PermissionsState, error) {
+// DoRead reads ACLs for the parent object identified by newState.ObjectID rather
+// than the (possibly stale) deployment state id. After an out-of-band recreate of
+// the parent, state still points at the gone object_id while planning has already
+// resolved newState.ObjectID to the new identifier; reading from the new identifier
+// avoids a permanent drift where the old object_id keeps returning ACL data on V1
+// permissions APIs.
+func (r *ResourcePermissions) DoRead(ctx context.Context, id string, newState *PermissionsState) (*PermissionsState, error) {
+	if newState != nil && newState.ObjectID != "" {
+		id = newState.ObjectID
+	}
+
 	extractedType, extractedID, err := parsePermissionsID(id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/pipeline.go b/bundle/direct/dresources/pipeline.go
index 0861f0fad2f..22e17f3bf76 100644
--- a/bundle/direct/dresources/pipeline.go
+++ b/bundle/direct/dresources/pipeline.go
@@ -57,7 +57,7 @@ func (*ResourcePipeline) RemapState(remote *PipelineRemote) *pipelines.CreatePip
 	return &remote.CreatePipeline
 }
 
-func (r *ResourcePipeline) DoRead(ctx context.Context, id string) (*PipelineRemote, error) {
+func (r *ResourcePipeline) DoRead(ctx context.Context, id string, _ *pipelines.CreatePipeline) (*PipelineRemote, error) {
 	resp, err := r.client.Pipelines.GetByPipelineId(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/postgres_branch.go b/bundle/direct/dresources/postgres_branch.go
index 11c0e7a0fe0..2f9cbb70815 100644
--- a/bundle/direct/dresources/postgres_branch.go
+++ b/bundle/direct/dresources/postgres_branch.go
@@ -94,7 +94,7 @@ func makePostgresBranchRemote(branch *postgres.Branch) *PostgresBranchRemote {
 	}
 }
 
-func (r *ResourcePostgresBranch) DoRead(ctx context.Context, id string) (*PostgresBranchRemote, error) {
+func (r *ResourcePostgresBranch) DoRead(ctx context.Context, id string, _ *PostgresBranchState) (*PostgresBranchRemote, error) {
 	branch, err := r.client.Postgres.GetBranch(ctx, postgres.GetBranchRequest{Name: id})
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/postgres_catalog.go b/bundle/direct/dresources/postgres_catalog.go
index 35a1c0f0cc6..1ef6520a896 100644
--- a/bundle/direct/dresources/postgres_catalog.go
+++ b/bundle/direct/dresources/postgres_catalog.go
@@ -83,7 +83,7 @@ func makePostgresCatalogRemote(catalog *postgres.Catalog) *PostgresCatalogRemote
 	}
 }
 
-func (r *ResourcePostgresCatalog) DoRead(ctx context.Context, id string) (*PostgresCatalogRemote, error) {
+func (r *ResourcePostgresCatalog) DoRead(ctx context.Context, id string, _ *PostgresCatalogState) (*PostgresCatalogRemote, error) {
 	catalog, err := r.client.Postgres.GetCatalog(ctx, postgres.GetCatalogRequest{Name: id})
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/postgres_endpoint.go b/bundle/direct/dresources/postgres_endpoint.go
index 04dd6874f2c..6e23042f548 100644
--- a/bundle/direct/dresources/postgres_endpoint.go
+++ b/bundle/direct/dresources/postgres_endpoint.go
@@ -103,7 +103,7 @@ func makePostgresEndpointRemote(endpoint *postgres.Endpoint) *PostgresEndpointRe
 	}
 }
 
-func (r *ResourcePostgresEndpoint) DoRead(ctx context.Context, id string) (*PostgresEndpointRemote, error) {
+func (r *ResourcePostgresEndpoint) DoRead(ctx context.Context, id string, _ *PostgresEndpointState) (*PostgresEndpointRemote, error) {
 	endpoint, err := r.client.Postgres.GetEndpoint(ctx, postgres.GetEndpointRequest{Name: id})
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/postgres_project.go b/bundle/direct/dresources/postgres_project.go
index fc2ef631e38..87b1a8366d3 100644
--- a/bundle/direct/dresources/postgres_project.go
+++ b/bundle/direct/dresources/postgres_project.go
@@ -98,7 +98,7 @@ func makePostgresProjectRemote(project *postgres.Project) *PostgresProjectRemote
 	}
 }
 
-func (r *ResourcePostgresProject) DoRead(ctx context.Context, id string) (*PostgresProjectRemote, error) {
+func (r *ResourcePostgresProject) DoRead(ctx context.Context, id string, _ *PostgresProjectState) (*PostgresProjectRemote, error) {
 	project, err := r.client.Postgres.GetProject(ctx, postgres.GetProjectRequest{Name: id})
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/postgres_synced_table.go b/bundle/direct/dresources/postgres_synced_table.go
index 623605faf43..2e181583db6 100644
--- a/bundle/direct/dresources/postgres_synced_table.go
+++ b/bundle/direct/dresources/postgres_synced_table.go
@@ -83,7 +83,7 @@ func makePostgresSyncedTableRemote(syncedTable *postgres.SyncedTable) *PostgresS
 	}
 }
 
-func (r *ResourcePostgresSyncedTable) DoRead(ctx context.Context, id string) (*PostgresSyncedTableRemote, error) {
+func (r *ResourcePostgresSyncedTable) DoRead(ctx context.Context, id string, _ *PostgresSyncedTableState) (*PostgresSyncedTableRemote, error) {
 	syncedTable, err := r.client.Postgres.GetSyncedTable(ctx, postgres.GetSyncedTableRequest{Name: id})
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/quality_monitor.go b/bundle/direct/dresources/quality_monitor.go
index c66fed4e0bb..74b9aa8a660 100644
--- a/bundle/direct/dresources/quality_monitor.go
+++ b/bundle/direct/dresources/quality_monitor.go
@@ -65,7 +65,7 @@ func (*ResourceQualityMonitor) RemapState(info *catalog.MonitorInfo) *QualityMon
 	}
 }
 
-func (r *ResourceQualityMonitor) DoRead(ctx context.Context, id string) (*catalog.MonitorInfo, error) {
+func (r *ResourceQualityMonitor) DoRead(ctx context.Context, id string, _ *QualityMonitorState) (*catalog.MonitorInfo, error) {
 	//nolint:staticcheck // Direct quality_monitor resource still uses legacy monitor endpoints; v1 data-quality migration is separate work.
 	return r.client.QualityMonitors.Get(ctx, catalog.GetQualityMonitorRequest{
 		TableName: id,
diff --git a/bundle/direct/dresources/registered_model.go b/bundle/direct/dresources/registered_model.go
index 888870a7581..7fa5c9e21b3 100644
--- a/bundle/direct/dresources/registered_model.go
+++ b/bundle/direct/dresources/registered_model.go
@@ -46,7 +46,7 @@ func (*ResourceRegisteredModel) RemapState(model *catalog.RegisteredModelInfo) *
 	}
 }
 
-func (r *ResourceRegisteredModel) DoRead(ctx context.Context, id string) (*catalog.RegisteredModelInfo, error) {
+func (r *ResourceRegisteredModel) DoRead(ctx context.Context, id string, _ *catalog.CreateRegisteredModelRequest) (*catalog.RegisteredModelInfo, error) {
 	return r.client.RegisteredModels.Get(ctx, catalog.GetRegisteredModelRequest{
 		FullName:        id,
 		IncludeAliases:  false,
diff --git a/bundle/direct/dresources/schema.go b/bundle/direct/dresources/schema.go
index f082ea6c547..65322f7d0ab 100644
--- a/bundle/direct/dresources/schema.go
+++ b/bundle/direct/dresources/schema.go
@@ -33,7 +33,7 @@ func (*ResourceSchema) RemapState(info *catalog.SchemaInfo) *catalog.CreateSchem
 	}
 }
 
-func (r *ResourceSchema) DoRead(ctx context.Context, id string) (*catalog.SchemaInfo, error) {
+func (r *ResourceSchema) DoRead(ctx context.Context, id string, _ *catalog.CreateSchema) (*catalog.SchemaInfo, error) {
 	return r.client.Schemas.GetByFullName(ctx, id)
 }
 
diff --git a/bundle/direct/dresources/schema_test.go b/bundle/direct/dresources/schema_test.go
index d013610e052..96058b89c0c 100644
--- a/bundle/direct/dresources/schema_test.go
+++ b/bundle/direct/dresources/schema_test.go
@@ -40,7 +40,7 @@ func TestResourceSchema_DoUpdate_WithUnsupportedForceSendFields(t *testing.T) {
 	_, err = adapter.DoUpdate(ctx, id, config, &PlanEntry{})
 	require.NoError(t, err)
 
-	result, err := adapter.DoRead(ctx, id)
+	result, err := adapter.DoRead(ctx, id, config)
 	require.NoError(t, err)
 
 	result.CreatedAt = 0
diff --git a/bundle/direct/dresources/secret_scope.go b/bundle/direct/dresources/secret_scope.go
index c811dc84d77..ad3e1b33ac1 100644
--- a/bundle/direct/dresources/secret_scope.go
+++ b/bundle/direct/dresources/secret_scope.go
@@ -51,7 +51,7 @@ func (*ResourceSecretScope) RemapState(remote *workspace.SecretScope) *SecretSco
 // DoRead fetches the secret scope by name. Since the Secrets API does not provide
 // a "get by name" endpoint (see https://docs.databricks.com/api/workspace/secrets),
 // we must list all scopes and filter by name to check if the scope still exists.
-func (r *ResourceSecretScope) DoRead(ctx context.Context, id string) (*workspace.SecretScope, error) {
+func (r *ResourceSecretScope) DoRead(ctx context.Context, id string, _ *SecretScopeConfig) (*workspace.SecretScope, error) {
 	scopes, err := r.client.Secrets.ListScopesAll(ctx)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/secret_scope_acls.go b/bundle/direct/dresources/secret_scope_acls.go
index ef04cb7cb6a..482e71aa5f0 100644
--- a/bundle/direct/dresources/secret_scope_acls.go
+++ b/bundle/direct/dresources/secret_scope_acls.go
@@ -73,7 +73,7 @@ func (*ResourceSecretScopeAcls) KeyedSlices() map[string]any {
 	}
 }
 
-func (r *ResourceSecretScopeAcls) DoRead(ctx context.Context, id string) (*SecretScopeAclsState, error) {
+func (r *ResourceSecretScopeAcls) DoRead(ctx context.Context, id string, _ *SecretScopeAclsState) (*SecretScopeAclsState, error) {
 	// id is the scope name
 	currentAcls, err := r.client.Secrets.ListAclsAll(ctx, workspace.ListAclsRequest{
 		Scope: id,
diff --git a/bundle/direct/dresources/sql_warehouse.go b/bundle/direct/dresources/sql_warehouse.go
index 146dee5294d..d1cab5f5466 100644
--- a/bundle/direct/dresources/sql_warehouse.go
+++ b/bundle/direct/dresources/sql_warehouse.go
@@ -94,7 +94,7 @@ func (*ResourceSqlWarehouse) RemapState(warehouse *SqlWarehouseRemote) *SqlWareh
 }
 
 // DoRead reads the warehouse by id.
-func (r *ResourceSqlWarehouse) DoRead(ctx context.Context, id string) (*SqlWarehouseRemote, error) {
+func (r *ResourceSqlWarehouse) DoRead(ctx context.Context, id string, _ *SqlWarehouseState) (*SqlWarehouseRemote, error) {
 	warehouse, err := r.client.Warehouses.GetById(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/synced_database_table.go b/bundle/direct/dresources/synced_database_table.go
index d45c6fb3fc7..5ec75813e93 100644
--- a/bundle/direct/dresources/synced_database_table.go
+++ b/bundle/direct/dresources/synced_database_table.go
@@ -20,7 +20,7 @@ func (*ResourceSyncedDatabaseTable) PrepareState(input *resources.SyncedDatabase
 	return &input.SyncedDatabaseTable
 }
 
-func (r *ResourceSyncedDatabaseTable) DoRead(ctx context.Context, name string) (*database.SyncedDatabaseTable, error) {
+func (r *ResourceSyncedDatabaseTable) DoRead(ctx context.Context, name string, _ *database.SyncedDatabaseTable) (*database.SyncedDatabaseTable, error) {
 	return r.client.Database.GetSyncedDatabaseTableByName(ctx, name)
 }
 
diff --git a/bundle/direct/dresources/vector_search_endpoint.go b/bundle/direct/dresources/vector_search_endpoint.go
index 1178def16be..d1462bd2c63 100644
--- a/bundle/direct/dresources/vector_search_endpoint.go
+++ b/bundle/direct/dresources/vector_search_endpoint.go
@@ -57,7 +57,7 @@ func (*ResourceVectorSearchEndpoint) RemapState(remote *VectorSearchEndpointRemo
 	}
 }
 
-func (r *ResourceVectorSearchEndpoint) DoRead(ctx context.Context, id string) (*VectorSearchEndpointRemote, error) {
+func (r *ResourceVectorSearchEndpoint) DoRead(ctx context.Context, id string, _ *vectorsearch.CreateEndpoint) (*VectorSearchEndpointRemote, error) {
 	info, err := r.client.VectorSearchEndpoints.GetEndpointByEndpointName(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/vector_search_index.go b/bundle/direct/dresources/vector_search_index.go
index 48ee6f0f968..07014a7aeb7 100644
--- a/bundle/direct/dresources/vector_search_index.go
+++ b/bundle/direct/dresources/vector_search_index.go
@@ -106,7 +106,7 @@ func (*ResourceVectorSearchIndex) RemapState(remote *VectorSearchIndexRemote) *V
 	return state
 }
 
-func (r *ResourceVectorSearchIndex) DoRead(ctx context.Context, id string) (*VectorSearchIndexRemote, error) {
+func (r *ResourceVectorSearchIndex) DoRead(ctx context.Context, id string, _ *VectorSearchIndexState) (*VectorSearchIndexRemote, error) {
 	index, err := r.client.VectorSearchIndexes.GetIndexByIndexName(ctx, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/volume.go b/bundle/direct/dresources/volume.go
index 73bf7a79b40..97c4bd84de4 100644
--- a/bundle/direct/dresources/volume.go
+++ b/bundle/direct/dresources/volume.go
@@ -38,7 +38,7 @@ func (*ResourceVolume) RemapState(info *catalog.VolumeInfo) *catalog.CreateVolum
 	}
 }
 
-func (r *ResourceVolume) DoRead(ctx context.Context, id string) (*catalog.VolumeInfo, error) {
+func (r *ResourceVolume) DoRead(ctx context.Context, id string, _ *catalog.CreateVolumeRequestContent) (*catalog.VolumeInfo, error) {
 	return r.client.Volumes.ReadByName(ctx, id)
 }
 

From 74189482c5a973dc4b121f0aaf4e3dc9b2f7c931 Mon Sep 17 00:00:00 2001
From: Jan Rose <jan.rose@databricks.com>
Date: Fri, 12 Jun 2026 22:12:14 +0200
Subject: [PATCH 4/5] Add NEXT_CHANGELOG.md entry for #5587

Co-authored-by: Isaac
---
 NEXT_CHANGELOG.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md
index 9c5ce923dd5..4f53c00ec6e 100644
--- a/NEXT_CHANGELOG.md
+++ b/NEXT_CHANGELOG.md
@@ -12,6 +12,7 @@
 * direct: Fix resolving a resource reference that is used more than once within the same field ([#5558](https://github.com/databricks/cli/pull/5558)).
 * Bundle variable references now accept Unicode letters in path segments (e.g. `${var.变量}`). ([#5532](https://github.com/databricks/cli/pull/5532))
 * Ignore remote changes for vector search direct_access_index_spec.schema_json to prevent drift when the backend normalizes the schema ([#5481](https://github.com/databricks/cli/pull/5481)).
+* direct: Fix permanent drift on `permissions` when the parent resource is deleted and recreated out-of-band with the same name ([#5587](https://github.com/databricks/cli/pull/5587)).
 
 ### Dependency updates
 

From ee3d0b43b2bf2df20ea7de7d0a782cee82b43813 Mon Sep 17 00:00:00 2001
From: Jan Rose <jan.rose@databricks.com>
Date: Wed, 17 Jun 2026 18:38:50 +0200
Subject: [PATCH 5/5] [WIP] add recreate_cascade badness test for
 registered_models

---
 .../recreate_cascade/databricks.yml           |  17 +++
 .../recreate_cascade/out.test.toml            |   3 +
 .../recreate_cascade/output.txt               | 126 ++++++++++++++++++
 .../registered_models/recreate_cascade/script |   9 ++
 .../recreate_cascade/test.toml                |   6 +
 5 files changed, 161 insertions(+)
 create mode 100644 acceptance/bundle/resources/registered_models/recreate_cascade/databricks.yml
 create mode 100644 acceptance/bundle/resources/registered_models/recreate_cascade/out.test.toml
 create mode 100644 acceptance/bundle/resources/registered_models/recreate_cascade/output.txt
 create mode 100644 acceptance/bundle/resources/registered_models/recreate_cascade/script
 create mode 100644 acceptance/bundle/resources/registered_models/recreate_cascade/test.toml

diff --git a/acceptance/bundle/resources/registered_models/recreate_cascade/databricks.yml b/acceptance/bundle/resources/registered_models/recreate_cascade/databricks.yml
new file mode 100644
index 00000000000..2f47b217e1a
--- /dev/null
+++ b/acceptance/bundle/resources/registered_models/recreate_cascade/databricks.yml
@@ -0,0 +1,17 @@
+bundle:
+  name: registered-model-cascade-recreate
+
+resources:
+  schemas:
+    parent_schema:
+      name: myschema
+      catalog_name: main
+      storage_root: dbfs:/parent_storage_root_v1
+      comment: parent schema
+
+  registered_models:
+    child_model:
+      name: mymodel
+      catalog_name: main
+      schema_name: ${resources.schemas.parent_schema.name}
+      comment: child model
diff --git a/acceptance/bundle/resources/registered_models/recreate_cascade/out.test.toml b/acceptance/bundle/resources/registered_models/recreate_cascade/out.test.toml
new file mode 100644
index 00000000000..e90b6d5d1ba
--- /dev/null
+++ b/acceptance/bundle/resources/registered_models/recreate_cascade/out.test.toml
@@ -0,0 +1,3 @@
+Local = true
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/resources/registered_models/recreate_cascade/output.txt b/acceptance/bundle/resources/registered_models/recreate_cascade/output.txt
new file mode 100644
index 00000000000..6351b89c672
--- /dev/null
+++ b/acceptance/bundle/resources/registered_models/recreate_cascade/output.txt
@@ -0,0 +1,126 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/registered-model-cascade-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== Change schema's storage_root to trigger schema recreate
+>>> update_file.py databricks.yml dbfs:/parent_storage_root_v1 dbfs:/parent_storage_root_v2
+
+=== Plan should recreate BOTH the schema and its dependent registered_model, but currently only the schema is recreated
+>>> [CLI] bundle plan
+recreate schemas.parent_schema
+
+Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged
+
+>>> [CLI] bundle plan --output json
+{
+  "plan_version": 2,
+  "cli_version": "[DEV_VERSION]",
+  "lineage": "[UUID]",
+  "serial": 1,
+  "plan": {
+    "resources.registered_models.child_model": {
+      "depends_on": [
+        {
+          "node": "resources.schemas.parent_schema",
+          "label": "${resources.schemas.parent_schema.name}"
+        }
+      ],
+      "action": "skip",
+      "remote_state": {
+        "catalog_name": "main",
+        "comment": "child model",
+        "created_at": [UNIX_TIME_MILLIS][0],
+        "created_by": "[USERNAME]",
+        "full_name": "main.myschema.mymodel",
+        "metastore_id": "[UUID]",
+        "name": "mymodel",
+        "owner": "[USERNAME]",
+        "schema_name": "myschema",
+        "updated_at": [UNIX_TIME_MILLIS][0],
+        "updated_by": "[USERNAME]"
+      },
+      "changes": {
+        "created_at": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "created_by": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": ""
+        },
+        "full_name": {
+          "action": "skip",
+          "reason": "backend_default",
+          "remote": "main.myschema.mymodel"
+        },
+        "metastore_id": {
+          "action": "skip",
+          "reason": "backend_default",
+          "remote": "[UUID]"
+        },
+        "owner": {
+          "action": "skip",
+          "reason": "backend_default",
+          "remote": "[USERNAME]"
+        },
+        "updated_at": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "updated_by": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": ""
+        }
+      }
+    },
+    "resources.schemas.parent_schema": {
+      "action": "recreate",
+      "new_state": {
+        "value": {
+          "catalog_name": "main",
+          "comment": "parent schema",
+          "name": "myschema",
+          "storage_root": "dbfs:/parent_storage_root_v2"
+        }
+      },
+      "remote_state": {
+        "browse_only": false,
+        "catalog_name": "main",
+        "catalog_type": "MANAGED_CATALOG",
+        "comment": "parent schema",
+        "created_at": [UNIX_TIME_MILLIS][1],
+        "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.myschema",
+        "metastore_id": "[UUID]",
+        "name": "myschema",
+        "owner": "[USERNAME]",
+        "schema_id": "[UUID]",
+        "storage_root": "dbfs:/parent_storage_root_v1",
+        "updated_at": [UNIX_TIME_MILLIS][1],
+        "updated_by": "[USERNAME]"
+      },
+      "changes": {
+        "storage_root": {
+          "action": "recreate",
+          "reason": "immutable",
+          "old": "dbfs:/parent_storage_root_v1",
+          "new": "dbfs:/parent_storage_root_v2",
+          "remote": "dbfs:/parent_storage_root_v1"
+        }
+      }
+    }
+  }
+}
diff --git a/acceptance/bundle/resources/registered_models/recreate_cascade/script b/acceptance/bundle/resources/registered_models/recreate_cascade/script
new file mode 100644
index 00000000000..120747ce31a
--- /dev/null
+++ b/acceptance/bundle/resources/registered_models/recreate_cascade/script
@@ -0,0 +1,9 @@
+echo "*" > .gitignore
+trace $CLI bundle deploy
+
+title "Change schema's storage_root to trigger schema recreate"
+trace update_file.py databricks.yml "dbfs:/parent_storage_root_v1" "dbfs:/parent_storage_root_v2"
+
+title "Plan should recreate BOTH the schema and its dependent registered_model, but currently only the schema is recreated"
+trace $CLI bundle plan
+trace $CLI bundle plan --output json
diff --git a/acceptance/bundle/resources/registered_models/recreate_cascade/test.toml b/acceptance/bundle/resources/registered_models/recreate_cascade/test.toml
new file mode 100644
index 00000000000..1543f5e1907
--- /dev/null
+++ b/acceptance/bundle/resources/registered_models/recreate_cascade/test.toml
@@ -0,0 +1,6 @@
+Local = true
+Cloud = false
+RecordRequests = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+Badness = "Recreating a parent resource (schema) should propagate Recreate to dependents (registered_model) that reference it, but the planner does not. The dependent plans Skip while the parent is deleted underneath it. On a real workspace this either fails the parent delete (children still attached) or orphans the child. Tracked as a framework-level fix in bundle/direct/bundle_plan.go."