From b87d0fdf8bafc34326aaf58fc9265bc850d8ec2c Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 12:08:26 +0000
Subject: [PATCH 01/12] Persist the DMS version in the resource state serial

---
 acceptance/bin/print_state.py                 |  4 +-
 .../bundle/dms/declined-deploy/output.txt     |  1 +
 acceptance/bundle/dms/depends-on/output.txt   |  1 +
 acceptance/bundle/dms/empty-grants/output.txt |  1 +
 .../bundle/dms/failed-delete/output.txt       |  2 +
 .../bundle/dms/failed-recreate/output.txt     |  2 +
 .../dms/failed-update-permissions/output.txt  |  2 +
 .../bundle/dms/multiple-resources/output.txt  |  2 +
 acceptance/bundle/dms/no-drift/output.txt     |  2 +
 .../operation-upload-fails-once/output.txt    |  2 +
 acceptance/bundle/dms/provenance/output.txt   |  1 +
 .../bundle/dms/record-failure/output.txt      |  1 +
 acceptance/bundle/dms/stale-plan/output.txt   |  3 +-
 .../bundle/dms/successful-recreate/output.txt |  2 +
 bundle/direct/dstate/state.go                 | 28 +++---
 bundle/direct/dstate/state_test.go            | 93 +++++++++++++++++++
 16 files changed, 129 insertions(+), 18 deletions(-)

diff --git a/acceptance/bin/print_state.py b/acceptance/bin/print_state.py
index d40adbca178..580f3966838 100755
--- a/acceptance/bin/print_state.py
+++ b/acceptance/bin/print_state.py
@@ -184,8 +184,8 @@ def print_recorded_state(filename, target):
     # printed state matches a non-recording run without every caller piping through nostamp.
     data["state"] = scrub(get_recorded_state(target))
 
-    # The service owns the version and the file persists no serial, so take it from the deployment.
-    # Rebuilt in header order, since the file has no serial key to overwrite in place.
+    # The service owns the version; the file's serial is only a cache.
+    # Rebuild in header order with the authoritative serial from the deployment.
     last_version_id = get_last_version_id(target)
     serial = int(last_version_id) if last_version_id else 0
 
diff --git a/acceptance/bundle/dms/declined-deploy/output.txt b/acceptance/bundle/dms/declined-deploy/output.txt
index e1f52e355ff..3ef52219df5 100644
--- a/acceptance/bundle/dms/declined-deploy/output.txt
+++ b/acceptance/bundle/dms/declined-deploy/output.txt
@@ -97,6 +97,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/depends-on/output.txt b/acceptance/bundle/dms/depends-on/output.txt
index d25afa59255..c1b9beb1dec 100644
--- a/acceptance/bundle/dms/depends-on/output.txt
+++ b/acceptance/bundle/dms/depends-on/output.txt
@@ -89,6 +89,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/empty-grants/output.txt b/acceptance/bundle/dms/empty-grants/output.txt
index 4bf5a7efa43..0a6cbabcbce 100644
--- a/acceptance/bundle/dms/empty-grants/output.txt
+++ b/acceptance/bundle/dms/empty-grants/output.txt
@@ -161,6 +161,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt
index fd8d9a72e61..8203cdf34f2 100644
--- a/acceptance/bundle/dms/failed-delete/output.txt
+++ b/acceptance/bundle/dms/failed-delete/output.txt
@@ -90,6 +90,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -181,6 +182,7 @@ API message: Fault injected by test.
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/failed-recreate/output.txt b/acceptance/bundle/dms/failed-recreate/output.txt
index ddbd5707b6c..860460c3375 100644
--- a/acceptance/bundle/dms/failed-recreate/output.txt
+++ b/acceptance/bundle/dms/failed-recreate/output.txt
@@ -90,6 +90,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -176,6 +177,7 @@ Files: 2 uploaded, 0 deleted
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/failed-update-permissions/output.txt b/acceptance/bundle/dms/failed-update-permissions/output.txt
index c73181e7247..47ef6cd6eac 100644
--- a/acceptance/bundle/dms/failed-update-permissions/output.txt
+++ b/acceptance/bundle/dms/failed-update-permissions/output.txt
@@ -43,6 +43,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -97,6 +98,7 @@ Files: 1 uploaded, 0 deleted
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index f090abba0fe..10c8a496c2b 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -68,6 +68,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -133,6 +134,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/no-drift/output.txt b/acceptance/bundle/dms/no-drift/output.txt
index c0b7a35688e..c2fe06a747e 100644
--- a/acceptance/bundle/dms/no-drift/output.txt
+++ b/acceptance/bundle/dms/no-drift/output.txt
@@ -48,6 +48,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -197,6 +198,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index 4d417d6adb7..dc976dadead 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -37,6 +37,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -79,6 +80,7 @@ Files: 2 uploaded, 0 deleted
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/provenance/output.txt b/acceptance/bundle/dms/provenance/output.txt
index 7719a98aed7..4decd10218b 100644
--- a/acceptance/bundle/dms/provenance/output.txt
+++ b/acceptance/bundle/dms/provenance/output.txt
@@ -98,6 +98,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/record-failure/output.txt b/acceptance/bundle/dms/record-failure/output.txt
index a35bb951a32..9f61f4a9274 100644
--- a/acceptance/bundle/dms/record-failure/output.txt
+++ b/acceptance/bundle/dms/record-failure/output.txt
@@ -90,6 +90,7 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
diff --git a/acceptance/bundle/dms/stale-plan/output.txt b/acceptance/bundle/dms/stale-plan/output.txt
index 71c39e8e24b..e9fbc70f832 100644
--- a/acceptance/bundle/dms/stale-plan/output.txt
+++ b/acceptance/bundle/dms/stale-plan/output.txt
@@ -33,6 +33,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -73,7 +74,7 @@ Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
 
 >>> jq {serial} .databricks/bundle/default/resources.json
 {
-  "serial": null
+  "serial": 2
 }
 
 === Replaying the older plan is rejected: same deployment, outdated version
diff --git a/acceptance/bundle/dms/successful-recreate/output.txt b/acceptance/bundle/dms/successful-recreate/output.txt
index f7c42c00f00..f2ecc3f8cbc 100644
--- a/acceptance/bundle/dms/successful-recreate/output.txt
+++ b/acceptance/bundle/dms/successful-recreate/output.txt
@@ -90,6 +90,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 1,
  "features": {
   "deployment_history": {}
  },
@@ -192,6 +193,7 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
  "state_version": 3,
  "cli_version": "[CLI_VERSION]",
  "lineage": "[UUID]",
+ "serial": 2,
  "features": {
   "deployment_history": {}
  },
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index a0be3d80629..e0ea3f09c53 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -111,8 +111,8 @@ type DeploymentState struct {
 	// the deployment, which Open cannot know about.
 	DeploymentID string
 
-	// VersionID is the DMS counterpart of Data.Serial - same meaning, but owned by the service
-	// rather than the file, so it lives outside Data and is never written to resources.json.
+	// VersionID is the service-owned counter for recorded deployments. It is cached in
+	// resources.json as Serial, but Open always takes the authoritative value from the service.
 	VersionID int
 }
 
@@ -139,8 +139,8 @@ type Header struct {
 
 	Lineage string `json:"lineage"`
 
-	// Serial counts state writes. Omitted for recorded deployments, where the service's version is
-	// the source of truth and Open sets this from it.
+	// Serial counts state writes. For recorded deployments it caches the current DMS version;
+	// the service remains the source of truth.
 	Serial int `json:"serial,omitempty"`
 
 	// Features maps each feature flag this state depends on to a (currently empty)
@@ -401,9 +401,9 @@ func (db *DeploymentState) StateFeatures() map[string]struct{} {
 	return maps.Clone(db.Data.Features)
 }
 
-// GetSerial reports the counter a saved plan is validated against: the recorded version for a
-// deployment that records history, since its state file persists no serial of its own, and the
-// state serial otherwise. The two mean the same thing and advance together.
+// GetSerial reports the counter a saved plan is validated against: the service's version for a
+// deployment that records history, and the state serial otherwise. A recorded state's cached
+// serial is not used for plan validation.
 func (db *DeploymentState) GetSerial() int {
 	db.mu.Lock()
 	defer db.mu.Unlock()
@@ -465,9 +465,8 @@ type (
 // Open reads the deployment state from disk, recovering the WAL when withRecovery is set.
 // When withDeploymentHistory is set it builds a DMS client from wsClient and reads resources from
 // the service instead, with dmsDeploymentID the id the service holds (empty before the first
-// recorded deploy); lineage and serial still come from the file, since that is what the write path
-// increments. Open only reads through the client - InitializeOperationBuffer installs the write path once a
-// version exists.
+// recorded deploy). Lineage comes from the file; the version comes from dmsDeployment. Open only
+// reads through the client - InitializeOperationBuffer installs the write path once a version exists.
 func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, withDeploymentHistory WithDeploymentHistory, dmsDeployment OpenDmsArgs) error {
 	db.mu.Lock()
 	defer db.mu.Unlock()
@@ -603,8 +602,8 @@ Run "databricks bundle destroy" first, then deploy again with deployment history
 			}
 		}
 
-		// The service owns the version number, so it is read from the deployment rather than from
-		// the state file, which persists none. No deployment yet means no versions, hence zero.
+		// The service owns the version number; the state file only caches it.
+		// No deployment yet means no versions, hence zero.
 		db.VersionID = dmsDeployment.LastVersionID
 	}
 
@@ -992,9 +991,8 @@ func (db *DeploymentState) dataForFile() Database {
 			header.Features = make(map[string]struct{})
 		}
 		header.Features[FeatureDeploymentHistory] = struct{}{}
