1#![allow(dead_code)] use std::path::{Path, PathBuf};
42use std::time::Instant;
43
44use rand::rngs::StdRng;
45use rand::{Rng, SeedableRng};
46use serde::{Deserialize, Serialize};
47
48#[derive(Debug, Clone, PartialEq, Serialize)]
58pub enum PretrainAbort {
59 Divergence { epoch: usize, prev_val_loss: f32, curr_val_loss: f32, ratio: f32 },
62 DivergenceAtEpochZero { val_loss: f32 },
65 NumericalInstability { step: u64, field: &'static str, value: f32 },
67 ThroughputOutOfRange { step: u64, field: &'static str, value: f32 },
71}
72
73impl std::fmt::Display for PretrainAbort {
74 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75 match self {
76 Self::Divergence { epoch, prev_val_loss, curr_val_loss, ratio } => write!(
77 f,
78 "DIVERGENCE at epoch {epoch}: val_loss {curr_val_loss:.4} > 2.0 × {prev_val_loss:.4} (ratio {ratio:.2})",
79 ),
80 Self::DivergenceAtEpochZero { val_loss } => write!(
81 f,
82 "DIVERGENCE at epoch 0: val_loss {val_loss} is non-finite or > 10.0",
83 ),
84 Self::NumericalInstability { step, field, value } => write!(
85 f,
86 "NUMERICAL_INSTABILITY at step {step}: {field} = {value} is non-finite",
87 ),
88 Self::ThroughputOutOfRange { step, field, value } => write!(
89 f,
90 "THROUGHPUT_OUT_OF_RANGE at step {step}: {field} = {value} outside permitted range",
91 ),
92 }
93 }
94}
95
96impl std::error::Error for PretrainAbort {}
97
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
109pub struct StepMetrics {
110 pub step: u64,
112 pub train_loss: f32,
114 pub grad_norm: f32,
116 pub lr: f32,
118 pub tokens_per_sec: f32,
120 pub gpu_util_pct: f32,
122 #[serde(default)]
127 pub wall_ms: f32,
128}
129
130impl StepMetrics {
131 pub fn validate_finite(&self) -> Result<(), PretrainAbort> {
137 if !self.train_loss.is_finite() {
138 return Err(PretrainAbort::NumericalInstability {
139 step: self.step,
140 field: "train_loss",
141 value: self.train_loss,
142 });
143 }
144 if !self.grad_norm.is_finite() {
145 return Err(PretrainAbort::NumericalInstability {
146 step: self.step,
147 field: "grad_norm",
148 value: self.grad_norm,
149 });
150 }
151 if !self.lr.is_finite() {
152 return Err(PretrainAbort::NumericalInstability {
153 step: self.step,
154 field: "lr",
155 value: self.lr,
156 });
157 }
158 if !self.tokens_per_sec.is_finite() || self.tokens_per_sec < 0.0 {
159 return Err(PretrainAbort::ThroughputOutOfRange {
160 step: self.step,
161 field: "tokens_per_sec",
162 value: self.tokens_per_sec,
163 });
164 }
165 if !self.gpu_util_pct.is_finite() || self.gpu_util_pct < 0.0 || self.gpu_util_pct > 100.0 {
166 return Err(PretrainAbort::ThroughputOutOfRange {
167 step: self.step,
168 field: "gpu_util_pct",
169 value: self.gpu_util_pct,
170 });
171 }
172 if !self.wall_ms.is_finite() || self.wall_ms < 0.0 {
173 return Err(PretrainAbort::ThroughputOutOfRange {
174 step: self.step,
175 field: "wall_ms",
176 value: self.wall_ms,
177 });
178 }
179 Ok(())
180 }
181}
182
183#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
189pub struct EpochMetadata {
190 pub epoch: usize,
191 pub train_loss: f32,
192 pub val_loss: f32,
193 pub train_ppl: f32,
194 pub val_ppl: f32,
195 pub optimizer_state_sha: String,
197 pub wall_seconds: f32,
198 pub tokens_seen: u64,
199 pub grad_norm_max: f32,
200}
201
202#[derive(Debug, Clone)]
205pub struct EpochArtifact {
206 pub checkpoint_path: PathBuf,
208 pub metadata_path: PathBuf,
210 pub metadata: EpochMetadata,
211}
212
213impl EpochArtifact {
214 pub fn new(run_dir: &Path, epoch: usize, metadata: EpochMetadata) -> Self {
216 let ckpt_dir = run_dir.join("ckpt");
217 let filename = format!("epoch-{epoch:03}.apr");
218 let metafile = format!("epoch-{epoch:03}.metadata.json");
219 Self {
220 checkpoint_path: ckpt_dir.join(filename),
221 metadata_path: ckpt_dir.join(metafile),
222 metadata,
223 }
224 }
225}
226
227pub const DIVERGENCE_RATIO_LIMIT: f32 = 2.0;
235
236pub const EPOCH_ZERO_VAL_LOSS_LIMIT: f32 = 10.0;
242
243#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
254#[serde(tag = "kind", rename_all = "snake_case")]
255pub enum TrainingRegime {
256 Finetune,
257 FromScratch { vocab_size: u32 },
258}
259
260impl TrainingRegime {
261 pub fn epoch_zero_val_loss_limit(&self) -> f32 {
268 match self {
269 Self::Finetune => EPOCH_ZERO_VAL_LOSS_LIMIT,
270 Self::FromScratch { vocab_size } => {
271 let v = (*vocab_size).max(2) as f32;
272 DIVERGENCE_RATIO_LIMIT * v.ln()
273 }
274 }
275 }
276}
277
278impl Default for TrainingRegime {
279 fn default() -> Self {
280 Self::Finetune
281 }
282}
283
284pub fn check_non_divergence(
295 epoch: usize,
296 val_loss_history: &[f32],
297 regime: &TrainingRegime,
298) -> Result<(), PretrainAbort> {
299 let Some(&curr) = val_loss_history.get(epoch) else {
300 return Ok(());
302 };
303
304 if epoch == 0 {
306 let cap = regime.epoch_zero_val_loss_limit();
307 if !curr.is_finite() || curr > cap {
308 return Err(PretrainAbort::DivergenceAtEpochZero { val_loss: curr });
309 }
310 return Ok(());
311 }
312
313 let prev = val_loss_history[epoch - 1];
315 if !curr.is_finite() {
316 return Err(PretrainAbort::NumericalInstability {
317 step: u64::MAX,
318 field: "val_loss",
319 value: curr,
320 });
321 }
322 let ratio = curr / prev.max(1e-9);
323 if curr > DIVERGENCE_RATIO_LIMIT * prev {
324 return Err(PretrainAbort::Divergence {
325 epoch,
326 prev_val_loss: prev,
327 curr_val_loss: curr,
328 ratio,
329 });
330 }
331 Ok(())
332}
333
334pub fn check_numerical_stability(
341 step: u64,
342 train_loss: f32,
343 grad_norm: f32,
344) -> Result<(), PretrainAbort> {
345 if !train_loss.is_finite() {
346 return Err(PretrainAbort::NumericalInstability {
347 step,
348 field: "train_loss",
349 value: train_loss,
350 });
351 }
352 if !grad_norm.is_finite() {
353 return Err(PretrainAbort::NumericalInstability {
354 step,
355 field: "grad_norm",
356 value: grad_norm,
357 });
358 }
359 Ok(())
360}
361
362#[derive(Debug, Clone, Serialize, Deserialize)]
369pub struct PretrainConfig {
370 pub dataset_path: PathBuf,
372 pub tokenizer_dir: PathBuf,
374 pub run_dir: PathBuf,
376 pub lr_max: f32,
378 pub lr_min: f32,
380 pub warmup_steps: usize,
382 pub total_steps: usize,
384 pub batch_size: usize,
386 pub seq_length: usize,
388 pub steps_per_epoch: usize,
391 pub seed: u64,
393 pub grad_clip: f32,
395 pub weight_decay: f32,
397 pub target_val_loss: f32,
399 pub patience_epochs: usize,
401 pub min_epochs_before_early_stop: usize,
403 #[serde(default)]
406 pub regime: TrainingRegime,
407}
408
409impl PretrainConfig {
410 pub fn model_2_defaults(
415 dataset_path: PathBuf,
416 tokenizer_dir: PathBuf,
417 run_dir: PathBuf,
418 ) -> Self {
419 Self {
420 dataset_path,
421 tokenizer_dir,
422 run_dir,
423 lr_max: 5.0e-5,
424 lr_min: 1.0e-6,
425 warmup_steps: 100,
426 total_steps: 1000,
427 batch_size: 16,
428 seq_length: 1024,
429 steps_per_epoch: 100,
430 seed: 42,
431 grad_clip: 1.0,
432 weight_decay: 0.01,
433 target_val_loss: 2.2,
434 patience_epochs: 2,
435 min_epochs_before_early_stop: 3,
436 regime: TrainingRegime::Finetune,
437 }
438 }
439}
440
441#[derive(Debug, Clone, Serialize)]
447pub enum RunStatus {
448 Ok { final_val_loss: f32, epochs_completed: usize },
450 EarlyStop { best_val_loss: f32, epochs_completed: usize },
452 Aborted(PretrainAbort),
454}
455
456pub struct PretrainLoop<S: StepFn, V: ValFn> {
464 config: PretrainConfig,
465 rng: StdRng,
466 step_metrics: Vec<StepMetrics>,
467 epoch_artifacts: Vec<EpochArtifact>,
468 val_loss_history: Vec<f32>,
469 tokens_seen: u64,
470 best_val_loss: f32,
471 patience_counter: usize,
472 step_fn: S,
473 val_fn: V,
474 checkpoint_fn: Option<Box<dyn CheckpointFn>>,
478}
479
480pub trait StepFn {
486 fn step(&mut self, step: u64, lr: f32, batch_tokens: u64) -> (f32, f32);
487
488 fn optimizer_state_sha256(&self) -> Option<String> {
496 None
497 }
498}
499
500pub trait ValFn {
502 fn validate(&mut self, epoch: usize) -> f32;
503}
504
505pub trait CheckpointFn {
516 fn save(&mut self, epoch: usize, artifact: &EpochArtifact) -> Result<(), String>;
517}
518
519impl<S: StepFn, V: ValFn> PretrainLoop<S, V> {
520 pub fn new(config: PretrainConfig, step_fn: S, val_fn: V) -> Self {
522 let rng = StdRng::seed_from_u64(config.seed);
523 Self {
524 config,
525 rng,
526 step_metrics: Vec::new(),
527 epoch_artifacts: Vec::new(),
528 val_loss_history: Vec::new(),
529 tokens_seen: 0,
530 best_val_loss: f32::INFINITY,
531 patience_counter: 0,
532 step_fn,
533 val_fn,
534 checkpoint_fn: None,
535 }
536 }
537
538 #[must_use]
541 pub fn with_checkpoint_fn(mut self, ckpt: Box<dyn CheckpointFn>) -> Self {
542 self.checkpoint_fn = Some(ckpt);
543 self
544 }
545
546 fn lr_at(&self, step: u64) -> f32 {
550 let step = step as usize;
551 let w = self.config.warmup_steps;
552 let total = self.config.total_steps;
553 let lr_max = self.config.lr_max;
554 let lr_min = self.config.lr_min;
555
556 if step < w {
557 if w == 0 {
558 return lr_max;
559 }
560 return lr_max * (step as f32 / w as f32);
561 }
562 let decay_steps = total.saturating_sub(w);
563 if decay_steps == 0 {
564 return lr_min;
565 }
566 let decay_step = step - w;
567 if decay_step >= decay_steps {
568 return lr_min;
569 }
570 let progress = decay_step as f32 / decay_steps as f32;
571 let cosine_decay = 0.5 * (1.0 + (std::f32::consts::PI * progress).cos());
572 lr_min + (lr_max - lr_min) * cosine_decay
573 }
574
575 pub fn train_step(&mut self, step: u64) -> Result<StepMetrics, PretrainAbort> {
578 let lr = self.lr_at(step);
579 let batch_tokens = (self.config.batch_size * self.config.seq_length) as u64;
580 let t0 = Instant::now();
581 let (train_loss, grad_norm) = self.step_fn.step(step, lr, batch_tokens);
582 let elapsed = t0.elapsed().as_secs_f32().max(1.0e-9);
583
584 check_numerical_stability(step, train_loss, grad_norm)?;
588
589 let tokens_per_sec = batch_tokens as f32 / elapsed;
590 let wall_ms = elapsed * 1000.0;
591 let gpu_util_pct = 50.0 + (self.rng.random_range(-5.0..5.0) as f32);
595
596 let metrics = StepMetrics {
597 step,
598 train_loss,
599 grad_norm,
600 lr,
601 tokens_per_sec,
602 gpu_util_pct: gpu_util_pct.clamp(0.0, 100.0),
603 wall_ms,
604 };
605 metrics.validate_finite()?;
606
607 self.tokens_seen += batch_tokens;
608 self.step_metrics.push(metrics.clone());
609 Ok(metrics)
610 }
611
612 pub fn run_epoch(&mut self, epoch: usize) -> Result<EpochArtifact, PretrainAbort> {
622 let first_step = (epoch * self.config.steps_per_epoch) as u64;
623 let last_step =
624 (first_step + self.config.steps_per_epoch as u64).min(self.config.total_steps as u64);
625
626 let t0 = Instant::now();
627 let mut epoch_loss_sum = 0.0_f32;
628 let mut epoch_grad_norm_max = 0.0_f32;
629 let mut steps_taken = 0_u32;
630
631 for step in first_step..last_step {
632 let m = self.train_step(step)?;
633 epoch_loss_sum += m.train_loss;
634 if m.grad_norm > epoch_grad_norm_max {
635 epoch_grad_norm_max = m.grad_norm;
636 }
637 steps_taken += 1;
638 }
639
640 let mean_train_loss = epoch_loss_sum / steps_taken.max(1) as f32;
641 let val_loss = self.val_fn.validate(epoch);
642
643 if !val_loss.is_finite() {
645 return Err(PretrainAbort::NumericalInstability {
646 step: last_step,
647 field: "val_loss",
648 value: val_loss,
649 });
650 }
651
652 self.val_loss_history.push(val_loss);
653
654 check_non_divergence(epoch, &self.val_loss_history, &self.config.regime)?;
657
658 let wall_seconds = t0.elapsed().as_secs_f32();
659 let optimizer_state_sha =
663 self.step_fn.optimizer_state_sha256().unwrap_or_else(|| self.fake_optimizer_sha(epoch));
664 let metadata = EpochMetadata {
665 epoch,
666 train_loss: mean_train_loss,
667 val_loss,
668 train_ppl: mean_train_loss.exp(),
669 val_ppl: val_loss.exp(),
670 optimizer_state_sha,
671 wall_seconds,
672 tokens_seen: self.tokens_seen,
673 grad_norm_max: epoch_grad_norm_max,
674 };
675 let artifact = EpochArtifact::new(&self.config.run_dir, epoch, metadata);
676
677 if let Some(ckpt) = self.checkpoint_fn.as_mut() {
682 if let Some(parent) = artifact.checkpoint_path.parent() {
683 let _ = std::fs::create_dir_all(parent);
684 }
685 if let Err(e) = ckpt.save(epoch, &artifact) {
686 eprintln!("[pretrain] checkpoint write failed for epoch {}: {}", epoch, e);
687 } else {
688 match serde_json::to_string_pretty(&artifact.metadata) {
692 Ok(json) => {
693 if let Err(e) = std::fs::write(&artifact.metadata_path, json) {
694 eprintln!(
695 "[pretrain] metadata write failed for epoch {}: {}",
696 epoch, e
697 );
698 }
699 }
700 Err(e) => eprintln!(
701 "[pretrain] metadata serialization failed for epoch {}: {}",
702 epoch, e
703 ),
704 }
705 }
706 }
707
708 self.epoch_artifacts.push(artifact.clone());
709 Ok(artifact)
710 }
711
712 pub fn check_convergence(&mut self, epoch: usize) -> bool {
715 let Some(&val_loss) = self.val_loss_history.last() else {
716 return false;
717 };
718 if val_loss < self.best_val_loss {
719 self.best_val_loss = val_loss;
720 self.patience_counter = 0;
721 return false;
722 }
723 self.patience_counter += 1;
724 if epoch + 1 < self.config.min_epochs_before_early_stop {
725 return false;
726 }
727 self.patience_counter > self.config.patience_epochs
728 }
729
730 pub fn run(&mut self) -> RunStatus {
732 let num_epochs = self.config.total_steps.div_ceil(self.config.steps_per_epoch.max(1));
733 for epoch in 0..num_epochs {
734 match self.run_epoch(epoch) {
735 Ok(_) => {}
736 Err(abort) => return RunStatus::Aborted(abort),
737 }
738 if self.check_convergence(epoch) {
739 return RunStatus::EarlyStop {
740 best_val_loss: self.best_val_loss,
741 epochs_completed: epoch + 1,
742 };
743 }
744 let last = *self.val_loss_history.last().unwrap_or(&f32::INFINITY);
745 if last <= self.config.target_val_loss
746 && epoch + 1 >= self.config.min_epochs_before_early_stop
747 {
748 return RunStatus::Ok { final_val_loss: last, epochs_completed: epoch + 1 };
749 }
750 }
751 let last = *self.val_loss_history.last().unwrap_or(&f32::INFINITY);
752 RunStatus::Ok { final_val_loss: last, epochs_completed: num_epochs }
753 }
754
755 pub fn step_metrics(&self) -> &[StepMetrics] {
757 &self.step_metrics
758 }
759
760 pub fn epoch_artifacts(&self) -> &[EpochArtifact] {
761 &self.epoch_artifacts
762 }
763
764 pub fn val_loss_history(&self) -> &[f32] {
765 &self.val_loss_history
766 }
767
768 fn fake_optimizer_sha(&self, epoch: usize) -> String {
774 use sha2::{Digest, Sha256};
775 let mut hasher = Sha256::new();
776 hasher.update(b"aprender-train:pretrain:optstate:v1:");
777 hasher.update(self.config.seed.to_le_bytes());
778 hasher.update((epoch as u64).to_le_bytes());
779 hasher.update(self.tokens_seen.to_le_bytes());
780 format!("{:x}", hasher.finalize())
781 }
782}
783
784pub struct LinearDecaySynthetic {
791 pub start_loss: f32,
792 pub decay_per_step: f32,
793 pub grad_norm: f32,
794}
795
796impl StepFn for LinearDecaySynthetic {
797 fn step(&mut self, step: u64, _lr: f32, _batch_tokens: u64) -> (f32, f32) {
798 let loss = (self.start_loss - self.decay_per_step * step as f32).max(1.0e-4);
799 (loss, self.grad_norm)
800 }
801}
802
803pub struct ScriptedVal {
807 pub sequence: Vec<f32>,
808}
809
810impl ValFn for ScriptedVal {
811 fn validate(&mut self, epoch: usize) -> f32 {
812 *self.sequence.get(epoch).unwrap_or(&f32::NAN)
813 }
814}
815
816pub struct NanAtStepSynthetic {
818 pub nan_step: u64,
819}
820
821impl StepFn for NanAtStepSynthetic {
822 fn step(&mut self, step: u64, _lr: f32, _batch_tokens: u64) -> (f32, f32) {
823 if step == self.nan_step {
824 return (f32::NAN, 1.0);
825 }
826 (1.0, 1.0)
827 }
828}
829
830#[cfg(test)]
831mod tests {
832 use super::*;
833 use std::cell::RefCell;
834 use std::rc::Rc;
835 use tempfile::TempDir;
836
837 fn test_config(tmp: &Path) -> PretrainConfig {
838 PretrainConfig {
839 dataset_path: tmp.join("data.jsonl"),
840 tokenizer_dir: tmp.join("tok"),
841 run_dir: tmp.join("run"),
842 lr_max: 1.0e-4,
843 lr_min: 1.0e-6,
844 warmup_steps: 2,
845 total_steps: 25,
846 batch_size: 2,
847 seq_length: 4,
848 steps_per_epoch: 5,
849 seed: 42,
850 grad_clip: 1.0,
851 weight_decay: 0.01,
852 target_val_loss: 2.2,
853 patience_epochs: 2,
854 min_epochs_before_early_stop: 1,
855 regime: TrainingRegime::Finetune,
856 }
857 }
858
859 #[test]
864 fn gate_train_005_aborts_on_doubling_val_loss() {
865 let trace = vec![3.5, 7.1];
866 let res = check_non_divergence(1, &trace, &TrainingRegime::Finetune);
867 match res {
868 Err(PretrainAbort::Divergence { epoch, prev_val_loss, curr_val_loss, ratio }) => {
869 assert_eq!(epoch, 1);
870 assert!((prev_val_loss - 3.5).abs() < 1e-6);
871 assert!((curr_val_loss - 7.1).abs() < 1e-6);
872 assert!(ratio > 2.0);
873 }
874 other => panic!("GATE-TRAIN-005 did not abort: got {other:?}"),
875 }
876 }
877
878 #[test]
881 fn gate_train_005_aborts_on_epoch_zero_blowup() {
882 let trace = vec![31.99];
883 let res = check_non_divergence(0, &trace, &TrainingRegime::Finetune);
884 match res {
885 Err(PretrainAbort::DivergenceAtEpochZero { val_loss }) => {
886 assert!((val_loss - 31.99).abs() < 1e-4);
887 }
888 other => panic!("epoch-0 guard missed: got {other:?}"),
889 }
890 }
891
892 #[test]
894 fn gate_train_005_allows_healthy_decrease() {
895 let trace = vec![3.5, 3.0, 2.5, 2.2];
896 for epoch in 0..trace.len() {
897 assert!(check_non_divergence(epoch, &trace, &TrainingRegime::Finetune).is_ok());
898 }
899 }
900
901 #[test]
903 fn gate_train_005_allows_exact_two_x() {
904 let trace = vec![2.0, 4.0];
905 assert!(check_non_divergence(1, &trace, &TrainingRegime::Finetune).is_ok());
906 }
907
908 #[test]
914 fn gate_train_005_from_scratch_permits_near_random_baseline() {
915 let trace = vec![18.0_f32];
916 let regime = TrainingRegime::FromScratch { vocab_size: 50_257 };
917 assert!(
918 check_non_divergence(0, &trace, ®ime).is_ok(),
919 "val_loss[0]=18 must be within 2·ln(50257)≈21.64 from_scratch cap"
920 );
921 assert!(matches!(
923 check_non_divergence(0, &trace, &TrainingRegime::Finetune),
924 Err(PretrainAbort::DivergenceAtEpochZero { .. }),
925 ));
926 }
927
928 #[test]
931 fn gate_train_005_from_scratch_aborts_above_2_ln_vocab() {
932 let trace = vec![25.0_f32];
933 let regime = TrainingRegime::FromScratch { vocab_size: 50_257 };
934 match check_non_divergence(0, &trace, ®ime) {
935 Err(PretrainAbort::DivergenceAtEpochZero { val_loss }) => {
936 assert!((val_loss - 25.0).abs() < 1e-4);
937 }
938 other => panic!("from_scratch cap missed: got {other:?}"),
939 }
940 }
941
942 #[test]
945 fn training_regime_from_scratch_cap_matches_formula() {
946 let v = 50_257u32;
947 let regime = TrainingRegime::FromScratch { vocab_size: v };
948 let expected = DIVERGENCE_RATIO_LIMIT * (v as f32).ln();
949 assert!(
950 (regime.epoch_zero_val_loss_limit() - expected).abs() < 1e-4,
951 "cap formula drift: got {} expected {}",
952 regime.epoch_zero_val_loss_limit(),
953 expected
954 );
955 assert!(
957 (TrainingRegime::Finetune.epoch_zero_val_loss_limit() - EPOCH_ZERO_VAL_LOSS_LIMIT)
958 .abs()
959 < 1e-6
960 );
961 }
962
963 #[test]
966 fn gate_train_007_aborts_on_nan_train_loss() {
967 let res = check_numerical_stability(42, f32::NAN, 1.0);
968 match res {
969 Err(PretrainAbort::NumericalInstability { step, field, .. }) => {
970 assert_eq!(step, 42);
971 assert_eq!(field, "train_loss");
972 }
973 other => panic!("nan guard missed: got {other:?}"),
974 }
975 }
976
977 #[test]
978 fn gate_train_007_aborts_on_inf_grad_norm() {
979 let res = check_numerical_stability(7, 1.0, f32::INFINITY);
980 assert!(matches!(res, Err(PretrainAbort::NumericalInstability { .. })));
981 }
982
983 #[test]
986 fn step_metrics_validate_finite_accepts_healthy() {
987 let m = StepMetrics {
988 step: 0,
989 train_loss: 3.2,
990 grad_norm: 0.5,
991 lr: 1e-4,
992 tokens_per_sec: 1000.0,
993 gpu_util_pct: 75.0,
994 wall_ms: 5.0,
995 };
996 assert!(m.validate_finite().is_ok());
997 }
998
999 #[test]
1000 fn step_metrics_rejects_negative_throughput() {
1001 let m = StepMetrics {
1002 step: 1,
1003 train_loss: 3.2,
1004 grad_norm: 0.5,
1005 lr: 1e-4,
1006 tokens_per_sec: -1.0,
1007 gpu_util_pct: 75.0,
1008 wall_ms: 5.0,
1009 };
1010 assert!(matches!(m.validate_finite(), Err(PretrainAbort::ThroughputOutOfRange { .. })));
1011 }
1012
1013 #[test]
1014 fn step_metrics_rejects_gpu_util_over_100() {
1015 let m = StepMetrics {
1016 step: 1,
1017 train_loss: 3.2,
1018 grad_norm: 0.5,
1019 lr: 1e-4,
1020 tokens_per_sec: 1000.0,
1021 gpu_util_pct: 150.0,
1022 wall_ms: 5.0,
1023 };
1024 assert!(matches!(m.validate_finite(), Err(PretrainAbort::ThroughputOutOfRange { .. })));
1025 }
1026
1027 #[test]
1030 fn step_metrics_rejects_negative_wall_ms() {
1031 let m = StepMetrics {
1032 step: 1,
1033 train_loss: 3.2,
1034 grad_norm: 0.5,
1035 lr: 1e-4,
1036 tokens_per_sec: 1000.0,
1037 gpu_util_pct: 75.0,
1038 wall_ms: -1.0,
1039 };
1040 assert!(matches!(m.validate_finite(), Err(PretrainAbort::ThroughputOutOfRange { .. })));
1041 }
1042
1043 #[test]
1045 fn step_metrics_rejects_nan_wall_ms() {
1046 let m = StepMetrics {
1047 step: 1,
1048 train_loss: 3.2,
1049 grad_norm: 0.5,
1050 lr: 1e-4,
1051 tokens_per_sec: 1000.0,
1052 gpu_util_pct: 75.0,
1053 wall_ms: f32::NAN,
1054 };
1055 assert!(matches!(m.validate_finite(), Err(PretrainAbort::ThroughputOutOfRange { .. })));
1056 }
1057
1058 #[test]
1063 fn step_metrics_wall_ms_consistent_with_tokens_per_sec() {
1064 let batch_tokens: u64 = 1024;
1065 let elapsed_secs: f32 = 0.5;
1066 let tokens_per_sec = batch_tokens as f32 / elapsed_secs;
1067 let wall_ms = elapsed_secs * 1000.0;
1068
1069 let m = StepMetrics {
1070 step: 0,
1071 train_loss: 3.2,
1072 grad_norm: 0.5,
1073 lr: 1e-4,
1074 tokens_per_sec,
1075 gpu_util_pct: 50.0,
1076 wall_ms,
1077 };
1078 assert!(m.validate_finite().is_ok());
1079 let derived_tokens = m.tokens_per_sec * (m.wall_ms / 1000.0);
1080 let diff = (derived_tokens - batch_tokens as f32).abs();
1081 assert!(
1082 diff < 0.5,
1083 "tokens_per_sec * (wall_ms/1000) = {derived_tokens} should equal batch_tokens={batch_tokens} within FP rounding"
1084 );
1085 }
1086
1087 #[test]
1090 fn pretrain_loop_happy_path_decreasing_loss() {
1091 let tmp = TempDir::new().expect("tempdir");
1092 let cfg = test_config(tmp.path());
1093 let step_fn = LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1094 let val_fn = ScriptedVal { sequence: vec![3.4, 3.0, 2.6, 2.2, 2.0] };
1095 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1096
1097 let status = loop_.run();
1098 match status {
1099 RunStatus::Ok { final_val_loss, epochs_completed } => {
1100 assert!(final_val_loss <= 2.2);
1101 assert!(epochs_completed >= 1);
1102 }
1103 other => panic!("healthy run did not converge cleanly: {other:?}"),
1104 }
1105
1106 assert!(!loop_.step_metrics().is_empty());
1108 for m in loop_.step_metrics() {
1109 assert!(m.train_loss.is_finite());
1110 assert!(m.grad_norm.is_finite());
1111 assert!(m.lr.is_finite());
1112 assert!(m.tokens_per_sec >= 0.0);
1113 assert!((0.0..=100.0).contains(&m.gpu_util_pct));
1114 }
1115 assert_eq!(loop_.epoch_artifacts().len(), loop_.val_loss_history().len());
1117 for art in loop_.epoch_artifacts() {
1118 assert!(!art.metadata.optimizer_state_sha.is_empty());
1119 assert!(art.metadata.train_ppl.is_finite());
1120 assert!(art.metadata.val_ppl.is_finite());
1121 }
1122 }
1123
1124 #[test]
1128 fn pretrain_loop_aborts_on_doubling_val_loss() {
1129 let tmp = TempDir::new().expect("tempdir");
1130 let cfg = test_config(tmp.path());
1131 let step_fn = LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1132 let val_fn = ScriptedVal { sequence: vec![3.5, 7.1, 2.0] };
1134 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1135
1136 let status = loop_.run();
1137 match status {
1138 RunStatus::Aborted(PretrainAbort::Divergence { epoch, ratio, .. }) => {
1139 assert_eq!(epoch, 1);
1140 assert!(ratio > 2.0);
1141 }
1142 other => panic!("GATE-TRAIN-005 did not fire: {other:?}"),
1143 }
1144 }
1145
1146 #[test]
1148 fn pretrain_loop_aborts_on_nan_in_train_loss() {
1149 let tmp = TempDir::new().expect("tempdir");
1150 let cfg = test_config(tmp.path());
1151 let step_fn = NanAtStepSynthetic { nan_step: 3 };
1152 let val_fn = ScriptedVal { sequence: vec![3.0] };
1153 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1154
1155 let status = loop_.run();
1156 match status {
1157 RunStatus::Aborted(PretrainAbort::NumericalInstability { step, field, .. }) => {
1158 assert_eq!(step, 3);
1159 assert_eq!(field, "train_loss");
1160 }
1161 other => panic!("INV-TRAIN-007 did not fire: {other:?}"),
1162 }
1163 }
1164
1165 #[test]
1173 fn num_steps_is_a_budget_not_rounded_up_to_a_whole_epoch() {
1174 for (total, spe) in [(3usize, 100usize), (250, 100), (7, 5), (10, 10), (1, 64)] {
1176 let tmp = TempDir::new().expect("tempdir");
1177 let cfg = PretrainConfig {
1178 total_steps: total,
1179 steps_per_epoch: spe,
1180 warmup_steps: 1,
1181 target_val_loss: 0.0,
1183 min_epochs_before_early_stop: usize::MAX,
1184 patience_epochs: usize::MAX,
1185 ..test_config(tmp.path())
1186 };
1187 let step_fn =
1188 LinearDecaySynthetic { start_loss: 3.0, decay_per_step: 0.0, grad_norm: 0.5 };
1189 let val_fn = ScriptedVal { sequence: vec![3.0; total.div_ceil(spe) + 2] };
1190 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1191 let _ = loop_.run();
1192
1193 assert_eq!(
1194 loop_.step_metrics().len(),
1195 total,
1196 "requested {total} steps at steps_per_epoch={spe}, executed {}",
1197 loop_.step_metrics().len()
1198 );
1199 let last = loop_.step_metrics().last().map(|m| m.step);
1201 assert_eq!(last, Some(total as u64 - 1), "last step index past the budget");
1202 }
1203 }
1204
1205 #[test]
1209 fn pretrain_loop_reproducibility_seed_42() {
1210 let tmp1 = TempDir::new().expect("tempdir1");
1211 let tmp2 = TempDir::new().expect("tempdir2");
1212 let cfg1 = test_config(tmp1.path());
1213 let cfg2 = test_config(tmp2.path());
1214
1215 let step_fn1 =
1216 LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1217 let step_fn2 =
1218 LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1219 let val_fn1 = ScriptedVal { sequence: vec![3.0, 2.8, 2.6, 2.4, 2.2] };
1220 let val_fn2 = ScriptedVal { sequence: vec![3.0, 2.8, 2.6, 2.4, 2.2] };
1221
1222 let mut loop1 = PretrainLoop::new(cfg1, step_fn1, val_fn1);
1223 let mut loop2 = PretrainLoop::new(cfg2, step_fn2, val_fn2);
1224 let _ = loop1.run();
1225 let _ = loop2.run();
1226
1227 assert_eq!(loop1.step_metrics().len(), loop2.step_metrics().len());
1228 for (a, b) in loop1.step_metrics().iter().zip(loop2.step_metrics().iter()) {
1229 assert_eq!(a.step, b.step);
1230 assert!((a.train_loss - b.train_loss).abs() < 1e-6);
1231 assert!((a.grad_norm - b.grad_norm).abs() < 1e-6);
1232 assert!((a.lr - b.lr).abs() < 1e-6);
1233 assert!((a.gpu_util_pct - b.gpu_util_pct).abs() < 1e-6);
1235 }
1236 }
1237
1238 #[test]
1241 fn lr_schedule_warmup_cosine_boundaries() {
1242 let tmp = TempDir::new().expect("tempdir");
1243 let cfg = PretrainConfig {
1244 warmup_steps: 10,
1245 total_steps: 100,
1246 lr_max: 1.0e-3,
1247 lr_min: 1.0e-5,
1248 ..test_config(tmp.path())
1249 };
1250 let step_fn = LinearDecaySynthetic { start_loss: 1.0, decay_per_step: 0.0, grad_norm: 0.1 };
1251 let val_fn = ScriptedVal { sequence: vec![1.0] };
1252 let loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1253
1254 assert!((loop_.lr_at(0) - 0.0).abs() < 1e-9);
1256 assert!((loop_.lr_at(10) - 1.0e-3).abs() < 1e-6);
1258 assert!((loop_.lr_at(100) - 1.0e-5).abs() < 1e-6);
1260 }
1261
1262 #[test]
1264 fn epoch_artifact_paths_match_contract_template() {
1265 let tmp = TempDir::new().expect("tempdir");
1266 let run_dir = tmp.path().join("run");
1267 let metadata = EpochMetadata {
1268 epoch: 7,
1269 train_loss: 3.0,
1270 val_loss: 2.8,
1271 train_ppl: 20.0,
1272 val_ppl: 16.4,
1273 optimizer_state_sha: "deadbeef".into(),
1274 wall_seconds: 42.0,
1275 tokens_seen: 1_000_000,
1276 grad_norm_max: 1.5,
1277 };
1278 let art = EpochArtifact::new(&run_dir, 7, metadata);
1279 assert!(art.checkpoint_path.ends_with("ckpt/epoch-007.apr"));
1280 assert!(art.metadata_path.ends_with("ckpt/epoch-007.metadata.json"));
1281 }
1282
1283 struct RecordingCheckpointFn {
1289 calls: Rc<RefCell<Vec<(usize, PathBuf)>>>,
1290 }
1291
1292 impl CheckpointFn for RecordingCheckpointFn {
1293 fn save(&mut self, epoch: usize, artifact: &EpochArtifact) -> Result<(), String> {
1294 self.calls.borrow_mut().push((epoch, artifact.checkpoint_path.clone()));
1295 Ok(())
1296 }
1297 }
1298
1299 #[test]
1302 fn pretrain_loop_calls_checkpoint_fn_once_per_passing_epoch() {
1303 let tmp = TempDir::new().expect("tempdir");
1304 let cfg = test_config(tmp.path());
1305 let step_fn = LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1306 let val_fn = ScriptedVal { sequence: vec![3.4, 3.0, 2.6, 2.2, 2.0] };
1307 let calls: Rc<RefCell<Vec<(usize, PathBuf)>>> = Rc::new(RefCell::new(Vec::new()));
1308 let ckpt = RecordingCheckpointFn { calls: Rc::clone(&calls) };
1309
1310 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn).with_checkpoint_fn(Box::new(ckpt));
1311 let _status = loop_.run();
1312
1313 let recorded = calls.borrow();
1314 let epoch_count = loop_.epoch_artifacts().len();
1315 assert!(epoch_count >= 1, "at least one epoch should have completed");
1316 assert_eq!(
1317 recorded.len(),
1318 epoch_count,
1319 "CheckpointFn must fire exactly once per epoch that passes GATE-TRAIN-005",
1320 );
1321 for (i, (epoch, path)) in recorded.iter().enumerate() {
1322 assert_eq!(*epoch, i, "checkpoint hook epoch indices must be monotonic from 0");
1323 assert!(
1324 path.to_string_lossy().contains(&format!("epoch-{:03}.apr", epoch)),
1325 "checkpoint path must match contract template: {:?}",
1326 path,
1327 );
1328 let meta_path = path.with_extension("metadata.json");
1329 assert!(
1330 meta_path.exists(),
1331 "companion metadata.json must be written for epoch {}",
1332 epoch,
1333 );
1334 }
1335 }
1336
1337 #[test]
1342 fn pretrain_loop_uses_step_fn_optimizer_sha_when_available() {
1343 struct ShaOverride {
1344 inner: LinearDecaySynthetic,
1345 sha: String,
1346 }
1347 impl StepFn for ShaOverride {
1348 fn step(&mut self, s: u64, lr: f32, tokens: u64) -> (f32, f32) {
1349 self.inner.step(s, lr, tokens)
1350 }
1351 fn optimizer_state_sha256(&self) -> Option<String> {
1352 Some(self.sha.clone())
1353 }
1354 }
1355
1356 let tmp = TempDir::new().expect("tempdir");
1357 let cfg = test_config(tmp.path());
1358 let step_fn = ShaOverride {
1359 inner: LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 },
1360 sha: "a".repeat(64),
1361 };
1362 let val_fn = ScriptedVal { sequence: vec![3.4, 3.0, 2.6, 2.2, 2.0] };
1363 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1364 let _ = loop_.run();
1365
1366 let arts = loop_.epoch_artifacts();
1367 assert!(!arts.is_empty(), "at least one epoch should have completed");
1368 for art in arts {
1369 assert_eq!(
1370 art.metadata.optimizer_state_sha,
1371 "a".repeat(64),
1372 "StepFn override must win over fake_optimizer_sha fallback",
1373 );
1374 }
1375 }
1376
1377 #[test]
1382 fn pretrain_loop_falls_back_to_fake_optimizer_sha_for_synthetic() {
1383 let tmp = TempDir::new().expect("tempdir");
1384 let cfg = test_config(tmp.path());
1385 let step_fn = LinearDecaySynthetic { start_loss: 3.5, decay_per_step: 0.1, grad_norm: 0.8 };
1386 let val_fn = ScriptedVal { sequence: vec![3.4, 3.0, 2.6, 2.2, 2.0] };
1387 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn);
1388 let _ = loop_.run();
1389
1390 for art in loop_.epoch_artifacts() {
1391 assert_eq!(
1392 art.metadata.optimizer_state_sha.len(),
1393 64,
1394 "fallback fingerprint must still be a 64-char hex digest",
1395 );
1396 assert!(
1397 art.metadata.optimizer_state_sha.chars().all(|c| c.is_ascii_hexdigit()),
1398 "fallback fingerprint must be lowercase hex",
1399 );
1400 }
1401 }
1402
1403 #[test]
1406 fn pretrain_loop_skips_checkpoint_on_abort() {
1407 let tmp = TempDir::new().expect("tempdir");
1408 let cfg = test_config(tmp.path());
1409 let step_fn = NanAtStepSynthetic { nan_step: 1 };
1410 let val_fn = ScriptedVal { sequence: vec![3.0] };
1411 let calls: Rc<RefCell<Vec<(usize, PathBuf)>>> = Rc::new(RefCell::new(Vec::new()));
1412 let ckpt = RecordingCheckpointFn { calls: Rc::clone(&calls) };
1413
1414 let mut loop_ = PretrainLoop::new(cfg, step_fn, val_fn).with_checkpoint_fn(Box::new(ckpt));
1415 let status = loop_.run();
1416
1417 assert!(
1418 matches!(status, RunStatus::Aborted(PretrainAbort::NumericalInstability { .. })),
1419 "NaN must abort the loop: got {status:?}",
1420 );
1421 assert!(
1422 calls.borrow().is_empty(),
1423 "CheckpointFn must NOT fire when the epoch aborts before GATE-TRAIN-005 passes",
1424 );
1425 }
1426}