bundle: add bundle.managed_state config field + DMS test server

acceptance/bundle/managed-state/gate-config/databricks.yml

@@ -0,0 +1,7 @@
+bundle:
+  name: test-managed-state-gate-config
+  managed_state: true
+
+targets:
+  default:
+    default: true

acceptance/bundle/managed-state/gate-config/output.txt

@@ -0,0 +1,3 @@
+
+>>> [CLI] bundle validate -o json
+true

acceptance/bundle/managed-state/gate-config/script

@@ -0,0 +1 @@
+trace $CLI bundle validate -o json | jq '.bundle.managed_state'

acceptance/bundle/managed-state/gate-config/test.toml

@@ -0,0 +1 @@
+Ignore = [".databricks"]

acceptance/bundle/managed-state/gate-env/databricks.yml

@@ -0,0 +1,6 @@
+bundle:
+  name: test-managed-state-gate-env
+
+targets:
+  default:
+    default: true

acceptance/bundle/managed-state/gate-env/output.txt

@@ -0,0 +1,9 @@
+
+>>> DATABRICKS_BUNDLE_MANAGED_STATE=true [CLI] bundle validate -o json
+null
+
+>>> DATABRICKS_BUNDLE_MANAGED_STATE=false [CLI] bundle validate -o json
+null
+
+>>> [CLI] bundle validate -o json
+null

acceptance/bundle/managed-state/gate-env/script

@@ -0,0 +1,7 @@
+# The env var does not surface in bundle config dumps; it is consumed lazily
+# at deploy/destroy/bind/unbind time via ResolveManagedStateSetting. This test
+# pins that contract by showing the env var leaves .bundle.managed_state
+# unset in the validated config, regardless of the env var value.
+trace DATABRICKS_BUNDLE_MANAGED_STATE=true $CLI bundle validate -o json | jq '.bundle.managed_state'
+trace DATABRICKS_BUNDLE_MANAGED_STATE=false $CLI bundle validate -o json | jq '.bundle.managed_state'
+trace $CLI bundle validate -o json | jq '.bundle.managed_state'

acceptance/bundle/managed-state/gate-env/test.toml

@@ -0,0 +1 @@
+Ignore = [".databricks"]

bundle/config/bundle.go

