diff --git a/acceptance/bundle/deploy/snapshot-comparison/output.txt b/acceptance/bundle/deploy/snapshot-comparison/output.txt
index 1db9fa5024d..b5bb597e264 100644
--- a/acceptance/bundle/deploy/snapshot-comparison/output.txt
+++ b/acceptance/bundle/deploy/snapshot-comparison/output.txt
@@ -8,8 +8,6 @@ Deployment complete!
 
 === Run migrate on bundle 1
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 Success! Migrated 2 resources to direct engine state file: [TEST_TMP_DIR]/bundle1/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/help/bundle-deployment-migrate/output.txt b/acceptance/bundle/help/bundle-deployment-migrate/output.txt
index da2f95707bc..9312fb85ffc 100644
--- a/acceptance/bundle/help/bundle-deployment-migrate/output.txt
+++ b/acceptance/bundle/help/bundle-deployment-migrate/output.txt
@@ -13,7 +13,7 @@ Usage:
 
 Flags:
   -h, --help          help for migrate
-      --noplancheck   Skip running bundle plan before migration.
+      --noplancheck   No-op (kept for compatibility).
 
 Global Flags:
       --debug            enable debug logging
diff --git a/acceptance/bundle/migrate/basic/output.txt b/acceptance/bundle/migrate/basic/output.txt
index dafa3a4086e..cd45577a1ee 100644
--- a/acceptance/bundle/migrate/basic/output.txt
+++ b/acceptance/bundle/migrate/basic/output.txt
@@ -11,8 +11,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 3 unchanged
 Success! Migrated 3 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/dashboards/output.txt b/acceptance/bundle/migrate/dashboards/output.txt
index 19a4f1c7bb5..cfda4350ceb 100644
--- a/acceptance/bundle/migrate/dashboards/output.txt
+++ b/acceptance/bundle/migrate/dashboards/output.txt
@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/default-python/output.txt b/acceptance/bundle/migrate/default-python/output.txt
index 15d49805b47..fb554bc258d 100644
--- a/acceptance/bundle/migrate/default-python/output.txt
+++ b/acceptance/bundle/migrate/default-python/output.txt
@@ -22,16 +22,7 @@ Deployment complete!
 
 >>> print_state.py
 
->>> musterr [CLI] bundle deployment migrate
-Building python_artifact...
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Building python_artifact...
-update jobs.sample_job
-
-Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
-Error: 'databricks bundle plan' shows actions planned, aborting migration. Please run 'databricks bundle deploy' first to ensure your bundle is up to date, If actions persist after deploy, skip plan check with --noplancheck option
-
->>> [CLI] bundle deployment migrate --noplancheck
+>>> [CLI] bundle deployment migrate
 Building python_artifact...
 Success! Migrated 2 resources to direct engine state file: [TEST_TMP_DIR]/my_default_python/.databricks/bundle/dev/resources.json
 
diff --git a/acceptance/bundle/migrate/default-python/script b/acceptance/bundle/migrate/default-python/script
index c9b585dbea7..e3a0b9b3e05 100755
--- a/acceptance/bundle/migrate/default-python/script
+++ b/acceptance/bundle/migrate/default-python/script
@@ -5,8 +5,7 @@ cd my_default_python
 trace DATABRICKS_BUNDLE_ENGINE=terraform $CLI bundle deploy
 trace print_state.py > ../out.state_original.json
 
-trace musterr $CLI bundle deployment migrate
-trace $CLI bundle deployment migrate --noplancheck
+trace $CLI bundle deployment migrate
 
 trace print_state.py > ../out.state_after_migration.json
 trace jq '.. | .libraries? | select(.)' ../out.state_after_migration.json
diff --git a/acceptance/bundle/migrate/grants/output.txt b/acceptance/bundle/migrate/grants/output.txt
index 146787d549a..86acb6398f5 100644
--- a/acceptance/bundle/migrate/grants/output.txt
+++ b/acceptance/bundle/migrate/grants/output.txt
@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 6 unchanged
 Success! Migrated 6 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/permissions/output.txt b/acceptance/bundle/migrate/permissions/output.txt
index f85c8d7bdbf..51caca51104 100644
--- a/acceptance/bundle/migrate/permissions/output.txt
+++ b/acceptance/bundle/migrate/permissions/output.txt
@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 4 unchanged
 Success! Migrated 4 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/profile_arg/output.txt b/acceptance/bundle/migrate/profile_arg/output.txt
index a6def38363a..082feee15b1 100644
--- a/acceptance/bundle/migrate/profile_arg/output.txt
+++ b/acceptance/bundle/migrate/profile_arg/output.txt
@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate -p non_existent321
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan -p non_existent321", there should be no actions planned.
@@ -24,8 +22,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate -p non_existent321 -t prod
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 Success! Migrated 2 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/prod/resources.json
 
 Validate the migration by running "databricks bundle plan -t prod -p non_existent321", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/runas/output.txt b/acceptance/bundle/migrate/runas/output.txt
index 74b9a0217f3..f8a9b475bdc 100644
--- a/acceptance/bundle/migrate/runas/output.txt
+++ b/acceptance/bundle/migrate/runas/output.txt
@@ -81,20 +81,6 @@ Consider using a adding a top-level permissions section such as the following:
 See https://docs.databricks.com/dev-tools/bundles/permissions.html to learn more about permission configuration.
   in databricks.yml:5:3
 
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Recommendation: permissions section should explicitly include the current deployment identity '[USERNAME]' or one of its groups
-If it is not included, CAN_MANAGE permissions are only applied if the present identity is used to deploy.
-
-Consider using a adding a top-level permissions section such as the following:
-
-  permissions:
-    - user_name: [USERNAME]
-      level: CAN_MANAGE
-
-See https://docs.databricks.com/dev-tools/bundles/permissions.html to learn more about permission configuration.
-  in databricks.yml:5:3
-
-Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 Success! Migrated 2 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/production/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.
diff --git a/acceptance/bundle/migrate/var_arg/output.txt b/acceptance/bundle/migrate/var_arg/output.txt
index a7f8c0e5b2e..3c20819211d 100644
--- a/acceptance/bundle/migrate/var_arg/output.txt
+++ b/acceptance/bundle/migrate/var_arg/output.txt
@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate --var=job_name=Custom Job Name
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan --var 'job_name=Custom Job Name'", there should be no actions planned.
@@ -39,8 +37,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate --var job_name=Custom Job Name
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan --var 'job_name=Custom Job Name'", there should be no actions planned.
diff --git a/bundle/deploy/terraform/util.go b/bundle/deploy/terraform/util.go
index 7ca5e9a1d14..51df7f49cd0 100644
--- a/bundle/deploy/terraform/util.go
+++ b/bundle/deploy/terraform/util.go
@@ -51,6 +51,15 @@ type stateInstanceAttributes struct {
 }
 
 // Returns a mapping resourceKey -> stateInstanceAttributes
+// ParseResourcesStateFromBytes parses a terraform state file from already-read bytes.
+func ParseResourcesStateFromBytes(ctx context.Context, raw []byte) (ExportedResourcesMap, error) {
+	var state resourcesState
+	if err := json.Unmarshal(raw, &state); err != nil {
+		return nil, err
+	}
+	return resourcesStateToMap(ctx, &state)
+}
+
 func parseResourcesState(ctx context.Context, path string) (ExportedResourcesMap, error) {
 	rawState, err := os.ReadFile(path)
 	if err != nil {
@@ -59,12 +68,10 @@ func parseResourcesState(ctx context.Context, path string) (ExportedResourcesMap
 		}
 		return nil, err
 	}
-	var state resourcesState
-	err = json.Unmarshal(rawState, &state)
-	if err != nil {
-		return nil, err
-	}
+	return ParseResourcesStateFromBytes(ctx, rawState)
+}
 
