Add preliminary job_runs resource

acceptance/bundle/refschema/out.fields.txt

@@ -752,6 +752,45 @@ resources.genie_spaces.*.permissions[*].group_name	string	ALL
 resources.genie_spaces.*.permissions[*].level	iam.PermissionLevel	ALL
 resources.genie_spaces.*.permissions[*].service_principal_name	string	ALL
 resources.genie_spaces.*.permissions[*].user_name	string	ALL
+resources.job_runs.*.dbt_commands	[]string	ALL
+resources.job_runs.*.dbt_commands[*]	string	ALL
+resources.job_runs.*.id	string	INPUT
+resources.job_runs.*.idempotency_token	string	ALL
+resources.job_runs.*.jar_params	[]string	ALL
+resources.job_runs.*.jar_params[*]	string	ALL
+resources.job_runs.*.job_id	int64	ALL
+resources.job_runs.*.job_parameters	map[string]string	ALL
+resources.job_runs.*.job_parameters.*	string	ALL
+resources.job_runs.*.lifecycle	resources.Lifecycle	INPUT
+resources.job_runs.*.lifecycle.prevent_destroy	bool	INPUT
+resources.job_runs.*.modified_status	string	INPUT
+resources.job_runs.*.notebook_params	map[string]string	ALL
+resources.job_runs.*.notebook_params.*	string	ALL
+resources.job_runs.*.only	[]string	ALL
+resources.job_runs.*.only[*]	string	ALL
+resources.job_runs.*.performance_target	jobs.PerformanceTarget	ALL
+resources.job_runs.*.pipeline_params	*jobs.PipelineParams	ALL
+resources.job_runs.*.pipeline_params.full_refresh	bool	ALL
+resources.job_runs.*.pipeline_params.full_refresh_selection	[]string	ALL
+resources.job_runs.*.pipeline_params.full_refresh_selection[*]	string	ALL
+resources.job_runs.*.pipeline_params.refresh_flow_selection	[]string	ALL
+resources.job_runs.*.pipeline_params.refresh_flow_selection[*]	string	ALL
+resources.job_runs.*.pipeline_params.refresh_selection	[]string	ALL
+resources.job_runs.*.pipeline_params.refresh_selection[*]	string	ALL
+resources.job_runs.*.pipeline_params.reset_checkpoint_selection	[]string	ALL
+resources.job_runs.*.pipeline_params.reset_checkpoint_selection[*]	string	ALL
+resources.job_runs.*.python_named_params	map[string]string	ALL
+resources.job_runs.*.python_named_params.*	string	ALL
+resources.job_runs.*.python_params	[]string	ALL
+resources.job_runs.*.python_params[*]	string	ALL
+resources.job_runs.*.queue	*jobs.QueueSettings	ALL
+resources.job_runs.*.queue.enabled	bool	ALL
+resources.job_runs.*.run_id	int64	REMOTE
+resources.job_runs.*.spark_submit_params	[]string	ALL
+resources.job_runs.*.spark_submit_params[*]	string	ALL
+resources.job_runs.*.sql_params	map[string]string	ALL
+resources.job_runs.*.sql_params.*	string	ALL
+resources.job_runs.*.url	string	INPUT
 resources.jobs.*.budget_policy_id	string	ALL
 resources.jobs.*.continuous	*jobs.Continuous	ALL
 resources.jobs.*.continuous.pause_status	jobs.PauseStatus	ALL

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

@@ -0,0 +1,15 @@
+bundle:
+  name: job-runs-basic
+
+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/basic/output.txt

@@ -0,0 +1,34 @@
+
+=== deploy triggers exactly one run
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-basic/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== exactly one run-now request was made
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "job_id": [MY_JOB_ID]
+  }
+}
+
+=== the triggered run id is stored in state
+>>> read_id.py my_run
+[MY_RUN_ID]
+
+>>> read_id.py my_job
+[MY_JOB_ID]
+
+>>> [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-basic/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/basic/script

@@ -0,0 +1,15 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm out.requests.txt
+}
+trap cleanup EXIT
+
+title "deploy triggers exactly one run"
+trace $CLI bundle deploy
+
+title "exactly one run-now request was made"
+trace print_requests.py //jobs/run-now
+
+title "the triggered run id is stored in state"
+trace read_id.py my_run
+trace read_id.py my_job

acceptance/bundle/resources/job_runs/basic/test.toml

@@ -0,0 +1,4 @@
+# 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

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

@@ -0,0 +1,17 @@
+bundle:
+  name: job-runs-redeploy
+
+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}
+      job_parameters:
+        env: dev

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