@@ -50,6 +50,15 @@ type Bundle struct {
 	// Can be overridden with the DATABRICKS_BUNDLE_ENGINE environment variable.
 	Engine engine.EngineType `json:"engine,omitempty"`
 
+	// ManagedState opts the bundle into server-side deployment state management
+	// via the Deployment Metadata Service. When true, deployment locks and
+	// resource state are held by the control plane instead of the workspace
+	// filesystem. Can be overridden with the DATABRICKS_BUNDLE_MANAGED_STATE
+	// environment variable.
+	//
+	// Experimental: this surface is subject to change without notice.
+	ManagedState bool `json:"managed_state,omitempty"`
+
 	// Deployment section specifies deployment related configuration for bundle
 	Deployment Deployment `json:"deployment,omitempty"`
 

bundle/config/managedstate/managedstate.go

@@ -0,0 +1,97 @@
+// Package managedstate exposes the configuration surface that gates the
+// bundle's use of the Deployment Metadata Service (DMS) for server-managed
+// deployment state and locking.
+//
+// The setting can be controlled via two routes, in priority order:
+//  1. The bundle.managed_state field in databricks.yml.
+//  2. The DATABRICKS_BUNDLE_MANAGED_STATE environment variable.
+//
+// When neither route opts in, the bundle falls back to the historical
+// workspace-filesystem-based state and lock implementation. Use
+// cmd/bundle/utils.ResolveManagedStateSetting to combine the two sources
+// with location-aware source attribution.
+package managedstate
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/env"
+)
+
+// EnvVar is the environment variable that opts the bundle into managed
+// (server-side) deployment state.
+const EnvVar = "DATABRICKS_BUNDLE_MANAGED_STATE"
+
+// Default is used when the user has not opted in via config or env.
+const Default = false
+
+// Setting represents a resolved managed-state setting along with where it
+// came from. Source is a human-readable string used in log lines and error
+// messages (e.g. "bundle.managed_state setting at databricks.yml:3:5" or
+// "DATABRICKS_BUNDLE_MANAGED_STATE environment variable").
+type Setting struct {
+	Enabled bool
+	Source  string
+}
+
+// FromEnv reads the DATABRICKS_BUNDLE_MANAGED_STATE environment variable.
+//
+// Accepts the standard strconv.ParseBool spellings (1/0, t/T, true/True/TRUE,
+// f/F, false/False/FALSE) and additionally accepts "yes"/"no"/"y"/"n"
+// case-insensitively. An unset or empty value returns (false, false, nil) --
+// isSet=false signals that the env var was not set, not that it parsed to
+// false. Invalid values return an error.
+func FromEnv(ctx context.Context) (value, isSet bool, err error) {
+	raw := env.Get(ctx, EnvVar)
+	if raw == "" {
+		return false, false, nil
+	}
+
+	switch strings.ToLower(raw) {
+	case "yes", "y":
+		return true, true, nil
+	case "no", "n":
+		return false, true, nil
+	}
+
+	parsed, parseErr := strconv.ParseBool(raw)
+	if parseErr != nil {
+		return false, true, fmt.Errorf("unexpected setting for %s=%q (expected a boolean value)", EnvVar, raw)
+	}
+	return parsed, true, nil
+}
+
+// Resolve combines the bundle.managed_state config field and the
+// DATABRICKS_BUNDLE_MANAGED_STATE environment variable into a single
+// Setting with source attribution.
+//
+// Priority: configEnabled (i.e. bundle.managed_state=true in databricks.yml)
+// > env var > Default. configValue is the dyn.Value for the bundle root and
+// is used only for file/line/column source attribution; pass dyn.InvalidValue
+// if a location isn't available.
+func Resolve(ctx context.Context, configEnabled bool, configValue dyn.Value) (Setting, error) {
+	if configEnabled {
+		source := "bundle.managed_state setting"
+		v := dyn.GetValue(configValue, "bundle.managed_state")
+		if locs := v.Locations(); len(locs) > 0 {
+			loc := locs[0]
+			source = fmt.Sprintf("bundle.managed_state setting at %s:%d:%d", filepath.ToSlash(loc.File), loc.Line, loc.Column)
+		}
+		return Setting{Enabled: true, Source: source}, nil
+	}
+
+	envValue, isSet, err := FromEnv(ctx)
+	if err != nil {
+		return Setting{}, err
+	}
+	if isSet {
+		return Setting{Enabled: envValue, Source: EnvVar + " environment variable"}, nil
+	}
+
+	return Setting{Enabled: Default}, nil
+}

bundle/config/managedstate/managedstate_test.go

@@ -0,0 +1,55 @@
+package managedstate
+
+import (
+	"testing"
+
+	"github.com/databricks/cli/libs/env"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFromEnvNotSet(t *testing.T) {
+	value, isSet, err := FromEnv(t.Context())
+	require.NoError(t, err)
+	assert.False(t, isSet)
+	assert.False(t, value)
+}
+
+func TestFromEnvEmpty(t *testing.T) {
+	ctx := env.Set(t.Context(), EnvVar, "")
+	value, isSet, err := FromEnv(ctx)
+	require.NoError(t, err)
+	assert.False(t, isSet)
+	assert.False(t, value)
+}
+
+func TestFromEnvTruthy(t *testing.T) {
+	for _, raw := range []string{"true", "TRUE", "True", "1", "t", "T", "yes", "YES", "Yes", "y", "Y"} {
+		t.Run(raw, func(t *testing.T) {
+			ctx := env.Set(t.Context(), EnvVar, raw)
+			value, isSet, err := FromEnv(ctx)
+			require.NoError(t, err)
+			assert.True(t, isSet)
+			assert.True(t, value)
+		})
+	}
+}
+
+func TestFromEnvFalsy(t *testing.T) {
+	for _, raw := range []string{"false", "FALSE", "False", "0", "f", "F", "no", "NO", "No", "n", "N"} {
+		t.Run(raw, func(t *testing.T) {
+			ctx := env.Set(t.Context(), EnvVar, raw)
+			value, isSet, err := FromEnv(ctx)
+			require.NoError(t, err)
+			assert.True(t, isSet)
+			assert.False(t, value)
+		})
+	}
+}
+
+func TestFromEnvInvalid(t *testing.T) {
+	ctx := env.Set(t.Context(), EnvVar, "not-a-bool")
+	_, _, err := FromEnv(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), EnvVar)
+}

bundle/internal/schema/annotations.yml