-		// Under DMS the serial is managed by the deployment's last_version_id, so it is not
-		// persisted here; Open sets it from the service.
-		header.Serial = 0
+		// Cache the current version without making the file authoritative on the next Open.
+		header.Serial = db.VersionID
 
 		return Database{
 			Header: header,
diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index c8d78d150d6..490b290c630 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -7,6 +7,10 @@ import (
 	"testing"
 
 	"github.com/databricks/cli/internal/build"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/databricks/cli/libs/testserver"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -45,6 +49,95 @@ func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) {
 	assert.ErrorIs(t, err, os.ErrNotExist)
 }
 
+func TestDataForFileSerial(t *testing.T) {
+	tests := []struct {
+		name       string
+		recording  bool
+		versionID  int
+		wantSerial int
+	}{
+		{name: "direct state keeps its serial", versionID: 7, wantSerial: 3},
+		{name: "recorded state caches the current version", recording: true, versionID: 7, wantSerial: 7},
+		{name: "no recorded version omits the serial", recording: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db := DeploymentState{
+				Data:           NewDatabase("test-lineage", 3),
+				recordsHistory: tt.recording,
+				VersionID:      tt.versionID,
+			}
+			db.Data.State["resources.jobs.foo"] = ResourceEntry{ID: "123", State: json.RawMessage(`{"name":"foo"}`)}
+
+			snapshot := db.dataForFile()
+			assert.Equal(t, tt.wantSerial, snapshot.Serial)
+			assert.Equal(t, 3, db.Data.Serial)
+			assert.Len(t, db.Data.State, 1)
+			if tt.recording {
+				assert.Empty(t, snapshot.State)
+				assert.Contains(t, snapshot.Features, FeatureDeploymentHistory)
+			} else {
+				assert.Equal(t, db.Data.State, snapshot.State)
+			}
+			data, err := json.Marshal(snapshot)
+			require.NoError(t, err)
+			if tt.wantSerial == 0 {
+				assert.NotContains(t, string(data), `"serial"`)
+			}
+		})
+	}
+}
+
+func TestOpenUsesDMSVersionForSerial(t *testing.T) {
+	server := testserver.New(t)
+	server.Handle("GET", "/api/2.0/bundle/deployments/123/resources", func(req testserver.Request) any {
+		return bundledeployments.ListResourcesResponse{}
+	})
+	testserver.AddDefaultHandlers(server)
+	client, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "testtoken"})
+	require.NoError(t, err)
+
+	tests := []struct {
+		name          string
+		omitSerial    bool
+		cachedSerial  int
+		deploymentID  string
+		lastVersionID int
+	}{
+		{name: "legacy missing serial", omitSerial: true, deploymentID: "123", lastVersionID: 7},
+		{name: "legacy zero serial", deploymentID: "123", lastVersionID: 7},
+		{name: "cached serial behind service", cachedSerial: 2, deploymentID: "123", lastVersionID: 7},
+		{name: "cached serial ahead of service", cachedSerial: 99, deploymentID: "123", lastVersionID: 7},
+		{name: "no deployment has version zero", cachedSerial: 99},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), "resources.json")
+			seed := map[string]any{
+				"state_version": currentStateVersion,
+				"lineage":       "test-lineage",
+				"features":      map[string]struct{}{FeatureDeploymentHistory: {}},
+				"state":         map[string]ResourceEntry{},
+			}
+			if !tt.omitSerial {
+				seed["serial"] = tt.cachedSerial
+			}
+			data, err := json.Marshal(seed)
+			require.NoError(t, err)
+			require.NoError(t, os.WriteFile(path, data, 0o600))
+
+			ctx := cmdctx.SetWorkspaceClient(t.Context(), client)
+			var db DeploymentState
+			require.NoError(t, db.Open(ctx, path, WithRecovery(false), WithWrite(false), WithDeploymentHistory(true), OpenDmsArgs{
+				DeploymentID: tt.deploymentID, LastVersionID: tt.lastVersionID,
+			}))
+			assert.Equal(t, tt.lastVersionID, db.GetSerial())
+			assert.Equal(t, tt.cachedSerial, db.Data.Serial)
+			mustFinalize(t, &db)
+		})
+	}
+}
+
 func TestExportStateFromDataJobRunJobID(t *testing.T) {
 	data := Database{
 		State: map[string]ResourceEntry{

From beeca5749864379d3daf9175cb6c83c3d00e89e0 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 12:47:57 +0000
Subject: [PATCH 02/12] Clarify why DMS serial is duplicated in resource state

---
 bundle/direct/dstate/state.go | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index e0ea3f09c53..3b30a7a2362 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -111,8 +111,9 @@ type DeploymentState struct {
 	// the deployment, which Open cannot know about.
 	DeploymentID string
 
-	// VersionID is the service-owned counter for recorded deployments. It is cached in
-	// resources.json as Serial, but Open always takes the authoritative value from the service.
+	// VersionID is the service-owned counter for recorded deployments. It is also stored in
+	// resources.json as Serial so outdated local copies of state do not take precedence.
+	// Open always takes the authoritative value from the service.
 	VersionID int
 }
 
@@ -139,8 +140,8 @@ type Header struct {
 
 	Lineage string `json:"lineage"`
 
-	// Serial counts state writes. For recorded deployments it caches the current DMS version;
-	// the service remains the source of truth.
+	// Serial counts state writes. For recorded deployments it duplicates the current DMS version
+	// so outdated local copies of state do not take precedence; the service remains the source of truth.
 	Serial int `json:"serial,omitempty"`
 
 	// Features maps each feature flag this state depends on to a (currently empty)

From ec206eb08bd120cd6e6c9e2c1027a027beb01b15 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 12:52:30 +0000
Subject: [PATCH 03/12] Describe persisted DMS serial as a copy

---
 acceptance/bin/print_state.py |  2 +-
 bundle/direct/dstate/state.go | 21 +++++++++++----------
 2 files changed, 12 insertions(+), 11 deletions(-)

diff --git a/acceptance/bin/print_state.py b/acceptance/bin/print_state.py
index 580f3966838..6ad674342c3 100755
--- a/acceptance/bin/print_state.py
+++ b/acceptance/bin/print_state.py
@@ -184,7 +184,7 @@ def print_recorded_state(filename, target):
     # printed state matches a non-recording run without every caller piping through nostamp.
     data["state"] = scrub(get_recorded_state(target))
 
-    # The service owns the version; the file's serial is only a cache.
+    # The file's serial is a copy of the DMS version to ensure outdated resource state is not read.
     # Rebuild in header order with the authoritative serial from the deployment.
     last_version_id = get_last_version_id(target)
     serial = int(last_version_id) if last_version_id else 0
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 3b30a7a2362..1f9688bf453 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -111,8 +111,8 @@ type DeploymentState struct {
 	// the deployment, which Open cannot know about.
 	DeploymentID string
 
-	// VersionID is the service-owned counter for recorded deployments. It is also stored in
-	// resources.json as Serial so outdated local copies of state do not take precedence.
+	// VersionID is the service-owned counter for recorded deployments. A copy is stored in
+	// resources.json as Serial to ensure outdated resource state is not read.
 	// Open always takes the authoritative value from the service.
 	VersionID int
 }
@@ -140,8 +140,8 @@ type Header struct {
 
 	Lineage string `json:"lineage"`
 
-	// Serial counts state writes. For recorded deployments it duplicates the current DMS version
-	// so outdated local copies of state do not take precedence; the service remains the source of truth.
+	// Serial counts state writes. For recorded deployments it is a copy of the current DMS version
+	// to ensure outdated resource state is not read; the service remains the source of truth.
 	Serial int `json:"serial,omitempty"`
 
 	// Features maps each feature flag this state depends on to a (currently empty)
@@ -403,8 +403,8 @@ func (db *DeploymentState) StateFeatures() map[string]struct{} {
 }
 
 // GetSerial reports the counter a saved plan is validated against: the service's version for a
-// deployment that records history, and the state serial otherwise. A recorded state's cached
-// serial is not used for plan validation.
+// deployment that records history, and the state serial otherwise. When the deployment uses the
+// deployment metadata service, the serial from resources.json is not used for plan validation.
 func (db *DeploymentState) GetSerial() int {
 	db.mu.Lock()
 	defer db.mu.Unlock()
@@ -466,8 +466,8 @@ type (
 // Open reads the deployment state from disk, recovering the WAL when withRecovery is set.
 // When withDeploymentHistory is set it builds a DMS client from wsClient and reads resources from
 // the service instead, with dmsDeploymentID the id the service holds (empty before the first
-// recorded deploy). Lineage comes from the file; the version comes from dmsDeployment. Open only
-// reads through the client - InitializeOperationBuffer installs the write path once a version exists.
+// recorded deploy). Lineage comes from the file. Open only reads through the client -
+// InitializeOperationBuffer installs the write path once a version exists.
 func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, withDeploymentHistory WithDeploymentHistory, dmsDeployment OpenDmsArgs) error {
 	db.mu.Lock()
 	defer db.mu.Unlock()
@@ -603,7 +603,7 @@ Run "databricks bundle destroy" first, then deploy again with deployment history
 			}
 		}
 
-		// The service owns the version number; the state file only caches it.
+		// The service owns the version number; the state file stores a copy.
 		// No deployment yet means no versions, hence zero.
 		db.VersionID = dmsDeployment.LastVersionID
 	}
@@ -992,7 +992,8 @@ func (db *DeploymentState) dataForFile() Database {
 			header.Features = make(map[string]struct{})
 		}
 		header.Features[FeatureDeploymentHistory] = struct{}{}
-		// Cache the current version without making the file authoritative on the next Open.
+		// Store a copy of the current version to ensure outdated resource state is not read.
+		// Open still takes the authoritative version from the service.
 		header.Serial = db.VersionID
 
 		return Database{

From 9c50d7abe2bbc7909819381ab4b1247193c0e4f9 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 13:09:03 +0000
Subject: [PATCH 04/12] Test newer remote DMS state wins with history setting
 disabled

---
 .../dms/stale-local-state/databricks.yml.tmpl | 10 ++++++
 .../dms/stale-local-state/out.test.toml       |  2 ++
 .../bundle/dms/stale-local-state/output.txt   | 31 +++++++++++++++++
 .../bundle/dms/stale-local-state/script       | 33 +++++++++++++++++++
 .../bundle/dms/stale-local-state/test.toml    |  1 +
 5 files changed, 77 insertions(+)
 create mode 100644 acceptance/bundle/dms/stale-local-state/databricks.yml.tmpl
 create mode 100644 acceptance/bundle/dms/stale-local-state/out.test.toml
 create mode 100644 acceptance/bundle/dms/stale-local-state/output.txt
 create mode 100644 acceptance/bundle/dms/stale-local-state/script
 create mode 100644 acceptance/bundle/dms/stale-local-state/test.toml

diff --git a/acceptance/bundle/dms/stale-local-state/databricks.yml.tmpl b/acceptance/bundle/dms/stale-local-state/databricks.yml.tmpl
new file mode 100644
index 00000000000..a9b1eada1f9
--- /dev/null
+++ b/acceptance/bundle/dms/stale-local-state/databricks.yml.tmpl
@@ -0,0 +1,10 @@
+bundle:
+  name: dms-stale-local-state-$UNIQUE_NAME
+
+experimental:
+  deployment_history: true
+
+resources:
+  jobs:
+    foo:
+      name: foo
diff --git a/acceptance/bundle/dms/stale-local-state/out.test.toml b/acceptance/bundle/dms/stale-local-state/out.test.toml
new file mode 100644
index 00000000000..f68b4a89c36
--- /dev/null
+++ b/acceptance/bundle/dms/stale-local-state/out.test.toml
@@ -0,0 +1,2 @@
+Cloud = true
+EnvMatrix.DMS = [""]
diff --git a/acceptance/bundle/dms/stale-local-state/output.txt b/acceptance/bundle/dms/stale-local-state/output.txt
new file mode 100644
index 00000000000..32b359dbf87
--- /dev/null
+++ b/acceptance/bundle/dms/stale-local-state/output.txt
@@ -0,0 +1,31 @@
+
+=== Setup (LOG.setup): reconstruct full version 1 without a history marker, then deploy version 2
+>>> jq -c {serial, features} .databricks/resources.v1.json
+{"serial":1,"features":null}
+
+>>> jq -c {serial, features} .databricks/resources.v2.json
+{"serial":2,"features":{"deployment_history":{}}}
+
+=== 1. Config false and env false: newer remote state replaces the local copy and reads DMS
+>>> cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
+
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false [CLI] bundle summary --force-pull -o json
+Warn: Deployment history setting (false) does not match the existing state (true). Using the existing state.
+{"job_id":"[FOO_ID]"}
+
+>>> cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+
+>>> print_requests.py --dms //resources --get --oneline
+{"method": "GET", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/resources"}
+
+=== 2. Config removed and env false: restore the same snapshot, then read remote DMS state
+>>> cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
+
+>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false [CLI] bundle summary --force-pull -o json
+Warn: Deployment history setting (false) does not match the existing state (true). Using the existing state.
+{"job_id":"[FOO_ID]"}
+
+>>> cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+
+>>> print_requests.py --dms //resources --get --oneline
+{"method": "GET", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/resources"}
diff --git a/acceptance/bundle/dms/stale-local-state/script b/acceptance/bundle/dms/stale-local-state/script
new file mode 100644
index 00000000000..ee16e686128
--- /dev/null
+++ b/acceptance/bundle/dms/stale-local-state/script
@@ -0,0 +1,33 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+title "Setup (LOG.setup): reconstruct full version 1 without a history marker, then deploy version 2"
+{
+    $CLI bundle deploy
+    DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true print_state.py > .databricks/resources.v1.json
+    DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true replace_ids.py
+    update_file.py databricks.yml 'name: foo' 'name: updated'
+    $CLI bundle deploy
+    cp .databricks/bundle/default/resources.json .databricks/resources.v2.json
+    deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-stale-local-state-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+    add_repl "$deployment_id" DEPLOYMENT_ID
+} &> LOG.setup
+trace jq -c '{serial, features}' .databricks/resources.v1.json
+trace jq -c '{serial, features}' .databricks/resources.v2.json
+rm -f "$OUT_REQUESTS"
+
+title "1. Config false and env false: newer remote state replaces the local copy and reads DMS"
+update_file.py databricks.yml 'deployment_history: true' 'deployment_history: false'
+trace cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false $CLI bundle summary --force-pull -o json | jq -c '{job_id: .resources.jobs.foo.id}'
+trace cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+trace print_requests.py --dms //resources --get --oneline
+
+title "2. Config removed and env false: restore the same snapshot, then read remote DMS state"
+update_file.py databricks.yml $'experimental:\n  deployment_history: false\n\n' ''
+trace cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
+trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=false $CLI bundle summary --force-pull -o json | jq -c '{job_id: .resources.jobs.foo.id}'
+trace cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+trace print_requests.py --dms //resources --get --oneline
+
+DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle destroy --auto-approve &> LOG.cleanup
+rm -f "$OUT_REQUESTS"
diff --git a/acceptance/bundle/dms/stale-local-state/test.toml b/acceptance/bundle/dms/stale-local-state/test.toml
new file mode 100644
index 00000000000..fa9adf06e5f
--- /dev/null
+++ b/acceptance/bundle/dms/stale-local-state/test.toml
@@ -0,0 +1 @@
+EnvMatrix.DMS = [""]

From 1b371035a801dd3d2aeb70fb8a708cff50c54ecd Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 13:13:57 +0000
Subject: [PATCH 05/12] Only assert the authoritative DMS serial after opening
 state

---
 bundle/direct/dstate/state_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index 490b290c630..3760d472199 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -132,7 +132,6 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 				DeploymentID: tt.deploymentID, LastVersionID: tt.lastVersionID,
 			}))
 			assert.Equal(t, tt.lastVersionID, db.GetSerial())
-			assert.Equal(t, tt.cachedSerial, db.Data.Serial)
 			mustFinalize(t, &db)
 		})
 	}

From 473346462e0de7ced162d5c251a1525a2c136d09 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 16:26:46 +0000
Subject: [PATCH 06/12] Show remote and DMS serials after a failed destroy

---
 acceptance/bundle/dms/failed-delete/output.txt | 9 ++++++++-
 acceptance/bundle/dms/failed-delete/script     | 6 +++++-
 2 files changed, 13 insertions(+), 2 deletions(-)

diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt
index 8203cdf34f2..e3241b3d66a 100644
--- a/acceptance/bundle/dms/failed-delete/output.txt
+++ b/acceptance/bundle/dms/failed-delete/output.txt
@@ -176,7 +176,7 @@ API message: Fault injected by test.
   }
 ]
 