+func resourcesStateToMap(ctx context.Context, state *resourcesState) (ExportedResourcesMap, error) {
 	if state.Version != SupportedStateVersion {
 		return nil, fmt.Errorf("unsupported deployment state version: %d. Try re-deploying the bundle", state.Version)
 	}
@@ -131,5 +138,10 @@ func ParseResourcesState(ctx context.Context, b *bundle.Bundle) (ExportedResourc
 		return nil, err
 	}
 	filename, _ := b.StateFilenameTerraform(ctx)
-	return parseResourcesState(ctx, filepath.Join(cacheDir, filename))
+	return ParseResourcesStateFromPath(ctx, filepath.Join(cacheDir, filename))
+}
+
+// ParseResourcesStateFromPath parses a terraform state file at a known path.
+func ParseResourcesStateFromPath(ctx context.Context, path string) (ExportedResourcesMap, error) {
+	return parseResourcesState(ctx, path)
 }
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index 16b145f7af8..a63d70aee13 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -15,9 +15,7 @@ import (
 	"github.com/databricks/databricks-sdk-go"
 )
 
-type MigrateMode bool
-
-func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan, migrateMode MigrateMode) {
+func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan) {
 	if plan == nil {
 		panic("Planning is not done")
 	}
@@ -52,9 +50,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 
 		action := entry.Action
 		errorPrefix := fmt.Sprintf("cannot %s %s", action, resourceKey)
-		if migrateMode {
-			errorPrefix = "cannot migrate " + resourceKey
-		}
 
 		if action == deployplan.Undefined {
 			logdiag.LogError(ctx, fmt.Errorf("cannot deploy %s: unknown action %q", resourceKey, action))
@@ -82,10 +77,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		}
 
 		if action == deployplan.Delete {
-			if migrateMode {
-				logdiag.LogError(ctx, fmt.Errorf("%s: Unexpected delete action during migration", errorPrefix))
-				return false
-			}
 			err = d.Destroy(ctx, &b.StateDB)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
@@ -113,19 +104,8 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				return false
 			}
 
-			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)
-				if id == "" {
-					logdiag.LogError(ctx, fmt.Errorf("state entry not found for %q", resourceKey))
-					return false
-				}
-				err = b.StateDB.SaveState(resourceKey, id, sv.Value, entry.DependsOn)
-			} else {
-				// TODO: redo calcDiff to downgrade planned action if possible (?)
-				err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry)
-			}
-
+			// TODO: redo calcDiff to downgrade planned action if possible (?)
+			err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 5591626cd75..79e0b05648a 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -931,6 +931,12 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root
 	return p, nil
 }
 
+// ExtractReferences extracts all variable references from the config subtree rooted at node.
+// Returns a map from structpath string (field path within the resource) to template string.
+func ExtractReferences(root dyn.Value, node string) (map[string]string, error) {
+	return extractReferences(root, node)
+}
+
 func extractReferences(root dyn.Value, node string) (map[string]string, error) {
 	nodeType := config.GetResourceTypeFromKey(node)
 	refs := make(map[string]string)
diff --git a/bundle/migrate/build_state.go b/bundle/migrate/build_state.go
new file mode 100644
index 00000000000..1accf0e576e
--- /dev/null
+++ b/bundle/migrate/build_state.go
@@ -0,0 +1,224 @@
+package migrate
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/dyn/dynvar"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/structs/structaccess"
+	"github.com/databricks/cli/libs/structs/structpath"
+	"github.com/databricks/cli/libs/structs/structvar"
+)
+
+// BuildStateFromTF iterates over bundle resources, resolves cross-resource
+// references using TF state attributes, and writes each resource's state entry.
+// configRoot should be an un-interpolated config (with ${resources.*} references).
+func BuildStateFromTF(
+	ctx context.Context,
+	configRoot *config.Root,
+	adapters map[string]*dresources.Adapter,
+	stateDB *dstate.DeploymentState,
+	tfAttrs TFStateAttrs,
+	tfIDs terraform.ExportedResourcesMap,
+) error {
+	// Collect all resource nodes (same patterns as makePlan).
+	var nodes []string
+	patterns := []dyn.Pattern{
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey()),
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey(), dyn.Key("permissions")),
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey(), dyn.Key("grants")),
+	}
+	for _, pat := range patterns {
+		_, err := dyn.MapByPattern(
+			configRoot.Value(),
+			pat,
+			func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+				nodes = append(nodes, p.String())
+				return dyn.InvalidValue, nil
+			},
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	for _, node := range nodes {
+		idEntry, ok := tfIDs[node]
+		if !ok {
+			// Resource is in config but not in TF state (new resource); skip.
+			continue
+		}
+
+		group := config.GetResourceTypeFromKey(node)
+		if group == "" {
+			return fmt.Errorf("cannot determine resource type for %q", node)
+		}
+
+		adapter, ok := adapters[group]
+		if !ok {
+			log.Warnf(ctx, "unsupported resource type %q for %s, skipping", group, node)
+			continue
+		}
+
+		inputConfig, err := configRoot.GetResourceConfig(node)
+		if err != nil {
+			return fmt.Errorf("%s: getting config: %w", node, err)
+		}
+
+		baseRefs := map[string]string{}
+
+		switch {
+		case strings.HasSuffix(node, ".permissions"):
+			var sv *structvar.StructVar
+			if strings.HasPrefix(node, "resources.secret_scopes.") {
+				typedConfig, ok := inputConfig.(*[]resources.SecretScopePermission)
+				if !ok {
+					return fmt.Errorf("%s: expected *[]resources.SecretScopePermission, got %T", node, inputConfig)
+				}
+				sv, err = dresources.PrepareSecretScopeAclsInputConfig(*typedConfig, node)
+				if err != nil {
+					return fmt.Errorf("%s: preparing secret scope ACLs config: %w", node, err)
+				}
+			} else {
+				sv, err = dresources.PreparePermissionsInputConfig(inputConfig, node)
+				if err != nil {
+					return fmt.Errorf("%s: preparing permissions config: %w", node, err)
+				}
+			}
+			inputConfig = sv.Value
+			baseRefs = sv.Refs
+
+		case strings.HasSuffix(node, ".grants"):
+			sv, err := dresources.PrepareGrantsInputConfig(inputConfig, node)
+			if err != nil {
+				return fmt.Errorf("%s: preparing grants config: %w", node, err)
+			}
+			inputConfig = sv.Value
+			baseRefs = sv.Refs
+		}
+
+		newStateValue, err := adapter.PrepareState(inputConfig)
+		if err != nil {
+			return fmt.Errorf("%s: PrepareState: %w", node, err)
+		}
+
+		refs, err := direct.ExtractReferences(configRoot.Value(), node)
+		if err != nil {
+			return fmt.Errorf("%s: extracting references: %w", node, err)
+		}
+		maps.Copy(refs, baseRefs)
+
+		sv := structvar.NewStructVar(newStateValue, refs)
+
+		// Compute depends_on from cross-resource references before resolving them
+		// (resolution deletes entries from the refs map).
+		// Same logic as makePlan in bundle/direct/bundle_plan.go.
+		var dependsOn []deployplan.DependsOnEntry //nolint:prealloc
+		for _, refTemplate := range refs {
+			ref, ok := dynvar.NewRef(dyn.V(refTemplate))
+			if !ok {
+				continue
+			}
+			for _, targetPath := range ref.References() {
+				targetPathParsed, err := dyn.NewPathFromString(targetPath)
+				if err != nil {
+					continue
+				}
+				targetNodeDP, _ := config.GetNodeAndType(targetPathParsed)
+				targetNode := targetNodeDP.String()
+				fullRef := "${" + targetPath + "}"
+				found := false
+				for _, dep := range dependsOn {
+					if dep.Node == targetNode && dep.Label == fullRef {
+						found = true
+						break
+					}
+				}
+				if !found {
+					dependsOn = append(dependsOn, deployplan.DependsOnEntry{
+						Node:  targetNode,
+						Label: fullRef,
+					})
+				}
+			}
+		}
+		slices.SortFunc(dependsOn, func(a, b deployplan.DependsOnEntry) int {
+			if a.Node != b.Node {
+				return strings.Compare(a.Node, b.Node)
+			}
+			return strings.Compare(a.Label, b.Label)
+		})
+
+		// Resolve each reference using TF state.
+		// node format: "resources.<group>.<name>" or "resources.<group>.<name>.permissions"
+		parts := strings.SplitN(node, ".", 4)
+		var srcGroup, srcName string
+		if len(parts) >= 3 {
+			srcGroup = parts[1]
+			srcName = parts[2]
+		}
+
+		// Collect all field paths that need resolution (avoid modifying map during iteration).
+		type refEntry struct {
+			fieldPathStr string
+			refTemplate  string
+		}
+		var pendingRefs []refEntry
+		for fieldPathStr, refTemplate := range sv.Refs {
+			pendingRefs = append(pendingRefs, refEntry{fieldPathStr, refTemplate})
+		}
+
+		for _, pending := range pendingRefs {
+			fieldPath, err := structpath.ParsePath(pending.fieldPathStr)
+			if err != nil {
+				return fmt.Errorf("%s: parsing field path %q: %w", node, pending.fieldPathStr, err)
+			}
+
+			// ResolveFieldRef returns the fully resolved value for this field,
+			// using either Method A (TF state lookup) or Method B (template evaluation).
+			value, err := ResolveFieldRef(ctx, tfAttrs, srcGroup, srcName, fieldPath, pending.refTemplate)
+			if err != nil {
+				return fmt.Errorf("%s: cannot resolve field %q (template %q): %w", node, pending.fieldPathStr, pending.refTemplate, err)
+			}
+
+			// Set the resolved value directly and remove the ref entry.
+			if err := structaccess.Set(sv.Value, fieldPath, value); err != nil {
+				return fmt.Errorf("%s: cannot set resolved value for field %q: %w", node, pending.fieldPathStr, err)
+			}
+			delete(sv.Refs, pending.fieldPathStr)
+		}
+
+		if len(sv.Refs) > 0 {
+			return fmt.Errorf("%s: unresolved references: %v", node, sv.Refs)
+		}
+
+		// Handle etag for dashboards: look it up directly from TF state attributes.
+		// The "etag" field is a computed TF attribute not present in the bundle config,
+		// so it does not flow through PrepareState/ExtractReferences.
+		if etag, err := LookupTFField(tfAttrs, group, srcName, structpath.NewStringKey(nil, "etag")); err == nil && etag != nil {
+			if etagStr, ok := etag.(string); ok && etagStr != "" {
+				if err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etagStr); err != nil {
+					return fmt.Errorf("%s: cannot set etag: %w", node, err)
+				}
+			}
+		}
+
+		if err := stateDB.SaveState(node, idEntry.ID, sv.Value, dependsOn); err != nil {
+			return fmt.Errorf("%s: SaveState: %w", node, err)
+		}
+	}
+
+	return nil
+}
diff --git a/bundle/migrate/build_state_test.go b/bundle/migrate/build_state_test.go
new file mode 100644
index 00000000000..a7db542e09e
--- /dev/null
+++ b/bundle/migrate/build_state_test.go
@@ -0,0 +1,220 @@
+package migrate_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/bundle/migrate"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/dyn/convert"
+	"github.com/databricks/cli/libs/dyn/yamlloader"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// rootFromYAML builds a config.Root from a YAML snippet.
+// Template strings like "${resources.jobs.src.name}" are preserved in the
+// internal dyn.Value so BuildStateFromTF can find them via ExtractReferences.
+func rootFromYAML(t *testing.T, yaml string) config.Root {
+	t.Helper()
+	v, err := yamlloader.LoadYAML("test", bytes.NewBufferString(yaml))
+	require.NoError(t, err)
+	var root config.Root
+	require.NoError(t, convert.ToTyped(&root, v))
+	require.NoError(t, root.Mutate(func(_ dyn.Value) (dyn.Value, error) { return v, nil }))
+	return root
+}
+
+// runBuildStateFromTF is a helper that runs BuildStateFromTF, finalizes the
+// state, then reloads it so callers can inspect ResourceEntry (State + DependsOn).
+func runBuildStateFromTF(
+	t *testing.T,
+	yaml string,
+	tfAttrs migrate.TFStateAttrs,
+	tfIDs terraform.ExportedResourcesMap,
+) map[string]dstate.ResourceEntry {
+	t.Helper()
+
+	root := rootFromYAML(t, yaml)
+	adapters, err := dresources.InitAll(nil)
+	require.NoError(t, err)
+
+	statePath := filepath.Join(t.TempDir(), "resources.json")
+
+	var db dstate.DeploymentState
+	db.OpenWithData(statePath, dstate.NewDatabase("lineage", 1))
+	require.NoError(t, db.UpgradeToWrite())
+
+	err = migrate.BuildStateFromTF(t.Context(), &root, adapters, &db, tfAttrs, tfIDs)
+	require.NoError(t, err)
+
+	_, err = db.Finalize(t.Context())
+	require.NoError(t, err)
+
+	// Reload from disk to access the full ResourceEntry (State JSON + DependsOn).
+	raw, err := os.ReadFile(statePath)
+	require.NoError(t, err)
+	var data dstate.Database
+	require.NoError(t, json.Unmarshal(raw, &data))
+	return data.State
+}
+
+func TestBuildStateFromTF_BasicJob(t *testing.T) {
+	bundleYAML := `
+resources:
+  jobs:
+    my_job:
+      name: "hello"
+`
+	tfAttrs := migrate.TFStateAttrs{
+		"databricks_job": {
+			"my_job": json.RawMessage(`{"id": "123", "name": "hello"}`),
+		},
+	}
+	tfIDs := terraform.ExportedResourcesMap{
+		"resources.jobs.my_job": {ID: "123"},
+	}
+
+	state := runBuildStateFromTF(t, bundleYAML, tfAttrs, tfIDs)
+	entry, ok := state["resources.jobs.my_job"]
+	require.True(t, ok)
+	assert.Equal(t, "123", entry.ID)
+	assert.Empty(t, entry.DependsOn)
+}
+
+func TestBuildStateFromTF_ResourceNotInTFState_Skipped(t *testing.T) {
+	bundleYAML := `
+resources:
+  jobs:
+    new_job:
+      name: "new"
+    existing_job:
+      name: "existing"
+`
+	tfAttrs := migrate.TFStateAttrs{
+		"databricks_job": {
+			"existing_job": json.RawMessage(`{"id": "456", "name": "existing"}`),
+		},
+	}
+	tfIDs := terraform.ExportedResourcesMap{
+		"resources.jobs.existing_job": {ID: "456"},
+	}
+
+	state := runBuildStateFromTF(t, bundleYAML, tfAttrs, tfIDs)
+	assert.Contains(t, state, "resources.jobs.existing_job")
+	assert.NotContains(t, state, "resources.jobs.new_job")
+}
+
+func TestBuildStateFromTF_DependsOnComputedFromRefs(t *testing.T) {
+	bundleYAML := `
+resources:
+  pipelines:
+    src:
+      name: "source-pipeline"
+  jobs:
+    dst:
+      name: "${resources.pipelines.src.name}"
+`
+	tfAttrs := migrate.TFStateAttrs{
+		"databricks_pipeline": {
+			"src": json.RawMessage(`{"id": "p1", "name": "source-pipeline"}`),
+		},
+		"databricks_job": {
+			"dst": json.RawMessage(`{"id": "j1", "name": "source-pipeline"}`),
+		},
+	}
+	tfIDs := terraform.ExportedResourcesMap{
+		"resources.pipelines.src": {ID: "p1"},
+		"resources.jobs.dst":      {ID: "j1"},
+	}
+
+	state := runBuildStateFromTF(t, bundleYAML, tfAttrs, tfIDs)
+	entry, ok := state["resources.jobs.dst"]
+	require.True(t, ok)
+
+	// depends_on must point back to the referenced pipeline
+	require.Len(t, entry.DependsOn, 1)
+	assert.Equal(t, deployplan.DependsOnEntry{
+		Node:  "resources.pipelines.src",
+		Label: "${resources.pipelines.src.name}",
+	}, entry.DependsOn[0])
+
+	// resolved field value
+	var jobState map[string]any
+	require.NoError(t, json.Unmarshal(entry.State, &jobState))
+	assert.Equal(t, "source-pipeline", jobState["name"])
+}
+
+func TestBuildStateFromTF_NumericFieldReference(t *testing.T) {
+	// dst_job.max_concurrent_runs references src_job's int field.
+	// Verifies that the resolved value is stored as a number (not a string)
+	// and that depends_on is recorded.
+	bundleYAML := `
+resources:
+  jobs:
+    src_job:
+      name: "source"
+      max_concurrent_runs: 4
+    dst_job:
+      name: "dest"
+      max_concurrent_runs: "${resources.jobs.src_job.max_concurrent_runs}"
+`
+	tfAttrs := migrate.TFStateAttrs{
+		"databricks_job": {
+			"src_job": json.RawMessage(`{"id": "111", "name": "source", "max_concurrent_runs": 4}`),
+			"dst_job": json.RawMessage(`{"id": "222", "name": "dest",   "max_concurrent_runs": 4}`),
+		},
+	}
+	tfIDs := terraform.ExportedResourcesMap{
+		"resources.jobs.src_job": {ID: "111"},
+		"resources.jobs.dst_job": {ID: "222"},
+	}
+
+	state := runBuildStateFromTF(t, bundleYAML, tfAttrs, tfIDs)
+
+	entry, ok := state["resources.jobs.dst_job"]
+	require.True(t, ok)
+
+	// depends_on must point to src_job
+	require.Len(t, entry.DependsOn, 1)
+	assert.Equal(t, "resources.jobs.src_job", entry.DependsOn[0].Node)
+
+	// max_concurrent_runs must be stored as a number, not a string
+	var jobState map[string]any
+	require.NoError(t, json.Unmarshal(entry.State, &jobState))
+	assert.EqualValues(t, 4, jobState["max_concurrent_runs"])
+}
+
+func TestBuildStateFromTF_EtagStoredForDashboard(t *testing.T) {
+	// Etag is a top-level attribute in the TF dashboard state JSON; no separate map needed.
+	bundleYAML := `
+resources:
+  dashboards:
+    my_dash:
+      display_name: "My Dashboard"
+`
+	tfAttrs := migrate.TFStateAttrs{
+		"databricks_dashboard": {
+			"my_dash": json.RawMessage(`{"id": "d1", "display_name": "My Dashboard", "etag": "etag-abc123"}`),
+		},
+	}
+	tfIDs := terraform.ExportedResourcesMap{
+		"resources.dashboards.my_dash": {ID: "d1"},
+	}
+
+	state := runBuildStateFromTF(t, bundleYAML, tfAttrs, tfIDs)
+	entry, ok := state["resources.dashboards.my_dash"]
+	require.True(t, ok)
+
+	var dashState map[string]any
+	require.NoError(t, json.Unmarshal(entry.State, &dashState))
+	assert.Equal(t, "etag-abc123", dashState["etag"])
+}
diff --git a/bundle/migrate/resolve.go b/bundle/migrate/resolve.go
new file mode 100644
index 00000000000..8d5f6bc6b35
--- /dev/null
+++ b/bundle/migrate/resolve.go
@@ -0,0 +1,91 @@
+package migrate
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/dyn/dynvar"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/structs/structpath"
+)
+
+// evaluateTemplate evaluates a template string like "${resources.pipelines.bar.cluster[0].label}"
+// by looking up each ${...} reference from TF state.
+func evaluateTemplate(state TFStateAttrs, template string) (string, error) {
+	ref, ok := dynvar.NewRef(dyn.V(template))
+	if !ok {
+		return template, nil
+	}
+
+	result := template
+	for _, pathString := range ref.References() {
+		path, err := structpath.ParsePath(pathString)
+		if err != nil {
+			return "", fmt.Errorf("cannot parse reference path %q: %w", pathString, err)
+		}
+		// Expect resources.<group>.<name>.<field...>
+		if path.Len() < 4 {
+			return "", fmt.Errorf("unexpected reference format (too short): %q", pathString)
+		}
+		// Check first component is "resources"
+		firstNode := path.Prefix(1)
+		if firstNode.String() != "resources" {
+			return "", fmt.Errorf("unexpected reference format (expected resources.*): %q", pathString)
+		}
+
+		group := path.SkipPrefix(1).Prefix(1).String()
+		name := path.SkipPrefix(2).Prefix(1).String()
+		fieldPath := path.SkipPrefix(3)
+
+		value, err := LookupTFField(state, group, name, fieldPath)
+		if err != nil {
+			return "", fmt.Errorf("cannot look up %q: %w", pathString, err)
+		}
+
+		result = strings.ReplaceAll(result, "${"+pathString+"}", fmt.Sprintf("%v", value))
+	}
+	return result, nil
+}
+
+// ResolveFieldRef resolves a single reference for a field in resource (srcGroup, srcName).
+// fieldPath is the path of the field within the source resource (in DABs naming, from sv.Refs key).
+// refTemplate is the template string for that field, e.g. "${resources.pipelines.bar.cluster[0].label}".
+//
+// Two methods are tried:
+//   - Method A: read the field from the source resource's own TF state.
+//   - Method B: evaluate the template by reading each referenced field from TF state.
+//
+// Returns the reconciled value or an error if both methods fail.
+func ResolveFieldRef(ctx context.Context, state TFStateAttrs, srcGroup, srcName string, fieldPath *structpath.PathNode, refTemplate string) (any, error) {
+	// Method A: read field from source resource's TF state.
+	valueA, errA := LookupTFField(state, srcGroup, srcName, fieldPath)
+
+	// Method B: evaluate the template by looking up each reference.
+	valueB, errB := evaluateTemplate(state, refTemplate)
+
+	switch {
+	case errA == nil && errB == nil:
+		aStr := fmt.Sprintf("%v", valueA)
+		if aStr == valueB {
+			return valueA, nil
+		}
+		// Both succeeded but disagree: prefer longer string and warn.
+		if len(valueB) > len(aStr) {
+			log.Warnf(ctx, "resource %s.%s field %s: method A value %q and method B value %q disagree; using longer (method B)",
+				srcGroup, srcName, fieldPath, aStr, valueB)
+			return valueB, nil
+		}
+		log.Warnf(ctx, "resource %s.%s field %s: method A value %q and method B value %q disagree; using longer (method A)",
+			srcGroup, srcName, fieldPath, aStr, valueB)
+		return valueA, nil
+	case errA == nil:
+		return valueA, nil
+	case errB == nil:
+		return valueB, nil
+	default:
+		return nil, fmt.Errorf("%s.%s field %s: method A: %w; method B: %w",
+			srcGroup, srcName, fieldPath, errA, errB)
+	}
+}
diff --git a/bundle/migrate/resolve_test.go b/bundle/migrate/resolve_test.go
new file mode 100644
index 00000000000..221c873a73d
--- /dev/null
+++ b/bundle/migrate/resolve_test.go
@@ -0,0 +1,78 @@
+package migrate_test
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/databricks/cli/bundle/migrate"
+	"github.com/databricks/cli/libs/structs/structaccess"
+	"github.com/databricks/cli/libs/structs/structpath"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// state with src job having int and bool fields set.
+func testState() migrate.TFStateAttrs {
+	return migrate.TFStateAttrs{
+		"databricks_job": {
+			"src": json.RawMessage(`{
+				"id": "111",
+				"max_concurrent_runs": 4,
+				"always_running": true
+			}`),
+			"dst": json.RawMessage(`{
+				"id": "222",
+				"max_concurrent_runs": 4,
+				"always_running": true
+			}`),
+		},
+	}
+}
+
+// TestResolveFieldRefInt proves that when Method B (template evaluation) wins for
+// an int field, the returned string value is still usable: structaccess.Set must
+// parse it back to int and not error.
+func TestResolveFieldRefInt(t *testing.T) {
+	state := testState()
+	// Remove dst from state so Method A fails and Method B must be used.
+	delete(state["databricks_job"], "dst")
+
+	ctx := t.Context()
+	fieldPath, err := structpath.ParsePath("max_concurrent_runs")
+	require.NoError(t, err)
+
+	value, err := migrate.ResolveFieldRef(ctx, state, "jobs", "dst", fieldPath,
+		"${resources.jobs.src.max_concurrent_runs}")
+	require.NoError(t, err)
+
+	// Method B succeeds: returns string "4". Verify Set converts it to int.
+	type target struct {
+		MaxConcurrentRuns int `json:"max_concurrent_runs"`
+	}
+	s := &target{}
+	err = structaccess.Set(s, fieldPath, value)
+	assert.NoError(t, err, "Set should parse string %q into int field", value)
+	assert.Equal(t, 4, s.MaxConcurrentRuns)
+}
+
+// TestResolveFieldRefBool is the same for a bool field.
+func TestResolveFieldRefBool(t *testing.T) {
+	state := testState()
+	delete(state["databricks_job"], "dst")
+
+	ctx := t.Context()
+	fieldPath, err := structpath.ParsePath("always_running")
+	require.NoError(t, err)
+
+	value, err := migrate.ResolveFieldRef(ctx, state, "jobs", "dst", fieldPath,
+		"${resources.jobs.src.always_running}")
+	require.NoError(t, err)
+
+	type target struct {
+		AlwaysRunning bool `json:"always_running"`
+	}
+	s := &target{}
+	err = structaccess.Set(s, fieldPath, value)
+	assert.NoError(t, err, "Set should parse string %q into bool field", value)
+	assert.True(t, s.AlwaysRunning)
+}
diff --git a/bundle/migrate/tf_state.go b/bundle/migrate/tf_state.go
new file mode 100644
index 00000000000..72b64e77f19
--- /dev/null
+++ b/bundle/migrate/tf_state.go
@@ -0,0 +1,217 @@
+package migrate
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/terraform_dabs_map"
+	"github.com/databricks/cli/libs/structs/structpath"
+	tfjson "github.com/hashicorp/terraform-json"
+)
+
+// tfStateFieldAliases maps DABs group → DABs field name → TF state field name for
+// cases where a DABs state-computed field has a different name in TF state.
+// These fields are not captured by DABsToTerraformRenameMap because they are
+// state-only (not part of the bundle config struct).
+var tfStateFieldAliases = map[string]map[string]string{
+	// models.model_id is the numeric model ID; TF stores it as registered_model_id.
+	"models": {"model_id": "registered_model_id"},
+}
+
+// TFStateAttrs maps (tfResourceType → resourceName → raw JSON attributes).
+type TFStateAttrs map[string]map[string]json.RawMessage
+
+// TFStateMeta holds the top-level metadata from a terraform state file.
+type TFStateMeta struct {
+	Lineage string
+	Serial  int
+}
+
+// ParseTFStateFull reads the terraform state file once and returns the full
+// attribute map, the resource ID map, and the state metadata (lineage/serial).
+// Avoids reading and unmarshaling the file multiple times.
+func ParseTFStateFull(ctx context.Context, path string) (TFStateAttrs, terraform.ExportedResourcesMap, TFStateMeta, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, nil, TFStateMeta{}, nil
+		}
+		return nil, nil, TFStateMeta{}, err
+	}
+
+	var meta struct {
+		Lineage string `json:"lineage"`
+		Serial  int    `json:"serial"`
+	}
+	if err := json.Unmarshal(raw, &meta); err != nil {
+		return nil, nil, TFStateMeta{}, err
+	}
+
+	attrs, err := parseTFStateAttrsFromBytes(raw)
+	if err != nil {
+		return nil, nil, TFStateMeta{}, err
+	}
+	ids, err := terraform.ParseResourcesStateFromBytes(ctx, raw)
+	if err != nil {
+		return nil, nil, TFStateMeta{}, err
+	}
+	return attrs, ids, TFStateMeta{Lineage: meta.Lineage, Serial: meta.Serial}, nil
+}
+
+// ParseTFStateAttrs parses the terraform state file returning full attribute JSON per resource.
+//
+//deadcode:allow retained as standalone API for callers that only need attributes.
+func ParseTFStateAttrs(path string) (TFStateAttrs, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	return parseTFStateAttrsFromBytes(raw)
+}
+
+func parseTFStateAttrsFromBytes(raw []byte) (TFStateAttrs, error) {
+	var state struct {
+		Version   int `json:"version"`
+		Resources []struct {
+			Type      string              `json:"type"`
+			Name      string              `json:"name"`
+			Mode      tfjson.ResourceMode `json:"mode"`
+			Instances []struct {
+				Attributes json.RawMessage `json:"attributes"`
+			} `json:"instances"`
+		} `json:"resources"`
+	}
+	if err := json.Unmarshal(raw, &state); err != nil {
+		return nil, err
+	}
+	result := make(TFStateAttrs)
+	for _, r := range state.Resources {
+		if r.Mode != tfjson.ManagedResourceMode || len(r.Instances) == 0 {
+			continue
+		}
+		if result[r.Type] == nil {
+			result[r.Type] = make(map[string]json.RawMessage)
+		}
+		result[r.Type][r.Name] = r.Instances[0].Attributes
+	}
+	return result, nil
+}
+
+// LookupTFField looks up a field from TF state attributes for a bundle resource.
+// group is the DABs group (e.g. "pipelines"), name is the resource name.
+// fieldPath is the path to the field (may be in DABs or TF naming; both handled by DABsPathToTerraform).
+func LookupTFField(state TFStateAttrs, group, name string, fieldPath *structpath.PathNode) (any, error) {
+	tfType, ok := terraform.GroupToTerraformName[group]
+	if !ok {
+		return nil, fmt.Errorf("unknown resource group %q", group)
+	}
+
+	// Translate field path to TF naming.
+	// DABsPathToTerraform handles both DABs names (renames) and TF names (pass-through for unknowns).
+	// Returns error for known DABs-only fields that have no TF equivalent.
+	tfFieldPath, err := terraform_dabs_map.DABsPathToTerraform(group, fieldPath)
+	if err != nil {
+		return nil, err
+	}
+
+	attrsJSON, ok := state[tfType][name]
+	if !ok {
+		return nil, fmt.Errorf("%s.%s not found in TF state", tfType, name)
+	}
+
+	// Unmarshal into map[string]any to handle TF list-blocks: in TF state, single-block
+	// fields are stored as single-element arrays [{"field": "value"}], not as plain objects.
+	// Navigating via map avoids the json.Unmarshal type mismatch between []T in JSON and
+	// struct-typed schema fields.
+	var attrs map[string]any
+	if err := json.Unmarshal(attrsJSON, &attrs); err != nil {
+		return nil, fmt.Errorf("cannot parse TF state for %s.%s: %w", tfType, name, err)
+	}
+
+	value, err := navigateTFState(attrs, tfFieldPath)
+	if err == nil {
+		return value, nil
+	}
+
+	// Some DABs fields are top-level in TF state but DABsPathToTerraform added a
+	// wrapper prefix (e.g. "spec" for postgres resources). When the wrapped path
+	// fails, retry with the original unwrapped path.
+	if _, hasWrapper := terraform_dabs_map.DABsToTerraformWrappers[group]; hasWrapper {
+		if v, e := navigateTFState(attrs, fieldPath); e == nil {
+			return v, nil
+		}
+	}
+
+	// Apply state-only field aliases for fields whose DABs name differs from TF state name.
+	if aliases, ok := tfStateFieldAliases[group]; ok {
+		// Replace the first path segment if it matches a known alias.
+		if head, ok := fieldPath.StringKey(); ok {
+			if tfName, ok := aliases[head]; ok {
+				aliasPath := structpath.NewStringKey(nil, tfName)
+				if rest := fieldPath.SkipPrefix(1); rest != nil {
+					_ = rest // navigate through the alias root
+				}
+				// Translate aliased path with full DABsToTerraform for the renamed field.
+				if aliasFieldPath, e := terraform_dabs_map.DABsPathToTerraform(group, aliasPath); e == nil {
+					if v, e := navigateTFState(attrs, aliasFieldPath); e == nil {
+						return v, nil
+					}
+				}
+			}
+		}
+	}
+
+	return nil, err
+}
+
+// navigateTFState walks the TF state map using the given path.
+// TF stores single-block fields as single-element arrays ([{…}]).  When a string-key
+// step encounters a []any, it auto-descends into element [0] so callers can use plain
+// paths like "continuous.pause_status" even though TF stores them as [{"pause_status":…}].
+func navigateTFState(data map[string]any, path *structpath.PathNode) (any, error) {
+	var current any = data
+	for _, node := range path.AsSlice() {
+		if current == nil {
+			return nil, nil
+		}
+
+		if key, ok := node.StringKey(); ok {
+			// Auto-unwrap TF list-blocks: if the current value is a single-element
+			// array and the next step wants a map key, descend into element 0.
+			if arr, isArr := current.([]any); isArr {
+				if len(arr) == 0 {
+					return nil, nil
+				}
+				current = arr[0]
+			}
+			m, ok := current.(map[string]any)
+			if !ok {
+				return nil, fmt.Errorf("expected map at %q, got %T", key, current)
+			}
+			val, ok := m[key]
+			if !ok {
+				return nil, fmt.Errorf("%q: key not found", key)
+			}
+			current = val
+		} else if idx, ok := node.Index(); ok {
+			switch v := current.(type) {
+			case []any:
+				if idx < 0 || idx >= len(v) {
+					return nil, fmt.Errorf("index %d out of range (len %d)", idx, len(v))
+				}
+				current = v[idx]
+			default:
+				// TF [0] on a non-slice (already unwrapped) is a no-op.
+				if idx == 0 {
+					continue
+				}
+				return nil, fmt.Errorf("index %d: not a slice (%T)", idx, current)
+			}
+		}
+	}
+	return current, nil
+}
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index 3cac322f9e3..78a06ef732e 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -14,7 +14,6 @@ import (
 	"github.com/databricks/cli/bundle/deploy/metadata"
 	"github.com/databricks/cli/bundle/deploy/terraform"
 	"github.com/databricks/cli/bundle/deployplan"
-	"github.com/databricks/cli/bundle/direct"
 	"github.com/databricks/cli/bundle/libraries"
 	"github.com/databricks/cli/bundle/metrics"
 	"github.com/databricks/cli/bundle/permissions"
@@ -81,7 +80,7 @@ func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, ta
 		err   error
 	)
 	if targetEngine.IsDirect() {
-		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan, direct.MigrateMode(false))
+		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan)
 		state, err = b.DeploymentBundle.StateDB.Finalize(ctx)
 	} else {
 		bundle.ApplyContext(ctx, b, terraform.Apply())
diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go
index 95eec600dc2..e1196fbc7f4 100644
--- a/bundle/phases/destroy.go
+++ b/bundle/phases/destroy.go
@@ -12,7 +12,6 @@ import (
 	"github.com/databricks/cli/bundle/deploy/lock"
 	"github.com/databricks/cli/bundle/deploy/terraform"
 	"github.com/databricks/cli/bundle/deployplan"
-	"github.com/databricks/cli/bundle/direct"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/log"
@@ -76,7 +75,7 @@ func approvalForDestroy(ctx context.Context, b *bundle.Bundle, plan *deployplan.
 
 func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, engine engine.EngineType) {
 	if engine.IsDirect() {
-		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan, direct.MigrateMode(false))
+		b.DeploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan)
 	} else {
 		// Core destructive mutators for destroy. These require informed user consent.
 		bundle.ApplyContext(ctx, b, terraform.Apply())
diff --git a/bundle/statemgmt/upload_state_for_yaml_sync.go b/bundle/statemgmt/upload_state_for_yaml_sync.go
index 0399c7b31ff..77a010488a0 100644
--- a/bundle/statemgmt/upload_state_for_yaml_sync.go
+++ b/bundle/statemgmt/upload_state_for_yaml_sync.go
@@ -14,18 +14,16 @@ import (
 	"github.com/databricks/cli/bundle/config/mutator/resourcemutator"
 	"github.com/databricks/cli/bundle/deploy"
 	"github.com/databricks/cli/bundle/deploy/terraform"
-	"github.com/databricks/cli/bundle/deployplan"
-	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dresources"
 	"github.com/databricks/cli/bundle/direct/dstate"
 	"github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/bundle/migrate"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/dynvar"
 	"github.com/databricks/cli/libs/filer"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
-	"github.com/databricks/cli/libs/structs/structaccess"
-	"github.com/databricks/cli/libs/structs/structpath"
 )
 
 type uploadStateForYamlSync struct {
@@ -100,49 +98,31 @@ func uploadState(ctx context.Context, b *bundle.Bundle) error {
 }
 
 func (m *uploadStateForYamlSync) convertState(ctx context.Context, b *bundle.Bundle, snapshotPath string) (bool, error) {
-	terraformResources, err := terraform.ParseResourcesState(ctx, b)
+	_, localTerraformPath := b.StateFilenameTerraform(ctx)
+	tfAttrs, terraformResources, tfMeta, err := migrate.ParseTFStateFull(ctx, localTerraformPath)
 	if err != nil {
 		return false, fmt.Errorf("failed to parse terraform state: %w", err)
 	}
 
-	// ParseResourcesState returns nil when the terraform state file doesn't exist
+	// ParseTFStateFull returns nil IDs when the terraform state file doesn't exist
 	// (e.g. first deploy with no resources).
 	if terraformResources == nil {
 		return false, nil
 	}
 
-	_, localTerraformPath := b.StateFilenameTerraform(ctx)
-	data, err := os.ReadFile(localTerraformPath)
-	if err != nil {
-		return false, fmt.Errorf("failed to read terraform state: %w", err)
-	}
-
 	state := make(map[string]dstate.ResourceEntry)
-	etags := map[string]string{}
-
 	for key, resourceEntry := range terraformResources {
 		state[key] = dstate.ResourceEntry{
 			ID:    resourceEntry.ID,
 			State: json.RawMessage("{}"),
 		}
-		if resourceEntry.ETag != "" {
-			etags[key] = resourceEntry.ETag
-		}
-	}
-
-	var tfState struct {
-		Lineage string `json:"lineage"`
-		Serial  int    `json:"serial"`
-	}
-	if err := json.Unmarshal(data, &tfState); err != nil {
-		return false, err
 	}
 
-	migratedDB := dstate.NewDatabase(tfState.Lineage, tfState.Serial+1)
+	migratedDB := dstate.NewDatabase(tfMeta.Lineage, tfMeta.Serial+1)
 	migratedDB.State = state
 
-	deploymentBundle := &direct.DeploymentBundle{}
-	deploymentBundle.StateDB.OpenWithData(snapshotPath, migratedDB)
+	var stateDB dstate.DeploymentState
+	stateDB.OpenWithData(snapshotPath, migratedDB)
 
 	// Apply SecretScopeFixups so the config matches what the direct engine expects.
 	// This adds MANAGE ACL for the current user to all secret scopes, ensuring
@@ -152,9 +132,9 @@ func (m *uploadStateForYamlSync) convertState(ctx context.Context, b *bundle.Bun
 		return false, errors.New("failed to apply secret scope fixups")
 	}
 
-	// Get the dynamic value from b.Config and reverse the interpolation.
 	// b.Config has been modified by terraform.Interpolate which converts bundle-style
 	// references (${resources.pipelines.x.id}) to terraform-style (${databricks_pipeline.x.id}).
+	// BuildStateFromTF expects ${resources.*} references, so reverse the interpolation first.
 	interpolatedRoot := b.Config.Value()
 	uninterpolatedRoot, err := reverseInterpolate(interpolatedRoot)
 	if err != nil {
@@ -169,36 +149,20 @@ func (m *uploadStateForYamlSync) convertState(ctx context.Context, b *bundle.Bun
 		return false, fmt.Errorf("failed to create uninterpolated config: %w", err)
 	}
 
-	plan, err := deploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(ctx), &uninterpolatedConfig)
+	adapters, err := dresources.InitAll(nil)
 	if err != nil {
 		return false, err
 	}
 
-	for _, entry := range plan.Plan {
-		entry.Action = deployplan.Update
-	}
-
-	for key := range plan.Plan {
-		etag := etags[key]
-		if etag == "" {
-			continue
-		}
-		sv, ok := deploymentBundle.StateCache.Load(key)
-		if !ok {
-			continue
-		}
-		err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etag)
-		if err != nil {
-			log.Warnf(ctx, "Failed to set etag on %q: %v", key, err)
-		}
+	if err := stateDB.UpgradeToWrite(); err != nil {
+		return false, fmt.Errorf("upgrading state for apply: %w", err)
 	}
 
-	if err := deploymentBundle.StateDB.UpgradeToWrite(); err != nil {
-		return false, fmt.Errorf("upgrading state for apply: %w", err)
+	if err := migrate.BuildStateFromTF(ctx, &uninterpolatedConfig, adapters, &stateDB, tfAttrs, terraformResources); err != nil {
+		return false, err
 	}
 
-	deploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan, direct.MigrateMode(true))
-	if _, err := deploymentBundle.StateDB.Finalize(ctx); err != nil {
+	if _, err := stateDB.Finalize(ctx); err != nil {
 		return false, err
 	}
 
diff --git a/cmd/bundle/deployment/migrate.go b/cmd/bundle/deployment/migrate.go
index 801e46f7918..3c17e5c4be5 100644
--- a/cmd/bundle/deployment/migrate.go
+++ b/cmd/bundle/deployment/migrate.go
@@ -1,124 +1,55 @@
 package deployment
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
-	"os/exec"
 	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/bundle/config/mutator/resourcemutator"
-	"github.com/databricks/cli/bundle/deploy/terraform"
-	"github.com/databricks/cli/bundle/deployplan"
-	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dresources"
 	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/bundle/migrate"
 	"github.com/databricks/cli/cmd/bundle/utils"
 	"github.com/databricks/cli/cmd/root"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/env"
 	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/cli/libs/shellquote"
-	"github.com/databricks/cli/libs/structs/structaccess"
-	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/spf13/cobra"
 )
 
 const backupSuffix = ".backup"
 
-// runPlanCheck runs bundle plan and checks if there are any actions planned.
-// Returns error if plan fails or if there are actions planned.
-func runPlanCheck(cmd *cobra.Command, extraArgs []string, extraArgsStr string) error {
-	ctx := cmd.Context()
-
-	executable, err := os.Executable()
-	if err != nil {
-		return fmt.Errorf("failed to get executable path: %w", err)
-	}
-
-	args := []string{"bundle", "plan"}
-	args = append(args, extraArgs...)
-
-	planCmd := exec.CommandContext(ctx, executable, args...)
-	var stdout bytes.Buffer
-	planCmd.Stdout = &stdout
-	planCmd.Stderr = cmd.ErrOrStderr()
-
-	// Use the engine encoded in the state
-	planCmd.Env = append(os.Environ(), "DATABRICKS_BUNDLE_ENGINE=terraform")
-
-	err = planCmd.Run()
-
-	// Output the plan stdout as is
-	output := stdout.String()
-	fmt.Fprint(cmd.OutOrStdout(), output)
-
-	if err != nil {
-		msg := ""
-		if exitErr, ok := errors.AsType[*exec.ExitError](err); ok {
-			msg = fmt.Sprintf("exit code %d", exitErr.ExitCode())
-		} else {
-			msg = err.Error()
-		}
-		return fmt.Errorf("bundle plan failed with %s, aborting migration. To proceed with migration anyway, re-run the command with --noplancheck option", msg)
-	}
-
-	if !strings.Contains(output, "Plan:") {
-		return fmt.Errorf("cannot parse 'databricks bundle plan%s' output, aborting migration. Skip plan check with --noplancheck option", extraArgsStr)
-	}
-
-	if !strings.Contains(output, "Plan: 0 to add, 0 to change, 0 to delete") {
-		return fmt.Errorf("'databricks bundle plan%s' shows actions planned, aborting migration. Please run 'databricks bundle deploy%s' first to ensure your bundle is up to date, If actions persist after deploy, skip plan check with --noplancheck option", extraArgsStr, extraArgsStr)
-	}
-
-	return nil
-}
-
-func getCommonArgs(cmd *cobra.Command) ([]string, string) {
-	var args []string
+func getCommonArgs(cmd *cobra.Command) string {
 	var quotedArgs []string
 
 	if flag := cmd.Flag("target"); flag != nil && flag.Changed {
-		target := flag.Value.String()
-		if target != "" {
-			args = append(args, "-t")
-			args = append(args, target)
-			quotedArgs = append(quotedArgs, "-t")
-			quotedArgs = append(quotedArgs, shellquote.BashArg(target))
+		if target := flag.Value.String(); target != "" {
+			quotedArgs = append(quotedArgs, "-t", shellquote.BashArg(target))
 		}
 	}
 	if flag := cmd.Flag("profile"); flag != nil && flag.Changed {
-		profile := flag.Value.String()
-		if profile != "" {
-			args = append(args, "-p")
-			args = append(args, profile)
-			quotedArgs = append(quotedArgs, "-p")
-			quotedArgs = append(quotedArgs, shellquote.BashArg(profile))
+		if profile := flag.Value.String(); profile != "" {
+			quotedArgs = append(quotedArgs, "-p", shellquote.BashArg(profile))
 		}
 	}
 	if flag := cmd.Flag("var"); flag != nil && flag.Changed {
-		varValues, err := cmd.Flags().GetStringSlice("var")
-		if err == nil {
+		if varValues, err := cmd.Flags().GetStringSlice("var"); err == nil {
 			for _, v := range varValues {
-				args = append(args, "--var")
-				args = append(args, v)
-				quotedArgs = append(quotedArgs, "--var")
-				quotedArgs = append(quotedArgs, shellquote.BashArg(v))
+				quotedArgs = append(quotedArgs, "--var", shellquote.BashArg(v))
 			}
 		}
 	}
 
-	argsStr := ""
-
-	if len(quotedArgs) > 0 {
-		argsStr = " " + strings.Join(quotedArgs, " ")
+	if len(quotedArgs) == 0 {
+		return ""
 	}
-
-	return args, argsStr
+	return " " + strings.Join(quotedArgs, " ")
 }
 
 func newMigrateCommand() *cobra.Command {
@@ -136,11 +67,12 @@ to the workspace so that subsequent deploys of this bundle use direct deployment
 		Args: root.NoArgs,
 	}
 
-	var noPlanCheck bool
-	cmd.Flags().BoolVar(&noPlanCheck, "noplancheck", false, "Skip running bundle plan before migration.")
+	// --noplancheck kept for backward compatibility; the plan check was removed
+	// because the command no longer invokes the Terraform engine.
+	cmd.Flags().Bool("noplancheck", false, "No-op (kept for compatibility).")
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
-		extraArgs, extraArgsStr := getCommonArgs(cmd)
+		extraArgsStr := getCommonArgs(cmd)
 
 		// Clear the engine env var so migrate always uses terraform engine to read existing state,
 		// regardless of what the user may have set in their environment.
@@ -181,7 +113,7 @@ To start using direct engine, set "engine: direct" under bundle in your databric
 			return fmt.Errorf("reading %s: %w", localTerraformPath, err)
 		}
 
-		terraformResources, err := terraform.ParseResourcesState(ctx, b)
+		tfAttrs, terraformResources, _, err := migrate.ParseTFStateFull(ctx, localTerraformPath)
 		if err != nil {
 			return fmt.Errorf("failed to parse terraform state: %w", err)
 		}
@@ -201,33 +133,19 @@ To start using direct engine, set "engine: direct" under bundle in your databric
 			return fmt.Errorf("state file %s already exists", localPath)
 		}
 
-		// Run plan check unless --noplancheck is set
-		if !noPlanCheck {
-			cmdio.LogString(ctx, "Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:")
-			if err = runPlanCheck(cmd, extraArgs, extraArgsStr); err != nil {
-				return err
-			}
-		}
-
-		etags := map[string]string{}
-
 		state := make(map[string]dstate.ResourceEntry)
 		for key, resourceEntry := range terraformResources {
 			state[key] = dstate.ResourceEntry{
 				ID:    resourceEntry.ID,
 				State: json.RawMessage("{}"),
 			}
-			if resourceEntry.ETag != "" {
-				// dashboard:
-				etags[key] = resourceEntry.ETag
-			}
 		}
 
 		migratedDB := dstate.NewDatabase(stateDesc.Lineage, stateDesc.Serial+1)
 		migratedDB.State = state
 
-		deploymentBundle := &direct.DeploymentBundle{}
-		deploymentBundle.StateDB.OpenWithData(tempStatePath, migratedDB)
+		var stateDB dstate.DeploymentState
+		stateDB.OpenWithData(tempStatePath, migratedDB)
 
 		tempStatePathAutoRemove := true
 
@@ -245,43 +163,20 @@ To start using direct engine, set "engine: direct" under bundle in your databric
 			return root.ErrAlreadyPrinted
 		}
 
-		plan, err := deploymentBundle.CalculatePlan(ctx, b.WorkspaceClient(ctx), &b.Config)
+		adapters, err := dresources.InitAll(nil)
 		if err != nil {
 			return err
 		}
 
-		for _, entry := range plan.Plan {
-			// Force all actions to be "update" so that deploym below goes through every resource
-			entry.Action = deployplan.Update
-		}
-
-		// We need to copy ETag into new state.
-		// For most resources state consists of fully resolved local config snapshot + id.
-		// Dashboards are special in that they also store "etag" in state which is not provided by user but
-		// comes from remote state. If we don't store "etag" in state, we won't detect remote drift, because
-		// local=nil, remote="<some new etag>" which will be classified as a backend default and skipped.
-
-		for key := range plan.Plan {
-			etag := etags[key]
-			if etag == "" {
-				continue
-			}
-			sv, ok := deploymentBundle.StateCache.Load(key)
-			if !ok {
-				return fmt.Errorf("failed to read state for %q", key)
-			}
-			err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etag)
-			if err != nil {
-				return fmt.Errorf("failed to set etag on %q: %w", key, err)
-			}
+		if err := stateDB.UpgradeToWrite(); err != nil {
+			return fmt.Errorf("upgrading state for apply: %w", err)
 		}
 
-		if err := deploymentBundle.StateDB.UpgradeToWrite(); err != nil {
-			return fmt.Errorf("upgrading state for apply: %w", err)
+		if err := migrate.BuildStateFromTF(ctx, &b.Config, adapters, &stateDB, tfAttrs, terraformResources); err != nil {
+			return err
 		}
 
-		deploymentBundle.Apply(ctx, b.WorkspaceClient(ctx), plan, direct.MigrateMode(true))
-		if _, err := deploymentBundle.StateDB.Finalize(ctx); err != nil {
+		if _, err := stateDB.Finalize(ctx); err != nil {
 			logdiag.LogError(ctx, err)
 		}
 		if logdiag.HasError(ctx) {