[VPEX][10] Add --cluster-name target flag

acceptance/localenv/cluster-name-ambiguous/output.txt

@@ -0,0 +1,7 @@
+preflight  ok  check
+resolve    error  resolving cluster name "dup": there are 2 active clusters named "dup"; use --cluster-id to disambiguate
+fetch      pending
+merge      pending
+provision  pending
+validate   pending
+For more detail, re-run with --debug, or --output json to share a structured report.

acceptance/localenv/cluster-name-ambiguous/script

@@ -0,0 +1 @@
+musterr $CLI environments setup-local --cluster-name dup --dry-run

acceptance/localenv/cluster-name-ambiguous/test.toml

@@ -0,0 +1,20 @@
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = []
+
+[Env]
+DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST"
+
+# Two active clusters share the same name, so name→ID resolution is genuinely
+# ambiguous and errors as an actionable E_RESOLVE at the resolve phase.
+# (A terminated cluster sharing the name would be filtered out server-side via
+# ListClustersFilterBy, so it does not cause a spurious collision — see
+# GetClusterByName.)
+[[Server]]
+Pattern = "GET /api/2.1/clusters/list"
+Response.Body = '''
+{
+  "clusters": [
+    {"cluster_id": "cid-123", "cluster_name": "dup", "state": "RUNNING", "spark_version": "15.4.x-scala2.12"},
+    {"cluster_id": "cid-999", "cluster_name": "dup", "state": "RUNNING", "spark_version": "14.3.x-scala2.12"}
+  ]
+}
+'''

acceptance/localenv/cluster-name-check/output.txt

@@ -0,0 +1,13 @@
+
+>>> [CLI] environments setup-local --cluster-name my-cluster --dry-run
+preflight  ok  check
+resolve    ok  source=cluster envKey=dbr/15.4.x-scala2.12
+fetch      ok  source=[DATABRICKS_URL]/dbr/15.4.x-scala2.12/pyproject.toml fromCache=false
+merge      ok
+provision  ok
+validate   ok
+Plan: [TEST_TMP_DIR]/pyproject.toml
+  changed region: requires-python
+  changed region: tool.uv.constraint-dependencies
+  changed region: databricks-connect
+Check complete. No files were modified.

acceptance/localenv/cluster-name-check/script

@@ -0,0 +1 @@
+trace $CLI environments setup-local --cluster-name my-cluster --dry-run

acceptance/localenv/cluster-name-check/test.toml

@@ -0,0 +1,34 @@
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = []
+
+[Env]
+DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST"
+
+# --cluster-name resolves the name to an ID via the Clusters API (which lists
+# clusters and matches on name), then resolves identically to --cluster-id.
+[[Server]]
+Pattern = "GET /api/2.1/clusters/list"
+Response.Body = '''
+{
+  "clusters": [
+    {"cluster_id": "cid-123", "cluster_name": "my-cluster", "spark_version": "15.4.x-scala2.12"},
+    {"cluster_id": "cid-999", "cluster_name": "other-cluster", "spark_version": "14.3.x-scala2.12"}
+  ]
+}
+'''
+
+[[Server]]
+Pattern = "GET /dbr/15.4.x-scala2.12/pyproject.toml"
+Response.Body = '''
+[project]
+requires-python = ">=3.11"
+
+[dependency-groups]
+dev = ["databricks-connect~=15.4.0"]
+
+[tool.uv]
+constraint-dependencies = ["pyarrow<19"]
+'''
+
+[[Repls]]
+Old = 'uv uv \S+(?: \([^)]+\))?'
+New = 'uv [UV_VERSION]'

acceptance/localenv/flag-conflict/output.txt

@@ -1 +1 @@
-Error: if any flags in the group [cluster-id serverless-version job-id] are set none of the others can be; [cluster-id serverless-version] were all set
+Error: if any flags in the group [cluster-id cluster-name serverless-version job-id] are set none of the others can be; [cluster-id serverless-version] were all set

acceptance/localenv/help/output.txt

@@ -11,6 +11,7 @@ Usage:
 
 Flags:
       --cluster-id string           cluster ID to use as the compute target