-=== The state shows the resource still exists but the operation is marked failed
+=== The local state file records failed version 2
 >>> print_state.py --no-dms
 {
  "state_version": 3,
@@ -188,3 +188,10 @@ API message: Fault injected by test.
  },
  "state": {}
 }
+
+=== The remote checkpoint stays at serial 1 while DMS advances to version 2
+>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-failed-delete-[UNIQUE_NAME]/default/state/resources.json --format RAW
+{"serial":1}
+
+>>> print_state.py
+{"serial":2}
diff --git a/acceptance/bundle/dms/failed-delete/script b/acceptance/bundle/dms/failed-delete/script
index 438b2925eea..2b733207f0a 100644
--- a/acceptance/bundle/dms/failed-delete/script
+++ b/acceptance/bundle/dms/failed-delete/script
@@ -29,8 +29,12 @@ trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${d
 title "The failed operation records the error message"
 trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-operations "deployments/${deployment_id}/versions/2" -o json | jq -S 'map({resource_key, action_type, status, error_message})'
 
-title "The state shows the resource still exists but the operation is marked failed"
+title "The local state file records failed version 2"
 trace print_state.py --no-dms
 
+title "The remote checkpoint stays at serial 1 while DMS advances to version 2"
+trace MSYS_NO_PATHCONV=1 $CLI workspace export "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-failed-delete-${UNIQUE_NAME}/default/state/resources.json" --format RAW | jq -c '{serial}'
+trace print_state.py | jq -c '{serial}'
+
 # Drain any accumulated requests (plan, state reads) so the harness has nothing to diff.
 rm -f "$OUT_REQUESTS"

From f67caca6da6fc1f4dccfd0c43a7e14ffd7a2cfed Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 16:28:30 +0000
Subject: [PATCH 07/12] Use jq to read the deployment node ID in the serial
 test

---
 acceptance/bundle/dms/stale-local-state/script | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/acceptance/bundle/dms/stale-local-state/script b/acceptance/bundle/dms/stale-local-state/script
index ee16e686128..0af4dfae11e 100644
--- a/acceptance/bundle/dms/stale-local-state/script
+++ b/acceptance/bundle/dms/stale-local-state/script
@@ -8,7 +8,7 @@ title "Setup (LOG.setup): reconstruct full version 1 without a history marker, t
     update_file.py databricks.yml 'name: foo' 'name: updated'
     $CLI bundle deploy
     cp .databricks/bundle/default/resources.json .databricks/resources.v2.json
-    deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-stale-local-state-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+    deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-stale-local-state-${UNIQUE_NAME}/default/state/resources.deployment.json" -o json | jq -r '.object_id')
     add_repl "$deployment_id" DEPLOYMENT_ID
 } &> LOG.setup
 trace jq -c '{serial, features}' .databricks/resources.v1.json

