diff --git a/acceptance/bundle/migrate/auto-migrate-clean/output.txt b/acceptance/bundle/migrate/auto-migrate-clean/output.txt
index cd07b8b89c6..0526bae7622 100644
--- a/acceptance/bundle/migrate/auto-migrate-clean/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-clean/output.txt
@@ -17,10 +17,10 @@ direct_drymigrate_warnings false
 >>> [CLI] bundle deploy
 Warn: Direct engine requested in bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11 but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
-Migrated 1 resource to direct deployment engine.
 Files: 3 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+Migrating state to direct deployment engine (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
+Migrated 1 resource to direct deployment engine.
 
 >>> print_migration_telemetry
 direct_migrated_via_config true
diff --git a/acceptance/bundle/migrate/auto-migrate-conversion-failure/databricks.yml b/acceptance/bundle/migrate/auto-migrate-conversion-failure/databricks.yml
new file mode 100644
index 00000000000..25e8ccbf913
--- /dev/null
+++ b/acceptance/bundle/migrate/auto-migrate-conversion-failure/databricks.yml
@@ -0,0 +1,13 @@
+bundle:
+  name: test-bundle
+
+resources:
+  secret_scopes:
+    my_scope:
+      name: my-scope
+      permissions:
+        # An unknown level: SecretScopeFixups rejects it when it prepares the
+        # config for the direct engine, so the state conversion fails while the
+        # terraform deploy itself succeeds.
+        - level: BOGUS
+          user_name: someone@example.com
diff --git a/acceptance/bundle/migrate/auto-migrate-conversion-failure/out.test.toml b/acceptance/bundle/migrate/auto-migrate-conversion-failure/out.test.toml
new file mode 100644
index 00000000000..0938e678987
--- /dev/null
+++ b/acceptance/bundle/migrate/auto-migrate-conversion-failure/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/migrate/auto-migrate-conversion-failure/output.txt b/acceptance/bundle/migrate/auto-migrate-conversion-failure/output.txt
new file mode 100644
index 00000000000..7fa3a09a55e
--- /dev/null
+++ b/acceptance/bundle/migrate/auto-migrate-conversion-failure/output.txt
@@ -0,0 +1,46 @@
+
+=== Not opted in: the conversion failure is reported as a failed dry run
+>>> DATABRICKS_BUNDLE_ENGINE=terraform [CLI] bundle deploy
+Warning: invalid value "BOGUS" for enum field. Valid values are [READ WRITE MANAGE]
+  at resources.secret_scopes.my_scope.permissions[0].level
+
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Created secret_scopes.my_scope
+Created secret_scopes.my_scope.permissions
+Files: 5 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+Warn: post-deploy dry-run migration to direct: unknown permission level "BOGUS" for secret scope
+Warn: The warnings above are from a dry-run migration to the direct deployment engine (https://docs.databricks.com/aws/en/dev-tools/bundles/direct).
+Your deployment is not affected and works normally, but you may experience these issues when migrating to the direct deployment engine.
+Please forward these warnings to dabs-feedback@databricks.com
+
+>>> print_migration_telemetry
+direct_drymigrate_success false
+direct_drymigrate_warnings false
+direct_migrate_error_template unknown permission level %q for secret scope
+
+=== Opted in: the same failure stops the automatic migration
+>>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
+Warning: invalid value "BOGUS" for enum field. Valid values are [READ WRITE MANAGE]
+  at resources.secret_scopes.my_scope.permissions[0].level
+
+Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Files: 2 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+Warn: post-deploy dry-run migration to direct: unknown permission level "BOGUS" for secret scope
+Warn: The warnings above are from a dry-run migration to the direct deployment engine (https://docs.databricks.com/aws/en/dev-tools/bundles/direct).
+Your deployment is not affected and works normally, but you may experience these issues when migrating to the direct deployment engine.
+Please forward these warnings to dabs-feedback@databricks.com
+Warn: Direct engine was requested but the dry-run migration reported issues; automatic migration to the direct deployment engine is stopped. Address the issues above or run "databricks bundle deployment migrate" manually.
+
+>>> print_migration_telemetry
+direct_migrate_error true
+direct_migrate_error_template unknown permission level %q for secret scope
+
+=== State is still terraform, so nothing was migrated
+
+>>> find .databricks/bundle -name resources.json -type f
+
+>>> find .databricks/bundle -name terraform.tfstate* -type f
+.databricks/bundle/default/terraform/terraform.tfstate
diff --git a/acceptance/bundle/migrate/auto-migrate-conversion-failure/script b/acceptance/bundle/migrate/auto-migrate-conversion-failure/script
new file mode 100644
index 00000000000..efad1c5fab8
--- /dev/null
+++ b/acceptance/bundle/migrate/auto-migrate-conversion-failure/script
@@ -0,0 +1,20 @@
+export DATABRICKS_BUNDLE_ENGINE=
+
+# An unknown permission level is only a warning for the terraform deploy, but
+# SecretScopeFixups rejects it while preparing the config for the direct engine.
+# That is a state conversion failure: no API call is involved, so unlike the
+# commit failures it cannot be produced by injecting a fault.
+
+title "Not opted in: the conversion failure is reported as a failed dry run"
+trace DATABRICKS_BUNDLE_ENGINE=terraform $CLI bundle deploy
+trace print_migration_telemetry
+rm -f out.requests.txt
+
+title "Opted in: the same failure stops the automatic migration"
+trace DATABRICKS_BUNDLE_ENGINE=direct $CLI bundle deploy
+trace print_migration_telemetry
+rm -f out.requests.txt
+
+title "State is still terraform, so nothing was migrated\n"
+trace find .databricks/bundle -name "resources.json" -type f
+trace find .databricks/bundle -name "terraform.tfstate*" -type f
diff --git a/acceptance/bundle/migrate/auto-migrate-conversion-failure/test.toml b/acceptance/bundle/migrate/auto-migrate-conversion-failure/test.toml
new file mode 100644
index 00000000000..50abf12f70b
--- /dev/null
+++ b/acceptance/bundle/migrate/auto-migrate-conversion-failure/test.toml
@@ -0,0 +1,3 @@
+# The unknown permission level is deliberate; the enum warning is part of what
+# this test asserts.
+Ignore = [".databricks"]
diff --git a/acceptance/bundle/migrate/auto-migrate-direct-only-envvar/output.txt b/acceptance/bundle/migrate/auto-migrate-direct-only-envvar/output.txt
index 1254e8c2433..e4502558839 100644
--- a/acceptance/bundle/migrate/auto-migrate-direct-only-envvar/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-direct-only-envvar/output.txt
@@ -13,10 +13,10 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 >>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
 Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
-Migrated 1 resource to direct deployment engine.
 Files: 3 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
+Migrated 1 resource to direct deployment engine.
 
 >>> print_migration_telemetry
 direct_migrated_via_env true
diff --git a/acceptance/bundle/migrate/auto-migrate-direct-only/output.txt b/acceptance/bundle/migrate/auto-migrate-direct-only/output.txt
index 0811eaca457..a0103ed982b 100644
--- a/acceptance/bundle/migrate/auto-migrate-direct-only/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-direct-only/output.txt
@@ -23,10 +23,10 @@ json.plan.resources.instance_pools.pool.action = "skip";
 >>> [CLI] bundle deploy
 Warn: Direct engine requested in bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11 but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
-Migrated 1 resource to direct deployment engine.
 Files: 3 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+Migrating state to direct deployment engine (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
+Migrated 1 resource to direct deployment engine.
 
 >>> print_migration_telemetry
 direct_migrated_via_config true
diff --git a/acceptance/bundle/migrate/auto-migrate-empty-tfstate/output.txt b/acceptance/bundle/migrate/auto-migrate-empty-tfstate/output.txt
index 56061ed71c2..a84f340e2c7 100644
--- a/acceptance/bundle/migrate/auto-migrate-empty-tfstate/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-empty-tfstate/output.txt
@@ -3,9 +3,9 @@
 >>> [CLI] bundle deploy
 Warn: Direct engine requested in bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11 but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Removing empty terraform state; direct engine will be used on the next deploy (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
 Files: 4 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 0 unchanged
+Removing empty terraform state; direct engine will be used on the next deploy (opted in via bundle.engine setting at [TEST_TMP_DIR]/databricks.yml:3:11)...
 
 === Terraform state is renamed to .backup; no resources.json (empty state, nothing to persist)
 
diff --git a/acceptance/bundle/migrate/auto-migrate-envvar/output.txt b/acceptance/bundle/migrate/auto-migrate-envvar/output.txt
index 8e6b0cc136c..9c741646a7f 100644
--- a/acceptance/bundle/migrate/auto-migrate-envvar/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-envvar/output.txt
@@ -14,10 +14,10 @@ direct_drymigrate_warnings false
 >>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
 Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
-Migrated 1 resource to direct deployment engine.
 Files: 2 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
+Migrated 1 resource to direct deployment engine.
 
 >>> print_migration_telemetry
 direct_migrated_via_env true
diff --git a/acceptance/bundle/migrate/auto-migrate-push-failure/output.txt b/acceptance/bundle/migrate/auto-migrate-push-failure/output.txt
index 74412d94a92..be57bcbea10 100644
--- a/acceptance/bundle/migrate/auto-migrate-push-failure/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-push-failure/output.txt
@@ -14,13 +14,14 @@ direct_drymigrate_warnings false
 >>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
 Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
-Warn: automatic migration to direct engine failed: pushing direct state to workspace: access denied: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/state/resources.json
 Files: 2 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
+Warn: automatic migration to direct engine failed: pushing direct state to workspace: access denied: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/state/resources.json
 
 >>> print_migration_telemetry
 direct_migrate_commit_error true
+direct_migrate_commit_error_template pushing direct state to workspace: access denied [403 INJECTED]
 
 === Local state was NOT rewritten (still terraform)
 
@@ -33,10 +34,10 @@ direct_migrate_commit_error true
 >>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
 Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
-Migrated 1 resource to direct deployment engine.
 Files: 2 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
+Migrated 1 resource to direct deployment engine.
 
 >>> print_migration_telemetry
 direct_migrated_via_env true
diff --git a/acceptance/bundle/migrate/auto-migrate-tfbackup-failure/output.txt b/acceptance/bundle/migrate/auto-migrate-tfbackup-failure/output.txt
index 19980cd3060..53e558b0c39 100644
--- a/acceptance/bundle/migrate/auto-migrate-tfbackup-failure/output.txt
+++ b/acceptance/bundle/migrate/auto-migrate-tfbackup-failure/output.txt
@@ -14,13 +14,14 @@ direct_drymigrate_warnings false
 >>> DATABRICKS_BUNDLE_ENGINE=direct [CLI] bundle deploy
 Warn: Direct engine requested in DATABRICKS_BUNDLE_ENGINE environment variable but the existing state uses "terraform". Deploying on "terraform"; will attempt to migrate the state to the direct engine after this deploy.
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
-Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
-Warn: automatic migration to direct engine failed: pushing direct state to workspace: deleting remote terraform state: Fault injected by test.
 Files: 2 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 1 unchanged
+Migrating state to direct deployment engine (opted in via DATABRICKS_BUNDLE_ENGINE environment variable)...
+Warn: automatic migration to direct engine failed: pushing direct state to workspace: deleting remote terraform state: Fault injected by test.
 
 >>> print_migration_telemetry
 direct_migrate_commit_error true
+direct_migrate_commit_error_template pushing direct state to workspace: deleting remote terraform state: %w [403 INJECTED]
 
 === Local state was NOT rewritten (still terraform)
 
diff --git a/acceptance/bundle/migrate/script.prepare b/acceptance/bundle/migrate/script.prepare
index 9de2f19d437..27fb59bb8a4 100644
--- a/acceptance/bundle/migrate/script.prepare
+++ b/acceptance/bundle/migrate/script.prepare
@@ -1,7 +1,12 @@
-# Filter print_telemetry_bool_values output to auto-migration keys
-# (direct_drymigrate_*, direct_migrate_*, direct_migrated_via_*). Callers
-# clear out.requests.txt themselves; some tests want to keep it for a
-# subsequent print_requests.py assertion.
+# Filter telemetry output to auto-migration keys (direct_drymigrate_*,
+# direct_migrate_*, direct_migrated_via_*), booleans first and then the
+# PII-free error template that says which failure a direct_migrate_error or
+# direct_drymigrate_success=false was.
+#
+# Call it right after the deploy whose telemetry you want, so a block covers
+# exactly one deploy. It does not clear out.requests.txt: callers do that
+# themselves, since some want it kept for a subsequent print_requests.py.
 print_migration_telemetry() {
     print_telemetry_bool_values | grep '^direct_' || true
+    print_telemetry_error_templates | grep '^direct_' || true
 }
diff --git a/acceptance/script.prepare b/acceptance/script.prepare
index d71d37aae9b..1dedb40b1da 100644
--- a/acceptance/script.prepare
+++ b/acceptance/script.prepare
@@ -123,6 +123,12 @@ print_telemetry_bool_values() {
     jq -r 'select(.path? == "/telemetry-ext") | (.body.protoLogs // [])[] | fromjson | ( (.entry // .) | (.databricks_cli_log.bundle_deploy_event.experimental.bool_values // []) ) | map("\(.key) \(.value)") | .[]' out.requests.txt | grep -v '^engine_terraform_' | sort
 }
 
+# Print the PII-free error templates the deploy event carries, one "<field>
+# <template>" line each, skipping the ones this deploy did not set.
+print_telemetry_error_templates() {
+    jq -r 'select(.path? == "/telemetry-ext") | (.body.protoLogs // [])[] | fromjson | ( (.entry // .) | (.databricks_cli_log.bundle_deploy_event.experimental // {}) ) | to_entries | map(select((.key | endswith("_template")) and (.value // "") != "")) | map("\(.key) \(.value)") | .[]' out.requests.txt | sort
+}
+
 sethome() {
     local home="$1"
     mkdir -p "$home"
diff --git a/bundle/bundle.go b/bundle/bundle.go
index bcceb752088..898301787ab 100644
--- a/bundle/bundle.go
+++ b/bundle/bundle.go
@@ -85,6 +85,12 @@ type Metrics struct {
 	ExecutionTimes              []protos.IntMapEntry
 	LocalCacheMeasurementsMs    []protos.IntMapEntry // Local cache measurements stored as milliseconds
 
+	// PII-free descriptions of the errors this deploy hit, reported without
+	// scrubbing. Each is a libs/safeerr message template; see the matching fields
+	// on protos.BundleDeployExperimental for what each one covers.
+	DirectMigrateErrorTemplate       string
+	DirectMigrateCommitErrorTemplate string
+
 	// StateEngine is the engine that ran the deploy, set in deployCore. Empty when
 	// telemetry is emitted without a deploy having run.
 	StateEngine engine.EngineType
diff --git a/bundle/config/mutator/resourcemutator/secret_scope_fixups.go b/bundle/config/mutator/resourcemutator/secret_scope_fixups.go
index e85584da3f0..cf8a9326ebd 100644
--- a/bundle/config/mutator/resourcemutator/secret_scope_fixups.go
+++ b/bundle/config/mutator/resourcemutator/secret_scope_fixups.go
@@ -2,7 +2,6 @@ package resourcemutator
 
 import (
 	"context"
-	"fmt"
 	"slices"
 	"strings"
 
@@ -12,6 +11,7 @@ import (
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/iamutil"
+	"github.com/databricks/cli/libs/safeerr"
 	"github.com/databricks/databricks-sdk-go/service/iam"
 )
 
@@ -63,7 +63,7 @@ func collapsePermissions(scope *resources.SecretScope) error {
 	for _, perm := range scope.Permissions {
 		// Validate permission level
 		if _, ok := permissionRank[perm.Level]; !ok {
-			return fmt.Errorf("unknown permission level %q for secret scope", perm.Level)
+			return safeerr.Errorf("unknown permission level %q for secret scope", perm.Level)
 		}
 
 		// Add a prefix to retain the original principal type. In practice collisions here should
@@ -76,7 +76,7 @@ func collapsePermissions(scope *resources.SecretScope) error {
 		} else if perm.ServicePrincipalName != "" {
 			principal = "sp:" + perm.ServicePrincipalName
 		} else {
-			return fmt.Errorf("missing principal in permissions for secret scope %q", scope.Name)
+			return safeerr.Errorf("missing principal in permissions for secret scope %q", scope.Name)
 		}
 
 		existing, exists := principalPermissions[principal]
@@ -120,14 +120,18 @@ func collapsePermissions(scope *resources.SecretScope) error {
 	return nil
 }
 
-func (m *secretScopeFixups) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+// ApplySecretScopeFixups is the mutator's body, exported so a caller that wants
+// the failure as an error rather than a diagnostic can have it. The migration to
+// the direct engine is one: it reports the error's template to telemetry, which a
+// diagnostic's summary cannot carry. The returned key names the offending scope.
+func ApplySecretScopeFixups(b *bundle.Bundle, eng engine.EngineType) (string, error) {
 	// Secret scopes by default have the current user as a MANAGE ACL. We need to add it to the client ACL list
 	// to prevent a phantom persistent diff.
 	// We do not need to do this in terraform because terraform naively always applies the config during ACL
 	// creation without checking if the ACL already exists.
 	// https://github.com/databricks/terraform-provider-databricks/blob/5cb5d3fa46bc4843be1a4c4bce89296eaa2e14fc/secrets/resource_secret_acl.go#L43
-	if !m.engine.IsDirect() {
-		return nil
+	if !eng.IsDirect() {
+		return "", nil
 	}
 
 	// Secret scopes assigns the create MANAGE ACL on it by default. So we always add it to
@@ -140,19 +144,27 @@ func (m *secretScopeFixups) Apply(ctx context.Context, b *bundle.Bundle) diag.Di
 		currentUser := b.Config.Workspace.CurrentUser.User
 
 		addManageForCurrentUser(scope, currentUser)
-		err := collapsePermissions(scope)
-		if err != nil {
-			return diag.Diagnostics{
-				{
-					Severity:  diag.Error,
-					Summary:   "Failed to collapse permissions for secret scope",
-					Detail:    err.Error(),
-					Paths:     []dyn.Path{dyn.MustPathFromString("resources.secret_scopes." + key)},
-					Locations: []dyn.Location{b.Config.GetLocation("resources.secret_scopes." + key)},
-				},
-			}
+		if err := collapsePermissions(scope); err != nil {
+			return key, err
 		}
 	}
 
-	return nil
+	return "", nil
+}
+
+func (m *secretScopeFixups) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	key, err := ApplySecretScopeFixups(b, m.engine)
+	if err == nil {
+		return nil
+	}
+
+	return diag.Diagnostics{
+		{
+			Severity:  diag.Error,
+			Summary:   "Failed to collapse permissions for secret scope",
+			Detail:    err.Error(),
+			Paths:     []dyn.Path{dyn.MustPathFromString("resources.secret_scopes." + key)},
+			Locations: []dyn.Location{b.Config.GetLocation("resources.secret_scopes." + key)},
+		},
+	}
 }
diff --git a/bundle/config/resource_key.go b/bundle/config/resource_key.go
new file mode 100644
index 00000000000..e1988f56b00
--- /dev/null
+++ b/bundle/config/resource_key.go
@@ -0,0 +1,40 @@
+package config
+
+import "strings"
+
+// ResourceKey wraps a resource key (e.g. "resources.jobs.my_job" or
+// "resources.jobs.my_job.permissions") for use as an error argument. It formats
+// as the full key, so error messages are unchanged, but it reports only its
+// resource type to telemetry — the resource name is user-authored and therefore
+// PII, while the type is a value the CLI itself defines.
+//
+// Pass it wherever a resource key is interpolated into a safeerr error:
+//
+//	safeerr.Errorf("%s: SaveState: %w", config.ResourceKey(node), err)
+//	  message:  resources.jobs.my_job: SaveState: ...
+//	  template: jobs.*: SaveState: %w
+type ResourceKey string
+
+func (k ResourceKey) String() string {
+	return string(k)
+}
+
+// SafeString implements safeerr.SafeStringer, standing in for the key with its
+// name replaced by "*". A key this package cannot parse reports nothing beyond
+// the redaction marker, since an unrecognized shape may be anything at all.
+func (k ResourceKey) SafeString() string {
+	resourceType := GetResourceTypeFromKey(string(k))
+	if resourceType == "" {
+		return "*"
+	}
+
+	// GetResourceTypeFromKey collapses a sub-resource into "<group>.<kind>"
+	// (e.g. "jobs.permissions"), but in the key itself the kind trails the
+	// name. Rebuild the key's own shape so the stand-in reads like the value.
+	// The "resources." prefix is dropped: every key carries it, so it is noise.
+	group, kind, hasKind := strings.Cut(resourceType, ".")
+	if hasKind {
+		return group + ".*." + kind
+	}
+	return group + ".*"
+}
diff --git a/bundle/config/resource_key_test.go b/bundle/config/resource_key_test.go
new file mode 100644
index 00000000000..88a32861b70
--- /dev/null
+++ b/bundle/config/resource_key_test.go
@@ -0,0 +1,57 @@
+package config
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/databricks/cli/libs/safeerr"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResourceKeySafeString(t *testing.T) {
+	tests := []struct {
+		key  string
+		want string
+	}{
+		{key: "resources.jobs.my_job", want: "jobs.*"},
+		{key: "resources.pipelines.my_pipeline", want: "pipelines.*"},
+		{key: "resources.jobs.my_job.permissions", want: "jobs.*.permissions"},
+		{key: "resources.schemas.my_schema.grants", want: "schemas.*.grants"},
+		{key: "resources.secret_scopes.my scope.permissions", want: "secret_scopes.*.permissions"},
+
+		// Shapes GetResourceTypeFromKey does not recognize report nothing.
+		{key: "resources.jobs", want: "*"},
+		{key: "jobs.my_job", want: "*"},
+		{key: "", want: "*"},
+		{key: "/Workspace/Users/someone@example.com/x", want: "*"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.key, func(t *testing.T) {
+			assert.Equal(t, tt.want, ResourceKey(tt.key).SafeString())
+		})
+	}
+}
+
+// TestResourceKeyFormatsAsTheFullKey is what keeps error messages unchanged
+// when a call site starts passing ResourceKey instead of a bare string.
+func TestResourceKeyFormatsAsTheFullKey(t *testing.T) {
+	const key = "resources.jobs.my_job"
+
+	for _, format := range []string{"%s", "%q", "%v"} {
+		t.Run(format, func(t *testing.T) {
+			assert.Equal(t,
+				fmt.Sprintf(format, key),
+				fmt.Sprintf(format, ResourceKey(key)))
+		})
+	}
+}
+
+func TestResourceKeyInSafeerr(t *testing.T) {
+	err := safeerr.Errorf("%s: SaveState: %w",
+		ResourceKey("resources.jobs.my_job"), safeerr.New("disk full"))
+
+	assert.Equal(t, "resources.jobs.my_job: SaveState: disk full", err.Error())
+	assert.Equal(t, "jobs.*: SaveState: disk full", safeerr.ErrorTemplate(err))
+	assert.NotContains(t, safeerr.ErrorTemplate(err), "my_job")
+}
diff --git a/bundle/migrate/build_state.go b/bundle/migrate/build_state.go
index e8b382b370d..e5238368ec0 100644
--- a/bundle/migrate/build_state.go
+++ b/bundle/migrate/build_state.go
@@ -2,7 +2,6 @@ package migrate
 
 import (
 	"context"
-	"fmt"
 	"maps"
 	"slices"
 	"strings"
@@ -15,6 +14,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/dynvar"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/safeerr"
 	"github.com/databricks/cli/libs/structs/structaccess"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/structs/structvar"
@@ -64,7 +64,7 @@ func BuildStateFromTF(
 
 		group := config.GetResourceTypeFromKey(node)
 		if group == "" {
-			return warningsSeen, fmt.Errorf("cannot determine resource type for %q", node)
+			return warningsSeen, safeerr.Errorf("cannot determine resource type for %q", config.ResourceKey(node))
 		}
 
 		adapter, ok := adapters[group]
@@ -76,22 +76,22 @@ func BuildStateFromTF(
 
 		inputConfig, err := configRoot.GetResourceConfig(node)
 		if err != nil {
-			return warningsSeen, fmt.Errorf("%s: getting config: %w", node, err)
+			return warningsSeen, safeerr.Errorf("%s: getting config: %w", config.ResourceKey(node), err)
 		}
 
 		inputSV, err := adapter.PrepareInputConfig(inputConfig, node)
 		if err != nil {
-			return warningsSeen, fmt.Errorf("%s: PrepareInputConfig: %w", node, err)
+			return warningsSeen, safeerr.Errorf("%s: PrepareInputConfig: %w", config.ResourceKey(node), err)
 		}
 
 		newStateValue, err := adapter.PrepareState(inputSV.Value)
 		if err != nil {
-			return warningsSeen, fmt.Errorf("%s: PrepareState: %w", node, err)
+			return warningsSeen, safeerr.Errorf("%s: PrepareState: %w", config.ResourceKey(node), err)
 		}
 
 		refs, err := direct.ExtractReferences(configRoot.Value(), node, adapter.StateType())
 		if err != nil {
-			return warningsSeen, fmt.Errorf("%s: extracting references: %w", node, err)
+			return warningsSeen, safeerr.Errorf("%s: extracting references: %w", config.ResourceKey(node), err)
 		}
 		maps.Copy(refs, inputSV.Refs)
 
@@ -142,7 +142,7 @@ func BuildStateFromTF(
 		// is absent there (model_serving_endpoints, database_instances).
 		if _, ok := sv.Refs["object_id"]; ok {
 			if err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "object_id"), id); err != nil {
-				return warningsSeen, fmt.Errorf("%s: setting object_id: %w", node, err)
+				return warningsSeen, safeerr.Errorf("%s: setting object_id: %w", config.ResourceKey(node), err)
 			}
 			delete(sv.Refs, "object_id")
 		}
@@ -169,14 +169,14 @@ func BuildStateFromTF(
 		for _, pending := range pendingRefs {
 			fieldPath, err := structpath.ParsePath(pending.fieldPathStr)
 			if err != nil {
-				return warningsSeen, fmt.Errorf("%s: parsing field path %q: %w", node, pending.fieldPathStr, err)
+				return warningsSeen, safeerr.Errorf("%s: parsing field path %q: %w", config.ResourceKey(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, warned, err := ResolveFieldRef(ctx, tfAttrs, srcGroup, srcName, fieldPath, pending.refTemplate, warnPrefix)
 			if err != nil {
-				return warningsSeen, fmt.Errorf("%s: cannot resolve field %q (template %q): %w", node, pending.fieldPathStr, pending.refTemplate, err)
+				return warningsSeen, safeerr.Errorf("%s: cannot resolve field %q (template %q): %w", config.ResourceKey(node), pending.fieldPathStr, pending.refTemplate, err)
 			}
 			if warned {
 				warningsSeen = true
@@ -184,13 +184,13 @@ func BuildStateFromTF(
 
 			// Set the resolved value directly and remove the ref entry.
 			if err := structaccess.Set(sv.Value, fieldPath, value); err != nil {
-				return warningsSeen, fmt.Errorf("%s: cannot set resolved value for field %q: %w", node, pending.fieldPathStr, err)
+				return warningsSeen, safeerr.Errorf("%s: cannot set resolved value for field %q: %w", config.ResourceKey(node), pending.fieldPathStr, err)
 			}
 			delete(sv.Refs, pending.fieldPathStr)
 		}
 
 		if len(sv.Refs) > 0 {
-			return warningsSeen, fmt.Errorf("%s: unresolved references: %v", node, sv.Refs)
+			return warningsSeen, safeerr.Errorf("%s: unresolved references: %v", config.ResourceKey(node), sv.Refs)
 		}
 
 		// Handle etag for dashboards: read it directly from TF state attributes.
@@ -200,13 +200,13 @@ func BuildStateFromTF(
 		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 warningsSeen, fmt.Errorf("%s: cannot set etag: %w", node, err)
+					return warningsSeen, safeerr.Errorf("%s: cannot set etag: %w", config.ResourceKey(node), err)
 				}
 			}
 		}
 
 		if err := stateDB.SaveState(node, id, sv.Value, dependsOn); err != nil {
-			return warningsSeen, fmt.Errorf("%s: SaveState: %w", node, err)
+			return warningsSeen, safeerr.Errorf("%s: SaveState: %w", config.ResourceKey(node), err)
 		}
 	}
 
diff --git a/bundle/migrate/build_state_test.go b/bundle/migrate/build_state_test.go
index a495436008c..4b46fe37565 100644
--- a/bundle/migrate/build_state_test.go
+++ b/bundle/migrate/build_state_test.go
@@ -15,6 +15,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/convert"
 	"github.com/databricks/cli/libs/dyn/yamlloader"
+	"github.com/databricks/cli/libs/safeerr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -303,3 +304,101 @@ resources:
 		})
 	}
 }
+
+// buildStateErrFromTF is runBuildStateFromTF's failure counterpart: it returns
+// the error instead of requiring success.
+func buildStateErrFromTF(
+	t *testing.T,
+	yaml string,
+	tfAttrs migrate.TFStateAttrs,
+	tfIDs map[string]string,
+) error {
+	t.Helper()
+
+	root := rootFromYAML(t, yaml)
+	adapters, err := dresources.InitAll(nil)
+	require.NoError(t, err)
+
+	var db dstate.DeploymentState
+	db.OpenWithData(filepath.Join(t.TempDir(), "resources.json"), dstate.NewDatabase("lineage", 1))
+	require.NoError(t, db.UpgradeToWrite())
+
+	_, err = migrate.BuildStateFromTF(t.Context(), &root, adapters, &db, tfAttrs, tfIDs, "")
+	return err
+}
+
+// TestBuildStateFromTFErrors covers the ways a conversion fails, and pins the
+// message against the template reported to telemetry: the message names the
+// resource and field, the template names neither but still says which failure
+// it was. Together these are what a migration failure population can be broken
+// down by.
+func TestBuildStateFromTFErrors(t *testing.T) {
+	tests := []struct {
+		name string
+
+		yaml    string
+		tfAttrs migrate.TFStateAttrs
+		tfIDs   map[string]string
+
+		// wantErrContains is a substring of the user-facing message; secrets
+		// below must appear there and must not appear in the template.
+		wantErrContains string
+		wantTemplate    string
+	}{
+		{
+			name: "referenced resource missing from TF state",
+			yaml: `
+resources:
+  pipelines:
+    src_secret:
+      name: "source"
+  jobs:
+    dst_secret:
+      name: "${resources.pipelines.src_secret.name}"
+`,
+			tfAttrs: migrate.TFStateAttrs{
+				"databricks_job": {"dst_secret": json.RawMessage(`{"id": "j1"}`)},
+			},
+			tfIDs:           map[string]string{"resources.jobs.dst_secret": "j1"},
+			wantErrContains: "databricks_pipeline.src_secret not found in TF state",
+			wantTemplate:    `jobs.*: cannot resolve field %q (template %q): jobs.%s field %s: method A: %q: key not found; method B: cannot look up %q: databricks_pipeline.%s not found in TF state`,
+		},
+		{
+			name: "referenced field absent from TF attributes",
+			yaml: `
+resources:
+  pipelines:
+    src_secret:
+      name: "source"
+  jobs:
+    dst_secret:
+      name: "${resources.pipelines.src_secret.name}"
+`,
+			tfAttrs: migrate.TFStateAttrs{
+				"databricks_pipeline": {"src_secret": json.RawMessage(`{"id": "p1"}`)},
+				"databricks_job":      {"dst_secret": json.RawMessage(`{"id": "j1"}`)},
+			},
+			tfIDs: map[string]string{
+				"resources.pipelines.src_secret": "p1",
+				"resources.jobs.dst_secret":      "j1",
+			},
+			wantErrContains: "key not found",
+			wantTemplate:    `jobs.*: cannot resolve field %q (template %q): jobs.%s field %s: method A: %q: key not found; method B: cannot look up %q: %q: key not found`,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := buildStateErrFromTF(t, tc.yaml, tc.tfAttrs, tc.tfIDs)
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tc.wantErrContains)
+			assert.Equal(t, tc.wantTemplate, safeerr.ErrorTemplate(err))
+
+			// The names the message carries must not reach the template.
+			for _, secret := range []string{"src_secret", "dst_secret"} {
+				assert.Contains(t, err.Error(), secret, "message should name the resource")
+				assert.NotContains(t, safeerr.ErrorTemplate(err), secret, "template must not")
+			}
+		})
+	}
+}
diff --git a/bundle/migrate/resolve.go b/bundle/migrate/resolve.go
index 23d0180e3c6..15b2a31f3f8 100644
--- a/bundle/migrate/resolve.go
+++ b/bundle/migrate/resolve.go
@@ -8,6 +8,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/dynvar"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/safeerr"
 	"github.com/databricks/cli/libs/structs/structpath"
 )
 
@@ -23,16 +24,16 @@ func evaluateTemplate(state TFStateAttrs, template string) (string, error) {
 	for _, pathString := range ref.References() {
 		path, err := structpath.ParsePath(pathString)
 		if err != nil {
-			return "", fmt.Errorf("cannot parse reference path %q: %w", pathString, err)
+			return "", safeerr.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)
+			return "", safeerr.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)
+			return "", safeerr.Errorf("unexpected reference format (expected resources.*): %q", pathString)
 		}
 
 		group := path.SkipPrefix(1).Prefix(1).String()
@@ -41,7 +42,7 @@ func evaluateTemplate(state TFStateAttrs, template string) (string, error) {
 
 		value, err := LookupTFField(state, group, name, fieldPath)
 		if err != nil {
-			return "", fmt.Errorf("cannot look up %q: %w", pathString, err)
+			return "", safeerr.Errorf("cannot look up %q: %w", pathString, err)
 		}
 
 		result = strings.ReplaceAll(result, "${"+pathString+"}", fmt.Sprintf("%v", value))
@@ -88,7 +89,9 @@ func ResolveFieldRef(ctx context.Context, state TFStateAttrs, srcGroup, srcName
 	case errB == nil:
 		return valueB, false, nil
 	default:
-		return nil, false, fmt.Errorf("%s.%s field %s: method A: %w; method B: %w",
-			srcGroup, srcName, fieldPath, errA, errB)
+		// srcGroup is a resource type the CLI defines, so it is safe to report;
+		// the resource name and field path are not.
+		return nil, false, safeerr.Errorf("%s.%s field %s: method A: %w; method B: %w",
+			safeerr.Safe(srcGroup), srcName, fieldPath, errA, errB)
 	}
 }
diff --git a/bundle/migrate/tf_state.go b/bundle/migrate/tf_state.go
index 55634234051..04077b05411 100644
--- a/bundle/migrate/tf_state.go
+++ b/bundle/migrate/tf_state.go
@@ -5,11 +5,11 @@ 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/safeerr"
 	"github.com/databricks/cli/libs/structs/structpath"
 	tfjson "github.com/hashicorp/terraform-json"
 )
@@ -94,7 +94,7 @@ func parseTFStateAttrsFromRaw(s *rawTFState) TFStateAttrs {
 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)
+		return nil, safeerr.Errorf("unknown resource group %q", safeerr.Safe(group))
 	}
 
 	// Translate field path to TF naming.
@@ -107,7 +107,7 @@ func LookupTFField(state TFStateAttrs, group, name string, fieldPath *structpath
 
 	attrsJSON, ok := state[tfType][name]
 	if !ok {
-		return nil, fmt.Errorf("%s.%s not found in TF state", tfType, name)
+		return nil, safeerr.Errorf("%s.%s not found in TF state", safeerr.Safe(tfType), name)
 	}
 
 	// Unmarshal into map[string]any to handle TF list-blocks: in TF state, single-block
@@ -120,7 +120,7 @@ func LookupTFField(state TFStateAttrs, group, name string, fieldPath *structpath
 	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 nil, safeerr.Errorf("cannot parse TF state for %s.%s: %w", safeerr.Safe(tfType), name, err)
 	}
 
 	return navigateTFState(attrs, tfFieldPath)
@@ -148,18 +148,18 @@ func navigateTFState(data map[string]any, path *structpath.PathNode) (any, error
 			}
 			m, ok := current.(map[string]any)
 			if !ok {
-				return nil, fmt.Errorf("expected map at %q, got %T", key, current)
+				return nil, safeerr.Errorf("expected map at %q, got %T", key, current)
 			}
 			val, ok := m[key]
 			if !ok {
-				return nil, fmt.Errorf("%q: key not found", key)
+				return nil, safeerr.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))
+					return nil, safeerr.Errorf("index %d out of range (len %d)", safeerr.Safe(idx), safeerr.Safe(len(v)))
 				}
 				current = v[idx]
 			default:
@@ -167,7 +167,7 @@ func navigateTFState(data map[string]any, path *structpath.PathNode) (any, error
 				if idx == 0 {
 					continue
 				}
-				return nil, fmt.Errorf("index %d: not a slice (%T)", idx, current)
+				return nil, safeerr.Errorf("index %d: not a slice (%T)", safeerr.Safe(idx), current)
 			}
 		}
 	}
diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go
index bc06575baf3..a7c0f6dbd07 100644
--- a/bundle/phases/deploy.go
+++ b/bundle/phases/deploy.go
@@ -77,7 +77,7 @@ func approvalForDeploy(ctx context.Context, b *bundle.Bundle, plan *deployplan.P
 	return cmdio.AskYesOrNo(ctx, "Would you like to proceed?")
 }
 
-func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, stateEngine engine.EngineType, requestedEngine engine.EngineSetting) {
+func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, stateEngine engine.EngineType) {
 	// Apply resources and capture post-apply state.
 	// For direct: Finalize flushes the WAL to disk and returns the state;
 	// called even if Apply failed so partial progress is saved.
@@ -116,15 +116,6 @@ func deployCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, st
 		metadata.Upload(),
 		statemgmt.UploadStateForYamlSync(stateEngine),
 	)
-
-	// Once the deploy is complete, dry-run the migration to the direct engine
-	// and record the outcome in telemetry. If the user has opted in to the
-	// direct engine (via bundle.engine or DATABRICKS_BUNDLE_ENGINE) and the
-	// dry-run is clean, the migration is committed; otherwise nothing is
-	// written and the deploy is unaffected.
-	if !stateEngine.IsDirect() && !logdiag.HasError(ctx) {
-		statemgmt.MigrateToDirect(ctx, b, requestedEngine)
-	}
 }
 
 // logFileSummary reports what the file sync did. Separate from the resource summary
