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    /// Total environment steps to train for.
75    #[serde(default)]
76    #[schemars(example = Some(1_000_000))]
77    pub total_timesteps: Option<i64>,
78    /// Optimizer learning rate.
79    #[serde(default)]
80    #[schemars(example = Some(0.0003))]
81    pub learning_rate: Option<f64>,
82    #[serde(default)]
83    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
84    pub linear_lr_schedule: Option<bool>,
85    /// Number of environment steps collected per environment before each policy update.
86    #[serde(default)]
87    #[schemars(example = Some(2048))]
88    pub n_steps: Option<i64>,
89    /// Minibatch size for each gradient update.
90    #[serde(default)]
91    #[schemars(example = Some(64))]
92    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
93    pub batch_size: Option<i64>,
94    /// Number of epochs over each collected rollout per policy update.
95    #[serde(default)]
96    #[schemars(example = Some(10))]
97    pub n_epochs: Option<i64>,
98    /// Discount factor for future rewards.
99    #[serde(default)]
100    #[schemars(example = Some(0.99))]
101    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
102    pub gamma: Option<f64>,
103    /// GAE lambda - trades off bias vs. variance in advantage estimation.
104    #[serde(default)]
105    #[schemars(example = Some(0.95))]
106    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
107    pub gae_lambda: Option<f64>,
108    /// PPO's policy-update clipping range.
109    #[serde(default)]
110    #[schemars(example = Some(0.2))]
111    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
112    pub clip_range: Option<f64>,
113    /// Entropy bonus coefficient - encourages exploration.
114    #[serde(default)]
115    #[schemars(example = Some(0.0))]
116    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
117    pub ent_coef: Option<f64>,
118    /// Value-function loss coefficient.
119    #[serde(default)]
120    #[schemars(example = Some(0.5))]
121    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
122    pub vf_coef: Option<f64>,
123    /// Maximum gradient norm for gradient clipping.
124    #[serde(default)]
125    #[schemars(example = Some(0.5))]
126    #[allow(dead_code, reason = "schema/deserialize-only, see `policy`'s doc comment above")]
127    pub max_grad_norm: Option<f64>,
128    /// Number of evaluations with no improvement to wait before stopping training early.
129    #[serde(default)]
130    #[schemars(example = Some(10))]
131    pub patience: Option<i64>,
132    /// Minimum improvement in the evaluation metric to count as progress, paired with `patience`
133    /// for early stopping.
134    #[serde(default)]
135    #[schemars(example = Some(0.01))]
136    pub min_delta: Option<f64>,
137    /// Number of parallel environments to run during training.
138    #[serde(default)]
139    #[schemars(example = Some(8))]
140    pub n_envs: Option<i64>,
141
142    /// Number of hyperparameter-search trials to run for a tuning job.
143    #[serde(default)]
144    #[schemars(example = Some(20))]
145    pub n_trials: Option<i64>,
146    /// Per-trial timestep budget during a tuning job - typically much shorter than a full
147    /// training run's `total_timesteps`.
148    #[serde(default)]
149    #[schemars(example = Some(50_000))]
150    pub tune_timesteps: Option<i64>,
151    /// Number of episodes to run per evaluation.
152    #[serde(default)]
153    #[schemars(example = Some(5))]
154    pub eval_eps: Option<i64>,
155    /// Hyperparameter search space to explore for a tuning job. If omitted, a default search
156    /// space is used.
157    #[serde(default)]
158    pub search_space: Option<Vec<SearchSpaceEntry>>,
159
160    #[serde(skip)]
161    #[schemars(skip)]
162    pub source_path: PathBuf,
163}
164
165#[derive(Debug, Clone, Serialize)]
166pub struct CreateJobRequest {
167    pub name: String,
168    #[serde(skip_serializing_if = "Option::is_none")]
169    pub description: Option<String>,
170    pub algorithm: String,
171    pub robot_id: String,
172    #[serde(skip_serializing_if = "Option::is_none")]
173    pub project_id: Option<String>,
174    #[serde(skip_serializing_if = "Option::is_none")]
175    pub project_name: Option<String>,
176    #[serde(skip_serializing_if = "Option::is_none")]
177    pub env_name: Option<String>,
178    #[serde(skip_serializing_if = "Option::is_none")]
179    pub task_name: Option<String>,
180    #[serde(skip_serializing_if = "Option::is_none")]
181    pub total_timesteps: Option<i64>,
182    #[serde(skip_serializing_if = "Option::is_none")]
183    pub n_steps: Option<i64>,
184    #[serde(skip_serializing_if = "Option::is_none")]
185    pub n_epochs: Option<i64>,
186    #[serde(skip_serializing_if = "Option::is_none")]
187    pub learning_rate: Option<f64>,
188    #[serde(skip_serializing_if = "Option::is_none")]
189    pub patience: Option<i64>,
190    #[serde(skip_serializing_if = "Option::is_none")]
191    pub min_delta: Option<f64>,
192    #[serde(skip_serializing_if = "Option::is_none")]
193    pub n_envs: Option<i64>,
194    #[serde(skip_serializing_if = "Option::is_none")]
195    pub config_yaml: Option<String>,
196    #[serde(skip_serializing_if = "Option::is_none")]
197    pub resume_run_id: Option<String>,
198    #[serde(skip_serializing_if = "Option::is_none")]
199    pub n_trials: Option<i64>,
200    #[serde(skip_serializing_if = "Option::is_none")]
201    pub tune_timesteps: Option<i64>,
202    #[serde(skip_serializing_if = "Option::is_none")]
203    pub eval_eps: Option<i64>,
204    #[serde(skip_serializing_if = "Option::is_none")]
205    pub search_space: Option<Vec<SearchSpaceEntry>>,
206}
207
208/// Read `configs/rl/tasks/<task_name>/task.json`'s `env_name` field, relative to `project_root`.
209/// Returns `None` on any failure (missing file, invalid JSON, missing/non-string field).
210fn env_name_from_task_json(project_root: &Path, task_name: &str) -> Option<String> {
211    task_json_field(project_root, task_name, "env_name")
212}
213
214fn task_json_field(project_root: &Path, task_name: &str, field: &str) -> Option<String> {
215    let path = project_root
216        .join(mecha10_cli_core::paths::project::RL_TASKS_DIR)
217        .join(task_name)
218        .join("task.json");
219    let content = std::fs::read_to_string(path).ok()?;
220    let value: serde_json::Value = serde_json::from_str(&content).ok()?;
221    value.get(field)?.as_str().map(str::to_string)
222}
223
224/// Drop `value` if it equals `sentinel` - the "unset" placeholder some YAML/env fields use
225/// instead of a real `null` (e.g. `-1` / `-1.0`).
226///
227/// This is a *wire-building* helper, used only by [`JobDefinition::to_create_job_request`] to
228/// decide what literal value to put on the dispatch request body: a field left at its
229/// placeholder is omitted so the remote side applies its own default, rather than being sent
230/// `-1` as if that were a real number. It is deliberately **not** used for validation
231/// presence/requiredness checks anymore - [`JobDefinition::validation_fields`]/
232/// [`JobDefinition::validation_fields_for_algorithm`] pass raw values straight through to
233/// `mecha10-job-validation`, which decides per-field (via its own schema, not a blanket strip
234/// applied by every caller) whether a given field's `-1`/`-1.0` is a meaningful value (e.g.
235/// `patience`/`min_delta` for `ppo`, where it means "disabled") or a "not set" placeholder. Using
236/// this same blanket helper for *both* purposes used to be the bug: it stripped `patience`/
237/// `min_delta`'s real `-1`/`-1.0` before validation ever saw it, making a correctly-configured
238/// `ppo` job (the scaffolded `train.yml` template's own `patience: -1`/`min_delta: -1.0`) look
239/// like it was missing required fields. See PR #1639 (an ad hoc, special-cased attempt at fixing
240/// just this) and its revert PR #1641 (determined this needed real per-field schema design).
241fn non_sentinel<T: PartialEq>(value: Option<T>, sentinel: T) -> Option<T> {
242    value.filter(|v| *v != sentinel)
243}
244
245fn non_sentinel_i64(value: Option<i64>) -> Option<i64> {
246    non_sentinel(value, -1)
247}
248
249fn non_sentinel_f64(value: Option<f64>) -> Option<f64> {
250    non_sentinel(value, -1.0)
251}
252
253impl JobDefinition {
254    /// This job's fields in `mecha10-job-validation`'s shared vocabulary (see
255    /// [`mecha10_job_validation::JobFields`]). Raw values, not sentinel-stripped: whether a
256    /// `-1`/`-1.0` on a given field counts as "set" or "absent" is a per-field schema question
257    /// `mecha10-job-validation` answers itself (see its `SentinelConvention`), not something a
258    /// caller should pre-decide by stripping the value before validation ever sees it - see
259    /// [`non_sentinel_i64`]'s doc comment for why that used to be wrong.
260    fn validation_fields(&self, resume_run_id: Option<&str>) -> mecha10_job_validation::JobFields {
261        mecha10_job_validation::JobFields {
262            task_name: self.task_name.clone(),
263            algorithm: self.algorithm.clone(),
264            resume_run_id: resume_run_id.map(str::to_string),
265            tune_run_id: None,
266            n_trials: self.n_trials,
267            tune_timesteps: self.tune_timesteps,
268            eval_eps: self.eval_eps,
269        }
270    }
271
272    /// This job's per-algorithm hyperparameter fields in `mecha10-job-validation`'s shared
273    /// vocabulary (see [`mecha10_job_validation::AlgorithmFields`]) - the by-algorithm
274    /// counterpart to [`Self::validation_fields`]. Also raw, unstripped values, for the same
275    /// reason: `patience`/`min_delta`'s `-1`/`-1.0` is a real "disabled" value for `ppo` (the
276    /// scaffolded `train.yml` template's own setting), and only `mecha10-job-validation`'s schema
277    /// knows that - stripping it here before validation runs is exactly the bug this method used
278    /// to have (see [`non_sentinel_i64`]'s doc comment).
279    fn validation_fields_for_algorithm(&self) -> mecha10_job_validation::AlgorithmFields {
280        mecha10_job_validation::AlgorithmFields {
281            total_timesteps: self.total_timesteps,
282            learning_rate: self.learning_rate,
283            n_steps: self.n_steps,
284            n_epochs: self.n_epochs,
285            patience: self.patience,
286            min_delta: self.min_delta,
287        }
288    }
289
290    /// Validate this job against `mecha10-job-validation`'s flow_type requirement table and its
291    /// by-algorithm hyperparameter requirement table, plus a `task_name`-exists check against
292    /// the project's own task catalog. Algorithm-awareness (e.g. `ppo` needing `n_steps`, `act`
293    /// needing none of these fields) lives entirely in `mecha10-job-validation`'s
294    /// `AlgorithmField` table - this method only composes the two generic validators' results,
295    /// the same way `compose_node_schema` composes a base schema with a node's own custom
296    /// schema rather than validating them separately.
297    pub fn validate(&self, project_root: &Path, resume_run_id: Option<&str>) -> Result<()> {
298        let missing_basic =
299            mecha10_job_validation::validate_job(self.flow_type, &self.validation_fields(resume_run_id));
300        let missing_algorithm = mecha10_job_validation::validate_algorithm_fields(
301            self.algorithm.as_deref().unwrap_or_default(),
302            self.flow_type,
303            &self.validation_fields_for_algorithm(),
304        );
305
306        if !missing_basic.is_empty() || !missing_algorithm.is_empty() {
307            let details = missing_basic
308                .iter()
309                .map(|m| m.to_string())
310                .chain(missing_algorithm.iter().map(|m| m.to_string()))
311                .collect::<Vec<_>>()
312                .join("; ");
313            bail!("job '{}' is missing required fields: {details}", self.name);
314        }
315
316        if let Some(task_name) = self.task_name.as_deref() {
317            mecha10_job_validation::task_exists(project_root, task_name)
318                .with_context(|| format!("job '{}' has an invalid task_name", self.name))?;
319        }
320
321        Ok(())
322    }
323
324    /// Resolve which `mecha10_rl.envs` registry name this job targets: an explicit `env_name`
325    /// wins, else it's read from `task_name`'s own `task.json`. `None` when neither is available.
326    pub fn resolve_env_name(&self, project_root: &Path) -> Option<String> {
327        self.env_name.clone().or_else(|| {
328            self.task_name
329                .as_deref()
330                .and_then(|task_name| env_name_from_task_json(project_root, task_name))
331        })
332    }
333
334    /// Whether this job can be dispatched remotely with today's `CreateJobRequest`/`create_job`
335    /// contract. `tune-then-train`/`train-from-tune` aren't dispatchable yet and stay local-only.
336    pub fn supports_remote(&self) -> bool {
337        matches!(self.flow_type, FlowType::Train | FlowType::ResumeRun | FlowType::Tune)
338    }
339
340    /// Build the remote dispatch request body. Only meaningful when `supports_remote()` is true;
341    /// callers should check that first.
342    pub fn to_create_job_request(
343        &self,
344        project_root: &Path,
345        robot_id: &str,
346        project_name: &str,
347        resume_run_id: Option<&str>,
348    ) -> Result<CreateJobRequest> {
349        let config_yaml = std::fs::read_to_string(&self.source_path).with_context(|| {
350            format!(
351                "failed to read {} - this job definition's own file must exist before dispatching remotely",
352                self.source_path.display()
353            )
354        })?;
355
356        Ok(CreateJobRequest {
357            name: self.name.clone(),
358            description: self.description.clone(),
359            algorithm: self.algorithm.clone().unwrap_or_default(),
360            robot_id: robot_id.to_string(),
361            project_id: Some(robot_id.to_string()),
362            project_name: Some(project_name.to_string()),
363            env_name: self.resolve_env_name(project_root),
364            task_name: self.task_name.clone(),
365            total_timesteps: non_sentinel_i64(self.total_timesteps),
366            n_steps: non_sentinel_i64(self.n_steps),
367            n_epochs: non_sentinel_i64(self.n_epochs),
368            learning_rate: non_sentinel_f64(self.learning_rate),
369            patience: non_sentinel_i64(self.patience),
370            min_delta: non_sentinel_f64(self.min_delta),
371            n_envs: non_sentinel_i64(self.n_envs),
372            config_yaml: Some(config_yaml),
373            resume_run_id: resume_run_id
374                .map(str::trim)
375                .filter(|id| !id.is_empty())
376                .map(str::to_string),
377            n_trials: non_sentinel_i64(self.n_trials),
378            tune_timesteps: non_sentinel_i64(self.tune_timesteps),
379            eval_eps: non_sentinel_i64(self.eval_eps),
380            search_space: self.search_space.clone(),
381        })
382    }
383}
384
385/// Load every `.yml`/`.yaml` job definition under `{project_root}/configs/rl/jobs/`, sorted by
386/// file name for a stable menu order. Returns an empty list (not an error) when the directory
387/// doesn't exist - `mecha10 train` treats that as "no jobs configured yet" and points the user at
388/// `mecha10 init`/the bundled defaults rather than failing.
389pub fn load_job_catalog(project_root: &Path) -> Result<Vec<JobDefinition>> {
390    let jobs_dir = project_root.join(JOB_CATALOG_DIR);
391    if !jobs_dir.exists() {
392        return Ok(Vec::new());
393    }
394
395    let mut paths: Vec<PathBuf> = std::fs::read_dir(&jobs_dir)
396        .with_context(|| format!("failed to read {}", jobs_dir.display()))?
397        .filter_map(|entry| entry.ok())
398        .map(|entry| entry.path())
399        .filter(|path| {
400            path.extension()
401                .and_then(|ext| ext.to_str())
402                .is_some_and(|ext| ext == "yml" || ext == "yaml")
403        })
404        .collect();
405    paths.sort();
406
407    let mut jobs = Vec::with_capacity(paths.len());
408    for path in paths {
409        let content = std::fs::read_to_string(&path).with_context(|| format!("failed to read {}", path.display()))?;
410        let mut job: JobDefinition = serde_yaml::from_str(&content)
411            .with_context(|| format!("failed to parse job definition {}", path.display()))?;
412        job.source_path = path;
413        jobs.push(job);
414    }
415
416    Ok(jobs)
417}
418
419#[cfg(test)]
420mod tests {
421    use super::*;
422    use tempfile::TempDir;
423
424    fn write_job(dir: &Path, file_name: &str, contents: &str) {
425        std::fs::write(dir.join(file_name), contents).unwrap();
426    }
427
428    fn write_task_json(project_root: &Path, task_name: &str, contents: &str) {
429        let task_dir = project_root
430            .join(mecha10_cli_core::paths::project::RL_TASKS_DIR)
431            .join(task_name);
432        std::fs::create_dir_all(&task_dir).unwrap();
433        std::fs::write(task_dir.join("task.json"), contents).unwrap();
434    }
435
436    #[test]
437    fn resolve_env_name_prefers_an_explicit_env_name_over_the_tasks_own() {
438        let dir = TempDir::new().unwrap();
439        write_task_json(dir.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
440
441        let job = JobDefinition {
442            env_name: Some("rover_arena".to_string()),
443            task_name: Some("find_target".to_string()),
444            ..base_job()
445        };
446        assert_eq!(job.resolve_env_name(dir.path()), Some("rover_arena".to_string()));
447    }
448
449    #[test]
450    fn resolve_env_name_falls_back_to_the_tasks_own_task_json() {
451        let dir = TempDir::new().unwrap();
452        write_task_json(
453            dir.path(),
454            "find_target",
455            r#"{"env_name": "basic_arena", "model": "humanoid"}"#,
456        );
457
458        let job = JobDefinition {
459            task_name: Some("find_target".to_string()),
460            ..base_job()
461        };
462        assert_eq!(job.resolve_env_name(dir.path()), Some("basic_arena".to_string()));
463    }
464
465    #[test]
466    fn resolve_env_name_is_none_when_task_has_no_env_name_field() {
467        let dir = TempDir::new().unwrap();
468        write_task_json(dir.path(), "find_target", r#"{"model": "humanoid"}"#);
469
470        let job = JobDefinition {
471            task_name: Some("find_target".to_string()),
472            ..base_job()
473        };
474        assert_eq!(job.resolve_env_name(dir.path()), None);
475    }
476
477    #[test]
478    fn resolve_env_name_is_none_without_an_env_name_or_a_task_name() {
479        let dir = TempDir::new().unwrap();
480        let job = base_job();
481        assert_eq!(job.resolve_env_name(dir.path()), None);
482    }
483
484    /// A minimal, fully-`None`/default `JobDefinition` for tests to `..base_job()` from, so each
485    /// test only spells out the fields it actually cares about.
486    fn base_job() -> JobDefinition {
487        JobDefinition {
488            name: "job".to_string(),
489            description: None,
490            flow_type: FlowType::Train,
491            env_name: None,
492            task_name: None,
493            algorithm: Some("ppo".to_string()),
494            policy: None,
495            seed: None,
496            environment_map: None,
497            total_timesteps: None,
498            learning_rate: None,
499            linear_lr_schedule: None,
500            n_steps: None,
501            batch_size: None,
502            n_epochs: None,
503            gamma: None,
504            gae_lambda: None,
505            clip_range: None,
506            ent_coef: None,
507            vf_coef: None,
508            max_grad_norm: None,
509            patience: None,
510            min_delta: None,
511            n_envs: None,
512            n_trials: None,
513            tune_timesteps: None,
514            eval_eps: None,
515            search_space: None,
516            source_path: PathBuf::new(),
517        }
518    }
519
520    /// A `ppo` job with every one of its required hyperparameters set (per
521    /// `mecha10_job_validation::AlgorithmField::requirement`) except `total_timesteps`/
522    /// `n_steps`, which individual tests below null out to isolate the field they're checking.
523    fn ppo_job_with_hyperparams_set() -> JobDefinition {
524        JobDefinition {
525            task_name: Some("find_target".to_string()),
526            algorithm: Some("ppo".to_string()),
527            total_timesteps: Some(1000),
528            learning_rate: Some(0.0003),
529            n_steps: Some(2048),
530            n_epochs: Some(10),
531            patience: Some(10),
532            min_delta: Some(0.01),
533            ..base_job()
534        }
535    }
536
537    #[test]
538    fn validate_does_not_require_any_hyperparameter_fields_for_a_non_ppo_algorithm() {
539        let temp = TempDir::new().unwrap();
540        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
541
542        let job = JobDefinition {
543            task_name: Some("find_target".to_string()),
544            algorithm: Some("act".to_string()),
545            total_timesteps: Some(1000),
546            n_steps: None,
547            ..base_job()
548        };
549
550        assert!(job.validate(temp.path(), None).is_ok());
551    }
552
553    /// Regression test for the bug this crate's base/by-algorithm split exists to prevent from
554    /// recurring: `act` has no `total_timesteps` concept at all (it reads `total_steps`
555    /// instead, an untyped field today), so a `Train`-flow `act` job that omits
556    /// `total_timesteps` entirely must still pass `validate()` - it used to fail, because the
557    /// old hand-rolled check required `total_timesteps` unconditionally for every algorithm.
558    #[test]
559    fn validate_does_not_require_total_timesteps_for_an_act_train_job() {
560        let temp = TempDir::new().unwrap();
561        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
562
563        let job = JobDefinition {
564            flow_type: FlowType::Train,
565            task_name: Some("find_target".to_string()),
566            algorithm: Some("act".to_string()),
567            total_timesteps: None,
568            n_steps: None,
569            ..base_job()
570        };
571
572        assert!(job.validate(temp.path(), None).is_ok());
573    }
574
575    #[test]
576    fn validate_still_requires_n_steps_for_ppo() {
577        let temp = TempDir::new().unwrap();
578        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
579
580        let job = JobDefinition {
581            n_steps: None,
582            ..ppo_job_with_hyperparams_set()
583        };
584
585        let err = job.validate(temp.path(), None).unwrap_err();
586        assert!(
587            err.to_string().contains("n_steps"),
588            "expected error to mention n_steps, got: {err}"
589        );
590    }
591
592    #[test]
593    fn validate_passes_for_ppo_with_every_required_hyperparameter_set() {
594        let temp = TempDir::new().unwrap();
595        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
596
597        let job = ppo_job_with_hyperparams_set();
598
599        assert!(job.validate(temp.path(), None).is_ok());
600    }
601
602    /// Regression test for the original reported bug: the scaffolded `train.yml` template sets
603    /// `patience: -1` / `min_delta: -1.0` to mean "no early-stopping limit" - a real, intended
604    /// value, not an unset field. `validate()` must accept that, not report `patience`/
605    /// `min_delta` as missing the way it used to when every numeric field (including these two)
606    /// was blanket-stripped of `-1`/`-1.0` before validation ever ran. See PR #1639/#1641 for the
607    /// reverted ad hoc attempt at fixing just this.
608    #[test]
609    fn validate_passes_for_ppo_with_patience_and_min_delta_at_their_disabled_sentinel_value() {
610        let temp = TempDir::new().unwrap();
611        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
612
613        let job = JobDefinition {
614            patience: Some(-1),
615            min_delta: Some(-1.0),
616            ..ppo_job_with_hyperparams_set()
617        };
618
619        assert!(
620            job.validate(temp.path(), None).is_ok(),
621            "a ppo job with patience: -1 / min_delta: -1.0 (the scaffolded train.yml template's \
622             own values) must validate - those are meaningful 'disabled' values, not absence"
623        );
624    }
625
626    /// The flip side of the regression test above: `total_timesteps` has no sentinel-for-disabled
627    /// convention, so `-1` there must still count as absent/missing, the same as if it were
628    /// never set at all. This guards against over-generalizing "sentinels are fine everywhere"
629    /// while fixing the `patience`/`min_delta` bug.
630    #[test]
631    fn validate_still_rejects_total_timesteps_set_to_the_placeholder_sentinel_value() {
632        let temp = TempDir::new().unwrap();
633        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
634
635        let job = JobDefinition {
636            total_timesteps: Some(-1),
637            ..ppo_job_with_hyperparams_set()
638        };
639
640        let err = job.validate(temp.path(), None).unwrap_err();
641        assert!(
642            err.to_string().contains("total_timesteps"),
643            "expected error to mention total_timesteps, got: {err}"
644        );
645    }
646
647    /// `act` has no opinion about `patience`/`min_delta` at all (every field is `NotApplicable`
648    /// for it, per `mecha10_job_validation::AlgorithmField::requirement`), but the scaffolded
649    /// `train-act.yml` template sets them to `-1`/`-1.0` anyway (same as `train.yml`'s ppo job).
650    /// A correctly-configured `act` job must still validate regardless of what those two fields
651    /// are set to.
652    #[test]
653    fn validate_passes_for_act_with_patience_and_min_delta_set_to_the_sentinel_value() {
654        let temp = TempDir::new().unwrap();
655        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
656
657        let job = JobDefinition {
658            flow_type: FlowType::Train,
659            task_name: Some("find_target".to_string()),
660            algorithm: Some("act".to_string()),
661            total_timesteps: Some(1000),
662            patience: Some(-1),
663            min_delta: Some(-1.0),
664            ..base_job()
665        };
666
667        assert!(job.validate(temp.path(), None).is_ok());
668    }
669
670    /// A `ppo` `resume-run` job restores its rollout length (`n_steps`) from the checkpoint it
671    /// resumes, so `n_steps` specifically is *not* required for that flow - matching the real,
672    /// currently-shipping `resume-train.yml` job template (`algorithm: ppo`,
673    /// `flow_type: resume-run`, no `n_steps`). This carve-out is encoded in
674    /// `mecha10_job_validation::AlgorithmField::requirement`, which takes `flow_type` for exactly
675    /// this reason; it predates this crate as a hand-rolled `if` in the CLI
676    /// (`flow_type != ResumeRun && algorithm == "ppo" && n_steps missing`).
677    #[test]
678    fn validate_does_not_require_n_steps_for_a_ppo_resume_run() {
679        let temp = TempDir::new().unwrap();
680        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
681
682        let without_n_steps = JobDefinition {
683            flow_type: FlowType::ResumeRun,
684            n_steps: None,
685            ..ppo_job_with_hyperparams_set()
686        };
687        assert!(
688            without_n_steps.validate(temp.path(), Some("run-123")).is_ok(),
689            "a ppo resume-run job without n_steps must still validate"
690        );
691    }
692
693    /// The `n_steps` carve-out above is `n_steps`-specific: every *other* `ppo` hyperparameter is
694    /// still required for a `resume-run`. Dropping `total_timesteps` must still fail validation.
695    #[test]
696    fn validate_still_requires_total_timesteps_for_a_ppo_resume_run() {
697        let temp = TempDir::new().unwrap();
698        write_task_json(temp.path(), "find_target", r#"{"env_name": "basic_arena"}"#);
699
700        let missing_total_timesteps = JobDefinition {
701            flow_type: FlowType::ResumeRun,
702            n_steps: None,
703            total_timesteps: None,
704            ..ppo_job_with_hyperparams_set()
705        };
706        let err = missing_total_timesteps
707            .validate(temp.path(), Some("run-123"))
708            .unwrap_err();
709        assert!(
710            err.to_string().contains("total_timesteps"),
711            "expected error to mention total_timesteps, got: {err}"
712        );
713        assert!(
714            !err.to_string().contains("n_steps"),
715            "n_steps should remain exempt for a ppo resume-run, got: {err}"
716        );
717    }
718
719    #[test]
720    fn load_job_catalog_returns_empty_when_directory_missing() {
721        let temp = TempDir::new().unwrap();
722        let jobs = load_job_catalog(temp.path()).unwrap();
723        assert!(jobs.is_empty());
724    }
725
726    #[test]
727    fn load_job_catalog_parses_yaml_files_sorted_by_name() {
728        let temp = TempDir::new().unwrap();
729        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
730        std::fs::create_dir_all(&jobs_dir).unwrap();
731
732        write_job(
733            &jobs_dir,
734            "train.yml",
735            "name: train\nflow_type: train\nalgorithm: ppo\ntotal_timesteps: 1000000\n",
736        );
737        write_job(
738            &jobs_dir,
739            "hypertune.yml",
740            "name: hypertune\nflow_type: tune\nalgorithm: ppo\nn_trials: 10\n",
741        );
742
743        let jobs = load_job_catalog(temp.path()).unwrap();
744
745        assert_eq!(jobs.len(), 2);
746        assert_eq!(jobs[0].name, "hypertune");
747        assert_eq!(jobs[0].flow_type, FlowType::Tune);
748        assert_eq!(jobs[1].name, "train");
749        assert_eq!(jobs[1].flow_type, FlowType::Train);
750        assert_eq!(jobs[1].total_timesteps, Some(1000000));
751    }
752
753    #[test]
754    fn load_job_catalog_ignores_non_yaml_files() {
755        let temp = TempDir::new().unwrap();
756        let jobs_dir = temp.path().join(JOB_CATALOG_DIR);
757        std::fs::create_dir_all(&jobs_dir).unwrap();
758
759        write_job(&jobs_dir, "README.md", "not a job");
760        write_job(
761            &jobs_dir,
762            "train.yml",
763            "name: train\nflow_type: train\nalgorithm: ppo\n",
764        );
765
766        let jobs = load_job_catalog(temp.path()).unwrap();
767        assert_eq!(jobs.len(), 1);
768    }
769
770    #[test]
771    fn supports_remote_for_train_resume_run_and_tune() {
772        let base = base_job();
773        assert!(base.supports_remote());
774
775        let resume = JobDefinition {
776            flow_type: FlowType::ResumeRun,
777            ..base.clone()
778        };
779        assert!(resume.supports_remote());
780
781        let tune = JobDefinition {
782            flow_type: FlowType::Tune,
783            ..base.clone()
784        };
785        assert!(tune.supports_remote());
786
787        let tune_then_train = JobDefinition {
788            flow_type: FlowType::TuneThenTrain,
789            ..base.clone()
790        };
791        assert!(!tune_then_train.supports_remote());
792
793        let train_from_tune = JobDefinition {
794            flow_type: FlowType::TrainFromTune,
795            ..base
796        };
797        assert!(!train_from_tune.supports_remote());
798    }
799
800    #[test]
801    fn to_create_job_request_forwards_resume_run_id() {
802        let temp = TempDir::new().unwrap();
803        let jobs_dir = temp.path().join("configs/rl/jobs");
804        std::fs::create_dir_all(&jobs_dir).unwrap();
805        let source_path = jobs_dir.join("resume-train.yml");
806        std::fs::write(&source_path, "name: resume-train\nflow_type: resume-run\n").unwrap();
807
808        let job = JobDefinition {
809            name: "resume-train".to_string(),
810            flow_type: FlowType::ResumeRun,
811            source_path,
812            ..base_job()
813        };
814
815        let req = job
816            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("  run-abc123  "))
817            .unwrap();
818
819        assert_eq!(req.resume_run_id.as_deref(), Some("run-abc123"));
820    }
821
822    #[test]
823    fn to_create_job_request_omits_empty_resume_run_id() {
824        let temp = TempDir::new().unwrap();
825        let jobs_dir = temp.path().join("configs/rl/jobs");
826        std::fs::create_dir_all(&jobs_dir).unwrap();
827        let source_path = jobs_dir.join("resume-train.yml");
828        std::fs::write(&source_path, "name: resume-train\nflow_type: resume-run\n").unwrap();
829
830        let job = JobDefinition {
831            name: "resume-train".to_string(),
832            flow_type: FlowType::ResumeRun,
833            source_path,
834            ..base_job()
835        };
836
837        let req = job
838            .to_create_job_request(temp.path(), "my-project", "my-robot", Some("   "))
839            .unwrap();
840
841        assert_eq!(req.resume_run_id, None);
842    }
843
844    /// Contract test against `mecha10-job-validation`'s `JobField` list: every field applicable
845    /// to a remotely-dispatchable flow type must have a matching wire key on this struct.
846    #[test]
847    fn create_job_request_has_a_field_for_every_dispatchable_job_field() {
848        let dispatchable = [FlowType::Train, FlowType::ResumeRun, FlowType::Tune];
849
850        let req = CreateJobRequest {
851            name: "job".to_string(),
852            description: None,
853            algorithm: "ppo".to_string(),
854            robot_id: "robot-1".to_string(),
855            project_id: None,
856            project_name: None,
857            env_name: None,
858            task_name: Some("find_target".to_string()),
859            total_timesteps: Some(1_000_000),
860            n_steps: None,
861            n_epochs: None,
862            learning_rate: None,
863            patience: None,
864            min_delta: None,
865            n_envs: None,
866            config_yaml: None,
867            resume_run_id: Some("run-123".to_string()),
868            n_trials: Some(10),
869            tune_timesteps: Some(500_000),
870            eval_eps: Some(5),
871            search_space: None,
872        };
873        let json = serde_json::to_value(&req).expect("CreateJobRequest serializes");
874
875        for field in mecha10_job_validation::all_fields() {
876            let applicable = dispatchable
877                .iter()
878                .any(|&ft| field.requirement(ft) != mecha10_job_validation::FieldRequirement::NotApplicable);
879            if applicable {
880                assert!(
881                    json.get(field.wire_name()).is_some(),
882                    "CreateJobRequest has no '{}' field for {field:?}, which is applicable to a \
883                     dispatchable flow type",
884                    field.wire_name()
885                );
886            }
887        }
888    }
889
890    /// Contract test against `mecha10-job-validation`'s `AlgorithmField` list: every
891    /// per-algorithm hyperparameter field applicable to one of these algorithms must have a
892    /// matching wire key on this struct - the by-algorithm counterpart to
893    /// `create_job_request_has_a_field_for_every_dispatchable_job_field` above.
894    #[test]
895    fn create_job_request_has_a_field_for_every_algorithm_field() {
896        let algorithms = ["ppo", "sac", "td3", "a2c"];
897        let flow_types = [FlowType::Train, FlowType::ResumeRun, FlowType::Tune];
898
899        let req = CreateJobRequest {
900            name: "job".to_string(),
901            description: None,
902            algorithm: "ppo".to_string(),
903            robot_id: "robot-1".to_string(),
904            project_id: None,
905            project_name: None,
906            env_name: None,
907            task_name: Some("find_target".to_string()),
908            total_timesteps: Some(1_000_000),
909            n_steps: Some(2048),
910            n_epochs: Some(10),
911            learning_rate: Some(0.0003),
912            patience: Some(5),
913            min_delta: Some(0.01),
914            n_envs: None,
915            config_yaml: None,
916            resume_run_id: Some("run-123".to_string()),
917            n_trials: Some(10),
918            tune_timesteps: Some(500_000),
919            eval_eps: Some(5),
920            search_space: None,
921        };
922        let json = serde_json::to_value(&req).expect("CreateJobRequest serializes");
923
924        for field in mecha10_job_validation::all_algorithm_fields() {
925            let applicable = algorithms.iter().any(|&algorithm| {
926                flow_types.iter().any(|&ft| {
927                    field.requirement(algorithm, ft) != mecha10_job_validation::FieldRequirement::NotApplicable
928                })
929            });
930            if applicable {
931                assert!(
932                    json.get(field.wire_name()).is_some(),
933                    "CreateJobRequest has no '{}' field for {field:?}, which is applicable to a \
934                     supported algorithm",
935                    field.wire_name()
936                );
937            }
938        }
939    }
940
941    #[test]
942    fn to_create_job_request_forwards_tune_fields() {
943        let temp = TempDir::new().unwrap();
944        let jobs_dir = temp.path().join("configs/rl/jobs");
945        std::fs::create_dir_all(&jobs_dir).unwrap();
946        let source_path = jobs_dir.join("hypertune.yml");
947        std::fs::write(&source_path, "name: hypertune\nflow_type: tune\n").unwrap();
948
949        let job = JobDefinition {
950            name: "hypertune".to_string(),
951            flow_type: FlowType::Tune,
952            n_trials: Some(20),
953            tune_timesteps: Some(250_000),
954            eval_eps: Some(3),
955            source_path,
956            ..base_job()
957        };
958
959        let req = job
960            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
961            .unwrap();
962
963        assert_eq!(req.n_trials, Some(20));
964        assert_eq!(req.tune_timesteps, Some(250_000));
965        assert_eq!(req.eval_eps, Some(3));
966    }
967
968    #[test]
969    fn to_create_job_request_forwards_n_envs() {
970        let temp = TempDir::new().unwrap();
971        let jobs_dir = temp.path().join("configs/rl/jobs");
972        std::fs::create_dir_all(&jobs_dir).unwrap();
973        let source_path = jobs_dir.join("train.yml");
974        std::fs::write(&source_path, "name: train\nflow_type: train\nn_envs: 8\n").unwrap();
975
976        let job = JobDefinition {
977            name: "train".to_string(),
978            n_envs: Some(8),
979            source_path,
980            ..base_job()
981        };
982
983        let req = job
984            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
985            .unwrap();
986
987        assert_eq!(req.n_envs, Some(8));
988    }
989
990    #[test]
991    fn to_create_job_request_forwards_search_space() {
992        let temp = TempDir::new().unwrap();
993        let jobs_dir = temp.path().join("configs/rl/jobs");
994        std::fs::create_dir_all(&jobs_dir).unwrap();
995        let source_path = jobs_dir.join("hypertune.yml");
996        let yaml = "name: hypertune\nflow_type: tune\nsearch_space:\n  - name: learning_rate\n    \
997                    type: float\n    low: 1e-5\n    high: 1e-3\n    log: true\n  - name: n_steps\n    \
998                    type: categorical\n    choices: [1024, 2048, 4096]\n  - name: n_epochs\n    \
999                    type: int\n    low: 5\n    high: 20\n";
1000        std::fs::write(&source_path, yaml).unwrap();
1001        let job: JobDefinition = serde_yaml::from_str(yaml).unwrap();
1002        let job = JobDefinition { source_path, ..job };
1003
1004        let req = job
1005            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1006            .unwrap();
1007
1008        let search_space = req.search_space.expect("search_space forwarded");
1009        assert_eq!(search_space.len(), 3);
1010        assert_eq!(search_space[0].name, "learning_rate");
1011        assert_eq!(search_space[0].type_, SearchSpaceType::Float);
1012        assert_eq!(search_space[0].low, Some(1e-5));
1013        assert_eq!(search_space[0].high, Some(1e-3));
1014        assert_eq!(search_space[0].log, Some(true));
1015        assert_eq!(search_space[1].name, "n_steps");
1016        assert_eq!(search_space[1].type_, SearchSpaceType::Categorical);
1017        assert_eq!(
1018            search_space[1].choices,
1019            Some(vec![
1020                serde_json::json!(1024),
1021                serde_json::json!(2048),
1022                serde_json::json!(4096)
1023            ])
1024        );
1025        assert_eq!(search_space[2].name, "n_epochs");
1026        assert_eq!(search_space[2].type_, SearchSpaceType::Int);
1027        assert_eq!(search_space[2].low, Some(5.0));
1028        assert_eq!(search_space[2].high, Some(20.0));
1029    }
1030
1031    #[test]
1032    fn to_create_job_request_omits_search_space_when_unset() {
1033        let temp = TempDir::new().unwrap();
1034        let jobs_dir = temp.path().join("configs/rl/jobs");
1035        std::fs::create_dir_all(&jobs_dir).unwrap();
1036        let source_path = jobs_dir.join("train.yml");
1037        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
1038
1039        let job = JobDefinition {
1040            name: "train".to_string(),
1041            source_path,
1042            ..base_job()
1043        };
1044
1045        let req = job
1046            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1047            .unwrap();
1048
1049        assert_eq!(req.search_space, None);
1050    }
1051
1052    #[test]
1053    fn to_create_job_request_drops_sentinel_values() {
1054        let temp = TempDir::new().unwrap();
1055        let jobs_dir = temp.path().join("configs/rl/jobs");
1056        std::fs::create_dir_all(&jobs_dir).unwrap();
1057        let source_path = jobs_dir.join("train.yml");
1058        std::fs::write(&source_path, "name: train\nflow_type: train\nenv_name: rover\n").unwrap();
1059
1060        let job = JobDefinition {
1061            name: "train".to_string(),
1062            description: Some("desc".to_string()),
1063            env_name: Some("rover".to_string()),
1064            total_timesteps: Some(-1),
1065            n_steps: Some(2048),
1066            n_epochs: Some(-1),
1067            learning_rate: Some(-1.0),
1068            patience: Some(50),
1069            min_delta: Some(-1.0),
1070            n_envs: Some(-1),
1071            source_path,
1072            ..base_job()
1073        };
1074
1075        let req = job
1076            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1077            .unwrap();
1078
1079        assert_eq!(req.name, "train");
1080        assert_eq!(req.algorithm, "ppo");
1081        assert_eq!(req.robot_id, "my-project");
1082        assert_eq!(req.project_id.as_deref(), Some("my-project"));
1083        assert_eq!(req.project_name.as_deref(), Some("my-robot"));
1084        assert_eq!(req.total_timesteps, None);
1085        assert_eq!(req.n_steps, Some(2048));
1086        assert_eq!(req.n_epochs, None);
1087        assert_eq!(req.learning_rate, None);
1088        assert_eq!(req.patience, Some(50));
1089        assert_eq!(req.min_delta, None);
1090        assert_eq!(req.n_envs, None);
1091        assert_eq!(
1092            req.config_yaml.as_deref(),
1093            Some("name: train\nflow_type: train\nenv_name: rover\n")
1094        );
1095        assert_eq!(req.task_name, None);
1096    }
1097
1098    #[test]
1099    fn to_create_job_request_forwards_task_name() {
1100        let temp = TempDir::new().unwrap();
1101        let jobs_dir = temp.path().join("configs/rl/jobs");
1102        std::fs::create_dir_all(&jobs_dir).unwrap();
1103        let source_path = jobs_dir.join("train.yml");
1104        std::fs::write(&source_path, "name: train\nflow_type: train\n").unwrap();
1105
1106        let job = JobDefinition {
1107            name: "train".to_string(),
1108            task_name: Some("navigate".to_string()),
1109            source_path,
1110            ..base_job()
1111        };
1112
1113        let req = job
1114            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1115            .unwrap();
1116
1117        assert_eq!(req.task_name.as_deref(), Some("navigate"));
1118    }
1119
1120    #[test]
1121    fn to_create_job_request_errors_when_the_jobs_own_file_is_missing() {
1122        let temp = TempDir::new().unwrap();
1123        let job = JobDefinition {
1124            name: "train".to_string(),
1125            source_path: temp.path().join("configs/rl/jobs/train.yml"),
1126            ..base_job()
1127        };
1128
1129        let err = job
1130            .to_create_job_request(temp.path(), "my-project", "my-robot", None)
1131            .unwrap_err();
1132        assert!(err.to_string().contains("configs/rl/jobs/train.yml"));
1133    }
1134}
mecha10_cli/services/train_jobs.rs

mecha10_cli/services/
train_jobs.rs
