Add idempotency_token to job_runs resource

acceptance/bundle/resources/job_runs/basic/output.txt

@@ -68,6 +68,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID]
   }
 }

acceptance/bundle/resources/job_runs/idempotent_recreate/databricks.yml

@@ -0,0 +1,15 @@
+bundle:
+  name: job-runs-idempotent-recreate
+
+resources:
+  jobs:
+    my_job:
+      name: my-job
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/test
+
+  job_runs:
+    my_run:
+      job_id: ${resources.jobs.my_job.id}

acceptance/bundle/resources/job_runs/idempotent_recreate/output.txt

@@ -0,0 +1,61 @@
+
+=== initial deploy triggers the run
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== drop the job run from local state but keep the job (simulates a lost run id)
+>>> [CLI] bundle plan -o json
+{
+  "depends_on": [
+    {
+      "node": "resources.jobs.my_job",
+      "label": "${resources.jobs.my_job.id}"
+    }
+  ],
+  "action": "create",
+  "new_state": {
+    "value": {
+      "job_id": [NUMID]
+    }
+  }
+}
+
+=== redeploy re-issues run-now; the token dedupes to the existing run
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== run-now was issued twice with the same token, but no duplicate run was created
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
+    "job_id": [NUMID]
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
+    "job_id": [NUMID]
+  }
+}
+run id unchanged after recreate: the idempotency token deduped to the existing run
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.job_runs.my_run
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-idempotent-recreate/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/idempotent_recreate/script

@@ -0,0 +1,33 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+STATE=.databricks/bundle/default/resources.json
+run_id() {
+    jq -r '.state["resources.job_runs.my_run"].__id__' "$STATE"
+}
+
+title "initial deploy triggers the run"
+trace $CLI bundle deploy
+before=$(run_id)
+
+title "drop the job run from local state but keep the job (simulates a lost run id)"
+# The job stays deployed, so job_id -- and therefore the derived idempotency
+# token -- is unchanged, which is what lets the retried run-now dedupe.
+jq 'del(.state["resources.job_runs.my_run"])' "$STATE" > "$STATE.new"
+mv "$STATE.new" "$STATE"
+trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"]'
+
+title "redeploy re-issues run-now; the token dedupes to the existing run"
+trace $CLI bundle deploy
+after=$(run_id)
+
+title "run-now was issued twice with the same token, but no duplicate run was created"
+trace print_requests.py //jobs/run-now
+if [ "$before" = "$after" ]; then
+    echo "run id unchanged after recreate: the idempotency token deduped to the existing run"
+else
+    echo "run id changed ($before -> $after): a duplicate run was created"
+fi

acceptance/bundle/resources/job_runs/job_parameters/output.txt

@@ -11,6 +11,7 @@ Deployment complete!
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [NUMID],
     "job_parameters": {
       "env": "prod"

acceptance/bundle/resources/job_runs/redeploy/output.txt

@@ -30,6 +30,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID],
     "job_parameters": {
       "env": "dev"
@@ -118,6 +119,7 @@ Resources:
   "method": "POST",
   "path": "/api/2.2/jobs/run-now",
   "body": {
+    "idempotency_token": "[IDEMPOTENCY_TOKEN]",
     "job_id": [MY_JOB_ID],
     "job_parameters": {
       "env": "prod"

acceptance/bundle/resources/job_runs/test.toml

@@ -0,0 +1,11 @@
+# job_runs is a direct-engine-only resource; the Terraform provider has no
+# equivalent, so restrict the matrix to direct.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+RecordRequests = true
+
+# idempotency_token is a SHA-256 of the RunNow request, and the fake assigns
+# job_id from a time-based counter, so the hash differs every run. Mask it so the
+# run-now request goldens stay stable.
+[[Repls]]
+Old = '[0-9a-f]{64}'
+New = '[IDEMPOTENCY_TOKEN]'

bundle/config/validate/validate_job_run_idempotency_token.go

@@ -0,0 +1,48 @@
+package validate
+
+import (
+	"cmp"
+	"context"
+	"slices"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+func ValidateJobRunIdempotencyToken() bundle.ReadOnlyMutator {
+	return &validateJobRunIdempotencyToken{}
+}
+
+type validateJobRunIdempotencyToken struct{ bundle.RO }
+
+func (v *validateJobRunIdempotencyToken) Name() string {
+	return "validate:validate_job_run_idempotency_token"
+}
+
+func (v *validateJobRunIdempotencyToken) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	// idempotency_token is computed automatically from the run configuration
+	// (SHA-256 of the RunNow request) so retries dedupe. A user-provided value
+	// would be overwritten, so reject it up front.
+	for name, jr := range b.Config.Resources.JobRuns {
+		if jr.IdempotencyToken == "" {
+			continue
+		}
+		path := "resources.job_runs." + name + ".idempotency_token"
+		diags = append(diags, diag.Diagnostic{
+			Severity:  diag.Error,
+			Summary:   "idempotency_token is computed automatically and must not be set in bundle configuration",
+			Paths:     []dyn.Path{dyn.MustPathFromString(path)},
+			Locations: b.Config.GetLocations(path),
+		})
+	}
+
+	// Map iteration order is randomized; sort by path for stable output.
+	slices.SortFunc(diags, func(x, y diag.Diagnostic) int {
+		return cmp.Compare(x.Paths[0].String(), y.Paths[0].String())
+	})
+
+	return diags
+}

bundle/config/validate/validate_job_run_idempotency_token_test.go

@@ -0,0 +1,57 @@
+package validate
+
+import (
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func jobRunBundle(token string) *bundle.Bundle {
+	return &bundle.Bundle{
+		Config: config.Root{
+			Resources: config.Resources{
+				JobRuns: map[string]*resources.JobRun{
+					"my_run": {RunNow: jobs.RunNow{JobId: 1, IdempotencyToken: token}},
+				},
+			},
+		},
+	}
+}
+
+func TestValidateJobRunIdempotencyTokenRejectsUserValue(t *testing.T) {
+	diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), jobRunBundle("x"))
+	require.Len(t, diags, 1)
+	assert.Equal(t, diag.Error, diags[0].Severity)
+	assert.Equal(t, "idempotency_token is computed automatically and must not be set in bundle configuration", diags[0].Summary)
+	assert.Equal(t, "resources.job_runs.my_run.idempotency_token", diags[0].Paths[0].String())
+}
+
+func TestValidateJobRunIdempotencyTokenAllowsUnset(t *testing.T) {
+	diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), jobRunBundle(""))
+	require.Empty(t, diags)
+}
+
+func TestValidateJobRunIdempotencyTokenReportsAllSorted(t *testing.T) {
+	b := &bundle.Bundle{
+		Config: config.Root{
+			Resources: config.Resources{
+				JobRuns: map[string]*resources.JobRun{
+					"b_run": {RunNow: jobs.RunNow{JobId: 1, IdempotencyToken: "x"}},
+					"a_run": {RunNow: jobs.RunNow{JobId: 2, IdempotencyToken: "y"}},
+				},
+			},
+		},
+	}
+
+	diags := ValidateJobRunIdempotencyToken().Apply(t.Context(), b)
+	require.Len(t, diags, 2)
+	// Sorted by path, so a_run comes before b_run regardless of map order.
+	assert.Equal(t, "resources.job_runs.a_run.idempotency_token", diags[0].Paths[0].String())
+	assert.Equal(t, "resources.job_runs.b_run.idempotency_token", diags[1].Paths[0].String())
+}