@@ -310,7 +301,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 		return
 	}
 	if haveApproval {
-		deployCore(ctx, b, plan, stateEngine, requestedEngine)
+		deployCore(ctx, b, plan, stateEngine)
 	} else {
 		cmdio.LogString(ctx, "Deployment cancelled!")
 		return
@@ -330,6 +321,23 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 	// was intended rather than what was applied.
 	filesReported = true
 	logDeploySummary(ctx, b, plan)
+
+	// Dry-run the migration to the direct engine and record the outcome in
+	// telemetry. If the user opted in (via bundle.engine or
+	// DATABRICKS_BUNDLE_ENGINE) and the dry-run is clean, the migration is
+	// committed; otherwise nothing is written and the deploy is unaffected.
+	//
+	// Last, after the deploy has reported what it did: this is post-deploy work,
+	// and its warnings read as belonging to the deploy if they precede the
+	// summary.
+	//
+	// Gated on the deploy alone, which the early return above already guarantees
+	// — not on the postdeploy script. The resources were applied before that
+	// script ran, so the state is worth migrating even if it failed, the same
+	// reasoning that prints the summary regardless.
+	if !stateEngine.IsDirect() {
+		statemgmt.MigrateToDirect(ctx, b, requestedEngine)
+	}
 }
 
 func RunPlan(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) *deployplan.Plan {
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index cbe74f467fe..bb22cc376ef 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -305,6 +305,11 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 				ComplexVariableCount:         complexVariableCount,
 				LookupVariableCount:          lookupVariableCount,
 				BundleMutatorExecutionTimeMs: getExecutionTimes(b),
+
+				// Own alignment group: these names are long enough that sharing one
+				// would re-align every field above them.
+				DirectMigrateErrorTemplate:       b.Metrics.DirectMigrateErrorTemplate,
+				DirectMigrateCommitErrorTemplate: b.Metrics.DirectMigrateCommitErrorTemplate,
 			},
 		},
 	})
diff --git a/bundle/statemgmt/direct_migration.go b/bundle/statemgmt/direct_migration.go
index a13b7effd46..1eee7a1616f 100644
--- a/bundle/statemgmt/direct_migration.go
+++ b/bundle/statemgmt/direct_migration.go
@@ -19,10 +19,12 @@ import (
 	"github.com/databricks/cli/bundle/metrics"
 	"github.com/databricks/cli/bundle/migrate"
 	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/filer"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/cli/libs/safeerr"
 )
 
 // warnPrefix labels warnings emitted by the post-deploy dry-run so they are not
@@ -55,6 +57,7 @@ func MigrateToDirect(ctx context.Context, b *bundle.Bundle, requestedEngine engi
 	tfState, err := migrate.ParseTFStateFull(ctx, localTerraformPath)
 	if err != nil {
 		log.Warnf(ctx, "%sfailed to parse terraform state: %v", warnPrefix, err)
+		recordErrorTemplate(&b.Metrics.DirectMigrateErrorTemplate, err)
 		if requestedEngine.Type == engine.EngineDirect {
 			b.Metrics.SetBoolValue(metrics.DirectMigrateError, true)
 			log.Warnf(ctx, "%s", autoMigrateStoppedNotice)
@@ -84,6 +87,7 @@ func MigrateToDirect(ctx context.Context, b *bundle.Bundle, requestedEngine engi
 		cmdio.LogString(ctx, "Removing empty terraform state; direct engine will be used on the next deploy (opted in via "+requestedEngine.Source+")...")
 		if err := backupTerraformState(ctx, b); err != nil {
 			b.Metrics.SetBoolValue(metrics.DirectMigrateCommitError, true)
+			recordErrorTemplate(&b.Metrics.DirectMigrateCommitErrorTemplate, err)
 			log.Warnf(ctx, "automatic migration to direct engine failed: %v", err)
 			return
 		}
@@ -106,6 +110,7 @@ func MigrateToDirect(ctx context.Context, b *bundle.Bundle, requestedEngine engi
 
 	if err != nil {
 		log.Warnf(ctx, "%s%v", warnPrefix, err)
+		recordErrorTemplate(&b.Metrics.DirectMigrateErrorTemplate, err)
 	}
 	if hasWarnings || err != nil {
 		log.Warnf(ctx, "%s", feedbackNotice)
@@ -136,6 +141,7 @@ func MigrateToDirect(ctx context.Context, b *bundle.Bundle, requestedEngine engi
 
 	if err := commitMigration(ctx, b, tempStatePath, resourceCount); err != nil {
 		b.Metrics.SetBoolValue(metrics.DirectMigrateCommitError, true)
+		recordErrorTemplate(&b.Metrics.DirectMigrateCommitErrorTemplate, err)
 		log.Warnf(ctx, "automatic migration to direct engine failed: %v", err)
 		return
 	}
@@ -143,6 +149,17 @@ func MigrateToDirect(ctx context.Context, b *bundle.Bundle, requestedEngine engi
 	recordAutoMigrateSource(b, requestedEngine)
 }
 
+// recordErrorTemplate stores a PII-free description of err in target, which is
+// the metric for the same failure class as the boolean recorded alongside it —
+// so a query tells a conversion failure from a commit failure by field, without
+// joining against the booleans. Recorded for both populations, the opt-in one
+// and the dry run, since the booleans only say that a migration failed.
+func recordErrorTemplate(target *string, err error) {
+	if template := diag.ErrorTemplate(err); template != "" {
+		*target = template
+	}
+}
+
 // recordDryRunNoop records dry-run telemetry for a no-op case (no state, or
 // state with no managed resources) when the user did NOT opt in. On opt-in
 // paths the caller uses direct_migrate_* keys instead.
@@ -185,20 +202,20 @@ func backupTerraformState(ctx context.Context, b *bundle.Bundle) error {
 	remoteTerraformPath, localTerraformPath := b.StateFilenameTerraform(ctx)
 	reader, err := f.Read(ctx, remoteTerraformPath)
 	if err != nil && !errors.Is(err, fs.ErrNotExist) {
-		return fmt.Errorf("reading remote terraform state %s: %w", remoteTerraformPath, err)
+		return safeerr.Errorf("reading remote terraform state %s: %w", remoteTerraformPath, err)
 	}
 	if err == nil {
 		defer reader.Close()
 		if err := f.Write(ctx, remoteTerraformPath+".backup", reader, filer.OverwriteIfExists); err != nil {
-			return fmt.Errorf("writing remote terraform backup: %w", err)
+			return safeerr.Errorf("writing remote terraform backup: %w", err)
 		}
 		if err := f.Delete(ctx, remoteTerraformPath); err != nil {
-			return fmt.Errorf("deleting remote terraform state: %w", err)
+			return safeerr.Errorf("deleting remote terraform state: %w", err)
 		}
 	}
 
 	if err := os.Rename(localTerraformPath, localTerraformPath+".backup"); err != nil && !errors.Is(err, fs.ErrNotExist) {
-		return fmt.Errorf("renaming local terraform state to %s.backup: %w", localTerraformPath, err)
+		return safeerr.Errorf("renaming local terraform state to %s.backup: %w", localTerraformPath, err)
 	}
 	return nil
 }
@@ -256,12 +273,15 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 	var stateDB dstate.DeploymentState
 	stateDB.OpenWithData(tempStatePath, 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
-	// the migrated state and config agree on .permissions entries.
-	bundle.ApplyContext(ctx, b, resourcemutator.SecretScopeFixups(engine.EngineDirect))
-	if logdiag.HasError(ctx) {
-		return tempStatePath, resourceCount, false, errors.New("failed to apply secret scope fixups")
+	// Apply the secret scope fixups so the config matches what the direct engine
+	// expects. This adds MANAGE ACL for the current user to all secret scopes,
+	// ensuring the migrated state and config agree on .permissions entries.
+	//
+	// Called directly rather than through the mutator so the failure arrives as an
+	// error: a diagnostic's summary would reach telemetry as a generic "failed to
+	// apply secret scope fixups", losing which fixup failed.
+	if _, err := resourcemutator.ApplySecretScopeFixups(b, engine.EngineDirect); err != nil {
+		return tempStatePath, resourceCount, false, err
 	}
 
 	// b.Config has been modified by terraform.Interpolate which converts bundle-style
@@ -269,7 +289,7 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 	// BuildStateFromTF expects ${resources.*} references, so reverse the interpolation first.
 	uninterpolatedRoot, err := reverseInterpolate(b.Config.Value())
 	if err != nil {
-		return tempStatePath, resourceCount, false, fmt.Errorf("failed to reverse interpolation: %w", err)
+		return tempStatePath, resourceCount, false, safeerr.Errorf("failed to reverse interpolation: %w", err)
 	}
 
 	var uninterpolatedConfig config.Root
@@ -277,7 +297,7 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 		return uninterpolatedRoot, nil
 	})
 	if err != nil {
-		return tempStatePath, resourceCount, false, fmt.Errorf("failed to create uninterpolated config: %w", err)
+		return tempStatePath, resourceCount, false, safeerr.Errorf("failed to create uninterpolated config: %w", err)
 	}
 
 	adapters, err := dresources.InitAll(nil)
@@ -286,7 +306,7 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 	}
 
 	if err := stateDB.UpgradeToWrite(); err != nil {
-		return tempStatePath, resourceCount, false, fmt.Errorf("upgrading state for apply: %w", err)
+		return tempStatePath, resourceCount, false, safeerr.Errorf("upgrading state for apply: %w", err)
 	}
 
 	// warnPrefix labels the conversion's warnings as coming from the background dry run.
@@ -301,7 +321,7 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 
 	// BuildStateFromTF reports some failures via logdiag instead of returning an error.
 	if logdiag.HasError(ctx) {
-		return tempStatePath, resourceCount, hasWarnings, errors.New("state conversion failed")
+		return tempStatePath, resourceCount, hasWarnings, safeerr.New("state conversion failed")
 	}
 
 	return tempStatePath, resourceCount, hasWarnings, nil
@@ -320,13 +340,13 @@ func commitMigration(ctx context.Context, b *bundle.Bundle, tempStatePath string
 	// "file is missing"; treat it as a hard failure to avoid renaming over
 	// something we couldn't read.
 	if _, err := os.Stat(localDirectPath); err == nil {
-		return fmt.Errorf("state file %s already exists", localDirectPath)
+		return safeerr.Errorf("state file %s already exists", localDirectPath)
 	} else if !errors.Is(err, fs.ErrNotExist) {
-		return fmt.Errorf("stat %s: %w", localDirectPath, err)
+		return safeerr.Errorf("stat %s: %w", localDirectPath, err)
 	}
 
 	if err := pushDirectState(ctx, b, tempStatePath); err != nil {
-		return fmt.Errorf("pushing direct state to workspace: %w", err)
+		return safeerr.Errorf("pushing direct state to workspace: %w", err)
 	}
 
 	// Remote is now authoritative for direct engine; make local match. Local
@@ -336,13 +356,13 @@ func commitMigration(ctx context.Context, b *bundle.Bundle, tempStatePath string
 	// on failure so telemetry reflects what actually happened here (the
 	// migration is complete on the workspace but not on this checkout).
 	if err := os.MkdirAll(filepath.Dir(localDirectPath), 0o700); err != nil {
-		return fmt.Errorf("workspace migrated but creating local state directory failed: %w", err)
+		return safeerr.Errorf("workspace migrated but creating local state directory failed: %w", err)
 	}
 	if err := os.Rename(tempStatePath, localDirectPath); err != nil {
-		return fmt.Errorf("workspace migrated but writing local direct state failed: %w", err)
+		return safeerr.Errorf("workspace migrated but writing local direct state failed: %w", err)
 	}
 	if err := os.Rename(localTerraformPath, localTerraformPath+".backup"); err != nil && !errors.Is(err, fs.ErrNotExist) {
-		return fmt.Errorf("workspace migrated but backing up local terraform state failed: %w", err)
+		return safeerr.Errorf("workspace migrated but backing up local terraform state failed: %w", err)
 	}
 
 	suffix := "s"
@@ -388,16 +408,16 @@ func pushDirectState(ctx context.Context, b *bundle.Bundle, localPath string) er
 		return nil
 	}
 	if err != nil {
-		return fmt.Errorf("reading remote terraform state %s: %w", remoteTerraformPath, err)
+		return safeerr.Errorf("reading remote terraform state %s: %w", remoteTerraformPath, err)
 	}
 	defer reader.Close()
 
 	if err := f.Write(ctx, remoteTerraformPath+".backup", reader, filer.OverwriteIfExists); err != nil {
-		return fmt.Errorf("writing remote terraform backup: %w", err)
+		return safeerr.Errorf("writing remote terraform backup: %w", err)
 	}
 
 	if err := f.Delete(ctx, remoteTerraformPath); err != nil {
-		return fmt.Errorf("deleting remote terraform state: %w", err)
+		return safeerr.Errorf("deleting remote terraform state: %w", err)
 	}
 
 	return nil
diff --git a/libs/diag/diagnostic.go b/libs/diag/diagnostic.go
index fe7090462d4..d94658cc2f2 100644
--- a/libs/diag/diagnostic.go
+++ b/libs/diag/diagnostic.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/safeerr"
 )
 
 type Diagnostic struct {
@@ -54,6 +55,34 @@ func FromErr(err error) Diagnostics {
 	}
 }
 
+// ErrorTemplate returns a PII-free description of err for telemetry: the
+// message template of a safeerr error, or a typed error's own stand-in, plus the
+// safe fields of any API error at the end of its chain. The halves are
+// independent: a CLI error carries a template but
+// may wrap no API error, and an API error reached without any safeerr wrapping
+// has safe fields but no template.
+func ErrorTemplate(err error) string {
+	template := safeerr.ErrorTemplate(err)
+	if template == "" {
+		// Not raised through safeerr, but a typed error can still describe itself
+		// — libs/filer's errors do. This covers the call sites not yet converted,
+		// which is most of them.
+		if ss, ok := err.(safeerr.SafeStringer); ok {
+			template = ss.SafeString()
+		}
+	}
+	apiDescription := SafeAPIErrorDescription(err)
+
+	switch {
+	case apiDescription == "":
+		return template
+	case template == "":
+		return apiDescription
+	default:
+		return template + " [" + apiDescription + "]"
+	}
+}
+
 // FromErr returns a new warning diagnostic from the specified error, if any.
 func WarningFromErr(err error) Diagnostics {
 	if err == nil {
diff --git a/libs/diag/error_template_test.go b/libs/diag/error_template_test.go
new file mode 100644
index 00000000000..65f8e20dc2c
--- /dev/null
+++ b/libs/diag/error_template_test.go
@@ -0,0 +1,83 @@
+package diag
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/databricks/cli/libs/safeerr"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/stretchr/testify/assert"
+)
+
+// apiError builds an SDK error whose message carries user data, the way a real
+// one does.
+func apiError(code string, status int) error {
+	return &apierr.APIError{
+		ErrorCode:  code,
+		StatusCode: status,
+		Message:    "User alice@example.com cannot access /Workspace/Users/alice@example.com/job",
+	}
+}
+
+func TestSafeAPIErrorDescription(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{name: "not an api error", err: errors.New("boom"), want: ""},
+		{name: "code and status", err: apiError("PERMISSION_DENIED", http.StatusForbidden), want: "403 PERMISSION_DENIED"},
+		{name: "code only", err: apiError("RESOURCE_DOES_NOT_EXIST", 0), want: "RESOURCE_DOES_NOT_EXIST"},
+		{name: "status only", err: apiError("", http.StatusInternalServerError), want: "500"},
+		{name: "nothing safe", err: apiError("", 0), want: ""},
+		{name: "wrapped", err: fmt.Errorf("deploying: %w", apiError("QUOTA_EXCEEDED", 429)), want: "429 QUOTA_EXCEEDED"},
+
+		// A code that does not have the shape of an enum member is dropped
+		// rather than trusted, so free text cannot ride along.
+		{name: "code with a path", err: apiError("cannot find /Workspace/Users/a@b.com/x", 404), want: "404"},
+		{name: "code with a quoted name", err: apiError(`job "Q4 forecast" missing`, 404), want: "404"},
+		{name: "code lowercase", err: apiError("permission_denied", 403), want: "403"},
+		{name: "code with a dot", err: apiError("PERMISSION.DENIED", 403), want: "403"},
+		{name: "code with a space", err: apiError("PERMISSION DENIED", 403), want: "403"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, SafeAPIErrorDescription(tt.err))
+		})
+	}
+}
+
+// TestFromErrErrorTemplateCarriesNoUserData is the property the field exists
+// for: whatever the summary holds, the template holds none of it.
+
+func TestFromErrNil(t *testing.T) {
+	assert.Nil(t, FromErr(nil))
+}
+
+// TestErrorTemplateMatchesFromErr keeps the exported helper and the field in
+// step, since callers holding an error use one and callers holding a diagnostic
+// use the other.
+
+// standInErr models a typed CLI error such as libs/filer's: user data in the
+// message, a PII-free classification as its stand-in.
+type standInErr struct{}
+
+func (standInErr) Error() string      { return "access denied: /Workspace/Users/a@b.com/x" }
+func (standInErr) SafeString() string { return "access denied" }
+
+// TestErrorTemplateUsesStandInWithoutSafeerr covers the call sites not raised
+// through safeerr, which is most of them: a typed error still describes itself.
+func TestErrorTemplateUsesStandInWithoutSafeerr(t *testing.T) {
+	assert.Equal(t, "access denied", ErrorTemplate(standInErr{}))
+
+	// Combined with an API error at the end of the chain.
+	err := safeerr.Errorf("pushing state: %w", standInErr{})
+	assert.Equal(t, "pushing state: access denied", ErrorTemplate(err))
+
+	// A stand-in never displaces a real template.
+	assert.Equal(t, "cannot update %s: %w",
+		ErrorTemplate(safeerr.Errorf("cannot update %s: %w", "resources.jobs.x", errors.New("boom"))))
+}
diff --git a/libs/diag/sdk_error.go b/libs/diag/sdk_error.go
index d190498d6bf..7ad07c37f8e 100644
--- a/libs/diag/sdk_error.go
+++ b/libs/diag/sdk_error.go
@@ -3,6 +3,7 @@ package diag
 import (
 	"errors"
 	"fmt"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -42,3 +43,41 @@ func FormatAPIErrorDetails(e error) string {
 	}
 	return fmt.Sprintf("Endpoint: %s\nHTTP Status: %s\nAPI error_code: %s\nAPI message: %s", endpoint, httpStatus, apiErr.ErrorCode, apiErr.Message)
 }
+
+// safeErrorCode matches the shape of a platform error code: SCREAMING_SNAKE_CASE
+// and nothing else. The code is documented as a closed enum
+// (RESOURCE_DOES_NOT_EXIST, PERMISSION_DENIED, ...) but that is a convention,
+// not a contract, and the field is filled in by whichever service handled the
+// request. Requiring this shape is what makes it structurally impossible for a
+// path, principal, quoted resource name, or sentence of free text to reach
+// telemetry through it, whatever a service decides to return.
+var safeErrorCode = regexp.MustCompile(`^[A-Z][A-Z0-9_]{0,63}$`)
+
+// SafeAPIErrorDescription returns a PII-free description of an SDK API error, or
+// "" when e is not one or carries nothing safe to report.
+//
+// Only the structured fields are reported. An API error's message is
+// user-authored data — it echoes resource names, workspace paths, principals,
+// and config values back to the caller — so it never appears here. What is left
+// is still the most useful part for aggregation: which platform error the
+// request failed with.
+func SafeAPIErrorDescription(e error) string {
+	apiErr, ok := errors.AsType[*apierr.APIError](e)
+	if !ok {
+		return ""
+	}
+
+	// Status first, then code, matching FormatAPIErrorSummary so the template and
+	// the user-facing message order the same two values the same way.
+	var parts []string
+	if apiErr.StatusCode != 0 {
+		parts = append(parts, strconv.Itoa(apiErr.StatusCode))
+	}
+	if safeErrorCode.MatchString(apiErr.ErrorCode) {
+		parts = append(parts, apiErr.ErrorCode)
+	}
+	if len(parts) == 0 {
+		return ""
+	}
+	return strings.Join(parts, " ")
+}
diff --git a/libs/filer/errors.go b/libs/filer/errors.go
index 8648502c229..7cd62c1b083 100644
--- a/libs/filer/errors.go
+++ b/libs/filer/errors.go
@@ -2,6 +2,25 @@ package filer
 
 import "io/fs"
 
+// Each error below is a fixed classification followed by the path it concerns.
+// The classification is a source literal and safe to report to telemetry; the
+// path is user data. Naming the literals here lets Error() and SafeString()
+// derive from one string, so the message and what telemetry sees cannot drift.
+//
+// SafeString implements safeerr.SafeStringer: an error wrapped with %w by a
+// safeerr error contributes its classification to that error's message
+// template. See libs/safeerr.
+const (
+	msgFileAlreadyExists = "file already exists"
+	msgFileDoesNotExist  = "file does not exist"
+	msgNoSuchDirectory   = "no such directory"
+	msgNotADirectory     = "not a directory"
+	msgNotAFile          = "not a file"
+	msgDirectoryNotEmpty = "directory not empty"
+	msgCannotDeleteRoot  = "unable to delete filer root"
+	msgPermissionDenied  = "access denied"
+)
+
 // fileAlreadyExistsError is returned when attempting to write a file at a path
 // that already exists, without using the OverwriteIfExists WriteMode flag.
 type fileAlreadyExistsError struct {
@@ -9,7 +28,11 @@ type fileAlreadyExistsError struct {
 }
 
 func (err fileAlreadyExistsError) Error() string {
-	return "file already exists: " + err.path
+	return msgFileAlreadyExists + ": " + err.path
+}
+
+func (err fileAlreadyExistsError) SafeString() string {
+	return msgFileAlreadyExists
 }
 
 func (err fileAlreadyExistsError) Is(other error) bool {
@@ -28,7 +51,11 @@ func (err fileDoesNotExistError) Is(other error) bool {
 }
 
 func (err fileDoesNotExistError) Error() string {
-	return "file does not exist: " + err.path
+	return msgFileDoesNotExist + ": " + err.path
+}
+
+func (err fileDoesNotExistError) SafeString() string {
+	return msgFileDoesNotExist
 }
 
 // noSuchDirectoryError is returned when attempting to write a file to a path
@@ -39,7 +66,11 @@ type noSuchDirectoryError struct {
 }
 
 func (err noSuchDirectoryError) Error() string {
-	return "no such directory: " + err.path
+	return msgNoSuchDirectory + ": " + err.path
+}
+
+func (err noSuchDirectoryError) SafeString() string {
+	return msgNoSuchDirectory
 }
 
 func (err noSuchDirectoryError) Is(other error) bool {
@@ -53,7 +84,11 @@ type notADirectory struct {
 }
 
 func (err notADirectory) Error() string {
-	return "not a directory: " + err.path
+	return msgNotADirectory + ": " + err.path
+}
+
+func (err notADirectory) SafeString() string {
+	return msgNotADirectory
 }
 
 func (err notADirectory) Is(other error) bool {
@@ -67,7 +102,11 @@ type notAFile struct {
 }
 
 func (err notAFile) Error() string {
-	return "not a file: " + err.path
+	return msgNotAFile + ": " + err.path
+}
+
+func (err notAFile) SafeString() string {
+	return msgNotAFile
 }
 
 func (err notAFile) Is(other error) bool {
@@ -82,7 +121,11 @@ type directoryNotEmptyError struct {
 }
 
 func (err directoryNotEmptyError) Error() string {
-	return "directory not empty: " + err.path
+	return msgDirectoryNotEmpty + ": " + err.path
+}
+
+func (err directoryNotEmptyError) SafeString() string {
+	return msgDirectoryNotEmpty
 }
 
 func (err directoryNotEmptyError) Is(other error) bool {
@@ -95,7 +138,12 @@ func (err directoryNotEmptyError) Is(other error) bool {
 type cannotDeleteRootError struct{}
 
 func (err cannotDeleteRootError) Error() string {
-	return "unable to delete filer root"
+	return msgCannotDeleteRoot
+}
+
+// SafeString is the whole message: this error carries no path.
+func (err cannotDeleteRootError) SafeString() string {
+	return msgCannotDeleteRoot
 }
 
 func (err cannotDeleteRootError) Is(other error) bool {
@@ -113,7 +161,11 @@ type permissionError struct {
 }
 
 func (err permissionError) Error() string {
-	return "access denied: " + err.path
+	return msgPermissionDenied + ": " + err.path
+}
+
+func (err permissionError) SafeString() string {
+	return msgPermissionDenied
 }
 
 func (err permissionError) Is(other error) bool {
diff --git a/libs/filer/errors_test.go b/libs/filer/errors_test.go
index e12fe946450..02c75a2a533 100644
--- a/libs/filer/errors_test.go
+++ b/libs/filer/errors_test.go
@@ -2,8 +2,10 @@ package filer
 
 import (
 	"io/fs"
+	"strings"
 	"testing"
 
+	"github.com/databricks/cli/libs/safeerr"
 	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -109,3 +111,48 @@ func TestPermissionError_Unwrap(t *testing.T) {
 	require.ErrorAs(t, err, &got)
 	assert.Equal(t, "MAX_CHILD_NODE_SIZE_EXCEEDED", got.ErrorCode)
 }
+
+// TestErrorSafeStringOmitsPath is the property that lets these errors be
+// reported to telemetry: the classification survives, the path does not.
+func TestErrorSafeStringOmitsPath(t *testing.T) {
+	const path = "/Workspace/Users/someone@example.com/secret_project"
+
+	tests := []struct {
+		err            error
+		wantSafeString string
+	}{
+		{err: fileAlreadyExistsError{path: path}, wantSafeString: "file already exists"},
+		{err: fileDoesNotExistError{path: path}, wantSafeString: "file does not exist"},
+		{err: noSuchDirectoryError{path: path}, wantSafeString: "no such directory"},
+		{err: notADirectory{path: path}, wantSafeString: "not a directory"},
+		{err: notAFile{path: path}, wantSafeString: "not a file"},
+		{err: directoryNotEmptyError{path: path}, wantSafeString: "directory not empty"},
+		{err: permissionError{path: path}, wantSafeString: "access denied"},
+		{err: cannotDeleteRootError{}, wantSafeString: "unable to delete filer root"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.wantSafeString, func(t *testing.T) {
+			safe, ok := tt.err.(interface{ SafeString() string })
+			require.True(t, ok, "%T must supply a stand-in", tt.err)
+
+			assert.Equal(t, tt.wantSafeString, safe.SafeString())
+			assert.NotContains(t, safe.SafeString(), path)
+
+			// The message still leads with the same classification.
+			assert.True(t, strings.HasPrefix(tt.err.Error(), tt.wantSafeString),
+				"%q should start with %q", tt.err.Error(), tt.wantSafeString)
+		})
+	}
+}
+
+// TestErrorSafeStringReachesTemplate covers the end-to-end path: a filer error
+// wrapped by safeerr contributes its classification to the template.
+func TestErrorSafeStringReachesTemplate(t *testing.T) {
+	const path = "/Workspace/Users/someone@example.com/x"
+	err := safeerr.Errorf("pushing direct state to workspace: %w", permissionError{path: path})
+
+	assert.Equal(t, "pushing direct state to workspace: access denied: "+path, err.Error())
+	assert.Equal(t, "pushing direct state to workspace: access denied", safeerr.ErrorTemplate(err))
+	assert.NotContains(t, safeerr.ErrorTemplate(err), path)
+}
diff --git a/libs/safeerr/safeerr.go b/libs/safeerr/safeerr.go
new file mode 100644
index 00000000000..0394d4080cc
--- /dev/null
+++ b/libs/safeerr/safeerr.go
@@ -0,0 +1,277 @@
+// Package safeerr provides errors that retain their message template: the
+// format string, with values that came from the user left as verbs. Telemetry
+// can then report which error occurred without shipping the values
+// interpolated into it.
+//
+// An error created here behaves exactly like the fmt.Errorf it wraps. Error()
+// returns the fully interpolated message and errors.Is/errors.AsType walk the
+// same chain, so converting a call site changes nothing for existing consumers.
+// The template is reachable only through ErrorTemplate.
+//
+// The safe/unsafe vocabulary follows github.com/cockroachdb/redact: every
+// value is unsafe unless the caller marks it Safe.
+// See https://github.com/cockroachdb/errors#pii-free-error-reporting
+package safeerr
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// Errorf formats an error exactly like fmt.Errorf and retains format as the
+// error's message template.
+//
+// format must be a string constant. It is the one part of the error reported to
+// telemetry, so building it dynamically defeats the point of the package.
+func Errorf(format string, args ...any) error {
+	if false {
+		// Tells the vet printf analyzer that this is a printf wrapper, which it
+		// cannot infer on its own because the call below does not forward args
+		// verbatim. fmt.Errorf rather than fmt.Sprintf, so %w is accepted here
+		// too. Documented at
+		// https://pkg.go.dev/golang.org/x/tools/go/analysis/passes/printf
+		_ = fmt.Errorf(format, args...)
+	}
+
+	return &templateError{
+		err:      fmt.Errorf(format, renderArgs(args)...),
+		template: format,
+		args:     templateArgs(args),
+	}
+}
+
+// New returns an error whose message is entirely literal, making the whole
+// message its own template.
+func New(text string) error {
+	return &templateError{err: errors.New(text), template: text, args: nil}
+}
+
+// Safe marks v as free of user data, allowing its value to appear in the
+// message template. Use it only for values drawn from a set the CLI itself
+// defines — field paths, resource groups, API status codes — never for names,
+// paths, or anything else that originates in the user's configuration.
+//
+// Marking a value Safe does not change the error message: the value is
+// formatted by its verb exactly as it would have been without the marker.
+//
+// Safe does not override a value's own SafeStringer stand-in; see SafeStringer.
+func Safe(v any) any {
+	return safeValue{v: v}
+}
+
+// SafeStringer is implemented by values that are only partly free of user data
+// and can supply a PII-free stand-in for themselves. The full value still
+// appears in the error message; only the stand-in reaches the template.
+//
+// A resource key is the motivating case: "resources.jobs.my_job" mixes a group
+// the CLI defines with a name the user chose, so it stands in for itself as
+// "jobs.*" — enough to tell a failing job from a failing pipeline without
+// reporting which one.
+//
+// An error may also be a SafeStringer: under %w its own template is preferred,
+// and the stand-in is used only when it has none. SafeStringer otherwise wins
+// over Safe, so wrapping such a value in Safe cannot put its user-supplied part
+// back into the template.
+type SafeStringer interface {
+	SafeString() string
+}
+
+// ErrorTemplate returns the message template of the outermost templated error
+// in err's chain, or "" when the chain holds none.
+//
+// Templates chain: a %w whose wrapped error is itself templated contributes its
+// own template in place of the verb, recursively. A %w wrapping any other error
+// stays a bare %w, since nothing about that error is known to be safe.
+func ErrorTemplate(err error) string {
+	te, ok := errors.AsType[*templateError](err)
+	if !ok {
+		return ""
+	}
+	return te.expand()
+}
+
+// safeValue marks one argument of Errorf as free of user data.
+type safeValue struct{ v any }
+
+type templateError struct {
+	// err is the fmt.Errorf result. It carries the interpolated message and
+	// the unwrap chain, so this type never reimplements either.
+	err error
+
+	// template is the format string passed to Errorf.
+	template string
+
+	// args holds only what expand can use: safe values and wrapped errors.
+	// Every other argument is nil, so no user data stays reachable here.
+	args []any
+}
+
+func (e *templateError) Error() string {
+	return e.err.Error()
+}
+
+// Unwrap returns the fmt.Errorf result rather than the wrapped error itself, so
+// an error built with several %w verbs keeps working: that value implements
+// Unwrap() []error, which errors.Is and errors.AsType traverse.
+func (e *templateError) Unwrap() error {
+	return e.err
+}
+
+// renderArgs strips Safe markers so the message fmt produces is identical to
+// the one a plain fmt.Errorf call with the same values would have produced.
+func renderArgs(args []any) []any {
+	out := make([]any, len(args))
+	for i, a := range args {
+		if s, ok := a.(safeValue); ok {
+			out[i] = s.v
+		} else {
+			out[i] = a
+		}
+	}
+	return out
+}
+
+// templateArgs keeps the arguments expand may substitute — safe values, the
+// stand-ins of SafeStringer values, and errors whose own template can be
+// chained in — and drops the rest.
+func templateArgs(args []any) []any {
+	out := make([]any, len(args))
+	for i, a := range args {
+		switch v := a.(type) {
+		case safeValue:
+			// A value that declares a stand-in keeps it even here: the type
+			// knows which part of itself is user data, so it outranks a
+			// call-site assertion that the whole value is safe.
+			if ss, ok := v.v.(SafeStringer); ok {
+				out[i] = safeValue{v: ss.SafeString()}
+			} else {
+				out[i] = v
+			}
+		case error:
+			out[i] = v
+		case SafeStringer:
+			// Resolve the stand-in now and keep only that, so the value it came
+			// from — which holds user data — is not retained by the error.
+			out[i] = safeValue{v: v.SafeString()}
+		}
+	}
+	return out
+}
+
+// expand walks the template, substituting safe values and chained templates and
+// leaving every other verb in place.
+func (e *templateError) expand() string {
+	var sb strings.Builder
+	argIndex := 0
+
+	for i := 0; i < len(e.template); {
+		if e.template[i] != '%' {
+			sb.WriteByte(e.template[i])
+			i++
+			continue
+		}
+
+		spec, verb, next, ok := parseVerb(e.template, i)
+		if !ok {
+			// A construct expand does not model shifts which argument each
+			// later verb consumes, so substituting past it could attribute a
+			// safe value to the wrong verb. Fall back to the format string,
+			// which is a source literal and safe to report on its own.
+			return e.template
+		}
+
+		i = next
+		if verb == '%' {
+			sb.WriteString(spec)
+			continue
+		}
+
+		sb.WriteString(e.substitute(spec, verb, argIndex))
+		argIndex++
+	}
+
+	return sb.String()
+}
+
+// substitute returns the text to emit for a single verb.
+func (e *templateError) substitute(spec string, verb byte, argIndex int) string {
+	if argIndex >= len(e.args) {
+		// More verbs than arguments; go vet reports the call itself.
+		return spec
+	}
+
+	arg := e.args[argIndex]
+	if s, ok := arg.(safeValue); ok {
+		arg = s.v
+	} else if verb != 'w' {
+		return spec
+	}
+
+	if verb == 'w' {
+		// %w is only valid in fmt.Errorf, so it is never rendered.
+		err, ok := arg.(error)
+		if !ok {
+			return spec
+		}
+		// The wrapped error's own template wins. Failing that, an error that
+		// supplies a stand-in contributes that instead, which is how the CLI's
+		// own typed errors report their classification without the path or name
+		// they carry. Only the wrapped error itself is consulted, not its chain.
+		if inner := ErrorTemplate(err); inner != "" {
+			return inner
+		}
+		if ss, ok := err.(SafeStringer); ok {
+			return ss.SafeString()
+		}
+		return spec
+	}
+
+	return fmt.Sprintf(spec, arg)
+}
+
+// verbFlags are the flag characters fmt accepts between '%' and the verb.
+const verbFlags = "+-# 0"
+
+// parseVerb parses the verb that starts at the '%' at index i, returning its
+// full spec (e.g. "%-10q"), the verb letter, and the index just past it.
+//
+// It reports false for a dangling '%' and for the two constructs that change
+// which argument a verb consumes: an explicit argument index (%[2]s) and a '*'
+// width or precision (%*d).
+func parseVerb(format string, i int) (spec string, verb byte, next int, ok bool) {
+	j := i + 1
+	for j < len(format) && strings.IndexByte(verbFlags, format[j]) >= 0 {
+		j++
+	}
+
+	j, ok = skipNumber(format, j)
+	if !ok {
+		return "", 0, 0, false
+	}
+
+	if j < len(format) && format[j] == '.' {
+		j, ok = skipNumber(format, j+1)
+		if !ok {
+			return "", 0, 0, false
+		}
+	}
+
+	if j >= len(format) {
+		return "", 0, 0, false
+	}
+
+	return format[i : j+1], format[j], j + 1, true
+}
+
+// skipNumber advances past a width or precision, rejecting the '*' and '['
+// forms that consume an argument of their own.
+func skipNumber(format string, j int) (int, bool) {
+	if j < len(format) && (format[j] == '*' || format[j] == '[') {
+		return 0, false
+	}
+	for j < len(format) && format[j] >= '0' && format[j] <= '9' {
+		j++
+	}
+	return j, true
+}
diff --git a/libs/safeerr/safeerr_test.go b/libs/safeerr/safeerr_test.go
new file mode 100644
index 00000000000..c10d2ce0b08
--- /dev/null
+++ b/libs/safeerr/safeerr_test.go
@@ -0,0 +1,471 @@
+package safeerr
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// unsafeValue appears in every argument that is not marked Safe, so a test can
+// assert it never reaches a template.
+const unsafeValue = "resources.jobs.my_secret_job"
+
+func TestErrorf(t *testing.T) {
+	tests := []struct {
+		name         string
+		format       string
+		args         []any
+		wantMessage  string
+		wantTemplate string
+	}{
+		{
+			name:         "no verbs",
+			format:       "state conversion failed",
+			wantMessage:  "state conversion failed",
+			wantTemplate: "state conversion failed",
+		},
+		{
+			name:         "unsafe string",
+			format:       "%s: getting config",
+			args:         []any{unsafeValue},
+			wantMessage:  unsafeValue + ": getting config",
+			wantTemplate: "%s: getting config",
+		},
+		{
+			name:         "unsafe int",
+			format:       "unsupported deployment state version: %d",
+			args:         []any{7},
+			wantMessage:  "unsupported deployment state version: 7",
+			wantTemplate: "unsupported deployment state version: %d",
+		},
+		{
+			name:         "safe value is substituted",
+			format:       "%s: cannot set resolved value for field %q",
+			args:         []any{unsafeValue, Safe("tasks[0].job_id")},
+			wantMessage:  unsafeValue + `: cannot set resolved value for field "tasks[0].job_id"`,
+			wantTemplate: `%s: cannot set resolved value for field "tasks[0].job_id"`,
+		},
+		{
+			name:         "only safe values",
+			format:       "cannot convert %s to %s",
+			args:         []any{Safe("string"), Safe("int64")},
+			wantMessage:  "cannot convert string to int64",
+			wantTemplate: "cannot convert string to int64",
+		},
+		{
+			name:         "escaped percent is preserved",
+			format:       "100%% of %s",
+			args:         []any{unsafeValue},
+			wantMessage:  "100% of " + unsafeValue,
+			wantTemplate: "100%% of %s",
+		},
+		{
+			name:         "flags and width on an unsafe verb",
+			format:       "%-12s|",
+			args:         []any{"job"},
+			wantMessage:  "job         |",
+			wantTemplate: "%-12s|",
+		},
+		{
+			name:         "flags and width on a safe verb",
+			format:       "%-12s|",
+			args:         []any{Safe("job")},
+			wantMessage:  "job         |",
+			wantTemplate: "job         |",
+		},
+		{
+			name:         "width and precision on a safe verb",
+			format:       "%08.3f",
+			args:         []any{Safe(3.5)},
+			wantMessage:  "0003.500",
+			wantTemplate: "0003.500",
+		},
+		{
+			name:         "plus v on an unsafe verb",
+			format:       "reading state: %+v",
+			args:         []any{struct{ Path string }{unsafeValue}},
+			wantMessage:  "reading state: {Path:" + unsafeValue + "}",
+			wantTemplate: "reading state: %+v",
+		},
+		{
+			name:         "safe bool",
+			format:       "recovery enabled: %v",
+			args:         []any{Safe(true)},
+			wantMessage:  "recovery enabled: true",
+			wantTemplate: "recovery enabled: true",
+		},
+		{
+			name:         "mixed safe and unsafe in order",
+			format:       "%s: %s for %s in %s",
+			args:         []any{Safe("jobs"), unsafeValue, Safe("PrepareState"), "/home/user/bundle"},
+			wantMessage:  "jobs: " + unsafeValue + " for PrepareState in /home/user/bundle",
+			wantTemplate: "jobs: %s for PrepareState in %s",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := Errorf(tt.format, tt.args...)
+			assert.Equal(t, tt.wantMessage, err.Error())
+			assert.Equal(t, tt.wantTemplate, ErrorTemplate(err))
+		})
+	}
+}
+
+// TestErrorfMessageMatchesFmt is the property that makes converting a call site
+// a no-op for every existing consumer, including acceptance test goldens.
+func TestErrorfMessageMatchesFmt(t *testing.T) {
+	tests := []struct {
+		name   string
+		format string
+		safe   []any
+		raw    []any
+	}{
+		{
+			name:   "quoted safe value",
+			format: "field %q",
+			safe:   []any{Safe("tasks[0].job_id")},
+			raw:    []any{"tasks[0].job_id"},
+		},
+		{
+			name:   "safe struct with plus v",
+			format: "%+v",
+			safe:   []any{Safe(struct{ A int }{1})},
+			raw:    []any{struct{ A int }{1}},
+		},
+		{
+			name:   "safe nil",
+			format: "%v",
+			safe:   []any{Safe(nil)},
+			raw:    []any{nil},
+		},
+		{
+			name:   "safe error with s verb",
+			format: "%s",
+			safe:   []any{Safe(fs.ErrNotExist)},
+			raw:    []any{fs.ErrNotExist},
+		},
+		{
+			name:   "too few arguments",
+			format: "%s and %s",
+			safe:   []any{Safe("one")},
+			raw:    []any{"one"},
+		},
+		{
+			name:   "too many arguments",
+			format: "%s",
+			safe:   []any{Safe("one"), Safe("two")},
+			raw:    []any{"one", "two"},
+		},
+		{
+			name:   "wrong verb for type",
+			format: "%d",
+			safe:   []any{Safe("not a number")},
+			raw:    []any{"not a number"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, fmt.Errorf(tt.format, tt.raw...).Error(), Errorf(tt.format, tt.safe...).Error())
+		})
+	}
+}
+
+func TestErrorTemplateChains(t *testing.T) {
+	inner := Errorf("cannot convert %s to %s", Safe("string"), Safe("int64"))
+	middle := Errorf("%s: cannot set resolved value for field %q: %w", unsafeValue, Safe("tasks[0].job_id"), inner)
+	outer := Errorf("%s: SaveState: %w", unsafeValue, middle)
+
+	assert.Equal(t,
+		unsafeValue+": SaveState: "+unsafeValue+`: cannot set resolved value for field "tasks[0].job_id": cannot convert string to int64`,
+		outer.Error())
+	assert.Equal(t,
+		`%s: SaveState: %s: cannot set resolved value for field "tasks[0].job_id": cannot convert string to int64`,
+		ErrorTemplate(outer))
+
+	// Each level still reports its own template.
+	assert.Equal(t, `%s: cannot set resolved value for field "tasks[0].job_id": cannot convert string to int64`, ErrorTemplate(middle))
+	assert.Equal(t, "cannot convert string to int64", ErrorTemplate(inner))
+}
+
+func TestErrorTemplateChainsThroughForeignError(t *testing.T) {
+	// A %w wrapping an error with no template keeps the bare verb.
+	err := Errorf("reading %s: %w", "/home/user/state.json", fs.ErrNotExist)
+
+	assert.Equal(t, "reading /home/user/state.json: "+fs.ErrNotExist.Error(), err.Error())
+	assert.Equal(t, "reading %s: %w", ErrorTemplate(err))
+	assert.ErrorIs(t, err, fs.ErrNotExist)
+}
+
+func TestErrorTemplateChainsThroughUntemplatedWrap(t *testing.T) {
+	// A plain fmt.Errorf in the middle of the chain contributes nothing, so the
+	// innermost template that is known reaches the top.
+	inner := Errorf("cannot look up %q", Safe("continuous.pause_status"))
+	err := fmt.Errorf("%s: %w", unsafeValue, inner)
+
+	assert.Equal(t, `cannot look up "continuous.pause_status"`, ErrorTemplate(err))
+}
+
+func TestErrorTemplateChainsSeveralWrappedErrors(t *testing.T) {
+	first := Errorf("group %s has no adapter", Safe("quality_monitors"))
+	err := Errorf("%s: %w and %w", unsafeValue, first, fs.ErrPermission)
+
+	assert.Equal(t, "%s: group quality_monitors has no adapter and %w", ErrorTemplate(err))
+	assert.ErrorIs(t, err, fs.ErrPermission)
+	assert.ErrorIs(t, err, first)
+}
+
+func TestErrorTemplateWithoutTemplate(t *testing.T) {
+	assert.Empty(t, ErrorTemplate(nil))
+	assert.Empty(t, ErrorTemplate(errors.New(unsafeValue)))
+	assert.Empty(t, ErrorTemplate(fmt.Errorf("reading %s", unsafeValue)))
+	assert.Empty(t, ErrorTemplate(fs.ErrNotExist))
+}
+
+func TestNew(t *testing.T) {
+	err := New("state conversion failed")
+	assert.Equal(t, "state conversion failed", err.Error())
+	assert.Equal(t, "state conversion failed", ErrorTemplate(err))
+}
+
+func TestNewDoesNotFormat(t *testing.T) {
+	// New takes a literal message, so verbs in it are neither expanded nor lost.
+	err := New("100% of %s attempts failed")
+	assert.Equal(t, "100% of %s attempts failed", err.Error())
+	assert.Equal(t, "100% of %s attempts failed", ErrorTemplate(err))
+}
+
+func TestErrorTemplateFallsBackToRawFormat(t *testing.T) {
+	tests := []struct {
+		name   string
+		format string
+		args   []any
+	}{
+		{name: "explicit argument index", format: "%[1]s and %[1]s", args: []any{Safe("jobs")}},
+		{name: "star width", format: "%*d", args: []any{Safe(5), Safe(42)}},
+		{name: "star precision", format: "%.*f", args: []any{Safe(2), Safe(3.5)}},
+		{name: "dangling percent", format: "done: 100%", args: nil},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// The format string is a source literal, so reporting it verbatim is
+			// safe even though its verbs were not analysed.
+			assert.Equal(t, tt.format, ErrorTemplate(Errorf(tt.format, tt.args...)))
+		})
+	}
+}
+
+func TestErrorsIs(t *testing.T) {
+	sentinel := errors.New("sentinel")
+	err := Errorf("%s: %w", unsafeValue, sentinel)
+
+	assert.ErrorIs(t, err, sentinel)
+	assert.ErrorIs(t, Errorf("outer: %w", err), sentinel)
+	assert.NotErrorIs(t, Errorf("no wrapping: %s", sentinel), sentinel)
+}
+
+func TestErrorsAsType(t *testing.T) {
+	err := Errorf("%s: %w", unsafeValue, &fs.PathError{Op: "open", Path: "/tmp/x", Err: fs.ErrNotExist})
+
+	pathErr, ok := errors.AsType[*fs.PathError](err)
+	require.True(t, ok)
+	assert.Equal(t, "open", pathErr.Op)
+}
+
+// TestTemplateNeverLeaksUnsafeValues is the security property of the package:
+// nothing that was not marked Safe reaches the template.
+func TestTemplateNeverLeaksUnsafeValues(t *testing.T) {
+	const secret = "SECRET-a1b2c3"
+
+	errs := []error{
+		Errorf("%s", secret),
+		Errorf("%q: %v", secret, errors.New(secret)),
+		Errorf("%s: %w", secret, errors.New(secret)),
+		Errorf("%s: %w", secret, Errorf("inner %s: %w", secret, fs.ErrNotExist)),
+		Errorf("%v", struct{ Name string }{secret}),
+		Errorf("%s and %s", Safe("jobs"), secret),
+		Errorf("%[1]s", secret),
+		Errorf("%*s", Safe(4), secret),
+	}
+
+	for _, err := range errs {
+		require.Contains(t, err.Error(), secret, "message should carry the value")
+		assert.NotContains(t, ErrorTemplate(err), secret, "template must not carry the value")
+	}
+}
+
+func TestParseVerb(t *testing.T) {
+	tests := []struct {
+		format   string
+		wantSpec string
+		wantVerb byte
+		wantNext int
+		wantOk   bool
+	}{
+		{format: "%s", wantSpec: "%s", wantVerb: 's', wantNext: 2, wantOk: true},
+		{format: "%%", wantSpec: "%%", wantVerb: '%', wantNext: 2, wantOk: true},
+		{format: "%+v", wantSpec: "%+v", wantVerb: 'v', wantNext: 3, wantOk: true},
+		{format: "%#v", wantSpec: "%#v", wantVerb: 'v', wantNext: 3, wantOk: true},
+		{format: "%-12q", wantSpec: "%-12q", wantVerb: 'q', wantNext: 5, wantOk: true},
+		{format: "%08.3f", wantSpec: "%08.3f", wantVerb: 'f', wantNext: 6, wantOk: true},
+		{format: "% d", wantSpec: "% d", wantVerb: 'd', wantNext: 3, wantOk: true},
+		{format: "%w", wantSpec: "%w", wantVerb: 'w', wantNext: 2, wantOk: true},
+		{format: "%[1]s", wantOk: false},
+		{format: "%*d", wantOk: false},
+		{format: "%.*f", wantOk: false},
+		{format: "%", wantOk: false},
+		{format: "%-", wantOk: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.format, func(t *testing.T) {
+			spec, verb, next, ok := parseVerb(tt.format, 0)
+			assert.Equal(t, tt.wantOk, ok)
+			if !tt.wantOk {
+				return
+			}
+			assert.Equal(t, tt.wantSpec, spec)
+			assert.Equal(t, string(tt.wantVerb), string(verb))
+			assert.Equal(t, tt.wantNext, next)
+		})
+	}
+}
+
+// TestErrorTemplateIsStableAcrossCalls guards against expand mutating state.
+func TestErrorTemplateIsStableAcrossCalls(t *testing.T) {
+	err := Errorf("%s: field %q: %w", unsafeValue, Safe("id"), Errorf("inner %d", Safe(2)))
+	first := ErrorTemplate(err)
+	assert.Equal(t, first, ErrorTemplate(err))
+	assert.Equal(t, first, ErrorTemplate(err))
+}
+
+// TestUnsafeArgsAreNotRetained checks that unsafe values are dropped at
+// construction rather than kept alive inside the error for later inspection.
+func TestUnsafeArgsAreNotRetained(t *testing.T) {
+	err := Errorf("%s: %d", unsafeValue, 42)
+
+	te, ok := errors.AsType[*templateError](err)
+	require.True(t, ok)
+	require.Len(t, te.args, 2)
+	for i, arg := range te.args {
+		assert.Nil(t, arg, "argument %d should not be retained", i)
+	}
+}
+
+func TestSafeArgsAreRetainedForTemplating(t *testing.T) {
+	err := Errorf("%s: %s", Safe("jobs"), unsafeValue)
+
+	te, ok := errors.AsType[*templateError](err)
+	require.True(t, ok)
+	require.Len(t, te.args, 2)
+	assert.Equal(t, safeValue{v: "jobs"}, te.args[0])
+	assert.Nil(t, te.args[1])
+}
+
+func TestErrorTemplateDeepChain(t *testing.T) {
+	err := New("root cause")
+	for range 5 {
+		err = Errorf("%s: %w", unsafeValue, err)
+	}
+
+	assert.Equal(t, strings.Repeat("%s: ", 5)+"root cause", ErrorTemplate(err))
+}
+
+// safeStringerKey stands in for a resource key: the full value carries a
+// user-chosen name, the stand-in keeps only the group.
+type safeStringerKey string
+
+func (k safeStringerKey) SafeString() string { return "jobs.*" }
+
+// standInOnlyErr is a typed error carrying user data in its message and a
+// PII-free classification as its stand-in, modelled on libs/filer's errors.
+type standInOnlyErr struct{}
+
+func (standInOnlyErr) Error() string      { return "access denied: /Workspace/Users/a@b.com/x" }
+func (standInOnlyErr) SafeString() string { return "access denied" }
+
+func TestSafeStringer(t *testing.T) {
+	key := safeStringerKey("resources.jobs.my_job")
+
+	tests := []struct {
+		name         string
+		format       string
+		args         []any
+		wantMessage  string
+		wantTemplate string
+	}{
+		{
+			name:         "s verb",
+			format:       "cannot update %s: %w",
+			args:         []any{key, fs.ErrPermission},
+			wantMessage:  "cannot update resources.jobs.my_job: " + fs.ErrPermission.Error(),
+			wantTemplate: "cannot update jobs.*: %w",
+		},
+		{
+			name:         "q verb quotes the stand-in like the value",
+			format:       "%q not found",
+			args:         []any{key},
+			wantMessage:  `"resources.jobs.my_job" not found`,
+			wantTemplate: `"jobs.*" not found`,
+		},
+		{
+			name:         "alongside Safe and unsafe args",
+			format:       "cannot %s %s: field %q: %s",
+			args:         []any{Safe("update"), key, Safe("tasks[0].job_id"), unsafeValue},
+			wantMessage:  "cannot update resources.jobs.my_job: field \"tasks[0].job_id\": " + unsafeValue,
+			wantTemplate: `cannot update jobs.*: field "tasks[0].job_id": %s`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := Errorf(tt.format, tt.args...)
+			assert.Equal(t, tt.wantMessage, err.Error())
+			assert.Equal(t, tt.wantTemplate, ErrorTemplate(err))
+		})
+	}
+}
+
+func TestSafeStringerValueIsNotRetained(t *testing.T) {
+	err := Errorf("%s", safeStringerKey("resources.jobs.my_job"))
+
+	te, ok := errors.AsType[*templateError](err)
+	require.True(t, ok)
+	require.Len(t, te.args, 1)
+	// Only the stand-in survives; the key itself is gone.
+	assert.Equal(t, safeValue{v: "jobs.*"}, te.args[0])
+}
+
+func TestSafeStringerErrorPrefersItsTemplate(t *testing.T) {
+	// A templated error under %w contributes its template, not a stand-in.
+	inner := Errorf("inner %d", Safe(1))
+	err := Errorf("outer: %w", inner)
+	assert.Equal(t, "outer: inner 1", ErrorTemplate(err))
+}
+
+func TestSafeStringerErrorFallsBackToStandIn(t *testing.T) {
+	// An error with no template of its own contributes its stand-in, which is how
+	// a typed error reports its classification without the path it carries.
+	err := Errorf("writing state: %w", standInOnlyErr{})
+	assert.Equal(t, "writing state: access denied: /Workspace/Users/a@b.com/x", err.Error())
+	assert.Equal(t, "writing state: access denied", ErrorTemplate(err))
+	assert.NotContains(t, ErrorTemplate(err), "/Workspace")
+
+	// Without a stand-in the verb stays, as before.
+	assert.Equal(t, "writing state: %w", ErrorTemplate(Errorf("writing state: %w", fs.ErrPermission)))
+}
+
+func TestSafeStringerOutranksSafe(t *testing.T) {
+	// Wrapping a SafeStringer in Safe must not put its user-supplied part back
+	// into the template.
+	err := Errorf("%s", Safe(safeStringerKey("resources.jobs.my_job")))
+	assert.Equal(t, "resources.jobs.my_job", err.Error())
+	assert.Equal(t, "jobs.*", ErrorTemplate(err))
+}
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index 69e01a30715..d8c753df685 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -113,6 +113,28 @@ type BundleDeployExperimental struct {
 
 	// Local cache measurements in milliseconds (compute duration, potential savings, etc.)
 	LocalCacheMeasurementsMs []IntMapEntry `json:"local_cache_measurements_ms,omitempty"`
+
+	// PII-free descriptions of why a post-deploy migration to the direct engine
+	// failed. Each is a message
+	// template produced by libs/safeerr — the format string of the error, with
+	// everything the user supplied left as a verb — plus the safe fields of any
+	// API error at the end of its chain.
+	//
+	// These are the aggregatable counterpart to BundleDeployEvent.ErrorMessage,
+	// which is scrubbed heuristically and still treated as privileged. They are
+	// composed of source literals and closed enums only, so they need no
+	// scrubbing and can be grouped on directly.
+
+	// DirectMigrateErrorTemplate describes a post-deploy migration to the direct
+	// engine whose state could not be read or converted. Set alongside
+	// direct_migrate_error on opt-in deploys and direct_drymigrate_success on the
+	// dry run.
+	DirectMigrateErrorTemplate string `json:"direct_migrate_error_template,omitempty"`
+
+	// DirectMigrateCommitErrorTemplate describes a migration whose state
+	// converted cleanly but could not be committed. Set alongside
+	// direct_migrate_commit_error.
+	DirectMigrateCommitErrorTemplate string `json:"direct_migrate_commit_error_template,omitempty"`
 }
 
 // BundleResourcesMetadata mirrors the universe proto. Per-resource-type counts