+      --cluster-name string         cluster name to use as the compute target (resolved to an ID via the Clusters API)
       --constraints-only            apply the Python version and constraints without adding the databricks-connect dependency
       --dry-run                     compute the plan without writing files or provisioning
   -h, --help                        help for setup-local

acceptance/localenv/json-error/output.txt

@@ -35,7 +35,7 @@
   "error": {
     "code": "E_NO_TARGET",
     "failurePhase": "resolve",
-    "message": "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --serverless-version / --job-id",
+    "message": "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id",
     "diskMutated": false
   },
   "durationMs": 0

acceptance/localenv/no-target/output.txt

@@ -1,5 +1,5 @@
 preflight  ok  uv [UV_VERSION]
-resolve    error  No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --serverless-version / --job-id
+resolve    error  No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id
 fetch      pending
 merge      pending
 provision  pending

cmd/environments/compute.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 
 	databricks "github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/service/compute"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 )
 
@@ -23,6 +24,47 @@ func (c sdkCompute) GetClusterSparkVersion(ctx context.Context, clusterID string
 	return d.SparkVersion, nil
 }
 
+// GetClusterByName resolves a cluster name to its ID and Spark version.
+//
+// It intentionally does not use the SDK's GetByClusterName, which lists clusters
+// in every state and errors on any name collision: a terminated cluster sharing
+// a name with a live one would then block resolution for a name the user
+// reasonably considers unique. Instead we list only non-terminated clusters and
+// match by name. A name that is still ambiguous among live clusters, or matches
+// none, is a genuine error surfaced to the caller as an actionable E_RESOLVE.
+func (c sdkCompute) GetClusterByName(ctx context.Context, name string) (string, string, error) {
+	clusters, err := c.w.Clusters.ListAll(ctx, compute.ListClustersRequest{
+		FilterBy: &compute.ListClustersFilterBy{
+			// TERMINATING is excluded alongside TERMINATED: a cluster on its way
+			// down is not a usable target, and keeping it would reintroduce the
+			// stale-collision problem this filter exists to avoid.
+			ClusterStates: []compute.State{
+				compute.StatePending,
+				compute.StateRunning,
+				compute.StateRestarting,
+				compute.StateResizing,
+			},
+		},
+	})
+	if err != nil {
+		return "", "", fmt.Errorf("list clusters to resolve name %q: %w", name, err)
+	}
+	var matches []compute.ClusterDetails
+	for _, cl := range clusters {
+		if cl.ClusterName == name {
+			matches = append(matches, cl)
+		}
+	}
+	switch len(matches) {
+	case 0:
+		return "", "", fmt.Errorf("no active cluster named %q", name)
+	case 1:
+		return matches[0].ClusterId, matches[0].SparkVersion, nil
+	default:
+		return "", "", fmt.Errorf("there are %d active clusters named %q; use --cluster-id to disambiguate", len(matches), name)
+	}
+}
+
 // GetJobSparkVersion inspects the job's configuration to determine compute type.
 //
 // A job is considered serverless when it has non-empty Environments (JobEnvironment

cmd/environments/sync.go