@@ -50,6 +50,9 @@ github.com/databricks/cli/bundle/config.Bundle:
       The Git version control details that are associated with your bundle.
     "markdown_description": |-
       The Git version control details that are associated with your bundle. For supported attributes see [\_](/dev-tools/bundles/settings.md#git).
+  "managed_state":
+    "description": |-
+      Whether the bundle uses server-side state management via the Deployment Metadata Service. When true, deployment locks and resource state are held by the control plane instead of the workspace filesystem. Takes priority over the `DATABRICKS_BUNDLE_MANAGED_STATE` environment variable. Experimental: this surface is subject to change without notice.
   "name":
     "description": |-
       The name of the bundle.

cmd/bundle/utils/process.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/config/managedstate"
 	"github.com/databricks/cli/bundle/config/mutator"
 	"github.com/databricks/cli/bundle/config/validate"
 	"github.com/databricks/cli/bundle/deploy/terraform"
@@ -374,6 +375,13 @@ func ResolveEngineSetting(ctx context.Context, b *bundle.Bundle) (engine.EngineS
 	return engine.EngineSetting{}, nil
 }
 
+// ResolveManagedStateSetting determines the effective managed-state setting by
+// combining bundle config and env var.
+// Priority: bundle.managed_state config > DATABRICKS_BUNDLE_MANAGED_STATE env var > Default.
+func ResolveManagedStateSetting(ctx context.Context, b *bundle.Bundle) (managedstate.Setting, error) {
+	return managedstate.Resolve(ctx, b.Config.Bundle.ManagedState, b.Config.Value())
+}
+
 func rejectDefinitions(ctx context.Context, b *bundle.Bundle) {
 	if b.Config.Definitions != nil {
 		v := dyn.GetValue(b.Config.Value(), "definitions")

cmd/bundle/utils/resolve_managed_state_test.go

@@ -0,0 +1,62 @@
+package utils
+
+import (
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/managedstate"
+	"github.com/databricks/cli/libs/env"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResolveManagedStateConfigTakesPriority(t *testing.T) {
+	ctx := env.Set(t.Context(), managedstate.EnvVar, "false")
+	b := &bundle.Bundle{Config: config.Root{Bundle: config.Bundle{ManagedState: true}}}
+	result, err := ResolveManagedStateSetting(ctx, b)
+	require.NoError(t, err)
+	assert.True(t, result.Enabled)
+	assert.Contains(t, result.Source, "bundle.managed_state")
+}
+
+func TestResolveManagedStateEnvVarUsedWhenNoConfig(t *testing.T) {
+	ctx := env.Set(t.Context(), managedstate.EnvVar, "true")
+	b := &bundle.Bundle{Config: config.Root{}}
+	result, err := ResolveManagedStateSetting(ctx, b)
+	require.NoError(t, err)
+	assert.True(t, result.Enabled)
+	assert.Contains(t, result.Source, managedstate.EnvVar)
+}
+
+func TestResolveManagedStateNothingSet(t *testing.T) {
+	b := &bundle.Bundle{Config: config.Root{}}
+	result, err := ResolveManagedStateSetting(t.Context(), b)
+	require.NoError(t, err)
+	assert.False(t, result.Enabled)
+	assert.Empty(t, result.Source)
+}
+
+func TestResolveManagedStateInvalidEnvVar(t *testing.T) {
+	ctx := env.Set(t.Context(), managedstate.EnvVar, "not-a-bool")
+	b := &bundle.Bundle{Config: config.Root{}}
+	_, err := ResolveManagedStateSetting(ctx, b)
+	require.Error(t, err)
+}
+
+func TestResolveManagedStateInvalidEnvVarIgnoredWhenConfigSet(t *testing.T) {
+	ctx := env.Set(t.Context(), managedstate.EnvVar, "not-a-bool")
+	b := &bundle.Bundle{Config: config.Root{Bundle: config.Bundle{ManagedState: true}}}
+	result, err := ResolveManagedStateSetting(ctx, b)
+	require.NoError(t, err)
+	assert.True(t, result.Enabled)
+}
+
+func TestResolveManagedStateEnvVarFalseExplicit(t *testing.T) {
+	ctx := env.Set(t.Context(), managedstate.EnvVar, "false")
+	b := &bundle.Bundle{Config: config.Root{}}
+	result, err := ResolveManagedStateSetting(ctx, b)
+	require.NoError(t, err)
+	assert.False(t, result.Enabled)
+	assert.Contains(t, result.Source, managedstate.EnvVar)
+}