From d7fed167927f36401f32565078624219046d5f42 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 14 Oct 2026 16:37:12 +0000
Subject: [PATCH 08/12] Reject resource state serials ahead of deployment
 history

---
 .../resources.v2.deployment_history.json      |  2 +-
 .../resources.v3.deployment_history.json      |  2 +-
 bundle/direct/dstate/state.go                 |  3 ++
 bundle/direct/dstate/state_test.go            | 29 ++++++++++++++-----
 4 files changed, 27 insertions(+), 9 deletions(-)

diff --git a/acceptance/bundle/state/feature_flags/accept_new/resources.v2.deployment_history.json b/acceptance/bundle/state/feature_flags/accept_new/resources.v2.deployment_history.json
index 294508f75e7..307c4cdfb55 100644
--- a/acceptance/bundle/state/feature_flags/accept_new/resources.v2.deployment_history.json
+++ b/acceptance/bundle/state/feature_flags/accept_new/resources.v2.deployment_history.json
@@ -5,6 +5,6 @@
  },
  "cli_version": "0.0.0-dev",
  "lineage": "test-lineage",
- "serial": 1,
+ "serial": 0,
  "state": {}
 }
diff --git a/acceptance/bundle/state/feature_flags/accept_new/resources.v3.deployment_history.json b/acceptance/bundle/state/feature_flags/accept_new/resources.v3.deployment_history.json
index 1e47559efd8..fd2fa4b1e35 100644
--- a/acceptance/bundle/state/feature_flags/accept_new/resources.v3.deployment_history.json
+++ b/acceptance/bundle/state/feature_flags/accept_new/resources.v3.deployment_history.json
@@ -5,6 +5,6 @@
  },
  "cli_version": "0.0.0-dev",
  "lineage": "test-lineage",