@@ -48,31 +48,34 @@ env-owned sections are refreshed, user-owned content is preserved).`,
 // addTargetFlags adds the shared target and mode flags to a command.
 func addTargetFlags(cmd *cobra.Command) {
 	cmd.Flags().String("cluster-id", "", "cluster ID to use as the compute target")
+	cmd.Flags().String("cluster-name", "", "cluster name to use as the compute target (resolved to an ID via the Clusters API)")
 	cmd.Flags().String("serverless-version", "", "serverless version to use as the compute target (e.g. v4)")
 	cmd.Flags().String("job-id", "", "job ID to use as the compute target")
 	cmd.Flags().Bool("constraints-only", false, "apply the Python version and constraints without adding the databricks-connect dependency")
 	cmd.Flags().Bool("dry-run", false, "compute the plan without writing files or provisioning")
 	cmd.Flags().String("constraint-source-url", "", "URL for the constraint source (overrides "+envConstraintSource+")")
 	// Hide constraint-source-url from casual --help output; it is a power-user escape hatch.
 	_ = cmd.Flags().MarkHidden("constraint-source-url")
-	cmd.MarkFlagsMutuallyExclusive("cluster-id", "serverless-version", "job-id")
+	cmd.MarkFlagsMutuallyExclusive("cluster-id", "cluster-name", "serverless-version", "job-id")
 }
 
 // runPipeline builds and runs the setup-local Pipeline.
 func runPipeline(cmd *cobra.Command) error {
 	ctx := cmd.Context()
 
 	cluster, _ := cmd.Flags().GetString("cluster-id")
+	clusterName, _ := cmd.Flags().GetString("cluster-name")
 	serverless, _ := cmd.Flags().GetString("serverless-version")
 	job, _ := cmd.Flags().GetString("job-id")
 	constraintsOnly, _ := cmd.Flags().GetBool("constraints-only")
 	check, _ := cmd.Flags().GetBool("dry-run")
 	constraintSource, _ := cmd.Flags().GetString("constraint-source-url")
 
 	targetFlags := libslocalenv.TargetFlags{
-		Cluster:    cluster,
-		Serverless: serverless,
-		Job:        job,
+		Cluster:     cluster,
+		ClusterName: clusterName,
+		Serverless:  serverless,
+		Job:         job,
 	}
 	// ValidateTargetFlags is kept despite MarkFlagsMutuallyExclusive above:
 	// it also validates the library path (no Cobra equivalent) and guards
@@ -100,11 +103,11 @@ func runPipeline(cmd *cobra.Command) error {
 	cacheDir = filepath.Join(cacheDir, "databricks", "localenv")
 
 	// The bundle is only a fallback: ResolveTarget consults it solely when no
-	// explicit --cluster-id/--serverless-version/--job-id flag is set. Skip the bundle load
+	// explicit --cluster-id/--cluster-name/--serverless-version/--job-id flag is set. Skip the bundle load
 	// entirely when a flag is present — it would otherwise re-run TryConfigureBundle
 	// (a second full load) and re-print any bundle load-time diagnostics for nothing.
 	var bt libslocalenv.BundleTarget
-	if cluster == "" && serverless == "" && job == "" {
+	if cluster == "" && clusterName == "" && serverless == "" && job == "" {
 		bt = bundleTarget(cmd)
 	}
 

libs/localenv/target.go

@@ -10,16 +10,21 @@ import (
 type ComputeClient interface {
 	// GetClusterSparkVersion returns the Spark version string for a cluster.
 	GetClusterSparkVersion(ctx context.Context, clusterID string) (string, error)
+	// GetClusterByName resolves a cluster name to its ID and Spark version. It
+	// errors when the name is unknown or ambiguous (more than one cluster shares
+	// the name), so the caller can surface an actionable E_RESOLVE.
+	GetClusterByName(ctx context.Context, name string) (clusterID, sparkVersion string, err error)
 	// GetJobSparkVersion returns either a Spark version (isServerless=false) or a
 	// serverless marker (isServerless=true) for a job, plus a recorded version string.
 	GetJobSparkVersion(ctx context.Context, jobID string) (sparkVersion string, isServerless bool, version string, err error)
 }
 
 // TargetFlags holds the mutually-exclusive compute target flags from the CLI.
 type TargetFlags struct {
-	Cluster    string
-	Serverless string
-	Job        string
+	Cluster     string
+	ClusterName string
+	Serverless  string
+	Job         string
 }
 
 // BundleTarget is the three-state result of reading the bundle's configured
@@ -32,15 +37,18 @@ type BundleTarget struct {
 
 // noTargetMessage is the actionable message shown when no target is selected,
 // matching spec §2.3.
-const noTargetMessage = "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --serverless-version / --job-id"
+const noTargetMessage = "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id"
 
-// ValidateTargetFlags returns an error if more than one of the three flags is set.
+// ValidateTargetFlags returns an error if more than one of the target flags is set.
 // Cobra marks them mutually exclusive too; this guards the library path.
 func ValidateTargetFlags(f TargetFlags) error {
 	var set []string
 	if f.Cluster != "" {
 		set = append(set, "--cluster-id")
 	}
+	if f.ClusterName != "" {
+		set = append(set, "--cluster-name")
+	}
 	if f.Serverless != "" {
 		set = append(set, "--serverless-version")
 	}
@@ -79,6 +87,22 @@ func ResolveTarget(ctx context.Context, f TargetFlags, c ComputeClient, bt Bundl
 		}, nil
 	}
 
+	if f.ClusterName != "" {
+		// Resolve the name to an ID via the Clusters API; from there it is
+		// identical to --cluster-id. An unknown or ambiguous name (two clusters
+		// sharing it) yields an actionable E_RESOLVE.
+		id, v, err := c.GetClusterByName(ctx, f.ClusterName)
+		if err != nil {
+			return nil, NewError(ErrResolve, err, "resolving cluster name %q", f.ClusterName)
+		}
+		return &TargetInfo{
+			Source:       "cluster",
+			ClusterID:    id,
+			SparkVersion: v,
+			EnvKey:       EnvKeyForSparkVersion(v),
+		}, nil
+	}
+
 	if f.Serverless != "" {
 		return &TargetInfo{
 			Source:            "serverless",

libs/localenv/target_test.go

@@ -12,12 +12,19 @@ import (
 type stubCompute struct {
 	clusterVersion string
 	clusterErr     error
+	byNameID       string
+	byNameVersion  string
+	byNameErr      error
 }
 
 func (s stubCompute) GetClusterSparkVersion(_ context.Context, _ string) (string, error) {
 	return s.clusterVersion, s.clusterErr
 }
 
+func (s stubCompute) GetClusterByName(_ context.Context, _ string) (string, string, error) {
+	return s.byNameID, s.byNameVersion, s.byNameErr
+}
+
 func (s stubCompute) GetJobSparkVersion(_ context.Context, _ string) (string, bool, string, error) {
 	return "", false, "", nil
 }
@@ -48,6 +55,36 @@ func TestResolveClusterFlagError(t *testing.T) {
 	assert.Equal(t, ErrResolve, pe.Code)
 }
 
+func TestResolveClusterNameFlag(t *testing.T) {
+	// --cluster-name resolves to an ID via the Clusters API, then behaves like
+	// --cluster-id: source=cluster, the resolved ID is reported, and the env key
+	// derives from the resolved cluster's Spark version.
+	c := stubCompute{byNameID: "cid-123", byNameVersion: "15.4.x-scala2.12"}
+	ti, err := ResolveTarget(t.Context(), TargetFlags{ClusterName: "my-cluster"}, c, BundleTarget{})
+	require.NoError(t, err)
+	assert.Equal(t, "cluster", ti.Source)
+	assert.Equal(t, "cid-123", ti.ClusterID)
+	assert.Equal(t, "15.4.x-scala2.12", ti.SparkVersion)
+	assert.Equal(t, "dbr/15.4.x-scala2.12", ti.EnvKey)
+}
+
+func TestResolveClusterNameFlagError(t *testing.T) {
+	// An unknown or ambiguous name surfaces as E_RESOLVE.
+	c := stubCompute{byNameErr: errors.New("there are 2 instances of ClusterDetails named 'dup'")}
+	_, err := ResolveTarget(t.Context(), TargetFlags{ClusterName: "dup"}, c, BundleTarget{})
+	var pe *PipelineError
+	require.ErrorAs(t, err, &pe)
+	assert.Equal(t, ErrResolve, pe.Code)
+}
+
+func TestResolveClusterIdAndNameMutuallyExclusive(t *testing.T) {
+	// The library path rejects setting both --cluster-id and --cluster-name.
+	_, err := ResolveTarget(t.Context(), TargetFlags{Cluster: "abc", ClusterName: "xyz"}, stubCompute{}, BundleTarget{})
+	var pe *PipelineError
+	require.ErrorAs(t, err, &pe)
+	assert.Equal(t, ErrResolve, pe.Code)
+}
+
 func TestResolveBundleNothingSelected(t *testing.T) {
 	_, err := ResolveTarget(t.Context(), TargetFlags{}, stubCompute{}, BundleTarget{Selected: false})
 	var pe *PipelineError
@@ -75,6 +112,10 @@ func (jobStubCompute) GetClusterSparkVersion(_ context.Context, _ string) (strin
 	return "", nil
 }
 
+func (jobStubCompute) GetClusterByName(_ context.Context, _ string) (string, string, error) {
+	return "", "", nil
+}
+
 func (s jobStubCompute) GetJobSparkVersion(_ context.Context, _ string) (string, bool, string, error) {
 	return s.sparkVersion, s.isServerless, s.version, nil
 }