bundle/direct/dresources/job_run.go

@@ -2,6 +2,9 @@ package dresources
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"strconv"
 
@@ -129,9 +132,18 @@ func (*ResourceJobRun) RemapState(remote *JobRunRemote) *JobRunState {
 }
 
 func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (string, *JobRunRemote, error) {
+	// Copy the request so the derived token is sent to the API but never
+	// persisted into state (state stays token-free, plans stay clean).
+	req := config.RunNow
+	token, err := idempotencyToken(req)
+	if err != nil {
+		return "", nil, err
+	}
+	req.IdempotencyToken = token
+
 	// RunNow returns only the new run id, so we return a nil remote and let the
 	// framework read it back via DoRead.
-	wait, err := r.client.Jobs.RunNow(ctx, config.RunNow)
+	wait, err := r.client.Jobs.RunNow(ctx, req)
 	if err != nil {
 		return "", nil, err
 	}
@@ -159,3 +171,17 @@ func parseRunID(id string) (int64, error) {
 	}
 	return result, nil
 }
+
+// idempotencyToken derives a stable token from the RunNow request so that a
+// retried run-now returns the existing run instead of creating a duplicate.
+// The token is the hex SHA-256 of the request's JSON with idempotency_token
+// cleared; hex SHA-256 is exactly 64 chars, the Jobs API maximum.
+func idempotencyToken(req jobs.RunNow) (string, error) {
+	req.IdempotencyToken = ""
+	canonical, err := json.Marshal(req)
+	if err != nil {
+		return "", err
+	}
+	sum := sha256.Sum256(canonical)
+	return hex.EncodeToString(sum[:]), nil
+}

bundle/direct/dresources/job_run_test.go

@@ -0,0 +1,46 @@
+package dresources
+
+import (
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIdempotencyTokenIsStableHex(t *testing.T) {
+	run := jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}}
+
+	got, err := idempotencyToken(run)
+	require.NoError(t, err)
+
+	// hex SHA-256 is always 64 lowercase hex chars (the Jobs API maximum).
+	assert.Regexp(t, "^[0-9a-f]{64}$", got)
+
+	// Deterministic: the same config yields the same token, so a retry dedupes.
+	again, err := idempotencyToken(run)
+	require.NoError(t, err)
+	assert.Equal(t, got, again)
+}
+
+func TestIdempotencyTokenIgnoresPresetToken(t *testing.T) {
+	a, err := idempotencyToken(jobs.RunNow{JobId: 123})
+	require.NoError(t, err)
+	b, err := idempotencyToken(jobs.RunNow{JobId: 123, IdempotencyToken: "user-supplied"})
+	require.NoError(t, err)
+
+	// The token is cleared before hashing, so a preset value cannot change it.
+	assert.Equal(t, a, b)
+}
+
+func TestIdempotencyTokenChangesWithConfig(t *testing.T) {
+	dev, err := idempotencyToken(jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "dev"}})
+	require.NoError(t, err)
+	prod, err := idempotencyToken(jobs.RunNow{JobId: 123, JobParameters: map[string]string{"env": "prod"}})
+	require.NoError(t, err)
+	otherJob, err := idempotencyToken(jobs.RunNow{JobId: 456})
+	require.NoError(t, err)
+
+	assert.NotEqual(t, dev, prod)     // different params --> different token
+	assert.NotEqual(t, dev, otherJob) // different job_id --> different token
+}

bundle/phases/initialize.go

@@ -177,6 +177,9 @@ func Initialize(ctx context.Context, b *bundle.Bundle) {
 		// They are set by the CLI to track the bundle deployment and must not be set by the user.
 		validate.ValidateDeploymentFields(),
 
+		// Validate that idempotency_token is not set on job runs. It is computed automatically and must not be set by the user.
+		validate.ValidateJobRunIdempotencyToken(),
+
 		// Reads (dynamic): * (strings) (searches for ${resources.*} references)
 		// Warns (TF engine) or errors (direct engine) when a cross-resource reference
 		// points to a Terraform-only field with no DABs equivalent.