mecha10_cli/services/
train_jobs.rs

1//! Job catalog for `mecha10 train` (LAB-1776).
2//!
3//! A "job" is a YAML file under a project's `configs/rl/jobs/` directory that maps to one of
4//! the Kestra `flows/ml-pipeline/*.yml` flow types plus an env/algorithm/hyperparameter
5//! selection. `mecha10-templates`'s RL bundle (`create_rl_training_bundle`) ships a standard set
6//! of these into every scaffolded project - a project owner extends the menu simply by adding
7//! more `.yml` files to that directory, no mecha10 core changes needed, same extensibility
8//! philosophy as the `mecha10_rl.envs`/`mecha10_rl.algorithms` registry
9//! (`rl_task_runner.registry`).
10//!
11//! LAB-2970: a job file is now a *complete* training config, not just job identity/overrides
12//! layered onto a separate `configs/rl/local.yml`. The old two-file design (this file's
13//! `timesteps`/`lr`/`epochs` overrides, sentinel-merged at runtime onto `local.yml`'s full
14//! `PPOConfig`) had two names for the same hyperparameter depending on which file you were
15//! reading; that overlap is gone - `JobDefinition` now carries every `PPOConfig` field directly,
16//! using `PPOConfig`'s own field names (`total_timesteps`, `learning_rate`, `n_epochs`, ...), and
17//! `local.yml` no longer exists. `mecha10 train`'s remote dispatch already forwarded the raw
18//! config file's content as `config_yaml`/`CONFIG_YAML_B64` for the trainer container to run
19//! with (see `to_create_job_request`) - that transport is unchanged, it now just reads the job's
20//! own file instead of a separate `local.yml`.
21//!
22//! Field names/sentinel conventions (`-1`/`-1.0` meaning "use this field's built-in default")
23//! deliberately mirror `flows/ml-pipeline/train.yml`'s `inputs:` and
24//! `mecha10-api/src/jobs/types.rs::CreateJobRequest`, so a job definition maps 1:1 onto either
25//! dispatch path without translation surprises.
26
27use anyhow::{bail, Context, Result};
28use serde::{Deserialize, Serialize};
29use std::path::{Path, PathBuf};
30
31pub const JOB_CATALOG_DIR: &str = "configs/rl/jobs";
32
33pub use mecha10_job_validation::FlowType;
34
35/// One dimension of a hyperparameter search space for a tuning job.
36#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
37pub struct SearchSpaceEntry {
38    pub name: String,
39    #[serde(rename = "type")]
40    pub type_: SearchSpaceType,
41    #[serde(default)]
42    pub low: Option<f64>,
43    #[serde(default)]
44    pub high: Option<f64>,
45    #[serde(default)]
46    pub log: Option<bool>,
47    #[serde(default)]
48    pub choices: Option<Vec<serde_json::Value>>,
49}
50
51/// `search_space` entry kind - `float`/`int` take `low`/`high` (and optional `log`),
52/// `categorical` takes `choices`.
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, schemars::JsonSchema)]
54#[serde(rename_all = "lowercase")]
55pub enum SearchSpaceType {
56    Float,
57    Int,
58    Categorical,
59}
60
61/// A training job definition, loaded from `configs/rl/jobs/<name>.yml`.
62#[derive(Debug, Clone, Deserialize, schemars::JsonSchema)]
63#[allow(dead_code)]
64pub struct JobDefinition {
65    pub name: String,
66    #[serde(default)]
67    pub description: Option<String>,
68    pub flow_type: FlowType,
69    /// Name of the simulation environment to train in. If omitted, it's resolved from the
70    /// selected task.
71    #[serde(default)]
72    #[schemars(example = Some("warehouse_nav".to_string()))]
73    pub env_name: Option<String>,
74    /// Name of the task/curriculum to train on, selected from the project's task catalog.
75    #[serde(default)]
76    #[schemars(example = Some("patrol_curriculum".to_string()))]
77    pub task_name: Option<String>,
78    /// Name of the reinforcement learning algorithm to use for this job.
79    #[serde(default)]
80    #[schemars(example = Some("ppo".to_string()))]
81    pub algorithm: Option<String>,
82
83    /// Name of the policy network architecture to use. If omitted, it's auto-detected from the
84    /// environment.
85    #[serde(default)]
86    #[schemars(example = Some("MlpPolicy".to_string()))]
87    #[allow(dead_code, reason = "schema/deserialize-only, see doc comment above")]
88    pub policy: Option<String>,
89    #[serde(default)]
90    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
91    pub seed: Option<i64>,
92    #[serde(default)]
93    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
94    pub environment_map: Option<String>,
95    #[serde(default)]
96    pub total_timesteps: Option<i64>,
97    #[serde(default)]
98    pub learning_rate: Option<f64>,
99    #[serde(default)]
100    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
101    pub linear_lr_schedule: Option<bool>,
102    #[serde(default)]
103    pub n_steps: Option<i64>,
104    #[serde(default)]
105    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
106    pub batch_size: Option<i64>,
107    #[serde(default)]
108    pub n_epochs: Option<i64>,
109    #[serde(default)]
110    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
111    pub gamma: Option<f64>,
112    #[serde(default)]
113    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
114    pub gae_lambda: Option<f64>,
115    #[serde(default)]
116    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
117    pub clip_range: Option<f64>,
118    #[serde(default)]
119    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
120    pub ent_coef: Option<f64>,
121    #[serde(default)]
122    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
123    pub vf_coef: Option<f64>,
124    #[serde(default)]
125    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
126    pub max_grad_norm: Option<f64>,
127    /// Number of evaluations with no improvement to wait before stopping training early.
128    #[serde(default)]
129    #[schemars(example = Some(10))]
130    pub patience: Option<i64>,
131    #[serde(default)]
132    pub min_delta: Option<f64>,
133    /// Number of parallel environments to run during training.
134    #[serde(default)]
135    #[schemars(example = Some(8))]
136    pub n_envs: Option<i64>,
137
138    #[serde(default)]
139    pub n_trials: Option<i64>,
140    #[serde(default)]
141    pub tune_timesteps: Option<i64>,
142    #[serde(default)]
143    pub eval_eps: Option<i64>,
144    /// Hyperparameter search space to explore for a tuning job. If omitted, a default search
145    /// space is used.
146    #[serde(default)]
147    pub search_space: Option<Vec<SearchSpaceEntry>>,
148
149    #[serde(skip)]
150    #[schemars(skip)]
151    pub source_path: PathBuf,
152}
153
154#[derive(Debug, Clone, Serialize)]
155pub struct CreateJobRequest {
156    pub name: String,
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub description: Option<String>,
159    pub algorithm: String,
160    pub robot_id: String,
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub project_id: Option<String>,
163    #[serde(skip_serializing_if = "Option::is_none")]
164    pub project_name: Option<String>,
165    #[serde(skip_serializing_if = "Option::is_none")]
166    pub env_name: Option<String>,
167    #[serde(skip_serializing_if = "Option::is_none")]
168    pub task_name: Option<String>,
169    #[serde(skip_serializing_if = "Option::is_none")]
170    pub total_timesteps: Option<i64>,
171    #[serde(skip_serializing_if = "Option::is_none")]
172    pub n_steps: Option<i64>,
173    #[serde(skip_serializing_if = "Option::is_none")]
174    pub n_epochs: Option<i64>,
175    #[serde(skip_serializing_if = "Option::is_none")]
176    pub learning_rate: Option<f64>,
177    #[serde(skip_serializing_if = "Option::is_none")]
178    pub patience: Option<i64>,
179    #[serde(skip_serializing_if = "Option::is_none")]
180    pub min_delta: Option<f64>,
181    #[serde(skip_serializing_if = "Option::is_none")]
182    pub n_envs: Option<i64>,
183    #[serde(skip_serializing_if = "Option::is_none")]
184    pub config_yaml: Option<String>,
185    #[serde(skip_serializing_if = "Option::is_none")]
186    pub resume_run_id: Option<String>,
187    #[serde(skip_serializing_if = "Option::is_none")]
188    pub n_trials: Option<i64>,
189    #[serde(skip_serializing_if = "Option::is_none")]
190    pub tune_timesteps: Option<i64>,
191    #[serde(skip_serializing_if = "Option::is_none")]
192    pub eval_eps: Option<i64>,
193    #[serde(skip_serializing_if = "Option::is_none")]
194    pub search_space: Option<Vec<SearchSpaceEntry>>,
195}
196
197/// Read `configs/rl/tasks/<task_name>/task.json`'s `env_name` field, relative to `project_root`.
198/// Returns `None` on any failure (missing file, invalid JSON, missing/non-string field) - callers
199/// treat that the same as "no env_name available from the task", never a hard error: a job
200/// definition can still set `env_name` explicitly instead.
201fn env_name_from_task_json(project_root: &Path, task_name: &str) -> Option<String> {
202    task_json_field(project_root, task_name, "env_name")
203}
204
205fn task_json_field(project_root: &Path, task_name: &str, field: &str) -> Option<String> {
206    let path = project_root
207        .join(mecha10_cli_core::paths::project::RL_TASKS_DIR)
208        .join(task_name)
209        .join("task.json");
210    let content = std::fs::read_to_string(path).ok()?;
211    let value: serde_json::Value = serde_json::from_str(&content).ok()?;
212    value.get(field)?.as_str().map(str::to_string)
213}
214
215/// Drop `value` if it equals `sentinel` - the "unset" placeholder some YAML/env fields use
216/// instead of a real `null` (e.g. `-1` / `-1.0`).
217fn non_sentinel<T: PartialEq>(value: Option<T>, sentinel: T) -> Option<T> {
218    value.filter(|v| *v != sentinel)
219}
220
221fn non_sentinel_i64(value: Option<i64>) -> Option<i64> {
222    non_sentinel(value, -1)
223}
224
225fn non_sentinel_f64(value: Option<f64>) -> Option<f64> {
226    non_sentinel(value, -1.0)
227}
228
229impl JobDefinition {
230    /// This job's fields in `mecha10-job-validation`'s shared vocabulary (see
231    /// [`mecha10_job_validation::JobFields`]) - the CLI-specific hyperparameters
232    /// (`n_steps`/`learning_rate`/...) aren't included since that crate has no opinion on them.
233    fn validation_fields(&self, resume_run_id: Option<&str>) -> mecha10_job_validation::JobFields {
234        mecha10_job_validation::JobFields {
235            task_name: self.task_name.clone(),
236            algorithm: self.algorithm.clone(),
237            total_timesteps: non_sentinel_i64(self.total_timesteps),
238            resume_run_id: resume_run_id.map(str::to_string),
239            tune_run_id: None,
240            n_trials: non_sentinel_i64(self.n_trials),
241            tune_timesteps: non_sentinel_i64(self.tune_timesteps),
242            eval_eps: non_sentinel_i64(self.eval_eps),
243        }
244    }
245
246    /// Validate this job against `mecha10-job-validation`'s flow_type requirement table, plus a
247    /// `task_name`-exists check against the project's own task catalog
248    /// (`configs/rl/tasks/<task_name>/task.json`) - the validation pass remote dispatch should
249    /// run before doing anything else (see `run_remote` in `handlers/train.rs`). Replaces the
250    /// ad-hoc checks that used to live only in `handlers/train.rs` (e.g. the blank-`algorithm`
251    /// check for remote dispatch).
252    pub fn validate(&self, project_root: &Path, resume_run_id: Option<&str>) -> Result<()> {
253        let missing = mecha10_job_validation::validate_job(self.flow_type, &self.validation_fields(resume_run_id));
254        if !missing.is_empty() {
255            let details = missing.iter().map(|m| m.to_string()).collect::<Vec<_>>().join("; ");
256            bail!("job '{}' is missing required fields: {details}", self.name);
257        }
258
259        let mut missing_hyperparams = Vec::new();
260        if non_sentinel_i64(self.total_timesteps).is_none() {
261            missing_hyperparams.push("total_timesteps");
262        }
263        if self.flow_type != FlowType::ResumeRun && non_sentinel_i64(self.n_steps).is_none() {
264            missing_hyperparams.push("n_steps");
265        }
266        if !missing_hyperparams.is_empty() {
267            bail!(
268                "job '{}' is missing required field(s): {}",
269                self.name,
270                missing_hyperparams.join(", ")
271            );
272        }
273
274        if let Some(task_name) = self.task_name.as_deref() {
275            mecha10_job_validation::task_exists(project_root, task_name)
276                .with_context(|| format!("job '{}' has an invalid task_name", self.name))?;
277        }
278
279        Ok(())
280    }
281
282    /// Resolve which `mecha10_rl.envs` registry name this job targets: an explicit `env_name`
283    /// wins (back-compat for a job definition that still sets it), else it's read from
284    /// `task_name`'s own `task.json` (LAB env-belongs-with-the-task redesign - see
285    /// `train.yml.template`). `None` when neither is available - callers send that through
286    /// unchanged (an empty `ENV_NAME` locally, an omitted `env_name` for remote dispatch); the
287    /// Python side (`rl_task_runner.run`) applies the exact same fallback order independently,
288    /// so this is a CLI-side convenience/consistency resolution, not the only place it happens.
289    pub fn resolve_env_name(&self, project_root: &Path) -> Option<String> {
290        self.env_name.clone().or_else(|| {
291            self.task_name
292                .as_deref()
293                .and_then(|task_name| env_name_from_task_json(project_root, task_name))
294        })
295    }
296
297    /// Whether this job can be dispatched remotely with today's `CreateJobRequest`/`create_job`
298    /// contract. `train`, `resume-run`, and `tune` are wired server-side
299    /// (`mecha10-api/src/jobs/handlers.rs::train_flow_inputs`/`resume_flow_inputs`/
300    /// `tune_flow_inputs`); `tune-then-train`/`train-from-tune` aren't dispatchable from
301    /// `create_job` yet and stay local-only until the API grows dedicated support.
302    pub fn supports_remote(&self) -> bool {
303        matches!(self.flow_type, FlowType::Train | FlowType::ResumeRun | FlowType::Tune)
304    }
305
306    /// Build the remote dispatch request body. Only meaningful when `supports_remote()` is true;
307    /// callers should check that first. `algorithm` is required non-empty by `create_job` (unlike
308    /// local dispatch, where an empty `algorithm_name` falls back to the config file's value) -
309    /// callers should validate before sending.
310    pub fn to_create_job_request(
311        &self,
312        project_root: &Path,
313        robot_id: &str,
314        project_name: &str,
315        resume_run_id: Option<&str>,
316    ) -> Result<CreateJobRequest> {
317        let config_yaml = std::fs::read_to_string(&self.source_path).with_context(|| {
318            format!(
319                "failed to read {} - this job definition's own file must exist before dispatching remotely",
320                self.source_path.display()
321            )
322        })?;
323
324        Ok(CreateJobRequest {
325            name: self.name.clone(),
326            description: self.description.clone(),
327            algorithm: self.algorithm.clone().unwrap_or_default(),
328            robot_id: robot_id.to_string(),
329            project_id: Some(robot_id.to_string()),
330            project_name: Some(project_name.to_string()),
331            env_name: self.resolve_env_name(project_root),
332            task_name: self.task_name.clone(),
333            total_timesteps: non_sentinel_i64(self.total_timesteps),
334            n_steps: non_sentinel_i64(self.n_steps),
335            n_epochs: non_sentinel_i64(self.n_epochs),
336            learning_rate: non_sentinel_f64(self.learning_rate),
337            patience: non_sentinel_i64(self.patience),
338            min_delta: non_sentinel_f64(self.min_delta),
339            n_envs: non_sentinel_i64(self.n_envs),
340            config_yaml: Some(config_yaml),
341            resume_run_id: resume_run_id
342                .map(str::trim)
343                .filter(|id| !id.is_empty())
344                .map(str::to_string),
345            n_trials: non_sentinel_i64(self.n_trials),
346            tune_timesteps: non_sentinel_i64(self.tune_timesteps),
347            eval_eps: non_sentinel_i64(self.eval_eps),
348            search_space: self.search_space.clone(),
349        })
350    }
351}
352
353/// Load every `.yml`/`.yaml` job definition under `{project_root}/configs/rl/jobs/`, sorted by
354/// file name for a stable menu order. Returns an empty list (not an error) when the directory
355/// doesn't exist - `mecha10 train` treats that as "no jobs configured yet" and points the user at
356/// `mecha10 init`/the bundled defaults rather than failing.
357pub fn load_job_catalog(project_root: &Path) -> Result<Vec<JobDefinition>> {
358    let jobs_dir = project_root.join(JOB_CATALOG_DIR);
359    if !jobs_dir.exists() {
360        return Ok(Vec::new());
361    }
362
363    let mut paths: Vec<PathBuf> = std::fs::read_dir(&jobs_dir)
364        .with_context(|| format!("failed to read {}", jobs_dir.display()))?
365        .filter_map(|entry| entry.ok())
366        .map(|entry| entry.path())
367        .filter(|path| {
368            path.extension()
369                .and_then(|ext| ext.to_str())
370                .is_some_and(|ext| ext == "yml" || ext == "yaml")
371        })
372        .collect();
373    paths.sort();
374
375    let mut jobs = Vec::with_capacity(paths.len());
376    for path in paths {
377        let content = std::fs::read_to_string(&path).with_context(|| format!("failed to read {}", path.display()))?;
378        let mut job: JobDefinition = serde_yaml::from_str(&content)
379            .with_context(|| format!("failed to parse job definition {}", path.display()))?;
380        job.source_path = path;
381        jobs.push(job);
382    }
383
384    Ok(jobs)
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390    use tempfile::TempDir;
391
392    fn write_job(dir: &Path, file_name: &str, contents: &str) {
393        std::fs::write(dir.join(file_name), contents).unwrap();
394    }
395
396    fn write_task_json(project_root: &Path, task_name: &str, contents: &str) {
397        let task_dir = project_root
398            .join(mecha10_cli_core::paths::project::RL_TASKS_DIR)
399            .join(task_name);
400        std::fs::create_dir_all(&task_dir).unwrap();
401        std::fs::write(task_dir.join("task.json"), contents).unwrap();
402    }
403
404    #[test]
405    fn resolve_env_name_prefers_an_explicit_env_name_over_the_tasks_own() {
406        let dir = TempDir::new().unwrap();
407        write_task_json(dir.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
408
409        let job = JobDefinition {
410            env_name: Some("rover_arena".to_string()),
411            task_name: Some("find_target".to_string()),
412            ..base_job()
413        };
414        assert_eq!(job.resolve_env_name(dir.path()), Some("rover_arena".to_string()));
415    }
416
417    #[test]
418    fn resolve_env_name_falls_back_to_the_tasks_own_task_json() {
419        let dir = TempDir::new().unwrap();
420        write_task_json(
421            dir.path(),
422            "find_target",
423            r#"{"env_name": "basic_arena", "model": "humanoid"}"#,
424        );
425
426        let job = JobDefinition {
427            task_name: Some("find_target".to_string()),
428            ..base_job()
429        };
430        assert_eq!(job.resolve_env_name(dir.path()), Some("basic_arena".to_string()));
431    }
432
433    #[test]
434    fn resolve_env_name_is_none_when_task_has_no_env_name_field() {
435        let dir = TempDir::new().unwrap();
436        write_task_json(dir.path(), "find_target", r#"{"model": "humanoid"}"#);
437
438        let job = JobDefinition {
439            task_name: Some("find_target".to_string()),
440            ..base_job()
441        };
442        assert_eq!(job.resolve_env_name(dir.path()), None);
443    }
444
445    #[test]
446    fn resolve_env_name_is_none_without_an_env_name_or_a_task_name() {
447        let dir = TempDir::new().unwrap();
448        let job = base_job();
449        assert_eq!(job.resolve_env_name(dir.path()), None);
450    }
451
452    /// A minimal, fully-`None`/default `JobDefinition` for tests to `..base_job()` from, so each
453    /// test only spells out the fields it actually cares about.
454    fn base_job() -> JobDefinition {
455        JobDefinition {
456            name: "job".to_string(),
457            description: None,
458            flow_type: FlowType::Train,
459            env_name: None,
460            task_name: None,
461            algorithm: Some("ppo".to_string()),
462            policy: None,
463            seed: None,
464            environment_map: None,
465            total_timesteps: None,
466            learning_rate: None,
467            linear_lr_schedule: None,
468            n_steps: None,
469            batch_size: None,
470            n_epochs: None,
471            gamma: None,
472            gae_lambda: None,
473            clip_range: None,
474            ent_coef: None,
475            vf_coef: None,
476            max_grad_norm: None,
477            patience: None,
478            min_delta: None,
479            n_envs: None,
480            n_trials: None,
481            tune_timesteps: None,
482            eval_eps: None,
483            search_space: None,
484            source_path: PathBuf::new(),
485        }
486    }
487
488    #[test]
489    fn load_job_catalog_returns_empty_when_directory_missing() {
490        let temp = TempDir::new().unwrap();
491        let jobs = load_job_catalog(temp.path()).unwrap();
492        assert!(jobs.is_empty());
493    }
494
495    #[test]
496    fn load_job_catalog_parses_yaml_files_sorted_by_name() {
497        let temp = TempDir::new().unwrap();
498        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
499        std::fs::create_dir_all(&jobs_dir).unwrap();
500
501        write_job(
502            &jobs_dir,
503            "train.yml",
504            "name: train\nflow_type: train\nalgorithm: ppo\ntotal_timesteps: 1000000\n",
505        );
506        write_job(
507            &jobs_dir,
508            "hypertune.yml",
509            "name: hypertune\nflow_type: tune\nalgorithm: ppo\nn_trials: 10\n",
510        );
511
512        let jobs = load_job_catalog(temp.path()).unwrap();
513
514        assert_eq!(jobs.len(), 2);
515        assert_eq!(jobs[0].name, "hypertune");
516        assert_eq!(jobs[0].flow_type, FlowType::Tune);
517        assert_eq!(jobs[1].name, "train");
518        assert_eq!(jobs[1].flow_type, FlowType::Train);
519        assert_eq!(jobs[1].total_timesteps, Some(1000000));
520    }
521
522    #[test]
523    fn load_job_catalog_ignores_non_yaml_files() {
524        let temp = TempDir::new().unwrap();
525        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
526        std::fs::create_dir_all(&jobs_dir).unwrap();
527
528        write_job(&jobs_dir, "README.md", "not a job");
529        write_job(
530            &jobs_dir,
531            "train.yml",
532            "name: train\nflow_type: train\nalgorithm: ppo\n",
533        );
534
535        let jobs = load_job_catalog(temp.path()).unwrap();
536        assert_eq!(jobs.len(), 1);
537    }
538
539    #[test]
540    fn supports_remote_for_train_resume_run_and_tune() {
541        let base = base_job();
542        assert!(base.supports_remote());
543
544        let resume = JobDefinition {
545            flow_type: FlowType::ResumeRun,
546            ..base.clone()
547        };
548        assert!(resume.supports_remote());
549
550        let tune = JobDefinition {
551            flow_type: FlowType::Tune,
552            ..base.clone()
553        };
554        assert!(tune.supports_remote());
555
556        let tune_then_train = JobDefinition {
557            flow_type: FlowType::TuneThenTrain,
558            ..base.clone()
559        };
560        assert!(!tune_then_train.supports_remote());
561
562        let train_from_tune = JobDefinition {
563            flow_type: FlowType::TrainFromTune,
564            ..base
565        };
566        assert!(!train_from_tune.supports_remote());
567    }
568
569    #[test]
570    fn to_create_job_request_forwards_resume_run_id() {
571        let temp = TempDir::new().unwrap();
572        let jobs_dir = temp.path().join("configs/rl/jobs");
573        std::fs::create_dir_all(&jobs_dir).unwrap();
574        let source_path = jobs_dir.join("train-from-checkpoint.yml");
575        std::fs::write(&source_path, "name: train-from-checkpoint\nflow_type: resume-run\n").unwrap();
576
577        let job = JobDefinition {
578            name: "train-from-checkpoint".to_string(),
579            flow_type: FlowType::ResumeRun,
580            source_path,
581            ..base_job()
582        };
583
584        let req = job
585            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("  run-abc123  "))
586            .unwrap();
587
588        assert_eq!(req.resume_run_id.as_deref(), Some("run-abc123"));
589    }
590
591    #[test]
592    fn to_create_job_request_omits_empty_resume_run_id() {
593        let temp = TempDir::new().unwrap();
594        let jobs_dir = temp.path().join("configs/rl/jobs");
595        std::fs::create_dir_all(&jobs_dir).unwrap();
596        let source_path = jobs_dir.join("train-from-checkpoint.yml");
597        std::fs::write(&source_path, "name: train-from-checkpoint\nflow_type: resume-run\n").unwrap();
598
599        let job = JobDefinition {
600            name: "train-from-checkpoint".to_string(),
601            flow_type: FlowType::ResumeRun,
602            source_path,
603            ..base_job()
604        };
605
606        let req = job
607            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("   "))
608            .unwrap();
609
610        assert_eq!(req.resume_run_id, None);
611    }
612
613    /// Contract test against `mecha10-job-validation`'s `JobField` list (the one thing shared
614    /// with mecha10-api's own `CreateJobRequest`) - every field applicable to a flow type this
615    /// CLI actually dispatches remotely (`Train`/`ResumeRun`/`Tune` - see `supports_remote`) must
616    /// have a matching wire key on this struct. `TuneRunId` is excluded: it's only applicable to
617    /// `TrainFromTune`, which isn't remote-dispatchable yet (see `supports_remote`'s doc
618    /// comment), so this struct has no field for it - same reason `mecha10-api`'s own
619    /// `CreateJobRequest` doesn't either. A future `JobField` variant applicable to `Train`/
620    /// `ResumeRun`/`Tune` fails this test immediately instead of silently missing a field the
621    /// way `Tune`'s `n_trials`/`tune_timesteps`/`eval_eps` did before either side's
622    /// `CreateJobRequest` picked them up.
623    #[test]
624    fn create_job_request_has_a_field_for_every_dispatchable_job_field() {
625        let dispatchable = [FlowType::Train, FlowType::ResumeRun, FlowType::Tune];
626
627        let req = CreateJobRequest {
628            name: "job".to_string(),
629            description: None,
630            algorithm: "ppo".to_string(),
631            robot_id: "robot-1".to_string(),
632            project_id: None,
633            project_name: None,
634            env_name: None,
635            task_name: Some("find_target".to_string()),
636            total_timesteps: Some(1_000_000),
637            n_steps: None,
638            n_epochs: None,
639            learning_rate: None,
640            patience: None,
641            min_delta: None,
642            n_envs: None,
643            config_yaml: None,
644            resume_run_id: Some("run-123".to_string()),
645            n_trials: Some(10),
646            tune_timesteps: Some(500_000),
647            eval_eps: Some(5),
648            search_space: None,
649        };
650        let json = serde_json::to_value(&req).expect("CreateJobRequest serializes");
651
652        for field in mecha10_job_validation::all_fields() {
653            let applicable = dispatchable
654                .iter()
655                .any(|&ft| field.requirement(ft) != mecha10_job_validation::FieldRequirement::NotApplicable);
656            if applicable {
657                assert!(
658                    json.get(field.wire_name()).is_some(),
659                    "CreateJobRequest has no '{}' field for {field:?}, which is applicable to a \
660                     dispatchable flow type",
661                    field.wire_name()
662                );
663            }
664        }
665    }
666
667    #[test]
668    fn to_create_job_request_forwards_tune_fields() {
669        let temp = TempDir::new().unwrap();
670        let jobs_dir = temp.path().join("configs/rl/jobs");
671        std::fs::create_dir_all(&jobs_dir).unwrap();
672        let source_path = jobs_dir.join("hypertune.yml");
673        std::fs::write(&source_path, "name: hypertune\nflow_type: tune\n").unwrap();
674
675        let job = JobDefinition {
676            name: "hypertune".to_string(),
677            flow_type: FlowType::Tune,
678            n_trials: Some(20),
679            tune_timesteps: Some(250_000),
680            eval_eps: Some(3),
681            source_path,
682            ..base_job()
683        };
684
685        let req = job
686            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
687            .unwrap();
688
689        assert_eq!(req.n_trials, Some(20));
690        assert_eq!(req.tune_timesteps, Some(250_000));
691        assert_eq!(req.eval_eps, Some(3));
692    }
693
694    #[test]
695    fn to_create_job_request_forwards_n_envs() {
696        let temp = TempDir::new().unwrap();
697        let jobs_dir = temp.path().join("configs/rl/jobs");
698        std::fs::create_dir_all(&jobs_dir).unwrap();
699        let source_path = jobs_dir.join("train.yml");
700        std::fs::write(&source_path, "name: train\nflow_type: train\nn_envs: 8\n").unwrap();
701
702        let job = JobDefinition {
703            name: "train".to_string(),
704            n_envs: Some(8),
705            source_path,
706            ..base_job()
707        };
708
709        let req = job
710            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
711            .unwrap();
712
713        assert_eq!(req.n_envs, Some(8));
714    }
715
716    #[test]
717    fn to_create_job_request_forwards_search_space() {
718        let temp = TempDir::new().unwrap();
719        let jobs_dir = temp.path().join("configs/rl/jobs");
720        std::fs::create_dir_all(&jobs_dir).unwrap();
721        let source_path = jobs_dir.join("hypertune.yml");
722        let yaml = "name: hypertune\nflow_type: tune\nsearch_space:\n  - name: learning_rate\n    \
723                    type: float\n    low: 1e-5\n    high: 1e-3\n    log: true\n  - name: n_steps\n    \
724                    type: categorical\n    choices: [1024, 2048, 4096]\n  - name: n_epochs\n    \
725                    type: int\n    low: 5\n    high: 20\n";
726        std::fs::write(&source_path, yaml).unwrap();
727        let job: JobDefinition = serde_yaml::from_str(yaml).unwrap();
728        let job = JobDefinition { source_path, ..job };
729
730        let req = job
731            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
732            .unwrap();
733
734        let search_space = req.search_space.expect("search_space forwarded");
735        assert_eq!(search_space.len(), 3);
736        assert_eq!(search_space[0].name, "learning_rate");
737        assert_eq!(search_space[0].type_, SearchSpaceType::Float);
738        assert_eq!(search_space[0].low, Some(1e-5));
739        assert_eq!(search_space[0].high, Some(1e-3));
740        assert_eq!(search_space[0].log, Some(true));
741        assert_eq!(search_space[1].name, "n_steps");
742        assert_eq!(search_space[1].type_, SearchSpaceType::Categorical);
743        assert_eq!(
744            search_space[1].choices,
745            Some(vec![
746                serde_json::json!(1024),
747                serde_json::json!(2048),
748                serde_json::json!(4096)
749            ])
750        );
751        assert_eq!(search_space[2].name, "n_epochs");
752        assert_eq!(search_space[2].type_, SearchSpaceType::Int);
753        assert_eq!(search_space[2].low, Some(5.0));
754        assert_eq!(search_space[2].high, Some(20.0));
755    }
756
757    #[test]
758    fn to_create_job_request_omits_search_space_when_unset() {
759        let temp = TempDir::new().unwrap();
760        let jobs_dir = temp.path().join("configs/rl/jobs");
761        std::fs::create_dir_all(&jobs_dir).unwrap();
762        let source_path = jobs_dir.join("train.yml");
763        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
764
765        let job = JobDefinition {
766            name: "train".to_string(),
767            source_path,
768            ..base_job()
769        };
770
771        let req = job
772            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
773            .unwrap();
774
775        assert_eq!(req.search_space, None);
776    }
777
778    #[test]
779    fn to_create_job_request_drops_sentinel_values() {
780        let temp = TempDir::new().unwrap();
781        let jobs_dir = temp.path().join("configs/rl/jobs");
782        std::fs::create_dir_all(&jobs_dir).unwrap();
783        let source_path = jobs_dir.join("train.yml");
784        std::fs::write(&source_path, "name: train\nflow_type: train\nenv_name: rover\n").unwrap();
785
786        let job = JobDefinition {
787            name: "train".to_string(),
788            description: Some("desc".to_string()),
789            env_name: Some("rover".to_string()),
790            total_timesteps: Some(-1),
791            n_steps: Some(2048),
792            n_epochs: Some(-1),
793            learning_rate: Some(-1.0),
794            patience: Some(50),
795            min_delta: Some(-1.0),
796            n_envs: Some(-1),
797            source_path,
798            ..base_job()
799        };
800
801        let req = job
802            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
803            .unwrap();
804
805        assert_eq!(req.name, "train");
806        assert_eq!(req.algorithm, "ppo");
807        assert_eq!(req.robot_id, "my-project");
808        assert_eq!(req.project_id.as_deref(), Some("my-project"));
809        assert_eq!(req.project_name.as_deref(), Some("my-robot"));
810        assert_eq!(req.total_timesteps, None);
811        assert_eq!(req.n_steps, Some(2048));
812        assert_eq!(req.n_epochs, None);
813        assert_eq!(req.learning_rate, None);
814        assert_eq!(req.patience, Some(50));
815        assert_eq!(req.min_delta, None);
816        assert_eq!(req.n_envs, None);
817        assert_eq!(
818            req.config_yaml.as_deref(),
819            Some("name: train\nflow_type: train\nenv_name: rover\n")
820        );
821        assert_eq!(req.task_name, None);
822    }
823
824    #[test]
825    fn to_create_job_request_forwards_task_name() {
826        let temp = TempDir::new().unwrap();
827        let jobs_dir = temp.path().join("configs/rl/jobs");
828        std::fs::create_dir_all(&jobs_dir).unwrap();
829        let source_path = jobs_dir.join("train.yml");
830        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
831
832        let job = JobDefinition {
833            name: "train".to_string(),
834            task_name: Some("navigate".to_string()),
835            source_path,
836            ..base_job()
837        };
838
839        let req = job
840            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
841            .unwrap();
842
843        assert_eq!(req.task_name.as_deref(), Some("navigate"));
844    }
845
846    #[test]
847    fn to_create_job_request_errors_when_the_jobs_own_file_is_missing() {
848        let temp = TempDir::new().unwrap();
849        let job = JobDefinition {
850            name: "train".to_string(),
851            source_path: temp.path().join("configs/rl/jobs/train.yml"),
852            ..base_job()
853        };
854
855        let err = job
856            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
857            .unwrap_err();
858        assert!(err.to_string().contains("configs/rl/jobs/train.yml"));
859    }
860}
mecha10_cli/services/train_jobs.rs

mecha10_cli/services/
train_jobs.rs