- "serial": 1,
+ "serial": 0,
  "state": {}
 }
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 1f9688bf453..5bdc0829de4 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -559,6 +559,9 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe
 	// the config can neither start recording one that was not (its resources would be created a
 	// second time) nor stop recording one that is (the service still holds it).
 	_, recorded := db.Data.Features[FeatureDeploymentHistory]
+	if recording && recorded && db.Data.Serial > dmsDeployment.LastVersionID {
+		return fmt.Errorf("resource state serial (%d) is ahead of deployment history version (%d); the state may refer to a different deployment", db.Data.Serial, dmsDeployment.LastVersionID)
+	}
 	switch {
 	case recording && !recorded && len(db.Data.State) == 0:
 		if db.Data.Features == nil {
diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index 3760d472199..0b9a1f924c7 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -100,15 +100,21 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 	tests := []struct {
 		name          string
 		omitSerial    bool
-		cachedSerial  int
+		fileSerial    int
+		unmarked      bool
 		deploymentID  string
 		lastVersionID int
+		wantErr       string
 	}{
 		{name: "legacy missing serial", omitSerial: true, deploymentID: "123", lastVersionID: 7},
 		{name: "legacy zero serial", deploymentID: "123", lastVersionID: 7},
-		{name: "cached serial behind service", cachedSerial: 2, deploymentID: "123", lastVersionID: 7},
-		{name: "cached serial ahead of service", cachedSerial: 99, deploymentID: "123", lastVersionID: 7},
-		{name: "no deployment has version zero", cachedSerial: 99},
+		{name: "serial copy behind service", fileSerial: 2, deploymentID: "123", lastVersionID: 7},
+		{name: "serial copy equal to service", fileSerial: 7, deploymentID: "123", lastVersionID: 7},
+		{name: "serial copy ahead of service", fileSerial: 99, deploymentID: "123", lastVersionID: 7, wantErr: "resource state serial (99) is ahead of deployment history version (7)"},
+		{name: "no deployment has version zero"},
+		{name: "positive serial copy without deployment", fileSerial: 99, wantErr: "resource state serial (99) is ahead of deployment history version (0)"},
+		{name: "positive serial copy without recorded version", fileSerial: 1, deploymentID: "123", wantErr: "resource state serial (1) is ahead of deployment history version (0)"},
+		{name: "unmarked serial can be ahead of service", fileSerial: 99, unmarked: true, deploymentID: "123", lastVersionID: 7},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -120,7 +126,10 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 				"state":         map[string]ResourceEntry{},
 			}
 			if !tt.omitSerial {
-				seed["serial"] = tt.cachedSerial
+				seed["serial"] = tt.fileSerial
+			}
+			if tt.unmarked {
+				delete(seed, "features")
 			}
 			data, err := json.Marshal(seed)
 			require.NoError(t, err)
@@ -128,9 +137,15 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 
 			ctx := cmdctx.SetWorkspaceClient(t.Context(), client)
 			var db DeploymentState
-			require.NoError(t, db.Open(ctx, path, WithRecovery(false), WithWrite(false), WithDeploymentHistory(true), OpenDmsArgs{
+			err = db.Open(ctx, path, WithRecovery(false), WithWrite(false), WithDeploymentHistory(true), OpenDmsArgs{
 				DeploymentID: tt.deploymentID, LastVersionID: tt.lastVersionID,
-			}))
+			})
+			if tt.wantErr != "" {
+				require.ErrorContains(t, err, tt.wantErr)
+				assert.False(t, db.IsOpen())
+				return
+			}
+			require.NoError(t, err)
 			assert.Equal(t, tt.lastVersionID, db.GetSerial())
 			mustFinalize(t, &db)
 		})

From 356b242573c34a5a991db6d442d28e670acab94b Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Thu, 15 Oct 2026 15:41:38 +0000
Subject: [PATCH 09/12] Explain recovery from inconsistent deployment history
 state