@@ -0,0 +1,54 @@
+
+=== initial deploy triggers the first run
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> read_id.py my_job
+[MY_JOB_ID]
+
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "job_id": [MY_JOB_ID],
+    "job_parameters": {
+      "env": "dev"
+    }
+  }
+}
+
+=== change the run configuration and redeploy
+>>> update_file.py databricks.yml env: dev env: prod
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== the config change triggered a second, different run
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "job_id": [MY_JOB_ID],
+    "job_parameters": {
+      "env": "prod"
+    }
+  }
+}
+
+>>> [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-redeploy/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/redeploy/script

@@ -0,0 +1,17 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm out.requests.txt
+}
+trap cleanup EXIT
+
+title "initial deploy triggers the first run"
+trace $CLI bundle deploy
+trace read_id.py my_job
+trace print_requests.py //jobs/run-now
+
+title "change the run configuration and redeploy"
+trace update_file.py databricks.yml "env: dev" "env: prod"
+trace $CLI bundle deploy
+
+title "the config change triggered a second, different run"
+trace print_requests.py //jobs/run-now

acceptance/bundle/resources/job_runs/redeploy/test.toml

@@ -0,0 +1,4 @@
+# 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

bundle/config/mutator/resourcemutator/apply_bundle_permissions_test.go

@@ -31,6 +31,7 @@ var unsupportedResources = []string{
 	"postgres_catalogs",
 	"postgres_synced_tables",
 	"vector_search_indexes",
+	"job_runs",
 }
 
 func TestApplyBundlePermissions(t *testing.T) {

bundle/config/mutator/resourcemutator/apply_target_mode_test.go

@@ -92,6 +92,9 @@ func mockBundle(mode config.Mode) *bundle.Bundle {
 						},
 					},
 				},
+				JobRuns: map[string]*resources.JobRun{
+					"job_run1": {RunNow: jobs.RunNow{JobId: 1234}},
+				},
 				Pipelines: map[string]*resources.Pipeline{
 					"pipeline1": {CreatePipeline: pipelines.CreatePipeline{Name: "pipeline1", Continuous: true}},
 				},

bundle/config/mutator/resourcemutator/run_as.go

@@ -126,6 +126,17 @@ func validateRunAs(b *bundle.Bundle) diag.Diagnostics {
 		))
 	}
 
+	// Job runs execute under the triggered job's own identity; the RunNow API
+	// has no run_as field, so a differing bundle run_as cannot be honored.
+	if len(b.Config.Resources.JobRuns) > 0 {
+		diags = diags.Extend(reportRunAsNotSupported(
+			"job_runs",
+			b.Config.GetLocation("resources.job_runs"),
+			b.Config.Workspace.CurrentUser.UserName,
+			identity,
+		))
+	}
+
 	return diags
 }
 

bundle/config/mutator/resourcemutator/run_as_test.go

@@ -42,6 +42,7 @@ func allResourceTypes(t *testing.T) []string {
 		"experiments",
 		"external_locations",
 		"genie_spaces",
+		"job_runs",
 		"jobs",
 		"model_serving_endpoints",
 		"models",

bundle/config/resources.go

@@ -12,6 +12,7 @@ import (
 // Resources defines Databricks resources associated with the bundle.
 type Resources struct {
 	Jobs      map[string]*resources.Job      `json:"jobs,omitempty"`
+	JobRuns   map[string]*resources.JobRun   `json:"job_runs,omitempty"`
 	Pipelines map[string]*resources.Pipeline `json:"pipelines,omitempty"`
 
 	Models                map[string]*resources.MlflowModel          `json:"models,omitempty"`
@@ -94,6 +95,7 @@ func (r *Resources) AllResources() []ResourceGroup {
 	descriptions := SupportedResources()
 	return []ResourceGroup{
 		collectResourceMap(descriptions["jobs"], r.Jobs),
+		collectResourceMap(descriptions["job_runs"], r.JobRuns),
 		collectResourceMap(descriptions["pipelines"], r.Pipelines),
 		collectResourceMap(descriptions["models"], r.Models),
 		collectResourceMap(descriptions["experiments"], r.Experiments),
@@ -153,6 +155,7 @@ func (r *Resources) FindResourceByConfigKey(key string) (ConfigResource, error)
 func SupportedResources() map[string]resources.ResourceDescription {
 	return map[string]resources.ResourceDescription{
 		"jobs":                    (&resources.Job{}).ResourceDescription(),
+		"job_runs":                (&resources.JobRun{}).ResourceDescription(),
 		"pipelines":               (&resources.Pipeline{}).ResourceDescription(),
 		"models":                  (&resources.MlflowModel{}).ResourceDescription(),
 		"experiments":             (&resources.MlflowExperiment{}).ResourceDescription(),

bundle/config/resources/job_run.go

@@ -0,0 +1,71 @@
+package resources
+
+import (
+	"context"
+	"net/url"
+	"strconv"
+
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/marshal"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+)
+
+// JobRun is the bundle config for a triggered job run. The run is described by
+// the same fields as the Jobs RunNow API request, so we embed jobs.RunNow
+// directly instead of re-declaring them.
+type JobRun struct {
+	BaseResource
+	jobs.RunNow
+}
+
+func (r *JobRun) UnmarshalJSON(b []byte) error {
+	return marshal.Unmarshal(b, r)
+}
+
+func (r JobRun) MarshalJSON() ([]byte, error) {
+	return marshal.Marshal(r)
+}
+
+// Exists reports whether the run identified by id (a numeric run id) still
+// exists in the workspace. A run is the unit of existence here: once RunNow has
+// been called, the run is retrievable via GetRun for as long as the workspace
+// retains its history.
+func (r *JobRun) Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error) {
+	runID, err := strconv.ParseInt(id, 10, 64)
+	if err != nil {
+		return false, err
+	}
+	_, err = w.Jobs.GetRun(ctx, jobs.GetRunRequest{
+		RunId: runID,
+	})
+	if err != nil {
+		log.Debugf(ctx, "job run %s does not exist", id)
+		return false, err
+	}
+	return true, nil
+}
+
+func (r *JobRun) ResourceDescription() ResourceDescription {
+	return ResourceDescription{
+		SingularName:  "job_run",
+		PluralName:    "job_runs",
+		SingularTitle: "Job Run",
+		PluralTitle:   "Job Runs",
+	}
+}
+
+// GetName returns the in-product name. A run has no user-assigned name, so this
+// is empty.
+func (r *JobRun) GetName() string {
+	return ""
+}
+
+func (r *JobRun) GetURL() string {
+	return r.URL
+}
+
+// InitializeURL is a no-op for now: surfacing a stable run URL is deferred to a
+// later milestone.
+func (r *JobRun) InitializeURL(_ url.URL) {
+}

bundle/config/resources_test.go

@@ -125,6 +125,9 @@ func TestBundleResourcePluralNamesResolveInWorkspaceURLs(t *testing.T) {
 	// Resources that intentionally have no workspace URL.
 	noURL := map[string]bool{
 		"external_locations": true,
+		// A job run has no stable workspace URL yet; surfacing one is deferred
+		// to a later milestone (see JobRun.InitializeURL).
+		"job_runs":           true,
 		"postgres_branches":  true,
 		"postgres_endpoints": true,
 		"postgres_projects":  true,
@@ -157,6 +160,11 @@ func TestResourcesBindSupport(t *testing.T) {
 				JobSettings: jobs.JobSettings{},
 			},
 		},
+		JobRuns: map[string]*resources.JobRun{
+			"my_job_run": {
+				RunNow: jobs.RunNow{},
+			},
+		},
 		Pipelines: map[string]*resources.Pipeline{
 			"my_pipeline": {
 				CreatePipeline: pipelines.CreatePipeline{},
@@ -315,6 +323,7 @@ func TestResourcesBindSupport(t *testing.T) {
 	ctx := t.Context()
 	m := mocks.NewMockWorkspaceClient(t)
 	m.GetMockJobsAPI().EXPECT().Get(mock.Anything, mock.Anything).Return(nil, nil)
+	m.GetMockJobsAPI().EXPECT().GetRun(mock.Anything, mock.Anything).Return(nil, nil)
 	m.GetMockPipelinesAPI().EXPECT().Get(mock.Anything, mock.Anything).Return(nil, nil)
 	m.GetMockExperimentsAPI().EXPECT().GetExperiment(mock.Anything, mock.Anything).Return(nil, nil)
 	m.GetMockRegisteredModelsAPI().EXPECT().Get(mock.Anything, mock.Anything).Return(nil, nil)

bundle/deploy/terraform/lifecycle_test.go

@@ -18,6 +18,7 @@ func TestConvertLifecycleForAllResources(t *testing.T) {
 		"catalogs",
 		"external_locations",
 		"genie_spaces",
+		"job_runs",
 		"vector_search_endpoints",
 		"vector_search_indexes",
 	}