bundle/direct, libs/dms: fix DMS deploy/redeploy/destroy round-trip and record git info

bundle/direct/bundle_apply.go

@@ -93,6 +93,10 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		}
 
 		if action == deployplan.Delete {
+			// Capture the ID before the delete: DMS requires resource_id on a
+			// DELETE operation, but both Destroy and DeleteState drop it from state,
+			// so GetResourceID would return empty afterwards.
+			resourceID := b.StateDB.GetResourceID(resourceKey)
 			if entry.Gone {
 				// Planning confirmed the resource is already deleted remotely; only
 				// remove it from the state, without calling the delete API.
@@ -105,7 +109,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				return false
 			}
 			// Record the delete with DMS. State is nil: the resource is gone.
-			if err := opQueue.record(ctx, resourceKey, action, "", nil, nil); err != nil {
+			if err := opQueue.record(ctx, resourceKey, action, resourceID, nil, nil); err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false
 			}

bundle/direct/dstate/dms.go

@@ -1,10 +1,12 @@
 package dstate
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/libs/log"
@@ -82,11 +84,75 @@ func fetchDeploymentResources(ctx context.Context, client bundledeployments.Bund
 			}
 		}
 
+		// DMS stores state in a protobuf Struct, whose only numeric type is
+		// double, so integers come back as "1.0". The typed resource structs
+		// (e.g. jobs.JobSettings.MaxConcurrentRuns, num_workers) unmarshal those
+		// fields as int and reject the fractional form, so restore the integral
+		// doubles to integers before the state reaches them.
+		state, err := normalizeIntegralNumbers(recorded.State)
+		if err != nil {
+			return nil, fmt.Errorf("interpreting state recorded for %s: %w", key, err)
+		}
+
 		out[key] = ResourceEntry{
 			ID:        res.ResourceId,
-			State:     recorded.State,
+			State:     state,
 			DependsOn: recorded.DependsOn,
 		}
 	}
 	return out, nil
 }
+
+// normalizeIntegralNumbers rewrites JSON numbers that have no fractional part
+// (e.g. "1.0") as integers ("1"). DMS round-trips state through a protobuf
+// Struct whose only numeric type is double, so every integer it stores comes
+// back fractional; the typed resource structs unmarshal integer fields as int
+// and reject that form. Genuinely fractional numbers are left untouched.
+//
+// A nil or empty input is returned unchanged so an unrecorded resource keeps
+// its nil state rather than becoming "null".
+func normalizeIntegralNumbers(raw json.RawMessage) (json.RawMessage, error) {
+	if len(raw) == 0 {
+		return raw, nil
+	}
+
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
+	var v any
+	if err := dec.Decode(&v); err != nil {
+		return nil, err
+	}
+
+	return json.Marshal(normalizeValue(v))
+}
+
+// normalizeValue walks a decoded JSON value (with numbers as json.Number) and
+// converts every integral number to an int64, recursing into objects and
+// arrays. Non-numeric leaves are returned as-is.
+func normalizeValue(v any) any {
+	switch t := v.(type) {
+	case map[string]any:
+		for k, val := range t {
+			t[k] = normalizeValue(val)
+		}
+		return t
+	case []any:
+		for i, val := range t {
+			t[i] = normalizeValue(val)
+		}
+		return t
+	case json.Number:
+		// An integer already parses as int64; keep it. Otherwise the value is a
+		// double, and only its integral form needs rewriting - a real fraction
+		// must survive untouched (e.g. a float-typed config field).
+		if i, err := t.Int64(); err == nil {
+			return i
+		}
+		if f, err := t.Float64(); err == nil && f == math.Trunc(f) && !math.IsInf(f, 0) {
+			return int64(f)
+		}
+		return t
+	default:
+		return v
+	}
+}

bundle/direct/dstate/dms_test.go

@@ -66,3 +66,44 @@ func TestFetchDeploymentResourcesRejectsMalformedState(t *testing.T) {
 	_, err := fetchDeploymentResources(t.Context(), f, "dep-1")
 	assert.ErrorContains(t, err, "interpreting state recorded for resources.jobs.foo")
 }
+
+func TestFetchDeploymentResourcesNormalizesIntegralDoubles(t *testing.T) {
+	// DMS serializes state through a protobuf Struct, so integers come back as
+	// doubles ("1.0"). The typed job state unmarshals those fields as int, so
+	// they must be restored to integers on the way out.
+	recorded := json.RawMessage(`{"state":{"max_concurrent_runs":1.0,"tasks":[{"new_cluster":{"num_workers":2.0}}],"timeout_seconds":0.0}}`)
+	f := &fakeResourceLister{resources: []bundledeployments.Resource{
+		{ResourceKey: "jobs.foo", ResourceId: "123", State: &recorded},
+	}}
+
+	got, err := fetchDeploymentResources(t.Context(), f, "dep-1")
+	require.NoError(t, err)
+	assert.Equal(t, json.RawMessage(`{"max_concurrent_runs":1,"tasks":[{"new_cluster":{"num_workers":2}}],"timeout_seconds":0}`), got["resources.jobs.foo"].State)
+}
+
+func TestNormalizeIntegralNumbers(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"integral doubles become ints", `{"a":1.0,"b":2.0}`, `{"a":1,"b":2}`},
+		{"fractions are preserved", `{"a":1.5,"b":0.25}`, `{"a":1.5,"b":0.25}`},
+		{"nested objects and arrays", `{"tasks":[{"n":1.0},{"n":2.5}]}`, `{"tasks":[{"n":1},{"n":2.5}]}`},
+		{"large integral double", `{"id":1000000000000000.0}`, `{"id":1000000000000000}`},
+		{"non-numbers untouched", `{"s":"x","b":true,"z":null}`, `{"b":true,"s":"x","z":null}`},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := normalizeIntegralNumbers(json.RawMessage(tc.in))
+			require.NoError(t, err)
+			assert.JSONEq(t, tc.want, string(got))
+		})
+	}
+}
+
+func TestNormalizeIntegralNumbersEmptyInputUnchanged(t *testing.T) {
+	got, err := normalizeIntegralNumbers(nil)
+	require.NoError(t, err)
+	assert.Nil(t, got)
+}

bundle/direct/oprecorder.go

@@ -3,11 +3,14 @@
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
@@ -107,6 +110,22 @@
 		ResourceKey: dmsKey,
 		Operation:   operation,
 	})
+	// The CLI discards the response, so a failure to deserialize it does not mean
+	// the operation was not recorded: DMS serves sequence_id as a JSON string
+	// ("1") per proto3 int64 encoding, but bundledeployments.Operation.SequenceId
+	// is an int64 the SDK cannot parse from a string. This surfaces intermittently
+	// (only responses that carry sequence_id trip it) as a spurious deploy
+	// failure on a call the server accepted.
+	//
+	// The SDK emits this only on the 2xx body-parse path - a status >= 400 is
+	// mapped to *apierr.APIError before the body is read - so a "failed to
+	// unmarshal response body" error means the operation was recorded. Tolerate
+	// exactly that, and nothing broader: a transport error means the request may
+	// not have reached DMS and must still fail the deploy.
+	if err != nil && !errors.As(err, new(*apierr.APIError)) && strings.Contains(err.Error(), "failed to unmarshal response body") {
+		log.Debugf(ctx, "ignoring response deserialization error from CreateOperation for %s (operation was recorded): %v", dmsKey, err)
+		return nil
+	}
 	return err
 }
 

bundle/direct/oprecorder_test.go

@@ -2,10 +2,12 @@ package direct
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -16,13 +18,14 @@ type fakeOpClient struct {
 
 	mu       sync.Mutex
 	requests []bundledeployments.CreateOperationRequest
+	err      error
 }
 
 func (f *fakeOpClient) CreateOperation(ctx context.Context, req bundledeployments.CreateOperationRequest) (*bundledeployments.Operation, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.requests = append(f.requests, req)
-	return &bundledeployments.Operation{}, nil
+	return &bundledeployments.Operation{}, f.err
 }
 
 // uploadOne records a single operation through the given uploader, mirroring what
@@ -52,6 +55,41 @@ func TestOperationRecorderStripsResourcePrefix(t *testing.T) {
 	require.NotNil(t, req.Operation.State)
 }
 
