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 selection.
5
6use anyhow::{bail, Context, Result};
7use serde::{Deserialize, Serialize};
8use std::path::{Path, PathBuf};
9
10pub const JOB_CATALOG_DIR: &str = "configs/rl/jobs";
11
12pub use mecha10_job_validation::FlowType;
13
14/// One dimension of a hyperparameter search space for a tuning job.
15#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
16pub struct SearchSpaceEntry {
17    pub name: String,
18    #[serde(rename = "type")]
19    pub type_: SearchSpaceType,
20    #[serde(default)]
21    pub low: Option<f64>,
22    #[serde(default)]
23    pub high: Option<f64>,
24    #[serde(default)]
25    pub log: Option<bool>,
26    #[serde(default)]
27    pub choices: Option<Vec<serde_json::Value>>,
28}
29
30/// `search_space` entry kind - `float`/`int` take `low`/`high` (and optional `log`),
31/// `categorical` takes `choices`.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, schemars::JsonSchema)]
33#[serde(rename_all = "lowercase")]
34pub enum SearchSpaceType {
35    Float,
36    Int,
37    Categorical,
38}
39
40/// A training job definition, loaded from `configs/rl/jobs/<name>.yml`.
41#[derive(Debug, Clone, Deserialize, schemars::JsonSchema)]
42#[allow(dead_code)]
43pub struct JobDefinition {
44    pub name: String,
45    #[serde(default)]
46    pub description: Option<String>,
47    pub flow_type: FlowType,
48    /// Name of the simulation environment to train in. If omitted, it's resolved from the
49    /// selected task.
50    #[serde(default)]
51    #[schemars(example = Some("warehouse_nav".to_string()))]
52    pub env_name: Option<String>,
53    /// Name of the task/curriculum to train on, selected from the project's task catalog.
54    #[serde(default)]
55    #[schemars(example = Some("patrol_curriculum".to_string()))]
56    pub task_name: Option<String>,
57    /// Name of the reinforcement learning algorithm to use for this job.
58    #[serde(default)]
59    #[schemars(example = Some("ppo".to_string()))]
60    pub algorithm: Option<String>,
61
62    /// Name of the policy network architecture to use. If omitted, it's auto-detected from the
63    /// environment.
64    #[serde(default)]
65    #[schemars(example = Some("MlpPolicy".to_string()))]
66    #[allow(dead_code, reason = "schema/deserialize-only, see doc comment above")]
67    pub policy: Option<String>,
68    #[serde(default)]
69    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
70    pub seed: Option<i64>,
71    #[serde(default)]
72    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
73    pub environment_map: Option<String>,
74    #[serde(default)]
75    pub total_timesteps: Option<i64>,
76    #[serde(default)]
77    pub learning_rate: Option<f64>,
78    #[serde(default)]
79    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
80    pub linear_lr_schedule: Option<bool>,
81    #[serde(default)]
82    pub n_steps: Option<i64>,
83    #[serde(default)]
84    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
85    pub batch_size: Option<i64>,
86    #[serde(default)]
87    pub n_epochs: Option<i64>,
88    #[serde(default)]
89    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
90    pub gamma: Option<f64>,
91    #[serde(default)]
92    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
93    pub gae_lambda: Option<f64>,
94    #[serde(default)]
95    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
96    pub clip_range: Option<f64>,
97    #[serde(default)]
98    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
99    pub ent_coef: Option<f64>,
100    #[serde(default)]
101    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
102    pub vf_coef: Option<f64>,
103    #[serde(default)]
104    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
105    pub max_grad_norm: Option<f64>,
106    /// Number of evaluations with no improvement to wait before stopping training early.
107    #[serde(default)]
108    #[schemars(example = Some(10))]
109    pub patience: Option<i64>,
110    #[serde(default)]
111    pub min_delta: Option<f64>,
112    /// Number of parallel environments to run during training.
113    #[serde(default)]
114    #[schemars(example = Some(8))]
115    pub n_envs: Option<i64>,
116
117    #[serde(default)]
118    pub n_trials: Option<i64>,
119    #[serde(default)]
120    pub tune_timesteps: Option<i64>,
121    #[serde(default)]
122    pub eval_eps: Option<i64>,
123    /// Hyperparameter search space to explore for a tuning job. If omitted, a default search
124    /// space is used.
125    #[serde(default)]
126    pub search_space: Option<Vec<SearchSpaceEntry>>,
127
128    #[serde(skip)]
129    #[schemars(skip)]
130    pub source_path: PathBuf,
131}
132
133#[derive(Debug, Clone, Serialize)]
134pub struct CreateJobRequest {
135    pub name: String,
136    #[serde(skip_serializing_if = "Option::is_none")]
137    pub description: Option<String>,
138    pub algorithm: String,
139    pub robot_id: String,
140    #[serde(skip_serializing_if = "Option::is_none")]
141    pub project_id: Option<String>,
142    #[serde(skip_serializing_if = "Option::is_none")]
143    pub project_name: Option<String>,
144    #[serde(skip_serializing_if = "Option::is_none")]
145    pub env_name: Option<String>,
146    #[serde(skip_serializing_if = "Option::is_none")]
147    pub task_name: Option<String>,
148    #[serde(skip_serializing_if = "Option::is_none")]
149    pub total_timesteps: Option<i64>,
150    #[serde(skip_serializing_if = "Option::is_none")]
151    pub n_steps: Option<i64>,
152    #[serde(skip_serializing_if = "Option::is_none")]
153    pub n_epochs: Option<i64>,
154    #[serde(skip_serializing_if = "Option::is_none")]
155    pub learning_rate: Option<f64>,
156    #[serde(skip_serializing_if = "Option::is_none")]
157    pub patience: Option<i64>,
158    #[serde(skip_serializing_if = "Option::is_none")]
159    pub min_delta: Option<f64>,
160    #[serde(skip_serializing_if = "Option::is_none")]
161    pub n_envs: Option<i64>,
162    #[serde(skip_serializing_if = "Option::is_none")]
163    pub config_yaml: Option<String>,
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub resume_run_id: Option<String>,
166    #[serde(skip_serializing_if = "Option::is_none")]
167    pub n_trials: Option<i64>,
168    #[serde(skip_serializing_if = "Option::is_none")]
169    pub tune_timesteps: Option<i64>,
170    #[serde(skip_serializing_if = "Option::is_none")]
171    pub eval_eps: Option<i64>,
172    #[serde(skip_serializing_if = "Option::is_none")]
173    pub search_space: Option<Vec<SearchSpaceEntry>>,
174}
175
176/// Read `configs/rl/tasks/<task_name>/task.json`'s `env_name` field, relative to `project_root`.
177/// Returns `None` on any failure (missing file, invalid JSON, missing/non-string field).
178fn env_name_from_task_json(project_root: &Path, task_name: &str) -> Option<String> {
179    task_json_field(project_root, task_name, "env_name")
180}
181
182fn task_json_field(project_root: &Path, task_name: &str, field: &str) -> Option<String> {
183    let path = project_root
184        .join(mecha10_cli_core::paths::project::RL_TASKS_DIR)
185        .join(task_name)
186        .join("task.json");
187    let content = std::fs::read_to_string(path).ok()?;
188    let value: serde_json::Value = serde_json::from_str(&content).ok()?;
189    value.get(field)?.as_str().map(str::to_string)
190}
191
192/// Drop `value` if it equals `sentinel` - the "unset" placeholder some YAML/env fields use
193/// instead of a real `null` (e.g. `-1` / `-1.0`).
194///
195/// This is a *wire-building* helper, used only by [`JobDefinition::to_create_job_request`] to
196/// decide what literal value to put on the dispatch request body: a field left at its
197/// placeholder is omitted so the remote side applies its own default, rather than being sent
198/// `-1` as if that were a real number. It is deliberately **not** used for validation
199/// presence/requiredness checks anymore - [`JobDefinition::validation_fields`]/
200/// [`JobDefinition::validation_fields_for_algorithm`] pass raw values straight through to
201/// `mecha10-job-validation`, which decides per-field (via its own schema, not a blanket strip
202/// applied by every caller) whether a given field's `-1`/`-1.0` is a meaningful value (e.g.
203/// `patience`/`min_delta` for `ppo`, where it means "disabled") or a "not set" placeholder. Using
204/// this same blanket helper for *both* purposes used to be the bug: it stripped `patience`/
205/// `min_delta`'s real `-1`/`-1.0` before validation ever saw it, making a correctly-configured
206/// `ppo` job (the scaffolded `train.yml` template's own `patience: -1`/`min_delta: -1.0`) look
207/// like it was missing required fields. See PR #1639 (an ad hoc, special-cased attempt at fixing
208/// just this) and its revert PR #1641 (determined this needed real per-field schema design).
209fn non_sentinel<T: PartialEq>(value: Option<T>, sentinel: T) -> Option<T> {
210    value.filter(|v| *v != sentinel)
211}
212
213fn non_sentinel_i64(value: Option<i64>) -> Option<i64> {
214    non_sentinel(value, -1)
215}
216
217fn non_sentinel_f64(value: Option<f64>) -> Option<f64> {
218    non_sentinel(value, -1.0)
219}
220
221impl JobDefinition {
222    /// This job's fields in `mecha10-job-validation`'s shared vocabulary (see
223    /// [`mecha10_job_validation::JobFields`]). Raw values, not sentinel-stripped: whether a
224    /// `-1`/`-1.0` on a given field counts as "set" or "absent" is a per-field schema question
225    /// `mecha10-job-validation` answers itself (see its `SentinelConvention`), not something a
226    /// caller should pre-decide by stripping the value before validation ever sees it - see
227    /// [`non_sentinel_i64`]'s doc comment for why that used to be wrong.
228    fn validation_fields(&self, resume_run_id: Option<&str>) -> mecha10_job_validation::JobFields {
229        mecha10_job_validation::JobFields {
230            task_name: self.task_name.clone(),
231            algorithm: self.algorithm.clone(),
232            resume_run_id: resume_run_id.map(str::to_string),
233            tune_run_id: None,
234            n_trials: self.n_trials,
235            tune_timesteps: self.tune_timesteps,
236            eval_eps: self.eval_eps,
237        }
238    }
239
240    /// This job's per-algorithm hyperparameter fields in `mecha10-job-validation`'s shared
241    /// vocabulary (see [`mecha10_job_validation::AlgorithmFields`]) - the by-algorithm
242    /// counterpart to [`Self::validation_fields`]. Also raw, unstripped values, for the same
243    /// reason: `patience`/`min_delta`'s `-1`/`-1.0` is a real "disabled" value for `ppo` (the
244    /// scaffolded `train.yml` template's own setting), and only `mecha10-job-validation`'s schema
245    /// knows that - stripping it here before validation runs is exactly the bug this method used
246    /// to have (see [`non_sentinel_i64`]'s doc comment).
247    fn validation_fields_for_algorithm(&self) -> mecha10_job_validation::AlgorithmFields {
248        mecha10_job_validation::AlgorithmFields {
249            total_timesteps: self.total_timesteps,
250            learning_rate: self.learning_rate,
251            n_steps: self.n_steps,
252            n_epochs: self.n_epochs,
253            patience: self.patience,
254            min_delta: self.min_delta,
255        }
256    }
257
258    /// Validate this job against `mecha10-job-validation`'s flow_type requirement table and its
259    /// by-algorithm hyperparameter requirement table, plus a `task_name`-exists check against
260    /// the project's own task catalog. Algorithm-awareness (e.g. `ppo` needing `n_steps`, `act`
261    /// needing none of these fields) lives entirely in `mecha10-job-validation`'s
262    /// `AlgorithmField` table - this method only composes the two generic validators' results,
263    /// the same way `compose_node_schema` composes a base schema with a node's own custom
264    /// schema rather than validating them separately.
265    pub fn validate(&self, project_root: &Path, resume_run_id: Option<&str>) -> Result<()> {
266        let missing_basic =
267            mecha10_job_validation::validate_job(self.flow_type, &self.validation_fields(resume_run_id));
268        let missing_algorithm = mecha10_job_validation::validate_algorithm_fields(
269            self.algorithm.as_deref().unwrap_or_default(),
270            self.flow_type,
271            &self.validation_fields_for_algorithm(),
272        );
273
274        if !missing_basic.is_empty() || !missing_algorithm.is_empty() {
275            let details = missing_basic
276                .iter()
277                .map(|m| m.to_string())
278                .chain(missing_algorithm.iter().map(|m| m.to_string()))
279                .collect::<Vec<_>>()
280                .join("; ");
281            bail!("job '{}' is missing required fields: {details}", self.name);
282        }
283
284        if let Some(task_name) = self.task_name.as_deref() {
285            mecha10_job_validation::task_exists(project_root, task_name)
286                .with_context(|| format!("job '{}' has an invalid task_name", self.name))?;
287        }
288
289        Ok(())
290    }
291
292    /// Resolve which `mecha10_rl.envs` registry name this job targets: an explicit `env_name`
293    /// wins, else it's read from `task_name`'s own `task.json`. `None` when neither is available.
294    pub fn resolve_env_name(&self, project_root: &Path) -> Option<String> {
295        self.env_name.clone().or_else(|| {
296            self.task_name
297                .as_deref()
298                .and_then(|task_name| env_name_from_task_json(project_root, task_name))
299        })
300    }
301
302    /// Whether this job can be dispatched remotely with today's `CreateJobRequest`/`create_job`
303    /// contract. `tune-then-train`/`train-from-tune` aren't dispatchable yet and stay local-only.
304    pub fn supports_remote(&self) -> bool {
305        matches!(self.flow_type, FlowType::Train | FlowType::ResumeRun | FlowType::Tune)
306    }
307
308    /// Build the remote dispatch request body. Only meaningful when `supports_remote()` is true;
309    /// callers should check that first.
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    /// A `ppo` job with every one of its required hyperparameters set (per
489    /// `mecha10_job_validation::AlgorithmField::requirement`) except `total_timesteps`/
490    /// `n_steps`, which individual tests below null out to isolate the field they're checking.
491    fn ppo_job_with_hyperparams_set() -> JobDefinition {
492        JobDefinition {
493            task_name: Some("find_target".to_string()),
494            algorithm: Some("ppo".to_string()),
495            total_timesteps: Some(1000),
496            learning_rate: Some(0.0003),
497            n_steps: Some(2048),
498            n_epochs: Some(10),
499            patience: Some(10),
500            min_delta: Some(0.01),
501            ..base_job()
502        }
503    }
504
505    #[test]
506    fn validate_does_not_require_any_hyperparameter_fields_for_a_non_ppo_algorithm() {
507        let temp = TempDir::new().unwrap();
508        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
509
510        let job = JobDefinition {
511            task_name: Some("find_target".to_string()),
512            algorithm: Some("act".to_string()),
513            total_timesteps: Some(1000),
514            n_steps: None,
515            ..base_job()
516        };
517
518        assert!(job.validate(temp.path(), None).is_ok());
519    }
520
521    /// Regression test for the bug this crate's base/by-algorithm split exists to prevent from
522    /// recurring: `act` has no `total_timesteps` concept at all (it reads `total_steps`
523    /// instead, an untyped field today), so a `Train`-flow `act` job that omits
524    /// `total_timesteps` entirely must still pass `validate()` - it used to fail, because the
525    /// old hand-rolled check required `total_timesteps` unconditionally for every algorithm.
526    #[test]
527    fn validate_does_not_require_total_timesteps_for_an_act_train_job() {
528        let temp = TempDir::new().unwrap();
529        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
530
531        let job = JobDefinition {
532            flow_type: FlowType::Train,
533            task_name: Some("find_target".to_string()),
534            algorithm: Some("act".to_string()),
535            total_timesteps: None,
536            n_steps: None,
537            ..base_job()
538        };
539
540        assert!(job.validate(temp.path(), None).is_ok());
541    }
542
543    #[test]
544    fn validate_still_requires_n_steps_for_ppo() {
545        let temp = TempDir::new().unwrap();
546        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
547
548        let job = JobDefinition {
549            n_steps: None,
550            ..ppo_job_with_hyperparams_set()
551        };
552
553        let err = job.validate(temp.path(), None).unwrap_err();
554        assert!(
555            err.to_string().contains("n_steps"),
556            "expected error to mention n_steps, got: {err}"
557        );
558    }
559
560    #[test]
561    fn validate_passes_for_ppo_with_every_required_hyperparameter_set() {
562        let temp = TempDir::new().unwrap();
563        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
564
565        let job = ppo_job_with_hyperparams_set();
566
567        assert!(job.validate(temp.path(), None).is_ok());
568    }
569
570    /// Regression test for the original reported bug: the scaffolded `train.yml` template sets
571    /// `patience: -1` / `min_delta: -1.0` to mean "no early-stopping limit" - a real, intended
572    /// value, not an unset field. `validate()` must accept that, not report `patience`/
573    /// `min_delta` as missing the way it used to when every numeric field (including these two)
574    /// was blanket-stripped of `-1`/`-1.0` before validation ever ran. See PR #1639/#1641 for the
575    /// reverted ad hoc attempt at fixing just this.
576    #[test]
577    fn validate_passes_for_ppo_with_patience_and_min_delta_at_their_disabled_sentinel_value() {
578        let temp = TempDir::new().unwrap();
579        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
580
581        let job = JobDefinition {
582            patience: Some(-1),
583            min_delta: Some(-1.0),
584            ..ppo_job_with_hyperparams_set()
585        };
586
587        assert!(
588            job.validate(temp.path(), None).is_ok(),
589            "a ppo job with patience: -1 / min_delta: -1.0 (the scaffolded train.yml template's \
590             own values) must validate - those are meaningful 'disabled' values, not absence"
591        );
592    }
593
594    /// The flip side of the regression test above: `total_timesteps` has no sentinel-for-disabled
595    /// convention, so `-1` there must still count as absent/missing, the same as if it were
596    /// never set at all. This guards against over-generalizing "sentinels are fine everywhere"
597    /// while fixing the `patience`/`min_delta` bug.
598    #[test]
599    fn validate_still_rejects_total_timesteps_set_to_the_placeholder_sentinel_value() {
600        let temp = TempDir::new().unwrap();
601        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
602
603        let job = JobDefinition {
604            total_timesteps: Some(-1),
605            ..ppo_job_with_hyperparams_set()
606        };
607
608        let err = job.validate(temp.path(), None).unwrap_err();
609        assert!(
610            err.to_string().contains("total_timesteps"),
611            "expected error to mention total_timesteps, got: {err}"
612        );
613    }
614
615    /// `act` has no opinion about `patience`/`min_delta` at all (every field is `NotApplicable`
616    /// for it, per `mecha10_job_validation::AlgorithmField::requirement`), but the scaffolded
617    /// `train-act.yml` template sets them to `-1`/`-1.0` anyway (same as `train.yml`'s ppo job).
618    /// A correctly-configured `act` job must still validate regardless of what those two fields
619    /// are set to.
620    #[test]
621    fn validate_passes_for_act_with_patience_and_min_delta_set_to_the_sentinel_value() {
622        let temp = TempDir::new().unwrap();
623        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
624
625        let job = JobDefinition {
626            flow_type: FlowType::Train,
627            task_name: Some("find_target".to_string()),
628            algorithm: Some("act".to_string()),
629            total_timesteps: Some(1000),
630            patience: Some(-1),
631            min_delta: Some(-1.0),
632            ..base_job()
633        };
634
635        assert!(job.validate(temp.path(), None).is_ok());
636    }
637
638    /// A `ppo` `resume-run` job restores its rollout length (`n_steps`) from the checkpoint it
639    /// resumes, so `n_steps` specifically is *not* required for that flow - matching the real,
640    /// currently-shipping `resume-train.yml` job template (`algorithm: ppo`,
641    /// `flow_type: resume-run`, no `n_steps`). This carve-out is encoded in
642    /// `mecha10_job_validation::AlgorithmField::requirement`, which takes `flow_type` for exactly
643    /// this reason; it predates this crate as a hand-rolled `if` in the CLI
644    /// (`flow_type != ResumeRun && algorithm == "ppo" && n_steps missing`).
645    #[test]
646    fn validate_does_not_require_n_steps_for_a_ppo_resume_run() {
647        let temp = TempDir::new().unwrap();
648        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
649
650        let without_n_steps = JobDefinition {
651            flow_type: FlowType::ResumeRun,
652            n_steps: None,
653            ..ppo_job_with_hyperparams_set()
654        };
655        assert!(
656            without_n_steps.validate(temp.path(), Some("run-123")).is_ok(),
657            "a ppo resume-run job without n_steps must still validate"
658        );
659    }
660
661    /// The `n_steps` carve-out above is `n_steps`-specific: every *other* `ppo` hyperparameter is
662    /// still required for a `resume-run`. Dropping `total_timesteps` must still fail validation.
663    #[test]
664    fn validate_still_requires_total_timesteps_for_a_ppo_resume_run() {
665        let temp = TempDir::new().unwrap();
666        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
667
668        let missing_total_timesteps = JobDefinition {
669            flow_type: FlowType::ResumeRun,
670            n_steps: None,
671            total_timesteps: None,
672            ..ppo_job_with_hyperparams_set()
673        };
674        let err = missing_total_timesteps
675            .validate(temp.path(), Some("run-123"))
676            .unwrap_err();
677        assert!(
678            err.to_string().contains("total_timesteps"),
679            "expected error to mention total_timesteps, got: {err}"
680        );
681        assert!(
682            !err.to_string().contains("n_steps"),
683            "n_steps should remain exempt for a ppo resume-run, got: {err}"
684        );
685    }
686
687    #[test]
688    fn load_job_catalog_returns_empty_when_directory_missing() {
689        let temp = TempDir::new().unwrap();
690        let jobs = load_job_catalog(temp.path()).unwrap();
691        assert!(jobs.is_empty());
692    }
693
694    #[test]
695    fn load_job_catalog_parses_yaml_files_sorted_by_name() {
696        let temp = TempDir::new().unwrap();
697        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
698        std::fs::create_dir_all(&jobs_dir).unwrap();
699
700        write_job(
701            &jobs_dir,
702            "train.yml",
703            "name: train\nflow_type: train\nalgorithm: ppo\ntotal_timesteps: 1000000\n",
704        );
705        write_job(
706            &jobs_dir,
707            "hypertune.yml",
708            "name: hypertune\nflow_type: tune\nalgorithm: ppo\nn_trials: 10\n",
709        );
710
711        let jobs = load_job_catalog(temp.path()).unwrap();
712
713        assert_eq!(jobs.len(), 2);
714        assert_eq!(jobs[0].name, "hypertune");
715        assert_eq!(jobs[0].flow_type, FlowType::Tune);
716        assert_eq!(jobs[1].name, "train");
717        assert_eq!(jobs[1].flow_type, FlowType::Train);
718        assert_eq!(jobs[1].total_timesteps, Some(1000000));
719    }
720
721    #[test]
722    fn load_job_catalog_ignores_non_yaml_files() {
723        let temp = TempDir::new().unwrap();
724        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
725        std::fs::create_dir_all(&jobs_dir).unwrap();
726
727        write_job(&jobs_dir, "README.md", "not a job");
728        write_job(
729            &jobs_dir,
730            "train.yml",
731            "name: train\nflow_type: train\nalgorithm: ppo\n",
732        );
733
734        let jobs = load_job_catalog(temp.path()).unwrap();
735        assert_eq!(jobs.len(), 1);
736    }
737
738    #[test]
739    fn supports_remote_for_train_resume_run_and_tune() {
740        let base = base_job();
741        assert!(base.supports_remote());
742
743        let resume = JobDefinition {
744            flow_type: FlowType::ResumeRun,
745            ..base.clone()
746        };
747        assert!(resume.supports_remote());
748
749        let tune = JobDefinition {
750            flow_type: FlowType::Tune,
751            ..base.clone()
752        };
753        assert!(tune.supports_remote());
754
755        let tune_then_train = JobDefinition {
756            flow_type: FlowType::TuneThenTrain,
757            ..base.clone()
758        };
759        assert!(!tune_then_train.supports_remote());
760
761        let train_from_tune = JobDefinition {
762            flow_type: FlowType::TrainFromTune,
763            ..base
764        };
765        assert!(!train_from_tune.supports_remote());
766    }
767
768    #[test]
769    fn to_create_job_request_forwards_resume_run_id() {
770        let temp = TempDir::new().unwrap();
771        let jobs_dir = temp.path().join("configs/rl/jobs");
772        std::fs::create_dir_all(&jobs_dir).unwrap();
773        let source_path = jobs_dir.join("resume-train.yml");
774        std::fs::write(&source_path, "name: resume-train\nflow_type: resume-run\n").unwrap();
775
776        let job = JobDefinition {
777            name: "resume-train".to_string(),
778            flow_type: FlowType::ResumeRun,
779            source_path,
780            ..base_job()
781        };
782
783        let req = job
784            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("  run-abc123  "))
785            .unwrap();
786
787        assert_eq!(req.resume_run_id.as_deref(), Some("run-abc123"));
788    }
789
790    #[test]
791    fn to_create_job_request_omits_empty_resume_run_id() {
792        let temp = TempDir::new().unwrap();
793        let jobs_dir = temp.path().join("configs/rl/jobs");
794        std::fs::create_dir_all(&jobs_dir).unwrap();
795        let source_path = jobs_dir.join("resume-train.yml");
796        std::fs::write(&source_path, "name: resume-train\nflow_type: resume-run\n").unwrap();
797
798        let job = JobDefinition {
799            name: "resume-train".to_string(),
800            flow_type: FlowType::ResumeRun,
801            source_path,
802            ..base_job()
803        };
804
805        let req = job
806            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("   "))
807            .unwrap();
808
809        assert_eq!(req.resume_run_id, None);
810    }
811
812    /// Contract test against `mecha10-job-validation`'s `JobField` list: every field applicable
813    /// to a remotely-dispatchable flow type must have a matching wire key on this struct.
814    #[test]
815    fn create_job_request_has_a_field_for_every_dispatchable_job_field() {
816        let dispatchable = [FlowType::Train, FlowType::ResumeRun, FlowType::Tune];
817
818        let req = CreateJobRequest {
819            name: "job".to_string(),
820            description: None,
821            algorithm: "ppo".to_string(),
822            robot_id: "robot-1".to_string(),
823            project_id: None,
824            project_name: None,
825            env_name: None,
826            task_name: Some("find_target".to_string()),
827            total_timesteps: Some(1_000_000),
828            n_steps: None,
829            n_epochs: None,
830            learning_rate: None,
831            patience: None,
832            min_delta: None,
833            n_envs: None,
834            config_yaml: None,
835            resume_run_id: Some("run-123".to_string()),
836            n_trials: Some(10),
837            tune_timesteps: Some(500_000),
838            eval_eps: Some(5),
839            search_space: None,
840        };
841        let json = serde_json::to_value(&req).expect("CreateJobRequest serializes");
842
843        for field in mecha10_job_validation::all_fields() {
844            let applicable = dispatchable
845                .iter()
846                .any(|&ft| field.requirement(ft) != mecha10_job_validation::FieldRequirement::NotApplicable);
847            if applicable {
848                assert!(
849                    json.get(field.wire_name()).is_some(),
850                    "CreateJobRequest has no '{}' field for {field:?}, which is applicable to a \
851                     dispatchable flow type",
852                    field.wire_name()
853                );
854            }
855        }
856    }
857
858    #[test]
859    fn to_create_job_request_forwards_tune_fields() {
860        let temp = TempDir::new().unwrap();
861        let jobs_dir = temp.path().join("configs/rl/jobs");
862        std::fs::create_dir_all(&jobs_dir).unwrap();
863        let source_path = jobs_dir.join("hypertune.yml");
864        std::fs::write(&source_path, "name: hypertune\nflow_type: tune\n").unwrap();
865
866        let job = JobDefinition {
867            name: "hypertune".to_string(),
868            flow_type: FlowType::Tune,
869            n_trials: Some(20),
870            tune_timesteps: Some(250_000),
871            eval_eps: Some(3),
872            source_path,
873            ..base_job()
874        };
875
876        let req = job
877            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
878            .unwrap();
879
880        assert_eq!(req.n_trials, Some(20));
881        assert_eq!(req.tune_timesteps, Some(250_000));
882        assert_eq!(req.eval_eps, Some(3));
883    }
884
885    #[test]
886    fn to_create_job_request_forwards_n_envs() {
887        let temp = TempDir::new().unwrap();
888        let jobs_dir = temp.path().join("configs/rl/jobs");
889        std::fs::create_dir_all(&jobs_dir).unwrap();
890        let source_path = jobs_dir.join("train.yml");
891        std::fs::write(&source_path, "name: train\nflow_type: train\nn_envs: 8\n").unwrap();
892
893        let job = JobDefinition {
894            name: "train".to_string(),
895            n_envs: Some(8),
896            source_path,
897            ..base_job()
898        };
899
900        let req = job
901            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
902            .unwrap();
903
904        assert_eq!(req.n_envs, Some(8));
905    }
906
907    #[test]
908    fn to_create_job_request_forwards_search_space() {
909        let temp = TempDir::new().unwrap();
910        let jobs_dir = temp.path().join("configs/rl/jobs");
911        std::fs::create_dir_all(&jobs_dir).unwrap();
912        let source_path = jobs_dir.join("hypertune.yml");
913        let yaml = "name: hypertune\nflow_type: tune\nsearch_space:\n  - name: learning_rate\n    \
914                    type: float\n    low: 1e-5\n    high: 1e-3\n    log: true\n  - name: n_steps\n    \
915                    type: categorical\n    choices: [1024, 2048, 4096]\n  - name: n_epochs\n    \
916                    type: int\n    low: 5\n    high: 20\n";
917        std::fs::write(&source_path, yaml).unwrap();
918        let job: JobDefinition = serde_yaml::from_str(yaml).unwrap();
919        let job = JobDefinition { source_path, ..job };
920
921        let req = job
922            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
923            .unwrap();
924
925        let search_space = req.search_space.expect("search_space forwarded");
926        assert_eq!(search_space.len(), 3);
927        assert_eq!(search_space[0].name, "learning_rate");
928        assert_eq!(search_space[0].type_, SearchSpaceType::Float);
929        assert_eq!(search_space[0].low, Some(1e-5));
930        assert_eq!(search_space[0].high, Some(1e-3));
931        assert_eq!(search_space[0].log, Some(true));
932        assert_eq!(search_space[1].name, "n_steps");
933        assert_eq!(search_space[1].type_, SearchSpaceType::Categorical);
934        assert_eq!(
935            search_space[1].choices,
936            Some(vec![
937                serde_json::json!(1024),
938                serde_json::json!(2048),
939                serde_json::json!(4096)
940            ])
941        );
942        assert_eq!(search_space[2].name, "n_epochs");
943        assert_eq!(search_space[2].type_, SearchSpaceType::Int);
944        assert_eq!(search_space[2].low, Some(5.0));
945        assert_eq!(search_space[2].high, Some(20.0));
946    }
947
948    #[test]
949    fn to_create_job_request_omits_search_space_when_unset() {
950        let temp = TempDir::new().unwrap();
951        let jobs_dir = temp.path().join("configs/rl/jobs");
952        std::fs::create_dir_all(&jobs_dir).unwrap();
953        let source_path = jobs_dir.join("train.yml");
954        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
955
956        let job = JobDefinition {
957            name: "train".to_string(),
958            source_path,
959            ..base_job()
960        };
961
962        let req = job
963            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
964            .unwrap();
965
966        assert_eq!(req.search_space, None);
967    }
968
969    #[test]
970    fn to_create_job_request_drops_sentinel_values() {
971        let temp = TempDir::new().unwrap();
972        let jobs_dir = temp.path().join("configs/rl/jobs");
973        std::fs::create_dir_all(&jobs_dir).unwrap();
974        let source_path = jobs_dir.join("train.yml");
975        std::fs::write(&source_path, "name: train\nflow_type: train\nenv_name: rover\n").unwrap();
976
977        let job = JobDefinition {
978            name: "train".to_string(),
979            description: Some("desc".to_string()),
980            env_name: Some("rover".to_string()),
981            total_timesteps: Some(-1),
982            n_steps: Some(2048),
983            n_epochs: Some(-1),
984            learning_rate: Some(-1.0),
985            patience: Some(50),
986            min_delta: Some(-1.0),
987            n_envs: Some(-1),
988            source_path,
989            ..base_job()
990        };
991
992        let req = job
993            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
994            .unwrap();
995
996        assert_eq!(req.name, "train");
997        assert_eq!(req.algorithm, "ppo");
998        assert_eq!(req.robot_id, "my-project");
999        assert_eq!(req.project_id.as_deref(), Some("my-project"));
1000        assert_eq!(req.project_name.as_deref(), Some("my-robot"));
1001        assert_eq!(req.total_timesteps, None);
1002        assert_eq!(req.n_steps, Some(2048));
1003        assert_eq!(req.n_epochs, None);
1004        assert_eq!(req.learning_rate, None);
1005        assert_eq!(req.patience, Some(50));
1006        assert_eq!(req.min_delta, None);
1007        assert_eq!(req.n_envs, None);
1008        assert_eq!(
1009            req.config_yaml.as_deref(),
1010            Some("name: train\nflow_type: train\nenv_name: rover\n")
1011        );
1012        assert_eq!(req.task_name, None);
1013    }
1014
1015    #[test]
1016    fn to_create_job_request_forwards_task_name() {
1017        let temp = TempDir::new().unwrap();
1018        let jobs_dir = temp.path().join("configs/rl/jobs");
1019        std::fs::create_dir_all(&jobs_dir).unwrap();
1020        let source_path = jobs_dir.join("train.yml");
1021        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
1022
1023        let job = JobDefinition {
1024            name: "train".to_string(),
1025            task_name: Some("navigate".to_string()),
1026            source_path,
1027            ..base_job()
1028        };
1029
1030        let req = job
1031            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1032            .unwrap();
1033
1034        assert_eq!(req.task_name.as_deref(), Some("navigate"));
1035    }
1036
1037    #[test]
1038    fn to_create_job_request_errors_when_the_jobs_own_file_is_missing() {
1039        let temp = TempDir::new().unwrap();
1040        let job = JobDefinition {
1041            name: "train".to_string(),
1042            source_path: temp.path().join("configs/rl/jobs/train.yml"),
1043            ..base_job()
1044        };
1045
1046        let err = job
1047            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1048            .unwrap_err();
1049        assert!(err.to_string().contains("configs/rl/jobs/train.yml"));
1050    }
1051}
mecha10_cli/services/train_jobs.rs

mecha10_cli/services/
train_jobs.rs