---
 bundle/direct/dstate/state.go      | 2 +-
 bundle/direct/dstate/state_test.go | 5 +++++
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 5bdc0829de4..1dccace8387 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -560,7 +560,7 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe
 	// second time) nor stop recording one that is (the service still holds it).
 	_, recorded := db.Data.Features[FeatureDeploymentHistory]
 	if recording && recorded && db.Data.Serial > dmsDeployment.LastVersionID {
-		return fmt.Errorf("resource state serial (%d) is ahead of deployment history version (%d); the state may refer to a different deployment", db.Data.Serial, dmsDeployment.LastVersionID)
+		return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)\n\nDelete the local state file %q and deploy the bundle again with deployment history enabled (experimental.deployment_history: true) to proceed from deployment history version %d", db.Data.Serial, dmsDeployment.LastVersionID, filepath.ToSlash(path), dmsDeployment.LastVersionID)
 	}
 	switch {
 	case recording && !recorded && len(db.Data.State) == 0:
diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index 0b9a1f924c7..1d4239415a6 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -2,6 +2,7 @@ package dstate
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
@@ -142,6 +143,10 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 			})
 			if tt.wantErr != "" {
 				require.ErrorContains(t, err, tt.wantErr)
+				assert.ErrorContains(t, err, "inconsistent deployment state:")
+				assert.ErrorContains(t, err, fmt.Sprintf("Delete the local state file %q", filepath.ToSlash(path)))
+				assert.ErrorContains(t, err, "deployment history enabled (experimental.deployment_history: true)")
+				assert.ErrorContains(t, err, fmt.Sprintf("to proceed from deployment history version %d", tt.lastVersionID))
 				assert.False(t, db.IsOpen())
 				return
 			}

From 4ed7390a6d781e29bd63d1145618aeaf5994f6db Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Thu, 15 Oct 2026 16:00:00 +0000
Subject: [PATCH 10/12] Cover inconsistent DMS serial rejection and recovery

---
 .../bundle/dms/stale-local-state/output.txt      | 16 ++++++++++++++++
 acceptance/bundle/dms/stale-local-state/script   | 10 ++++++++++
 2 files changed, 26 insertions(+)

diff --git a/acceptance/bundle/dms/stale-local-state/output.txt b/acceptance/bundle/dms/stale-local-state/output.txt
index 32b359dbf87..5cfa68a8559 100644
--- a/acceptance/bundle/dms/stale-local-state/output.txt
+++ b/acceptance/bundle/dms/stale-local-state/output.txt
@@ -6,6 +6,22 @@
 >>> jq -c {serial, features} .databricks/resources.v2.json
 {"serial":2,"features":{"deployment_history":{}}}
 
+=== Local recorded serial 3 is ahead of DMS version 2
+>>> update_file.py .databricks/bundle/default/resources.json "serial": 2 "serial": 3
+
+>>> musterr [CLI] bundle deploy
+Error: inconsistent deployment state: resource state serial (3) is ahead of deployment history version (2)
+
+Delete the local state file "[TEST_TMP_DIR]/.databricks/bundle/default/resources.json" and deploy the bundle again with deployment history enabled (experimental.deployment_history: true) to proceed from deployment history version 2
+
+
+=== Delete the local copy and restore authoritative state
+>>> rm .databricks/bundle/default/resources.json
+
+>>> [CLI] bundle deploy -qq
+
+>>> cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+
 === 1. Config false and env false: newer remote state replaces the local copy and reads DMS
 >>> cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
 
diff --git a/acceptance/bundle/dms/stale-local-state/script b/acceptance/bundle/dms/stale-local-state/script
index 0af4dfae11e..f0bf7cdfbe5 100644
--- a/acceptance/bundle/dms/stale-local-state/script
+++ b/acceptance/bundle/dms/stale-local-state/script
@@ -13,6 +13,16 @@ title "Setup (LOG.setup): reconstruct full version 1 without a history marker, t
 } &> LOG.setup
 trace jq -c '{serial, features}' .databricks/resources.v1.json
 trace jq -c '{serial, features}' .databricks/resources.v2.json
+
+title "Local recorded serial 3 is ahead of DMS version 2"
+trace update_file.py .databricks/bundle/default/resources.json '"serial": 2' '"serial": 3'
+trace musterr $CLI bundle deploy
+
+title "Delete the local copy and restore authoritative state"
+trace rm .databricks/bundle/default/resources.json
+trace $CLI bundle deploy -qq
+trace cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
+
 rm -f "$OUT_REQUESTS"
 
 title "1. Config false and env false: newer remote state replaces the local copy and reads DMS"

From c8103759f3bc72bdf4eff8260debd8b7b3fcf8ab Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Fri, 16 Oct 2026 09:58:24 +0000
Subject: [PATCH 11/12] Remove recovery advice from inconsistent deployment
 state error

---
 acceptance/bundle/dms/stale-local-state/output.txt | 11 +----------
 acceptance/bundle/dms/stale-local-state/script     |  7 +------
 bundle/direct/dstate/state.go                      |  2 +-
 bundle/direct/dstate/state_test.go                 |  7 +------
 4 files changed, 4 insertions(+), 23 deletions(-)

diff --git a/acceptance/bundle/dms/stale-local-state/output.txt b/acceptance/bundle/dms/stale-local-state/output.txt
index 5cfa68a8559..5c2b2225a84 100644
--- a/acceptance/bundle/dms/stale-local-state/output.txt
+++ b/acceptance/bundle/dms/stale-local-state/output.txt
@@ -6,21 +6,12 @@
 >>> jq -c {serial, features} .databricks/resources.v2.json
 {"serial":2,"features":{"deployment_history":{}}}
 
-=== Local recorded serial 3 is ahead of DMS version 2
+=== Local recorded serial 3 ahead of DMS version 2 is rejected
 >>> update_file.py .databricks/bundle/default/resources.json "serial": 2 "serial": 3
 
 >>> musterr [CLI] bundle deploy
 Error: inconsistent deployment state: resource state serial (3) is ahead of deployment history version (2)
 
-Delete the local state file "[TEST_TMP_DIR]/.databricks/bundle/default/resources.json" and deploy the bundle again with deployment history enabled (experimental.deployment_history: true) to proceed from deployment history version 2
-
-
-=== Delete the local copy and restore authoritative state
->>> rm .databricks/bundle/default/resources.json
-
->>> [CLI] bundle deploy -qq
-
->>> cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
 
 === 1. Config false and env false: newer remote state replaces the local copy and reads DMS
 >>> cp .databricks/resources.v1.json .databricks/bundle/default/resources.json