+func TestOperationRecorderToleratesResponseDeserializationError(t *testing.T) {
+	// DMS returns sequence_id as a JSON string the SDK cannot parse into its
+	// int64 field, so CreateOperation can fail to deserialize a response the
+	// server accepted. The CLI discards the response, so this must not fail the
+	// deploy.
+	f := &fakeOpClient{err: errors.New("failed to unmarshal response body: invalid character '1' after top-level value")}
+	r := NewOperationRecorder(f, "dep-1", 2)
+
+	op, err := newRecordedOperation(deployplan.Create, "job-123", map[string]string{"name": "foo"}, nil)
+	require.NoError(t, err)
+	assert.NoError(t, r.upload(t.Context(), "resources.jobs.foo", op))
+	assert.Len(t, f.requests, 1)
+}
+
+func TestOperationRecorderPropagatesAPIError(t *testing.T) {
+	// A real API error (status >= 400) must still fail the deploy.
+	f := &fakeOpClient{err: &apierr.APIError{StatusCode: 400, ErrorCode: "INVALID_PARAMETER_VALUE", Message: "bad request"}}
+	r := NewOperationRecorder(f, "dep-1", 2)
+
+	op, err := newRecordedOperation(deployplan.Create, "job-123", map[string]string{"name": "foo"}, nil)
+	require.NoError(t, err)
+	assert.Error(t, r.upload(t.Context(), "resources.jobs.foo", op))
+}
+
+func TestOperationRecorderPropagatesTransportError(t *testing.T) {
+	// A transport error means the request may never have reached DMS, so it must
+	// not be swallowed like a response-deserialization error.
+	f := &fakeOpClient{err: errors.New("dial tcp: connection refused")}
+	r := NewOperationRecorder(f, "dep-1", 2)
+
+	op, err := newRecordedOperation(deployplan.Create, "job-123", map[string]string{"name": "foo"}, nil)
+	require.NoError(t, err)
+	assert.Error(t, r.upload(t.Context(), "resources.jobs.foo", op))
+}
+
 func TestNewRecordedOperationRecordsStateAsIs(t *testing.T) {
 	state := struct {
 		Name  string `json:"name"`

bundle/phases/dms.go

@@ -6,6 +6,7 @@ import (
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/libs/dms"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
 )
 
 // newDeploymentRecorder returns a dms.Recorder for the current deployment, or
@@ -39,5 +40,23 @@ func newDeploymentRecorder(ctx context.Context, b *bundle.Bundle, eng engine.Eng
 		statePath,
 		b.Config.Bundle.Target,
 		versionType,
+		gitInfo(b),
 	), nil
 }
+
+// gitInfo maps the bundle's resolved git details onto the DMS version's
+// GitInfo. The details come from the LoadGitDetails mutator (run in the
+// initialize phase, before any recorder), or from user-set bundle.git.* values.
+// Returns nil when none are known - the bundle is not in a git repository - so
+// DMS records no git info rather than empty strings.
+func gitInfo(b *bundle.Bundle) *bundledeployments.GitInfo {
+	g := b.Config.Bundle.Git
+	if g.OriginURL == "" && g.Branch == "" && g.Commit == "" {
+		return nil
+	}
+	return &bundledeployments.GitInfo{
+		OriginUrl: g.OriginURL,
+		Branch:    g.Branch,
+		Commit:    g.Commit,
+	}
+}

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/charmbracelet/huh v1.0.0 // MIT
 	github.com/charmbracelet/lipgloss v1.1.0 // MIT
 	github.com/charmbracelet/x/ansi v0.11.7 // MIT
-	github.com/databricks/databricks-sdk-go v0.160.0 // Apache-2.0
+	github.com/databricks/databricks-sdk-go v0.166.0 // Apache-2.0
 	github.com/google/jsonschema-go v0.4.3 // MIT
 	github.com/google/uuid v1.6.0 // BSD-3-Clause
 	github.com/gorilla/websocket v1.5.3 // BSD-2-Clause

go.sum

@@ -69,8 +69,8 @@ github.com/cyphar/filepath-securejoin v0.4.1 h1:JyxxyPEaktOD+GAnqIqTf9A8tHyAG22r
 github.com/cyphar/filepath-securejoin v0.4.1/go.mod h1:Sdj7gXlvMcPZsbhwhQ33GguGLDGQL7h7bg04C/+u9jI=
 github.com/danieljoos/wincred v1.2.3 h1:v7dZC2x32Ut3nEfRH+vhoZGvN72+dQ/snVXo/vMFLdQ=
 github.com/danieljoos/wincred v1.2.3/go.mod h1:6qqX0WNrS4RzPZ1tnroDzq9kY3fu1KwE7MRLQK4X0bs=
-github.com/databricks/databricks-sdk-go v0.160.0 h1:vwgT/11y2vMw41BxcKbUUqarg45lmoEdukk9yYJg5AM=
-github.com/databricks/databricks-sdk-go v0.160.0/go.mod h1:C5LNgGe6hGuRrTwoxFmuup3XtQQEaqtq0e+K8IFDIS4=
+github.com/databricks/databricks-sdk-go v0.166.0 h1:OrVvXMr6MFf3NXZn7EIddzpDE8E/er1TrLcWeoLtOwU=
+github.com/databricks/databricks-sdk-go v0.166.0/go.mod h1:C5LNgGe6hGuRrTwoxFmuup3XtQQEaqtq0e+K8IFDIS4=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=