diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md
index 989a1f273b0..d72fba264f6 100644
--- a/NEXT_CHANGELOG.md
+++ b/NEXT_CHANGELOG.md
@@ -16,6 +16,7 @@
 * Fix missing field descriptions in the bundle JSON schema for fields whose upstream API docs arrived after the field was first annotated (e.g. `vector_search_endpoints.*.target_qps`); stale placeholder markers no longer hide them ([#5588](https://github.com/databricks/cli/pull/5588)).
 * Fix `bundle deploy --plan` dropping a `postgres_role`'s `role_id`, which caused the role to be recreated on the next deploy ([#5672](https://github.com/databricks/cli/pull/5672)).
 * direct: Fix spurious cluster recreate when `apply_policy_default_values: true` is set ([#5693](https://github.com/databricks/cli/pull/5693)).
+* direct: New 'deployment migrate' implementation that parses terraform state instead of fetching resources state from the backend ([#5399](https://github.com/databricks/cli/pull/5399)).
 
 ### Dependency updates
 * Bump `github.com/databricks/databricks-sdk-go` from v0.141.0 to v0.147.0 ([#5636](https://github.com/databricks/cli/pull/5636)).
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/deploy/yaml-sync-empty-grants/output.txt b/acceptance/bundle/deploy/yaml-sync-empty-grants/output.txt
index fa2643665d8..e037b25fd29 100644
--- a/acceptance/bundle/deploy/yaml-sync-empty-grants/output.txt
+++ b/acceptance/bundle/deploy/yaml-sync-empty-grants/output.txt
@@ -3,6 +3,4 @@
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/yaml-sync-empty-grants/default/files...
 Deploying resources...
 Updating deployment state...
-Warn: Failed to create config snapshot: state conversion failed
-Warn: Config snapshot: state entry not found for "resources.schemas.schema1.grants"
 Deployment complete!
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/invariant/configs/job_run_job_ref.yml.tmpl b/acceptance/bundle/invariant/configs/job_run_job_ref.yml.tmpl
new file mode 100644
index 00000000000..c84f7cd5145
--- /dev/null
+++ b/acceptance/bundle/invariant/configs/job_run_job_ref.yml.tmpl
@@ -0,0 +1,34 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+resources:
+  jobs:
+    trigger_job:
+      name: test-trigger-$UNIQUE_NAME
+      # 9007199254740993 = 2^53+1, the smallest integer float64 cannot represent
+      # exactly (it rounds to 2^53 = ...992). Stored as a literal via PrepareState,
+      # so trigger_job keeps the exact value; watcher_job references it to force the
+      # migration to resolve it from TF state, where naive float64 parsing of JSON
+      # numbers would corrupt it. Local-only: deliberately out of range for the
+      # real backend (see no_run_job_ref_on_cloud in the parent test.toml).
+      max_concurrent_runs: 9007199254740993
+      tasks:
+        - task_key: notebook
+          notebook_task:
+            notebook_path: /Shared/notebook
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            instance_pool_id: $TEST_INSTANCE_POOL_ID
+            num_workers: 1
+
+    watcher_job:
+      name: test-watcher-$UNIQUE_NAME
+      # Resolved from trigger_job's TF state during migration. Exercises the
+      # json.Number parsing path end-to-end: with the fix the migrated state keeps
+      # 9007199254740993; without it the value is silently truncated to ...992.
+      max_concurrent_runs: ${resources.jobs.trigger_job.max_concurrent_runs}
+      tasks:
+        - task_key: run_trigger
+          run_job_task:
+            job_id: ${resources.jobs.trigger_job.id}
diff --git a/acceptance/bundle/invariant/continue_293/out.test.toml b/acceptance/bundle/invariant/continue_293/out.test.toml
index d9bdc5b830b..5c601542bce 100644
--- a/acceptance/bundle/invariant/continue_293/out.test.toml
+++ b/acceptance/bundle/invariant/continue_293/out.test.toml
@@ -19,6 +19,7 @@ EnvMatrix.INPUT_CONFIG = [
   "job_pydabs_1000_tasks.yml.tmpl",
   "job_cross_resource_ref.yml.tmpl",
   "job_permission_ref.yml.tmpl",
+  "job_run_job_ref.yml.tmpl",
   "job_with_depends_on.yml.tmpl",
   "job_with_permissions.yml.tmpl",
   "job_with_task.yml.tmpl",
diff --git a/acceptance/bundle/invariant/migrate/out.test.toml b/acceptance/bundle/invariant/migrate/out.test.toml
index d9bdc5b830b..5c601542bce 100644
--- a/acceptance/bundle/invariant/migrate/out.test.toml
+++ b/acceptance/bundle/invariant/migrate/out.test.toml
@@ -19,6 +19,7 @@ EnvMatrix.INPUT_CONFIG = [
   "job_pydabs_1000_tasks.yml.tmpl",
   "job_cross_resource_ref.yml.tmpl",
   "job_permission_ref.yml.tmpl",
+  "job_run_job_ref.yml.tmpl",
   "job_with_depends_on.yml.tmpl",
   "job_with_permissions.yml.tmpl",
   "job_with_task.yml.tmpl",
diff --git a/acceptance/bundle/invariant/no_drift/out.test.toml b/acceptance/bundle/invariant/no_drift/out.test.toml
index c4fb7e5bb95..e39444592d5 100644
--- a/acceptance/bundle/invariant/no_drift/out.test.toml
+++ b/acceptance/bundle/invariant/no_drift/out.test.toml
@@ -19,6 +19,7 @@ EnvMatrix.INPUT_CONFIG = [
   "job_pydabs_1000_tasks.yml.tmpl",
   "job_cross_resource_ref.yml.tmpl",
   "job_permission_ref.yml.tmpl",
+  "job_run_job_ref.yml.tmpl",
   "job_with_depends_on.yml.tmpl",
   "job_with_permissions.yml.tmpl",
   "job_with_task.yml.tmpl",
diff --git a/acceptance/bundle/invariant/test.toml b/acceptance/bundle/invariant/test.toml
index f0971991b6a..6a4f46c5f17 100644
--- a/acceptance/bundle/invariant/test.toml
+++ b/acceptance/bundle/invariant/test.toml
@@ -37,6 +37,7 @@ EnvMatrix.INPUT_CONFIG = [
     "job_pydabs_1000_tasks.yml.tmpl",
     "job_cross_resource_ref.yml.tmpl",
     "job_permission_ref.yml.tmpl",
+    "job_run_job_ref.yml.tmpl",
     "job_with_depends_on.yml.tmpl",
     "job_with_permissions.yml.tmpl",
     "job_with_task.yml.tmpl",
@@ -72,6 +73,11 @@ EnvMatrix.INPUT_CONFIG = [
 [EnvMatrixExclude]
 no_alert_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=alert.yml.tmpl"]
 
+# job_run_job_ref sets a health rule value to 2^53+1 to exercise large-integer
+# precision in state migration. The value is out of range for the real backend,
+# so this config is local-only (the mock server stores it verbatim).
+no_run_job_ref_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=job_run_job_ref.yml.tmpl"]
+
 # Postgres resources only work on AWS
 no_postgres_project_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_project.yml.tmpl"]
 no_postgres_branch_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_branch.yml.tmpl"]
diff --git a/acceptance/bundle/migrate/added/output.txt b/acceptance/bundle/migrate/added/output.txt
index 70e6f0bc192..9a77ac41c1c 100644
--- a/acceptance/bundle/migrate/added/output.txt
+++ b/acceptance/bundle/migrate/added/output.txt
@@ -5,14 +5,7 @@ Deploying resources...
 Updating deployment state...
 Deployment complete!
 
->>> musterr [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-create jobs.job_b
-
-Plan: 1 to add, 0 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
 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/added/script b/acceptance/bundle/migrate/added/script
index 0adc713936a..ecf7fec95b6 100644
--- a/acceptance/bundle/migrate/added/script
+++ b/acceptance/bundle/migrate/added/script
@@ -6,11 +6,8 @@ trace $CLI bundle deploy
 # Uncomment job_b (add it to config without deploying)
 update_file.py databricks.yml "#job_b" "job_b"
 
-# Should fail at plan check: job_b is "1 to add"
-trace musterr $CLI bundle deployment migrate
-
-# Should succeed: job_b skipped, will be created on next deploy
-trace $CLI bundle deployment migrate --noplancheck
+# job_b skipped, will be created on next deploy
+trace $CLI bundle deployment migrate
 
 # After migration: plan shows job_b as "to add"
 trace DATABRICKS_BUNDLE_ENGINE=direct $CLI bundle plan | contains.py "1 to add"
diff --git a/acceptance/bundle/migrate/basic/output.txt b/acceptance/bundle/migrate/basic/output.txt
index 85cdfebcc48..cc70cdaee02 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/removed/output.txt b/acceptance/bundle/migrate/removed/output.txt
index 8e4bcdf555c..9896acb4247 100644
--- a/acceptance/bundle/migrate/removed/output.txt
+++ b/acceptance/bundle/migrate/removed/output.txt
@@ -5,14 +5,7 @@ Deploying resources...
 Updating deployment state...
 Deployment complete!
 
->>> musterr [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-delete jobs.job_b
-
-Plan: 0 to add, 0 to change, 1 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
 Success! Migrated 2 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/removed/script b/acceptance/bundle/migrate/removed/script
index 99c6f5716ec..c6cea1e2c04 100644
--- a/acceptance/bundle/migrate/removed/script
+++ b/acceptance/bundle/migrate/removed/script
@@ -6,11 +6,8 @@ trace $CLI bundle deploy
 # Remove job_b from config without deploying the deletion
 grep -v job_b databricks.yml > databricks_tmp.yml && mv databricks_tmp.yml databricks.yml
 
-# Should fail at plan check: job_b is "1 to delete"
-trace musterr $CLI bundle deployment migrate
-
-# Should succeed: job_b's ID preserved in direct state for deletion on next deploy
-trace $CLI bundle deployment migrate --noplancheck
+# job_b's ID preserved in direct state for deletion on next deploy
+trace $CLI bundle deployment migrate
 
 # After migration: plan shows job_b as "to delete"
 trace DATABRICKS_BUNDLE_ENGINE=direct $CLI bundle plan | contains.py "1 to delete"
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 07384d6b262..3e078e8ae73 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 a9981ee63d8..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,20 +77,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		}
 
 		if action == deployplan.Delete {
-			if migrateMode {
-				// Resource is in terraform state but not in config. Preserve its ID in
-				// direct state so the next direct deploy will plan and execute deletion.
-				id := b.StateDB.GetResourceID(resourceKey)
-				if id == "" {
-					logdiag.LogError(ctx, fmt.Errorf("%s: internal error: no ID in state", errorPrefix))
-					return false
-				}
-				if err = b.StateDB.SaveState(resourceKey, id, json.RawMessage("{}"), entry.DependsOn); err != nil {
-					logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
-					return false
-				}
-				return true
-			}
 			err = d.Destroy(ctx, &b.StateDB)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
@@ -123,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 d890b8d5d7b..ca07b477f12 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -1029,6 +1029,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..315d1791666
--- /dev/null
+++ b/bundle/migrate/build_state.go
@@ -0,0 +1,225 @@
+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/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 map[string]string,
+) 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 {
+		id, ok := tfIDs[node]
+		if !ok {
+			// Resource is in config but not in TF state (new resource); skip.
+			log.Infof(ctx, "%s: not found in terraform state, skipping", node)
+			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: read it 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. Resources without
+		// an etag return an error from LookupTFField, which we treat as "no etag".
+		if v, err := LookupTFField(tfAttrs, group, srcName, structpath.NewStringKey(nil, "etag")); err == nil {
+			if etag, ok := v.(string); ok && etag != "" {
+				if err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etag); err != nil {
+					return fmt.Errorf("%s: cannot set etag: %w", node, err)
+				}
+			}
+		}
+
+		if err := stateDB.SaveState(node, 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..c3f6f5174fb
--- /dev/null
+++ b/bundle/migrate/build_state_test.go
@@ -0,0 +1,253 @@
+package migrate_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/databricks/cli/bundle/config"
+	"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
+}
+
+func runBuildStateFromTF(
+	t *testing.T,
+	yaml string,
+	tfAttrs migrate.TFStateAttrs,
+	tfIDs map[string]string,
+) 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)
+
+	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(t *testing.T) {
+	tests := []struct {
+		name         string
+		yaml         string
+		tfAttrs      migrate.TFStateAttrs
+		tfIDs        map[string]string
+		wantKey      string         // primary key to assert on
+		absentKey    string         // key that must NOT be in state
+		wantID       string         // expected entry.ID
+		wantState    map[string]any // expected fields in the state JSON (parsed via json.Unmarshal)
+		wantStateRaw string         // expected substring in raw state JSON bytes (use for large integers)
+		wantDeps     []deployplan.DependsOnEntry
+	}{
+		{
+			name: "basic job stored with ID",
+			yaml: `
+resources:
+  jobs:
+    my_job:
+      name: "hello"
+`,
+			tfAttrs: migrate.TFStateAttrs{
+				"databricks_job": {"my_job": json.RawMessage(`{"id": "123", "name": "hello"}`)},
+			},
+			tfIDs:   map[string]string{"resources.jobs.my_job": "123"},
+			wantKey: "resources.jobs.my_job",
+			wantID:  "123",
+		},
+		{
+			name: "resource not in TF state is skipped",
+			yaml: `
+resources:
+  jobs:
+    new_job:
+      name: "new"
+    existing_job:
+      name: "existing"
+`,
+			tfAttrs: migrate.TFStateAttrs{
+				"databricks_job": {"existing_job": json.RawMessage(`{"id": "456", "name": "existing"}`)},
+			},
+			tfIDs:     map[string]string{"resources.jobs.existing_job": "456"},
+			wantKey:   "resources.jobs.existing_job",
+			absentKey: "resources.jobs.new_job",
+			wantID:    "456",
+		},
+		{
+			name: "cross-resource ref: depends_on computed, field resolved",
+			yaml: `
+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: map[string]string{
+				"resources.pipelines.src": "p1",
+				"resources.jobs.dst":      "j1",
+			},
+			wantKey:   "resources.jobs.dst",
+			wantID:    "j1",
+			wantState: map[string]any{"name": "source-pipeline"},
+			wantDeps: []deployplan.DependsOnEntry{
+				{Node: "resources.pipelines.src", Label: "${resources.pipelines.src.name}"},
+			},
+		},
+		{
+			name: "numeric field reference stored as number not string",
+			yaml: `
+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: map[string]string{
+				"resources.jobs.src_job": "111",
+				"resources.jobs.dst_job": "222",
+			},
+			wantKey:   "resources.jobs.dst_job",
+			wantID:    "222",
+			wantState: map[string]any{"max_concurrent_runs": float64(4)},
+			wantDeps:  []deployplan.DependsOnEntry{{Node: "resources.jobs.src_job"}},
+		},
+		{
+			// 2^53+1 = 9007199254740993 cannot be represented exactly as float64.
+			// json.Unmarshal into map[string]any would produce 9007199254740992 (off by one).
+			// UseNumber preserves the original decimal string, so the value is exact.
+			name: "large integer run_job_task.job_id preserved exactly",
+			yaml: `
+resources:
+  jobs:
+    trigger_job:
+      name: "trigger"
+    watcher_job:
+      name: "watcher"
+      tasks:
+        - task_key: run_trigger
+          run_job_task:
+            job_id: "${resources.jobs.trigger_job.id}"
+`,
+			tfAttrs: migrate.TFStateAttrs{
+				"databricks_job": {
+					"trigger_job": json.RawMessage(`{"id": "9007199254740993", "name": "trigger", "max_concurrent_runs": 1}`),
+					"watcher_job": json.RawMessage(`{"id": "100", "name": "watcher", "task": [{"task_key": "run_trigger", "run_job_task": [{"job_id": 9007199254740993}]}]}`),
+				},
+			},
+			tfIDs: map[string]string{
+				"resources.jobs.trigger_job": "9007199254740993",
+				"resources.jobs.watcher_job": "100",
+			},
+			wantKey: "resources.jobs.watcher_job",
+			wantID:  "100",
+			// job_id must be stored as 9007199254740993, not 9007199254740992 (float64 truncation).
+			// Check raw bytes because json.Unmarshal would silently re-truncate when reading back.
+			wantStateRaw: `"job_id": 9007199254740993`,
+		},
+		{
+			name: "dashboard etag stored from TF attributes",
+			yaml: `
+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:     map[string]string{"resources.dashboards.my_dash": "d1"},
+			wantKey:   "resources.dashboards.my_dash",
+			wantID:    "d1",
+			wantState: map[string]any{"etag": "etag-abc123"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			state := runBuildStateFromTF(t, tc.yaml, tc.tfAttrs, tc.tfIDs)
+
+			if tc.absentKey != "" {
+				assert.NotContains(t, state, tc.absentKey)
+			}
+
+			entry, ok := state[tc.wantKey]
+			require.True(t, ok, "key %q not in state", tc.wantKey)
+			assert.Equal(t, tc.wantID, entry.ID)
+
+			if len(tc.wantState) > 0 {
+				var got map[string]any
+				require.NoError(t, json.Unmarshal(entry.State, &got))
+				for k, v := range tc.wantState {
+					assert.Equal(t, v, got[k], "state[%q]", k)
+				}
+			}
+
+			if tc.wantStateRaw != "" {
+				assert.Contains(t, string(entry.State), tc.wantStateRaw, "raw state JSON")
+			}
+
+			if tc.wantDeps != nil {
+				require.Len(t, entry.DependsOn, len(tc.wantDeps))
+				for i, want := range tc.wantDeps {
+					assert.Equal(t, want.Node, entry.DependsOn[i].Node)
+					if want.Label != "" {
+						assert.Equal(t, want.Label, entry.DependsOn[i].Label)
+					}
+				}
+			}
+		})
+	}
+}
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..55634234051
--- /dev/null
+++ b/bundle/migrate/tf_state.go
@@ -0,0 +1,175 @@
+package migrate
+
+import (
+	"bytes"
+	"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"
+)
+
+// TFStateAttrs maps (tfResourceType → resourceName → raw JSON attributes).
+type TFStateAttrs map[string]map[string]json.RawMessage
+
+// TFState holds everything parsed from a single terraform state file read.
+type TFState struct {
+	// Serial and Lineage are the top-level state metadata used to seed the direct state.
+	Serial  int
+	Lineage string
+
+	// Attrs maps (tfResourceType → resourceName → raw JSON attributes).
+	Attrs TFStateAttrs
+	// IDs maps bundle resource key → resource ID.
+	IDs map[string]string
+}
+
+// rawTFState is the on-disk terraform state format; it captures everything we need in one parse.
+type rawTFState struct {
+	Serial    int    `json:"serial"`
+	Lineage   string `json:"lineage"`
+	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"`
+}
+
+// ParseTFStateFull reads the terraform state file once and returns all parsed data.
+// Returns nil without error when the file does not exist (first deploy with no resources).
+func ParseTFStateFull(ctx context.Context, path string) (*TFState, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	// Parse once: lineage/serial live at the top level alongside the resources array,
+	// so a single unmarshal captures everything needed for both attrs and IDs.
+	var parsed rawTFState
+	if err := json.Unmarshal(raw, &parsed); err != nil {
+		return nil, err
+	}
+
+	attrs := parseTFStateAttrsFromRaw(&parsed)
+
+	exportedIDs, err := terraform.ParseResourcesStateFromBytes(ctx, raw)
+	if err != nil {
+		return nil, err
+	}
+	ids := make(map[string]string, len(exportedIDs))
+	for k, v := range exportedIDs {
+		ids[k] = v.ID
+	}
+	return &TFState{Attrs: attrs, IDs: ids, Lineage: parsed.Lineage, Serial: parsed.Serial}, nil
+}
+
+func parseTFStateAttrsFromRaw(s *rawTFState) TFStateAttrs {
+	result := make(TFStateAttrs)
+	for _, r := range s.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
+}
+
+// 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.
+	// UseNumber preserves large integers exactly; plain Unmarshal into interface{} uses
+	// float64 which loses precision for integers beyond 2^53.
+	var attrs map[string]any
+	dec := json.NewDecoder(bytes.NewReader(attrsJSON))
+	dec.UseNumber()
+	if err := dec.Decode(&attrs); err != nil {
+		return nil, fmt.Errorf("cannot parse TF state for %s.%s: %w", tfType, name, err)
+	}
+
+	return navigateTFState(attrs, tfFieldPath)
+}
+
+// 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 fd76151483c..840c7d821f1 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"
@@ -83,7 +82,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)
 		// Capture the finalized state for deploy telemetry. It carries each
 		// resource's state-size in bytes (from the WAL replay Finalize just
diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go
index 74049f26f42..fd580f5e971 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"
@@ -78,7 +77,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 6573d1dc56f..163c9fb4fdb 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 {
@@ -114,49 +112,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)
+	tfState, 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 {
+	if tfState == 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 {
+	for key, id := range tfState.IDs {
 		state[key] = dstate.ResourceEntry{
-			ID:    resourceEntry.ID,
+			ID:    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.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
@@ -166,9 +146,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 {
@@ -183,36 +163,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, tfState.Attrs, tfState.IDs); 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 2d54aafb1ed..39d9a0454d5 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,13 +113,13 @@ 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)
+		tfState, err := migrate.ParseTFStateFull(ctx, localTerraformPath)
 		if err != nil {
 			return fmt.Errorf("failed to parse terraform state: %w", err)
 		}
 
-		for key, resourceEntry := range terraformResources {
-			if resourceEntry.ID == "" {
+		for key, id := range tfState.IDs {
+			if id == "" {
 				return fmt.Errorf("failed to intepret terraform state for %s: missing ID", key)
 			}
 		}
@@ -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 {
+		for key, id := range tfState.IDs {
 			state[key] = dstate.ResourceEntry{
-				ID:    resourceEntry.ID,
+				ID:    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,55 +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 {
-			switch entry.Action {
-			case deployplan.Create:
-				// Resource is in config but not in terraform state; skip it during migration.
-				// It will be created on the first direct deploy.
-				entry.Action = deployplan.Skip
-			case deployplan.Delete:
-				// Resource is in terraform state but not in config. Keep as Delete so the
-				// apply migrate path can preserve its ID in direct state, allowing the next
-				// direct deploy to remove it.
-			default:
-				// Force existing resources to Update so migration reads their remote state
-				// and writes a full config snapshot.
-				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, tfState.Attrs, tfState.IDs); 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) {