diff --git a/acceptance/bundle/dms/stale-local-state/script b/acceptance/bundle/dms/stale-local-state/script
index f0bf7cdfbe5..38d0a3725b4 100644
--- a/acceptance/bundle/dms/stale-local-state/script
+++ b/acceptance/bundle/dms/stale-local-state/script
@@ -14,15 +14,10 @@ title "Setup (LOG.setup): reconstruct full version 1 without a history marker, t
 trace jq -c '{serial, features}' .databricks/resources.v1.json
 trace jq -c '{serial, features}' .databricks/resources.v2.json
 
-title "Local recorded serial 3 is ahead of DMS version 2"
+title "Local recorded serial 3 ahead of DMS version 2 is rejected"
 trace update_file.py .databricks/bundle/default/resources.json '"serial": 2' '"serial": 3'
 trace musterr $CLI bundle deploy
 
-title "Delete the local copy and restore authoritative state"
-trace rm .databricks/bundle/default/resources.json
-trace $CLI bundle deploy -qq
-trace cmp .databricks/resources.v2.json .databricks/bundle/default/resources.json
-
 rm -f "$OUT_REQUESTS"
 
 title "1. Config false and env false: newer remote state replaces the local copy and reads DMS"
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 1dccace8387..52573cf362b 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -560,7 +560,7 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe
 	// second time) nor stop recording one that is (the service still holds it).
 	_, recorded := db.Data.Features[FeatureDeploymentHistory]
 	if recording && recorded && db.Data.Serial > dmsDeployment.LastVersionID {
-		return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)\n\nDelete the local state file %q and deploy the bundle again with deployment history enabled (experimental.deployment_history: true) to proceed from deployment history version %d", db.Data.Serial, dmsDeployment.LastVersionID, filepath.ToSlash(path), dmsDeployment.LastVersionID)
+		return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)", db.Data.Serial, dmsDeployment.LastVersionID)
 	}
 	switch {
 	case recording && !recorded && len(db.Data.State) == 0:
diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index 1d4239415a6..dea5b7e713d 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -2,7 +2,6 @@ package dstate
 
 import (
 	"encoding/json"
-	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
@@ -142,11 +141,7 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 				DeploymentID: tt.deploymentID, LastVersionID: tt.lastVersionID,
 			})
 			if tt.wantErr != "" {
-				require.ErrorContains(t, err, tt.wantErr)
-				assert.ErrorContains(t, err, "inconsistent deployment state:")
-				assert.ErrorContains(t, err, fmt.Sprintf("Delete the local state file %q", filepath.ToSlash(path)))
-				assert.ErrorContains(t, err, "deployment history enabled (experimental.deployment_history: true)")
-				assert.ErrorContains(t, err, fmt.Sprintf("to proceed from deployment history version %d", tt.lastVersionID))
+				require.EqualError(t, err, "inconsistent deployment state: "+tt.wantErr)
 				assert.False(t, db.IsOpen())
 				return
 			}

From d2383c8e86eda921235bb7c00896013fd333f3f3 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Fri, 16 Oct 2026 10:00:28 +0000
Subject: [PATCH 12/12] Format DMS serial test cases with one field per line

---
 bundle/direct/dstate/state_test.go | 60 +++++++++++++++++++++++++-----
 1 file changed, 51 insertions(+), 9 deletions(-)

diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go
index dea5b7e713d..c8a50d736a8 100644
--- a/bundle/direct/dstate/state_test.go
+++ b/bundle/direct/dstate/state_test.go
@@ -106,15 +106,57 @@ func TestOpenUsesDMSVersionForSerial(t *testing.T) {
 		lastVersionID int
 		wantErr       string
 	}{
-		{name: "legacy missing serial", omitSerial: true, deploymentID: "123", lastVersionID: 7},
-		{name: "legacy zero serial", deploymentID: "123", lastVersionID: 7},
-		{name: "serial copy behind service", fileSerial: 2, deploymentID: "123", lastVersionID: 7},
-		{name: "serial copy equal to service", fileSerial: 7, deploymentID: "123", lastVersionID: 7},
-		{name: "serial copy ahead of service", fileSerial: 99, deploymentID: "123", lastVersionID: 7, wantErr: "resource state serial (99) is ahead of deployment history version (7)"},
-		{name: "no deployment has version zero"},
-		{name: "positive serial copy without deployment", fileSerial: 99, wantErr: "resource state serial (99) is ahead of deployment history version (0)"},
-		{name: "positive serial copy without recorded version", fileSerial: 1, deploymentID: "123", wantErr: "resource state serial (1) is ahead of deployment history version (0)"},
-		{name: "unmarked serial can be ahead of service", fileSerial: 99, unmarked: true, deploymentID: "123", lastVersionID: 7},
+		{
+			name:          "legacy missing serial",
+			omitSerial:    true,
+			deploymentID:  "123",
+			lastVersionID: 7,
+		},
+		{
+			name:          "legacy zero serial",
+			deploymentID:  "123",
+			lastVersionID: 7,
+		},
+		{
+			name:          "serial copy behind service",
+			fileSerial:    2,
+			deploymentID:  "123",
+			lastVersionID: 7,
+		},
+		{
+			name:          "serial copy equal to service",
+			fileSerial:    7,
+			deploymentID:  "123",
+			lastVersionID: 7,
+		},
+		{
+			name:          "serial copy ahead of service",
+			fileSerial:    99,
+			deploymentID:  "123",
+			lastVersionID: 7,
+			wantErr:       "resource state serial (99) is ahead of deployment history version (7)",
+		},
+		{
+			name: "no deployment has version zero",
+		},
+		{
+			name:       "positive serial copy without deployment",
+			fileSerial: 99,
+			wantErr:    "resource state serial (99) is ahead of deployment history version (0)",
+		},
+		{
+			name:         "positive serial copy without recorded version",
+			fileSerial:   1,
+			deploymentID: "123",
+			wantErr:      "resource state serial (1) is ahead of deployment history version (0)",
+		},
+		{
+			name:          "unmarked serial can be ahead of service",
+			fileSerial:    99,
+			unmarked:      true,
+			deploymentID:  "123",
+			lastVersionID: 7,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {