[bundle] Read deployment state from Deployment Metadata Service (step 4)

acceptance/bundle/dms/plan-and-summary/databricks.yml

@@ -0,0 +1,7 @@
+bundle:
+  name: dms-plan-and-summary
+
+resources:
+  jobs:
+    test_job:
+      name: test-job

acceptance/bundle/dms/plan-and-summary/output.txt

@@ -0,0 +1,61 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-plan-and-summary/default/files...
+Deploying resources...
+Deployment complete!
+
+>>> [CLI] bundle plan
+create jobs.test_job
+
+Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle summary
+Name: dms-plan-and-summary
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/dms-plan-and-summary/default
+Resources:
+  Jobs:
+    test_job:
+      Name: test-job
+      URL:  (not deployed)
+
+>>> print_requests.py --get //bundle ^//workspace-files ^//import-file
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments",
+  "q": {
+    "deployment_id": "[UUID]"
+  },
+  "body": {
+    "target_name": "default"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions",
+  "q": {
+    "version_id": "1"
+  },
+  "body": {
+    "cli_version": "[DEV_VERSION]",
+    "target_name": "default",
+    "version_type": "VERSION_TYPE_DEPLOY"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/1/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+{
+  "method": "GET",
+  "path": "/api/2.0/bundle/deployments/[UUID]/resources"
+}
+{
+  "method": "GET",
+  "path": "/api/2.0/bundle/deployments/[UUID]/resources"
+}

acceptance/bundle/dms/plan-and-summary/script

@@ -0,0 +1,17 @@
+# Deploy first so managed_service.json exists with a deployment_id.
+# Step 5 will report individual operations here; for now DMS only knows the
+# deployment exists.
+trace $CLI bundle deploy
+
+# bundle plan should hit DMS via ListResources. With no operations reported
+# yet, the resource list is empty and the job is planned as a create.
+trace $CLI bundle plan
+
+# bundle summary should also read from DMS. ListResources is empty, so the
+# job renders as (not deployed).
+trace $CLI bundle summary
+
+# Verify the metadata-service calls. Two ListResources lines confirm that
+# both plan and summary went through the DMS read path.
+trace print_requests.py --get //bundle ^//workspace-files ^//import-file
+print_requests.py --get > /dev/null 2>&1 || true

acceptance/bundle/dms/plan-and-summary/test.toml

@@ -0,0 +1,10 @@
+Ignore = [".databricks"]
+
+# Verify the state-read path against the deployment metadata service:
+#  - bundle deploy populates managed_service.json with a deployment_id.
+#  - bundle plan / summary read state from DMS via ListResources rather than
+#    from the workspace resources.json file.
+#
+# Step 4 does not yet ship operation reporting (that is step 5), so the DMS
+# resources list is empty after deploy. The test still exercises every
+# state-read code path end-to-end.

acceptance/bundle/dms/release-lock-error/output.txt

@@ -2,7 +2,6 @@
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-release-lock-error/fail-complete/files...
 Deploying resources...
-Updating deployment state...
 Deployment complete!
 Warn: Failed to release deployment lock: simulated complete version failure
 

bundle/bundle.go

@@ -141,6 +141,13 @@ type Bundle struct {
 	// (direct only) deployment implementation and state
 	DeploymentBundle direct.DeploymentBundle
 
+	// DeploymentID identifies the DMS-side deployment record for this bundle.
+	// Populated from the workspace managed_service.json during state pull when
+	// DATABRICKS_BUNDLE_MANAGED_STATE is set, and by the lock package after
+	// CreateDeployment. Empty when the deployment metadata service is not in
+	// use, or when DMS is enabled but no prior deployment exists.
+	DeploymentID string
+
 	// if true, we skip approval checks for deploy, destroy resources and delete
 	// files
 	AutoApprove bool

bundle/deploy/lock/deployment_metadata_service.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/deploy"
+	"github.com/databricks/cli/bundle/statemgmt"
 	"github.com/databricks/cli/internal/build"
 	"github.com/databricks/cli/libs/filer"
 	"github.com/databricks/cli/libs/log"
@@ -26,18 +27,6 @@ import (
 // renews the DMS-side lock lease while a deployment is in progress.
 const defaultHeartbeatInterval = 30 * time.Second
 
-// managedServiceFileName is the workspace state file where the lock package
-// persists the DMS deployment_id across CLI invocations. It is intentionally
-// scoped to this package for now; once the state-from-DMS path lands the
-// file (and accompanying struct) will move to bundle/statemgmt so both the
-// lock and state managers can share it.
-const managedServiceFileName = "managed_service.json"
-
-// managedServiceJSON is the on-disk shape of managedServiceFileName.
-type managedServiceJSON struct {
-	DeploymentID string `json:"deployment_id"`
-}
-
 // metadataServiceLock implements DeploymentLock against the bundle deployment
 // metadata service (DMS). The lock is acquired by creating a new Version
 // under the deployment; a background goroutine renews the lock lease via
@@ -91,6 +80,9 @@ func (l *metadataServiceLock) Acquire(ctx context.Context) error {
 
 	l.deploymentID = deploymentID
 	l.versionID = versionID
+	// Publish the deployment ID on the bundle so downstream code (e.g.
+	// statemgmt.LoadStateFromDMS) can address the right server-side record.
+	l.b.DeploymentID = deploymentID
 	l.stopHeartbeat = startHeartbeat(ctx, l.svc, deploymentID, versionID)
 
 	log.Infof(ctx, "Acquired deployment lock: deployment=%s version=%s", deploymentID, versionID)
@@ -218,23 +210,23 @@ func resolveDeploymentID(ctx context.Context, b *bundle.Bundle) (string, bool, e
 		return "", false, fmt.Errorf("failed to create state filer: %w", err)
 	}
 
-	reader, readErr := f.Read(ctx, managedServiceFileName)
+	reader, readErr := f.Read(ctx, statemgmt.ManagedServiceFileName)
 	if readErr == nil {
 		defer reader.Close()
 		data, err := io.ReadAll(reader)
 		if err != nil {
-			return "", false, fmt.Errorf("failed to read %s: %w", managedServiceFileName, err)
+			return "", false, fmt.Errorf("failed to read %s: %w", statemgmt.ManagedServiceFileName, err)
 		}
-		var sj managedServiceJSON
+		var sj statemgmt.ManagedServiceJSON
 		if err := json.Unmarshal(data, &sj); err != nil {
-			return "", false, fmt.Errorf("failed to parse %s: %w", managedServiceFileName, err)
+			return "", false, fmt.Errorf("failed to parse %s: %w", statemgmt.ManagedServiceFileName, err)
 		}
 		if sj.DeploymentID != "" {
 			return sj.DeploymentID, false, nil
 		}
 		// File exists but has no deployment_id — treat as fresh.
 	} else if !errors.Is(readErr, fs.ErrNotExist) {
-		return "", false, fmt.Errorf("failed to read %s: %w", managedServiceFileName, readErr)
+		return "", false, fmt.Errorf("failed to read %s: %w", statemgmt.ManagedServiceFileName, readErr)
 	}
 
 	return uuid.New().String(), true, nil
@@ -245,13 +237,13 @@ func writeDeploymentID(ctx context.Context, b *bundle.Bundle, deploymentID strin
 	if err != nil {
 		return fmt.Errorf("failed to create state filer: %w", err)
 	}
-	data, err := json.Marshal(managedServiceJSON{DeploymentID: deploymentID})
+	data, err := json.Marshal(statemgmt.ManagedServiceJSON{DeploymentID: deploymentID})
 	if err != nil {
-		return fmt.Errorf("failed to marshal %s: %w", managedServiceFileName, err)
+		return fmt.Errorf("failed to marshal %s: %w", statemgmt.ManagedServiceFileName, err)
 	}
-	if err := f.Write(ctx, managedServiceFileName, bytes.NewReader(data),
+	if err := f.Write(ctx, statemgmt.ManagedServiceFileName, bytes.NewReader(data),
 		filer.CreateParentDirectories, filer.OverwriteIfExists); err != nil {
-		return fmt.Errorf("failed to write %s: %w", managedServiceFileName, err)
+		return fmt.Errorf("failed to write %s: %w", statemgmt.ManagedServiceFileName, err)
 	}
 	return nil
 }

bundle/env/deployment_metadata.go

@@ -16,12 +16,6 @@ import (
 // behavior.
 const managedStateVariable = "DATABRICKS_BUNDLE_MANAGED_STATE"
 
-// ManagedState returns the raw value of DATABRICKS_BUNDLE_MANAGED_STATE if
-// set. Callers that only need a bool should use IsManagedState.
-func ManagedState(ctx context.Context) (string, bool) {
-	return get(ctx, []string{managedStateVariable})
-}
-
 // IsManagedState reports whether the DATABRICKS_BUNDLE_MANAGED_STATE
 // environment variable is set to a truthy value.
 func IsManagedState(ctx context.Context) bool {

bundle/statemgmt/managed_service_json.go

@@ -0,0 +1,18 @@
+package statemgmt
+
+// ManagedServiceFileName is the workspace state-directory file that records the
+// deployment metadata service (DMS) deployment_id for the bundle. The file
+// pins this bundle to a server-side deployment record across CLI invocations.
+// It is created by the lock package after the first CreateDeployment succeeds
+// and is read by both the lock package (when re-acquiring the lock) and the
+// statemgmt package (when loading resource state from DMS).
+//
+// resources.json continues to be written by the deploy path so that operators
+// who turn DATABRICKS_BUNDLE_MANAGED_STATE off again still have a usable local
+// state file. The fallback path is intentional, not accidental.
+const ManagedServiceFileName = "managed_service.json"
+
+// ManagedServiceJSON is the on-disk shape of ManagedServiceFileName.
+type ManagedServiceJSON struct {
+	DeploymentID string `json:"deployment_id"`
+}

bundle/statemgmt/state_dms.go

@@ -0,0 +1,64 @@
+package statemgmt
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	sdkbundle "github.com/databricks/databricks-sdk-go/service/bundle"
+)
+
+// LoadStateFromDMS populates the in-memory DeploymentState DB from the
+// deployment metadata service for the deployment identified by
+// b.DeploymentID. The state is opened in read mode via OpenWithData; the
+// historical local resources.json file is never touched on the DMS path.
+//
+// The path passed to OpenWithData is the local resources.json path: it is
+// only used for diagnostics and as the eventual write target if the deploy
+// path later upgrades the state to write mode (step 5 territory; today no
+// callers do that under DMS).
+//
+// When b.DeploymentID is empty the function is a no-op: this is the
+// "DMS enabled but no prior deployment" case, where the state is genuinely
+// empty until the lock package creates the deployment.
+func LoadStateFromDMS(ctx context.Context, b *bundle.Bundle) error {
+	if b.DeploymentID == "" {
+		// Initialize an empty state so subsequent reads (e.g. ExportState)
+		// don't panic on an unopened DB.
+		_, localPath := b.StateFilenameDirect(ctx)
+		b.DeploymentBundle.StateDB.OpenWithData(localPath, dstate.NewDatabase("", 0))
+		return nil
+	}
+
+	w := b.WorkspaceClient(ctx)
+	resources, err := w.Bundle.ListResourcesAll(ctx, sdkbundle.ListResourcesRequest{
+		Parent: "deployments/" + b.DeploymentID,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to list resources from deployment metadata service: %w", err)
+	}
+
+	data := dstate.NewDatabase("", 0)
+	for _, r := range resources {
+		// DMS reports resource keys without the "resources." prefix (e.g.
+		// "jobs.foo"); the local state DB uses the fully-qualified form
+		// ("resources.jobs.foo") as its map key, so prepend it here.
+		stateKey := "resources." + r.ResourceKey
+
+		var stateBytes json.RawMessage
+		if r.State != nil {
+			stateBytes = *r.State
+		}
+
+		data.State[stateKey] = dstate.ResourceEntry{
+			ID:    r.ResourceId,
+			State: stateBytes,
+		}
+	}
+
+	_, localPath := b.StateFilenameDirect(ctx)
+	b.DeploymentBundle.StateDB.OpenWithData(localPath, data)
+	return nil
+}