Carry a schedule onto the bundle job

acceptance/experimental/air/convert-to-dabs/output.txt

@@ -97,3 +97,58 @@ job and its uploaded files with:
 Error: environment.docker_image is not yet supported by convert-to-dabs
 
 Exit code: 1
+
+=== a schedule maps onto the bundle job (air run can't schedule; the bundle can)
+>>> [CLI] experimental air convert-to-dabs schedule.yaml --force
+Wrote a Databricks Asset Bundle to .:
+  databricks.yml
+  generated_artifacts/training_config.yaml
+  generated_artifacts/command.sh
+
+To deploy and run this workload as a bundle:
+  1. [CLI] bundle validate
+  2. [CLI] bundle deploy
+  3. [CLI] bundle run torchrun-a10-smoke-test
+
+bundle deploy uploads the code source and launch scripts automatically.
+To see what it deployed and where: [CLI] bundle summary
+
+Unlike `air run` (which submits an ephemeral run), bundle deploy creates a
+persistent job that is not garbage-collected. When you are done, remove the
+job and its uploaded files with:
+  [CLI] bundle destroy
+
+>>> cat databricks.yml
+bundle:
+  name: torchrun-a10-smoke-test
+sync:
+  paths:
+    - generated_artifacts
+targets:
+  dev:
+    mode: development
+    default: true
+resources:
+  jobs:
+    torchrun-a10-smoke-test:
+      name: torchrun-a10-smoke-test
+      tasks:
+        - task_key: torchrun-a10-smoke-test
+          environment_key: default
+          max_retries: 3
+          ai_runtime_task:
+            experiment: torchrun-a10-smoke-test
+            deployments:
+              - command_path: ./generated_artifacts/command.sh
+                compute:
+                  accelerator_type: GPU_1xA10
+                  accelerator_count: 1
+            code_source_path: ./src
+      environments:
+        - environment_key: default
+          spec:
+            environment_version: "4"
+      schedule:
+        quartz_cron_expression: 0 0 9 * * ?
+        timezone_id: America/Los_Angeles
+        pause_status: PAUSED

acceptance/experimental/air/convert-to-dabs/schedule.yaml

@@ -0,0 +1,13 @@
+experiment_name: torchrun-a10-smoke-test
+command: torchrun --nproc_per_node=1 train.py
+compute:
+  accelerator_type: GPU_1xA10
+  num_accelerators: 1
+code_source:
+  type: snapshot
+  snapshot:
+    root_path: ./src
+schedule:
+  quartz_cron_expression: "0 0 9 * * ?"
+  timezone_id: America/Los_Angeles
+  pause_status: PAUSED

acceptance/experimental/air/convert-to-dabs/script

@@ -18,3 +18,7 @@ trace $CLI experimental air convert-to-dabs train.yaml --force
 
 title "docker_image is not supported yet"
 errcode trace $CLI experimental air convert-to-dabs docker.yaml --output-dir generated-docker
+
+title "a schedule maps onto the bundle job (air run can't schedule; the bundle can)"
+trace $CLI experimental air convert-to-dabs schedule.yaml --force
+trace cat databricks.yml

acceptance/experimental/air/run/output.txt

@@ -22,7 +22,7 @@ Dry run: configuration for "smoke-test" is valid; not submitting.
 
 === override of an unknown field is rejected
 >>> [CLI] experimental air run -f valid.yaml --dry-run --override bogus=1
-Error: invalid --override "bogus": "bogus" is not a known field; available fields are: code_source, command, compute, env_variables, environment, experiment_name, idempotency_token, max_retries, mlflow_experiment_directory, mlflow_run_name, parameters, permissions, secrets, timeout_minutes, usage_policy_id, usage_policy_name
+Error: invalid --override "bogus": "bogus" is not a known field; available fields are: code_source, command, compute, env_variables, environment, experiment_name, idempotency_token, max_retries, mlflow_experiment_directory, mlflow_run_name, parameters, permissions, schedule, secrets, timeout_minutes, usage_policy_id, usage_policy_name
 
 Exit code: 1
 

experimental/air/cmd/convert_to_dabs.go

@@ -301,6 +301,19 @@ func buildBundleValue(ctx context.Context, cfg *runConfig, configPath, codeSourc
 	if perms := buildPermissionsValue(cfg.Permissions); perms.Kind() != dyn.KindInvalid {
 		job["permissions"] = nv(perms.MustSequence(), 5)
 	}
+	// schedule maps 1:1 onto the job's CronSchedule. `air run` can't honor it (it
+	// submits a one-time run); it exists so a converted bundle deploys as a recurring
+	// job. pause_status is omitted when unset so the Jobs default (UNPAUSED) applies.
+	if cfg.Schedule != nil {
+		sched := map[string]dyn.Value{
+			"quartz_cron_expression": nv(cfg.Schedule.QuartzCronExpression, 1),
+			"timezone_id":            nv(cfg.Schedule.TimezoneID, 2),
+		}
+		if cfg.Schedule.PauseStatus != "" {
+			sched["pause_status"] = nv(cfg.Schedule.PauseStatus, 3)
+		}
+		job["schedule"] = nv(sched, 6)
+	}
 
 	rootValue := map[string]dyn.Value{
 		"bundle": nv(map[string]dyn.Value{

experimental/air/cmd/convert_to_dabs_test.go

@@ -357,6 +357,48 @@ func TestConvertToDabsMapsUsagePolicyID(t *testing.T) {
 	assert.Equal(t, "12345678-90ab-cdef-1234-567890abcdef", get(t, root, "resources.jobs."+loaded.ExperimentName+".budget_policy_id").MustString())
 }
 
+// A schedule block maps 1:1 onto the bundle job's schedule (the field air run
+// can't honor but convert-to-dabs carries so bundle deploy schedules the job).
+func TestConvertToDabsMapsSchedule(t *testing.T) {
+	cfg := minimalConfig + `
+schedule:
+  quartz_cron_expression: "0 0 9 * * ?"
+  timezone_id: America/Los_Angeles
+  pause_status: PAUSED
+`
+	path := writeConfigFile(t, "run.yaml", cfg)
+	loaded, err := loadRunConfig(path)
+	require.NoError(t, err)
+
+	root, _, err := convertToDabs(t.Context(), loaded, path, filepath.Dir(path))
+	require.NoError(t, err)
+
+	sched := "resources.jobs." + loaded.ExperimentName + ".schedule"
+	assert.Equal(t, "0 0 9 * * ?", get(t, root, sched+".quartz_cron_expression").MustString())
+	assert.Equal(t, "America/Los_Angeles", get(t, root, sched+".timezone_id").MustString())
+	assert.Equal(t, "PAUSED", get(t, root, sched+".pause_status").MustString())
+}
+
+// pause_status is optional; when omitted it is left off the emitted schedule so
+// the Jobs default (UNPAUSED) applies.
+func TestConvertToDabsScheduleOmitsEmptyPauseStatus(t *testing.T) {
+	cfg := minimalConfig + `
+schedule:
+  quartz_cron_expression: "0 0 9 * * ?"
+  timezone_id: UTC
+`
+	path := writeConfigFile(t, "run.yaml", cfg)
+	loaded, err := loadRunConfig(path)
+	require.NoError(t, err)
+
+	root, _, err := convertToDabs(t.Context(), loaded, path, filepath.Dir(path))
+	require.NoError(t, err)
+
+	sched := "resources.jobs." + loaded.ExperimentName + ".schedule"
+	assert.True(t, has(root, sched+".quartz_cron_expression"))
+	assert.False(t, has(root, sched+".pause_status"), "empty pause_status must be omitted")
+}
+
 func TestConvertToDabsMapsPermissions(t *testing.T) {
 	cfg := minimalConfig + `
 permissions:

experimental/air/cmd/run.go

@@ -18,6 +18,7 @@ type runResult struct {
 	Status       string `json:"status"`
 	DryRun       bool   `json:"dry_run,omitempty"`
 	RunID        string `json:"run_id,omitempty"`
+	JobID        string `json:"job_id,omitempty"`
 	DashboardURL string `json:"dashboard_url,omitempty"`
 }
 
@@ -71,6 +72,13 @@ The workload is described by a YAML config file (see --file).`,
 			return renderEnvelope(ctx, runResult{Status: "DRY_RUN_OK", DryRun: true})
 		}
 
+		// A schedule turns the workload into a persistent, scheduled job instead of a
+		// one-time run: create (or update) the job and return, since there is no
+		// immediate run to submit or stream.
+		if cfg.Schedule != nil {
+			return runScheduled(ctx, cmd, cfg, file)
+		}
+
 		w := cmdctx.WorkspaceClient(ctx)
 		runID, dashboardURL, err := submitWorkload(ctx, w, cfg, file, idempotencyKey)
 		if err != nil {
@@ -128,6 +136,38 @@ The workload is described by a YAML config file (see --file).`,
 	return cmd
 }
 
+// runScheduled creates (or updates) a persistent, scheduled job for a workload
+// whose config carries a `schedule`. Unlike a submit, there is no immediate run
+// to stream, so --watch does not apply here.
+func runScheduled(ctx context.Context, cmd *cobra.Command, cfg *runConfig, configPath string) error {
+	w := cmdctx.WorkspaceClient(ctx)
+	jobID, jobURL, created, err := createScheduledJob(ctx, w, cfg, configPath)
+	if err != nil {
+		return err
+	}
+	jobIDStr := strconv.FormatInt(jobID, 10)
+
+	if root.OutputType(cmd) == flags.OutputJSON {
+		status := "SCHEDULED_UPDATED"
+		if created {
+			status = "SCHEDULED_CREATED"
+		}
+		return renderEnvelope(ctx, runResult{Status: status, JobID: jobIDStr, DashboardURL: jobURL})
+	}
+
+	verb := "Updated"
+	if created {
+		verb = "Created"
+	}
+	cmdio.LogString(ctx, fmt.Sprintf("%s scheduled job %s", verb, jobIDStr))
+	cmdio.LogString(ctx, "View at: "+jobURL)
+	cmdio.LogString(ctx, fmt.Sprintf("Runs on schedule: %s (%s)", cfg.Schedule.QuartzCronExpression, cfg.Schedule.TimezoneID))
+	if cfg.Schedule.PauseStatus == "PAUSED" {
+		cmdio.LogString(ctx, "The schedule is PAUSED; set pause_status: UNPAUSED (or unpause it in the Jobs UI) to activate it.")
+	}
+	return nil
+}
+
 // watchTerminalStatus resolves a watched run's final display state for the
 // closing --watch envelope. The run is terminal once streaming returns; if the
 // status can't be re-fetched, "UNKNOWN" is reported rather than guessing.

experimental/air/cmd/runconfig.go

@@ -57,6 +57,10 @@ type runConfig struct {
 	Permissions               []permission   `yaml:"permissions"`
 	UsagePolicyName           *string        `yaml:"usage_policy_name"`
 	UsagePolicyID             *string        `yaml:"usage_policy_id"`
+	// Schedule turns the workload into a recurring job. `air run` submits a one-time
+	// run and cannot honor it (see run.go); it is carried through convert-to-dabs,
+	// which emits it onto the bundle job where `bundle deploy` schedules it.
+	Schedule *scheduleConfig `yaml:"schedule"`
 }
 
 // validate runs structural validation over the whole config, returning the first
@@ -179,6 +183,12 @@ func (c *runConfig) validate() error {
 		}
 	}
 
+	if c.Schedule != nil {
+		if err := c.Schedule.validate(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -410,6 +420,30 @@ func (s *snapshotSourceConfig) validate() error {
 	return nil
 }
 
+// scheduleConfig mirrors the Jobs CronSchedule proto so it maps 1:1 onto the
+// bundle job's schedule block (see convert_to_dabs.go). pause_status is optional
+// and defaults to UNPAUSED, matching the Jobs default.
+type scheduleConfig struct {
+	QuartzCronExpression string `yaml:"quartz_cron_expression"`
+	TimezoneID           string `yaml:"timezone_id"`
+	PauseStatus          string `yaml:"pause_status"`
+}
+
+func (s *scheduleConfig) validate() error {
+	if strings.TrimSpace(s.QuartzCronExpression) == "" {
+		return errors.New("schedule.quartz_cron_expression is required")
+	}
+	if strings.TrimSpace(s.TimezoneID) == "" {
+		return errors.New("schedule.timezone_id is required (for example, 'America/Los_Angeles' or 'UTC')")
+	}
+	switch s.PauseStatus {
+	case "", "PAUSED", "UNPAUSED":
+	default:
+		return fmt.Errorf("schedule.pause_status must be PAUSED or UNPAUSED, got %q", s.PauseStatus)
+	}
+	return nil
+}
+
 // gitRef pins a snapshot to a specific git ref. branch and commit are mutually
 // exclusive; remote is only meaningful with branch.
 type gitRef struct {

experimental/air/cmd/runconfig_test.go

@@ -247,6 +247,18 @@ func TestRunConfigValidate_FieldRules(t *testing.T) {
 		// A name pasted into the id field gets pointed at the right field.
 		{"policy name in usage_policy_id", func(c *runConfig) { c.UsagePolicyID = str("team-a") }, "use usage_policy_name"},
 		{"uuid usage_policy_id alone is ok", func(c *runConfig) { c.UsagePolicyID = str("12345678-90ab-cdef-1234-567890abcdef") }, ""},
+		{"schedule missing cron", func(c *runConfig) {
+			c.Schedule = &scheduleConfig{TimezoneID: "UTC"}
+		}, "quartz_cron_expression is required"},
+		{"schedule missing timezone", func(c *runConfig) {
+			c.Schedule = &scheduleConfig{QuartzCronExpression: "0 0 9 * * ?"}
+		}, "timezone_id is required"},
+		{"schedule bad pause_status", func(c *runConfig) {
+			c.Schedule = &scheduleConfig{QuartzCronExpression: "0 0 9 * * ?", TimezoneID: "UTC", PauseStatus: "MAYBE"}
+		}, "pause_status must be PAUSED or UNPAUSED"},
+		{"schedule ok", func(c *runConfig) {
+			c.Schedule = &scheduleConfig{QuartzCronExpression: "0 0 9 * * ?", TimezoneID: "UTC"}
+		}, ""},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {