1use std::collections::{BTreeMap, BTreeSet};
10
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13
14use crate::campaign::stable_json_fingerprint;
15use crate::canonical::parse_typed_json;
16use crate::fold::FoldSet;
17use crate::metrics::ScoreSet;
18use crate::plan::CampaignSpec;
19use crate::runtime::InMemoryLineageRecorder;
20use crate::selection::{RefitStrategy, SelectionPolicy};
21use crate::training::{TrainingInfluenceKind, TrainingInfluenceManifest};
22
23pub const HPO_MANIFEST_SCHEMA_VERSION: u32 = 1;
24pub const N4MOPT_CHECKPOINT_SCHEMA_VERSION: u32 = 1;
25pub const N4MOPT_ARTIFACT_KIND: &str = "n4m_optimizer_checkpoint";
26pub const N4MOPT_FORMAT: &str = "N4MOPT";
27pub const METHODS_ABI_MAJOR: u32 = 2;
28pub const METHODS_RUNTIME_ABI_MINOR: u32 = 5;
29pub const METHODS_PLS_N4MM_MIN_ABI_MINOR: u32 = 0;
30pub const METHODS_PIPELINE_N4MM_MIN_ABI_MINOR: u32 = 5;
31pub const METHODS_N4MOPT_MIN_ABI_MINOR: u32 = 2;
32pub const METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR: u32 = 3;
33
34const fn methods_abi_major_default() -> u32 {
35 METHODS_ABI_MAJOR
36}
37
38const fn methods_n4mopt_min_abi_minor_default() -> u32 {
39 METHODS_N4MOPT_MIN_ABI_MINOR
40}
41pub const MAX_N4MOPT_CHECKPOINT_BYTES: usize = 64 * 1024 * 1024;
44
45pub fn methods_n4mm_abi_requirement(
51 artifact: &crate::runtime::ArtifactRef,
52) -> crate::Result<(u32, u32)> {
53 if artifact.kind != "n4m_model"
54 || artifact.backend != Some(crate::runtime::ArtifactBackend::Raw)
55 {
56 return Err(crate::DagMlError::RuntimeValidation(format!(
57 "native Methods artifact `{}` must be a raw n4m_model",
58 artifact.id
59 )));
60 }
61 let expected_minor = match artifact.controller_id.as_str() {
62 METHODS_PLS_CONTROLLER_ID => Some(
63 if artifact
64 .native_predictor_descriptor
65 .as_ref()
66 .is_some_and(|descriptor| descriptor.pipeline.is_some())
67 {
68 METHODS_PIPELINE_N4MM_MIN_ABI_MINOR
69 } else {
70 METHODS_PLS_N4MM_MIN_ABI_MINOR
71 },
72 ),
73 METHODS_RIDGE_CONTROLLER_ID => Some(METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR),
74 _ => None,
75 };
76 match (artifact.abi_major, artifact.abi_min_minor, expected_minor) {
77 (Some(METHODS_ABI_MAJOR), Some(minor), Some(expected)) if minor == expected => {
78 Ok((METHODS_ABI_MAJOR, minor))
79 }
80 (Some(METHODS_ABI_MAJOR), Some(minor), None) => Ok((METHODS_ABI_MAJOR, minor)),
81 (None, None, Some(METHODS_PLS_N4MM_MIN_ABI_MINOR)) => {
82 Ok((METHODS_ABI_MAJOR, METHODS_PLS_N4MM_MIN_ABI_MINOR))
83 }
84 (None, None, _) => Err(crate::DagMlError::RuntimeValidation(format!(
85 "native Methods artifact `{}` requires an explicit ABI minimum for controller `{}`",
86 artifact.id, artifact.controller_id
87 ))),
88 (major, minor, expected) => Err(crate::DagMlError::RuntimeValidation(format!(
89 "native Methods artifact `{}` declares ABI {:?}.{:?}; controller `{}` requires exactly {}.{}",
90 artifact.id,
91 major,
92 minor,
93 artifact.controller_id,
94 METHODS_ABI_MAJOR,
95 expected.unwrap_or_default()
96 ))),
97 }
98}
99
100pub fn validate_methods_abi_compatibility(
101 runtime_major: u32,
102 runtime_minor: u32,
103 required_major: u32,
104 required_min_minor: u32,
105) -> crate::Result<()> {
106 if runtime_major != required_major || runtime_minor < required_min_minor {
107 return Err(crate::DagMlError::RuntimeValidation(format!(
108 "Methods runtime ABI {runtime_major}.{runtime_minor} cannot consume payload requiring {required_major}.{required_min_minor}+"
109 )));
110 }
111 Ok(())
112}
113
114pub(crate) fn campaign_provenance_fingerprint(campaign: &CampaignSpec) -> crate::Result<String> {
120 let mut canonical = campaign.clone();
121 if let Some(serde_json::Value::Object(operation)) =
122 canonical.metadata.get_mut("methods_hpo_operation")
123 {
124 operation.remove("resume_package_json");
125 operation.remove("trials");
128 }
129 stable_json_fingerprint(&canonical)
130}
131
132pub type HpoResult<T> = std::result::Result<T, HpoError>;
133
134#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
137#[serde(deny_unknown_fields)]
138pub struct HpoNativeError {
139 pub status: i32,
140 pub kind: String,
141 pub message: String,
142 pub retryable: bool,
143}
144
145#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
146#[serde(tag = "code", rename_all = "snake_case")]
147pub enum HpoError {
148 MethodsOptimizerFeatureDisabled,
149 RuntimeConfiguration {
150 reason: String,
151 },
152 InvalidManifest {
153 reason: String,
154 },
155 InvalidSearchSpace {
156 reason: String,
157 },
158 InvalidTrial {
159 reason: String,
160 },
161 InvalidCheckpoint {
162 reason: String,
163 },
164 CheckpointBindingMismatch {
165 reason: String,
166 },
167 Native {
168 operation: String,
169 error: HpoNativeError,
170 },
171 PartialBatch {
172 committed: Vec<HpoTrial>,
173 error: HpoNativeError,
174 },
175 Evaluation {
176 reason: String,
177 },
178}
179
180impl std::fmt::Display for HpoError {
181 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
182 match self {
183 Self::MethodsOptimizerFeatureDisabled => f.write_str(
184 "Methods optimizer support is disabled; enable the published `methods-optimizer` feature",
185 ),
186 Self::RuntimeConfiguration { reason } => {
187 write!(f, "invalid Methods runtime configuration: {reason}")
188 }
189 Self::InvalidManifest { reason } => write!(f, "invalid HPO manifest: {reason}"),
190 Self::InvalidSearchSpace { reason } => write!(f, "invalid HPO search space: {reason}"),
191 Self::InvalidTrial { reason } => write!(f, "invalid native HPO trial: {reason}"),
192 Self::InvalidCheckpoint { reason } => write!(f, "invalid N4MOPT checkpoint: {reason}"),
193 Self::CheckpointBindingMismatch { reason } => write!(f, "checkpoint binding mismatch: {reason}"),
194 Self::Native { operation, error } => write!(f, "n4m {operation} failed ({}/{}): {}", error.kind, error.status, error.message),
195 Self::PartialBatch { committed, error } => write!(f, "n4m ask_batch committed {} trial(s), then failed ({}/{}): {}", committed.len(), error.kind, error.status, error.message),
196 Self::Evaluation { reason } => write!(f, "HPO evaluation failed: {reason}"),
197 }
198 }
199}
200impl std::error::Error for HpoError {}
201
202#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
203#[serde(deny_unknown_fields)]
204pub struct HpoStudyBinding {
205 pub controller_id: String,
206 pub study_id: String,
207 pub search_space_fingerprint: String,
208 pub optimizer_fingerprint: String,
209}
210
211impl HpoStudyBinding {
212 pub fn validate(&self) -> HpoResult<()> {
213 for (field, value) in [
214 ("controller_id", &self.controller_id),
215 ("study_id", &self.study_id),
216 ("search_space_fingerprint", &self.search_space_fingerprint),
217 ("optimizer_fingerprint", &self.optimizer_fingerprint),
218 ] {
219 if value.trim().is_empty() {
220 return Err(HpoError::InvalidManifest {
221 reason: format!("{field} must not be empty"),
222 });
223 }
224 }
225 Ok(())
226 }
227}
228
229#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
230#[serde(deny_unknown_fields)]
231pub struct MethodsHpoControllerManifest {
232 pub schema_version: u32,
233 pub binding: HpoStudyBinding,
234}
235
236impl MethodsHpoControllerManifest {
237 pub fn validate(&self) -> HpoResult<()> {
238 if self.schema_version != HPO_MANIFEST_SCHEMA_VERSION {
239 return Err(HpoError::InvalidManifest {
240 reason: format!(
241 "unsupported schema_version {}; expected {HPO_MANIFEST_SCHEMA_VERSION}",
242 self.schema_version
243 ),
244 });
245 }
246 self.binding.validate()
247 }
248}
249
250#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
253#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
254pub enum HpoParameter {
255 Int {
256 name: String,
257 low: i64,
258 high: i64,
259 step: i64,
260 log: bool,
261 },
262 Float {
263 name: String,
264 low: f64,
265 high: f64,
266 step: f64,
267 log: bool,
268 },
269 Categorical {
270 name: String,
271 values: Vec<HpoCategory>,
272 },
273 Ordinal {
274 name: String,
275 values: Vec<f64>,
276 },
277 SortedTuple {
278 name: String,
279 length: i32,
280 low: f64,
281 high: f64,
282 integer: bool,
283 },
284}
285
286impl HpoParameter {
287 fn name(&self) -> &str {
288 match self {
289 Self::Int { name, .. }
290 | Self::Float { name, .. }
291 | Self::Categorical { name, .. }
292 | Self::Ordinal { name, .. }
293 | Self::SortedTuple { name, .. } => name,
294 }
295 }
296 fn output_names(&self) -> Vec<String> {
297 match self {
298 Self::SortedTuple { name, length, .. } => (0..*length)
299 .map(|index| format!("{name}#{index}"))
300 .collect(),
301 _ => vec![self.name().to_string()],
302 }
303 }
304 fn validate(&self) -> HpoResult<()> {
305 if self.name().trim().is_empty() {
306 return Err(HpoError::InvalidSearchSpace {
307 reason: "parameter name must not be empty".to_string(),
308 });
309 }
310 match self {
311 Self::Int {
312 low, high, step, ..
313 } if low > high || *step <= 0 => Err(HpoError::InvalidSearchSpace {
314 reason: format!(
315 "integer parameter `{}` has invalid bounds or step",
316 self.name()
317 ),
318 }),
319 Self::Float {
320 low, high, step, ..
321 } if !low.is_finite()
322 || !high.is_finite()
323 || !step.is_finite()
324 || low > high
325 || *step < 0.0 =>
326 {
327 Err(HpoError::InvalidSearchSpace {
328 reason: format!(
329 "float parameter `{}` has invalid bounds or step",
330 self.name()
331 ),
332 })
333 }
334 Self::Categorical { values, .. } if values.is_empty() => {
335 Err(HpoError::InvalidSearchSpace {
336 reason: format!("categorical parameter `{}` has no values", self.name()),
337 })
338 }
339 Self::Ordinal { values, .. }
340 if values.is_empty() || values.iter().any(|value| !value.is_finite()) =>
341 {
342 Err(HpoError::InvalidSearchSpace {
343 reason: format!("ordinal parameter `{}` is invalid", self.name()),
344 })
345 }
346 Self::SortedTuple {
347 length, low, high, ..
348 } if *length <= 0 || !low.is_finite() || !high.is_finite() || low > high => {
349 Err(HpoError::InvalidSearchSpace {
350 reason: format!("sorted tuple parameter `{}` is invalid", self.name()),
351 })
352 }
353 _ => Ok(()),
354 }
355 }
356}
357
358#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
359#[serde(rename_all = "snake_case", untagged)]
360pub enum HpoCategory {
361 String(String),
362 Integer(i64),
363 Float(f64),
364 Boolean(bool),
365}
366
367#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
368#[serde(deny_unknown_fields)]
369pub struct HpoSearchSpace {
370 pub parameters: Vec<HpoParameter>,
371}
372
373impl HpoSearchSpace {
374 pub fn validate(&self) -> HpoResult<()> {
375 if self.parameters.is_empty() {
376 return Err(HpoError::InvalidSearchSpace {
377 reason: "search space has no parameters".to_string(),
378 });
379 }
380 let mut names = BTreeSet::new();
381 for parameter in &self.parameters {
382 parameter.validate()?;
383 for name in parameter.output_names() {
384 if !names.insert(name.clone()) {
385 return Err(HpoError::InvalidSearchSpace {
386 reason: format!("duplicate emitted parameter `{name}`"),
387 });
388 }
389 }
390 }
391 Ok(())
392 }
393 pub fn fingerprint(&self) -> HpoResult<String> {
396 self.validate()?;
397 let json = serde_json::to_string(self).map_err(|error| HpoError::InvalidSearchSpace {
398 reason: error.to_string(),
399 })?;
400 parse_typed_json(&json)
401 .and_then(|value| value.fingerprint())
402 .map_err(|error| HpoError::InvalidSearchSpace {
403 reason: format!("cannot canonically fingerprint search space: {error}"),
404 })
405 }
406}
407
408#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
412#[serde(deny_unknown_fields)]
413pub struct MethodsHpoStudyConfig {
414 pub controller_id: String,
415 pub study_id: String,
416 pub methods_abi: String,
420 pub search_space: HpoSearchSpace,
421 pub optimizer: HpoOptimizerConfig,
422}
423
424impl MethodsHpoStudyConfig {
425 #[cfg(feature = "methods-optimizer")]
426 fn methods_abi_identity(&self) -> HpoResult<String> {
427 if self.methods_abi.trim().is_empty() {
428 return Err(HpoError::InvalidManifest {
429 reason: "Methods ABI identity must be supplied by the native controller"
430 .to_string(),
431 });
432 }
433 Ok(self.methods_abi.clone())
434 }
435}
436
437#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
438#[serde(deny_unknown_fields)]
439pub struct HpoOptimizerConfig {
440 pub sampler: HpoSampler,
441 pub pruner: HpoPruner,
442 pub direction: HpoDirection,
443 pub metric: HpoMetric,
444 pub seed: u64,
445 pub n_startup_trials: i32,
446 pub max_resource: i32,
447 pub reduction_factor: i32,
448}
449
450impl HpoOptimizerConfig {
451 #[cfg(feature = "methods-optimizer")]
452 fn fingerprint(&self) -> HpoResult<String> {
453 let json = serde_json::to_string(self).map_err(|error| HpoError::InvalidManifest {
454 reason: error.to_string(),
455 })?;
456 parse_typed_json(&json)
457 .and_then(|value| value.fingerprint())
458 .map_err(|error| HpoError::InvalidManifest {
459 reason: format!("cannot canonically fingerprint optimizer configuration: {error}"),
460 })
461 }
462}
463
464#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
465#[serde(rename_all = "snake_case")]
466pub enum HpoSampler {
467 Random,
468 Sobol,
469 Lhs,
470 Ternary,
471 Ga,
472 Pso,
473 Cmaes,
474 Tpe,
475 GpEi,
476}
477#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
478#[serde(rename_all = "snake_case")]
479pub enum HpoPruner {
480 None,
481 Median,
482 Asha,
483 Hyperband,
484 Racing,
485}
486#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
487#[serde(rename_all = "snake_case")]
488pub enum HpoDirection {
489 Auto,
490 Minimize,
491 Maximize,
492}
493#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
494#[serde(rename_all = "snake_case")]
495pub enum HpoMetric {
496 Rmse,
497 Mse,
498 Mae,
499 R2,
500 Accuracy,
501 BalancedAccuracy,
502 F1,
503 Logloss,
504}
505
506#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
507#[serde(deny_unknown_fields)]
508pub struct HpoTrialParameter {
509 pub name: String,
510 pub value: f64,
511 #[serde(default)]
514 pub native_kind: Option<HpoNativeParameterKind>,
515 #[serde(default)]
516 pub category_type: Option<HpoCategoryType>,
517 #[serde(default)]
518 pub integer: bool,
519 pub active: bool,
520 pub category_index: Option<i32>,
521 pub category_label: Option<String>,
522}
523
524#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
525#[serde(rename_all = "snake_case")]
526pub enum HpoNativeParameterKind {
527 Int,
528 Float,
529 LogInt,
530 LogFloat,
531 Categorical,
532 Ordinal,
533 SortedTuple,
534}
535
536#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
537#[serde(rename_all = "snake_case")]
538pub enum HpoCategoryType {
539 String,
540 Integer,
541 Float,
542 Boolean,
543}
544
545#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
546#[serde(rename_all = "snake_case")]
547pub enum HpoTrialStatus {
548 Running,
549 Completed,
550 Pruned,
551 Failed,
552 Cancelled,
553}
554
555#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
556#[serde(deny_unknown_fields)]
557pub struct HpoFailure {
558 pub code: String,
559 pub message: String,
560 pub retryable: bool,
561}
562
563#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
564#[serde(deny_unknown_fields)]
565pub struct HpoIntermediate {
566 pub sequence: i64,
567 pub step: i32,
568 pub score: f64,
569 pub should_prune: bool,
570}
571
572#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
573#[serde(deny_unknown_fields)]
574pub struct HpoTrial {
575 pub id: i64,
576 pub ask_sequence: i64,
577 pub terminal_sequence: Option<i64>,
578 pub parameters: BTreeMap<String, HpoTrialParameter>,
579 pub parameter_order: Vec<String>,
580 pub status: HpoTrialStatus,
581 pub score: Option<f64>,
582 pub rung: i32,
583 pub duration: f64,
584 pub intermediates: Vec<HpoIntermediate>,
585 pub failure: Option<HpoFailure>,
586}
587
588#[cfg(any(test, feature = "methods-optimizer"))]
595fn canonical_hpo_terminal_ledger(trials: Vec<HpoTrial>) -> crate::Result<Vec<HpoTrial>> {
596 let json = serde_json::to_string(&trials)?;
597 parse_typed_json(&json).map_err(|error| {
598 crate::DagMlError::RuntimeValidation(format!(
599 "native Methods HPO terminal ledger has no strict TCV1 JSON preimage: {error}"
600 ))
601 })?;
602 Ok(serde_json::from_str(&json)?)
603}
604
605#[cfg(any(test, feature = "methods-optimizer"))]
606fn scores_within_one_ulp(left: f64, right: f64) -> bool {
607 if left == right {
608 return true;
609 }
610 if !left.is_finite() || !right.is_finite() {
611 return false;
612 }
613 let ordered = |value: f64| {
614 let bits = value.to_bits();
615 if bits & (1_u64 << 63) != 0 {
616 (!bits) as i128
617 } else {
618 (bits | (1_u64 << 63)) as i128
619 }
620 };
621 (ordered(left) - ordered(right)).abs() <= 1
622}
623
624#[cfg(any(test, feature = "methods-optimizer"))]
625fn optional_scores_within_one_ulp(left: Option<f64>, right: Option<f64>) -> bool {
626 match (left, right) {
627 (Some(left), Some(right)) => scores_within_one_ulp(left, right),
628 (None, None) => true,
629 _ => false,
630 }
631}
632
633#[cfg(any(test, feature = "methods-optimizer"))]
634fn hpo_terminal_trials_match(native: &[HpoTrial], persisted: &[HpoTrial]) -> bool {
635 native.len() == persisted.len()
636 && native.iter().zip(persisted).all(|(native, persisted)| {
637 native.id == persisted.id
638 && native.ask_sequence == persisted.ask_sequence
639 && native.terminal_sequence == persisted.terminal_sequence
640 && native.parameters == persisted.parameters
641 && native.parameter_order == persisted.parameter_order
642 && native.status == persisted.status
643 && optional_scores_within_one_ulp(native.score, persisted.score)
644 && native.rung == persisted.rung
645 && native.duration == persisted.duration
646 && native.failure == persisted.failure
647 && native.intermediates.len() == persisted.intermediates.len()
648 && native
649 .intermediates
650 .iter()
651 .zip(&persisted.intermediates)
652 .all(|(native, persisted)| {
653 native.sequence == persisted.sequence
654 && native.step == persisted.step
655 && native.should_prune == persisted.should_prune
656 && scores_within_one_ulp(native.score, persisted.score)
657 })
658 })
659}
660
661#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
664#[serde(deny_unknown_fields)]
665pub struct HpoBestTrial {
666 pub trial: HpoTrial,
667 pub score: f64,
668}
669
670#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
671#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)]
672pub enum HpoEvent {
673 Asked {
674 trial_id: i64,
675 },
676 Intermediate {
677 trial_id: i64,
678 step: i32,
679 score: f64,
680 should_prune: bool,
681 },
682 Terminal {
683 trial_id: i64,
684 status: HpoTrialStatus,
685 score: Option<f64>,
686 failure: Option<HpoFailure>,
687 },
688}
689
690pub struct HpoEvaluationBoundary<'a> {
694 pub folds: &'a FoldSet,
695 pub influence: &'a TrainingInfluenceManifest,
696 pub lineage: &'a mut InMemoryLineageRecorder,
697 pub scores: &'a mut ScoreSet,
698 pub selection: &'a SelectionPolicy,
699 pub refit_strategy: Option<RefitStrategy>,
700}
701
702impl HpoEvaluationBoundary<'_> {
703 pub fn validate(&self) -> HpoResult<()> {
704 self.folds
705 .validate()
706 .map_err(|error| HpoError::Evaluation {
707 reason: error.to_string(),
708 })?;
709 self.scores
710 .validate()
711 .map_err(|error| HpoError::Evaluation {
712 reason: error.to_string(),
713 })?;
714 self.selection
715 .validate()
716 .map_err(|error| HpoError::Evaluation {
717 reason: error.to_string(),
718 })?;
719 self.influence
720 .validate()
721 .map_err(|error| HpoError::Evaluation {
722 reason: error.to_string(),
723 })?;
724 if !self
725 .influence
726 .entries
727 .iter()
728 .any(|entry| entry.kind == TrainingInfluenceKind::HpoSelection)
729 {
730 return Err(HpoError::Evaluation {
731 reason: "training influence manifest has no hpo_selection entry".to_string(),
732 });
733 }
734 Ok(())
735 }
736}
737
738pub trait HpoEvaluator {
739 fn evaluate(
740 &mut self,
741 trial: &HpoTrial,
742 boundary: &mut HpoEvaluationBoundary<'_>,
743 ) -> HpoResult<HpoTerminal>;
744
745 fn evaluate_with_reporter(
749 &mut self,
750 trial: &HpoTrial,
751 boundary: &mut HpoEvaluationBoundary<'_>,
752 _reporter: &mut dyn HpoIntermediateReporter,
753 ) -> HpoResult<HpoTerminal> {
754 self.evaluate(trial, boundary)
755 }
756}
757
758pub trait HpoIntermediateReporter {
759 fn report(&mut self, step: i32, score: f64) -> HpoResult<HpoReportOutcome>;
760}
761
762#[derive(Clone, Debug, PartialEq)]
763pub enum HpoReportOutcome {
764 Continue,
765 Pruned(HpoTrial),
766}
767
768#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
769#[serde(tag = "state", rename_all = "snake_case")]
770pub enum HpoTerminal {
771 Completed { score: f64 },
772 Failed { failure: HpoFailure },
773 Pruned { failure: HpoFailure },
774 Cancelled { failure: HpoFailure },
775}
776
777#[derive(Clone, Debug, PartialEq)]
778pub struct HpoBatch {
779 pub trials: Vec<HpoTrial>,
780 pub native_error: Option<HpoNativeError>,
781}
782
783#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
784#[serde(deny_unknown_fields)]
785pub struct N4moptCheckpointArtifact {
786 pub schema_version: u32,
787 pub artifact_kind: String,
788 pub format: String,
789 #[serde(default = "methods_abi_major_default")]
790 pub abi_major: u32,
791 #[serde(default = "methods_n4mopt_min_abi_minor_default")]
792 pub abi_min_minor: u32,
793 pub binding: HpoStudyBinding,
794 pub methods_abi: String,
795 pub opaque_payload: Vec<u8>,
796 pub payload_sha256: String,
797}
798
799impl N4moptCheckpointArtifact {
800 #[cfg(feature = "methods-optimizer")]
801 fn new(
802 binding: HpoStudyBinding,
803 methods_abi: String,
804 opaque_payload: Vec<u8>,
805 ) -> HpoResult<Self> {
806 let value = Self {
807 schema_version: N4MOPT_CHECKPOINT_SCHEMA_VERSION,
808 artifact_kind: N4MOPT_ARTIFACT_KIND.to_string(),
809 format: N4MOPT_FORMAT.to_string(),
810 abi_major: METHODS_ABI_MAJOR,
811 abi_min_minor: METHODS_N4MOPT_MIN_ABI_MINOR,
812 binding,
813 methods_abi,
814 payload_sha256: payload_sha256(&opaque_payload),
815 opaque_payload,
816 };
817 value.validate()?;
818 Ok(value)
819 }
820 pub fn validate(&self) -> HpoResult<()> {
821 if self.schema_version != N4MOPT_CHECKPOINT_SCHEMA_VERSION
822 || self.artifact_kind != N4MOPT_ARTIFACT_KIND
823 || self.format != N4MOPT_FORMAT
824 || self.abi_major != METHODS_ABI_MAJOR
825 || self.abi_min_minor != METHODS_N4MOPT_MIN_ABI_MINOR
826 {
827 return Err(HpoError::InvalidCheckpoint {
828 reason: "checkpoint schema, kind, or format is invalid".to_string(),
829 });
830 }
831 self.binding.validate()?;
832 if self.methods_abi.trim().is_empty()
833 || self.opaque_payload.is_empty()
834 || self.opaque_payload.len() > MAX_N4MOPT_CHECKPOINT_BYTES
835 {
836 return Err(HpoError::InvalidCheckpoint {
837 reason: "checkpoint ABI/payload is invalid or exceeds the maximum size".to_string(),
838 });
839 }
840 if self.payload_sha256 != payload_sha256(&self.opaque_payload) {
841 return Err(HpoError::InvalidCheckpoint {
842 reason: "checkpoint payload SHA-256 differs from envelope".to_string(),
843 });
844 }
845 Ok(())
846 }
847}
848
849#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
855#[serde(deny_unknown_fields)]
856pub struct N4moptCheckpointReference {
857 pub artifact: crate::runtime::ArtifactRef,
858 pub binding: HpoStudyBinding,
859 pub methods_abi: String,
860 #[serde(default = "methods_abi_major_default")]
861 pub abi_major: u32,
862 #[serde(default = "methods_n4mopt_min_abi_minor_default")]
863 pub abi_min_minor: u32,
864}
865
866impl N4moptCheckpointReference {
867 pub fn validate(&self) -> HpoResult<()> {
868 self.binding.validate()?;
869 if self.methods_abi.trim().is_empty() {
870 return Err(HpoError::InvalidCheckpoint {
871 reason: "checkpoint reference has no Methods ABI identity".to_string(),
872 });
873 }
874 if self.abi_major != METHODS_ABI_MAJOR
875 || self.abi_min_minor != METHODS_N4MOPT_MIN_ABI_MINOR
876 || self.artifact.abi_major != Some(self.abi_major)
877 || self.artifact.abi_min_minor != Some(self.abi_min_minor)
878 {
879 return Err(HpoError::InvalidCheckpoint {
880 reason: format!(
881 "checkpoint reference must declare Methods ABI {}.{}+ on both envelope and artifact",
882 METHODS_ABI_MAJOR, METHODS_N4MOPT_MIN_ABI_MINOR
883 ),
884 });
885 }
886 self.artifact
887 .validate_portable()
888 .map_err(|error| HpoError::InvalidCheckpoint {
889 reason: format!("checkpoint archive reference is invalid: {error}"),
890 })?;
891 if self.artifact.kind != N4MOPT_ARTIFACT_KIND
892 || self.artifact.controller_id.as_str() != self.binding.controller_id
893 || self.artifact.backend != Some(crate::runtime::ArtifactBackend::Raw)
894 {
895 return Err(HpoError::InvalidCheckpoint {
896 reason: "checkpoint reference must be a raw Methods-owned N4MOPT artifact"
897 .to_string(),
898 });
899 }
900 Ok(())
901 }
902}
903
904fn payload_sha256(payload: &[u8]) -> String {
905 format!("{:x}", Sha256::digest(payload))
906}
907
908pub fn methods_optimizer_preflight() -> HpoResult<()> {
911 #[cfg(feature = "methods-optimizer")]
912 {
913 Ok(())
914 }
915 #[cfg(not(feature = "methods-optimizer"))]
916 {
917 Err(HpoError::MethodsOptimizerFeatureDisabled)
918 }
919}
920
921pub const METHODS_PLS_CONTROLLER_ID: &str = crate::runtime::NATIVE_PREDICTOR_METHODS_PLS_OWNER;
926
927pub const METHODS_RIDGE_CONTROLLER_ID: &str = crate::runtime::NATIVE_PREDICTOR_METHODS_RIDGE_OWNER;
933
934#[cfg(feature = "methods-optimizer")]
935#[derive(Clone, Debug, Eq, PartialEq, Deserialize)]
936#[serde(deny_unknown_fields)]
937struct MethodsPlsPipelineParamsV1 {
938 schema_version: u32,
939 pipeline_type: String,
940 savgol_window: i32,
941 savgol_poly_degree: i32,
942}
943
944#[cfg(feature = "methods-optimizer")]
945#[derive(Clone, Debug, Eq, PartialEq)]
946struct MethodsPlsParams {
947 n_components: i32,
948 pipeline: Option<MethodsPlsPipelineParamsV1>,
949}
950
951#[cfg(feature = "methods-optimizer")]
956pub fn validate_methods_pls_node_params(
957 params: &BTreeMap<String, serde_json::Value>,
958) -> crate::Result<()> {
959 methods_pls_params(params).map(|_| ())
960}
961
962#[cfg(feature = "methods-optimizer")]
963fn methods_pls_params(
964 params: &BTreeMap<String, serde_json::Value>,
965) -> crate::Result<MethodsPlsParams> {
966 let pipeline_shape =
967 params.contains_key("pipeline") && params.len() == 2 && params.contains_key("n_components");
968 let raw_shape = !params.contains_key("pipeline")
969 && params.len() == 1
970 && params.contains_key("n_components");
971 let legacy_sklearn_shape = !params.contains_key("pipeline")
972 && params.len() == 5
973 && ["copy", "max_iter", "n_components", "scale", "tol"]
974 .iter()
975 .all(|key| params.contains_key(*key));
976 if !pipeline_shape && !raw_shape && !legacy_sklearn_shape {
977 return Err(crate::DagMlError::RuntimeValidation(
978 "portable Methods PLS accepts only `n_components`, the exact historical sklearn PLS defaults, or the optional exact `pipeline` block".to_string(),
979 ));
980 }
981 if legacy_sklearn_shape
982 && (params["copy"].as_bool() != Some(true)
983 || params["max_iter"].as_i64() != Some(500)
984 || params["scale"].as_bool() != Some(true)
985 || params["tol"].as_f64() != Some(1.0e-6))
986 {
987 return Err(crate::DagMlError::RuntimeValidation(
988 "portable Methods PLS historical sklearn parameters must keep canonical defaults `copy=true`, `max_iter=500`, `scale=true`, and `tol=1e-6`".to_string(),
989 ));
990 }
991 let n_components = params["n_components"]
992 .as_i64()
993 .and_then(|value| i32::try_from(value).ok())
994 .filter(|value| *value > 0)
995 .ok_or_else(|| {
996 crate::DagMlError::RuntimeValidation(
997 "portable Methods PLS `n_components` must be a positive i32".to_string(),
998 )
999 })?;
1000 let pipeline = params
1001 .get("pipeline")
1002 .map(|value| {
1003 serde_json::from_value::<MethodsPlsPipelineParamsV1>(value.clone()).map_err(|error| {
1004 crate::DagMlError::RuntimeValidation(format!(
1005 "portable Methods PLS pipeline block is invalid: {error}"
1006 ))
1007 })
1008 })
1009 .transpose()?;
1010 if let Some(pipeline) = &pipeline {
1011 if pipeline.schema_version != 1
1012 || pipeline.pipeline_type
1013 != crate::runtime::NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1
1014 {
1015 return Err(crate::DagMlError::RuntimeValidation(
1016 "portable Methods PLS supports only the SNV -> Savitzky-Golay smooth pipeline v1"
1017 .to_string(),
1018 ));
1019 }
1020 if !(3..=501).contains(&pipeline.savgol_window)
1021 || pipeline.savgol_window % 2 == 0
1022 || pipeline.savgol_poly_degree < 0
1023 || pipeline.savgol_poly_degree >= pipeline.savgol_window
1024 {
1025 return Err(crate::DagMlError::RuntimeValidation(
1026 "portable Methods PLS pipeline has invalid Savitzky-Golay parameters".to_string(),
1027 ));
1028 }
1029 }
1030 Ok(MethodsPlsParams {
1031 n_components,
1032 pipeline,
1033 })
1034}
1035
1036#[cfg(feature = "methods-optimizer")]
1037fn validate_methods_pls_descriptor_against_params(
1038 params: &BTreeMap<String, serde_json::Value>,
1039 descriptor: Option<&crate::runtime::NativePredictorDescriptorV1>,
1040) -> crate::Result<()> {
1041 let params = methods_pls_params(params)?;
1042 let descriptor_pipeline = descriptor.and_then(|value| value.pipeline.as_ref());
1043 let matches_pipeline = match (¶ms.pipeline, descriptor_pipeline) {
1044 (None, None) => true,
1045 (Some(expected), Some(actual)) => {
1046 actual.pipeline_type == expected.pipeline_type
1047 && actual.savgol_window == expected.savgol_window
1048 && actual.savgol_poly_degree == expected.savgol_poly_degree
1049 }
1050 _ => false,
1051 };
1052 if !matches_pipeline
1053 || descriptor.is_some_and(|value| value.dimensions.n_components != params.n_components)
1054 {
1055 return Err(crate::DagMlError::RuntimeValidation(
1056 "portable Methods PLS parameters do not match the native predictor descriptor"
1057 .to_string(),
1058 ));
1059 }
1060 Ok(())
1061}
1062
1063#[cfg(feature = "methods-optimizer")]
1064pub fn inspect_methods_native_predictor_descriptor_v1(
1073 owner_controller: &crate::ControllerId,
1074 payload: &[u8],
1075) -> crate::Result<crate::runtime::NativePredictorDescriptorV1> {
1076 use crate::runtime::{
1077 NativePredictorDescriptorV1, NativePredictorDimensionsV1, NativePredictorPipelineV1,
1078 NativePredictorWriterAbiV1, NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1,
1079 NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1, NATIVE_PREDICTOR_FORMAT_N4MM,
1080 NATIVE_PREDICTOR_PIPELINE_FINGERPRINT_FNV1A64_V1,
1081 NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1,
1082 };
1083
1084 let info = n4m::inspect_n4mm(payload).map_err(|error| {
1085 crate::DagMlError::RuntimeValidation(format!(
1086 "native Methods predictor inspection failed: {error}"
1087 ))
1088 })?;
1089 let pipeline = info.pipeline.map(|pipeline| NativePredictorPipelineV1 {
1090 pipeline_type: NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1.to_string(),
1091 schema_version: pipeline.schema_version,
1092 operator_count: pipeline.operator_count,
1093 raw_n_features: pipeline.raw_n_features,
1094 model_n_features: pipeline.model_n_features,
1095 fingerprint_algorithm: NATIVE_PREDICTOR_PIPELINE_FINGERPRINT_FNV1A64_V1.to_string(),
1096 native_fingerprint: format!("{:016x}", pipeline.fingerprint),
1097 savgol_window: pipeline.savgol_window,
1098 savgol_poly_degree: pipeline.savgol_poly_degree,
1099 });
1100 let mut descriptor = NativePredictorDescriptorV1 {
1101 descriptor_type: NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1.to_string(),
1102 schema_version: NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1,
1103 artifact_sha256: format!("{:x}", Sha256::digest(payload)),
1104 owner_controller: owner_controller.clone(),
1105 format: NATIVE_PREDICTOR_FORMAT_N4MM.to_string(),
1106 format_version: info.format_version,
1107 writer_abi: NativePredictorWriterAbiV1 {
1108 major: info.writer_abi.0,
1109 minor: info.writer_abi.1,
1110 patch: info.writer_abi.2,
1111 },
1112 storage_algorithm: info.algorithm,
1113 capabilities: info.capabilities,
1114 dimensions: NativePredictorDimensionsV1 {
1115 training_samples: info.training_samples,
1116 n_features: info.n_features,
1117 n_targets: info.n_targets,
1118 n_components: info.n_components,
1119 },
1120 pipeline,
1121 descriptor_fingerprint: String::new(),
1122 };
1123 descriptor.descriptor_fingerprint = descriptor.compute_fingerprint()?;
1124 descriptor.validate()?;
1125 Ok(descriptor)
1126}
1127
1128#[cfg(feature = "methods-optimizer")]
1134#[derive(Clone, Debug, Eq, PartialEq)]
1135pub struct MethodsRuntime {
1136 library_path: std::path::PathBuf,
1137 abi_major: u32,
1138 abi_minor: u32,
1139}
1140
1141#[cfg(feature = "methods-optimizer")]
1142impl MethodsRuntime {
1143 pub fn configure(library_path: impl AsRef<std::path::Path>) -> HpoResult<Self> {
1144 let library_path = library_path.as_ref();
1145 if !library_path.is_absolute() {
1146 return Err(HpoError::RuntimeConfiguration {
1147 reason: "libn4m path must be absolute".to_string(),
1148 });
1149 }
1150 let canonical = std::fs::canonicalize(library_path).map_err(|error| {
1151 HpoError::RuntimeConfiguration {
1152 reason: format!(
1153 "cannot resolve libn4m path `{}`: {error}",
1154 library_path.display()
1155 ),
1156 }
1157 })?;
1158 let metadata =
1159 std::fs::metadata(&canonical).map_err(|error| HpoError::RuntimeConfiguration {
1160 reason: format!(
1161 "cannot inspect libn4m path `{}`: {error}",
1162 canonical.display()
1163 ),
1164 })?;
1165 if !metadata.is_file() {
1166 return Err(HpoError::RuntimeConfiguration {
1167 reason: format!(
1168 "libn4m path `{}` is not a regular file",
1169 canonical.display()
1170 ),
1171 });
1172 }
1173 n4m::configure_library(&canonical).map_err(|error| HpoError::RuntimeConfiguration {
1174 reason: format!("cannot load libn4m `{}`: {error}", canonical.display()),
1175 })?;
1176 n4m::Context::new().map_err(|error| HpoError::RuntimeConfiguration {
1180 reason: format!(
1181 "libn4m `{}` does not satisfy Methods ABI {}.{}: {error}",
1182 canonical.display(),
1183 METHODS_ABI_MAJOR,
1184 METHODS_RUNTIME_ABI_MINOR
1185 ),
1186 })?;
1187 Ok(Self {
1188 library_path: canonical,
1189 abi_major: METHODS_ABI_MAJOR,
1190 abi_minor: METHODS_RUNTIME_ABI_MINOR,
1191 })
1192 }
1193
1194 pub fn library_path(&self) -> &std::path::Path {
1195 &self.library_path
1196 }
1197
1198 fn ensure_n4mm_compatible(&self, artifact: &crate::runtime::ArtifactRef) -> crate::Result<()> {
1199 let (required_major, required_min_minor) = methods_n4mm_abi_requirement(artifact)?;
1200 validate_methods_abi_compatibility(
1201 self.abi_major,
1202 self.abi_minor,
1203 required_major,
1204 required_min_minor,
1205 )
1206 }
1207}
1208
1209#[cfg(feature = "methods-optimizer")]
1213pub struct MethodsHpoController {
1214 id: crate::ControllerId,
1215 _runtime: MethodsRuntime,
1216}
1217
1218#[cfg(feature = "methods-optimizer")]
1219impl MethodsHpoController {
1220 pub fn new(id: crate::ControllerId, runtime: MethodsRuntime) -> Self {
1221 Self {
1222 id,
1223 _runtime: runtime,
1224 }
1225 }
1226}
1227
1228#[cfg(feature = "methods-optimizer")]
1229impl crate::runtime::RuntimeController for MethodsHpoController {
1230 fn controller_id(&self) -> &crate::ControllerId {
1231 &self.id
1232 }
1233
1234 fn invoke(&self, task: &crate::runtime::NodeTask) -> crate::Result<crate::runtime::NodeResult> {
1235 Err(crate::DagMlError::RuntimeValidation(format!(
1236 "Methods HPO controller `{}` is training-owned and cannot execute graph task `{}` directly",
1237 self.id, task.node_plan.node_id
1238 )))
1239 }
1240
1241 fn create_tuner_session(
1242 &self,
1243 task: &crate::runtime::RuntimeHpoCampaignTask,
1244 context: &crate::runtime::RuntimeHpoExecutionContext,
1245 ) -> crate::Result<Box<dyn crate::runtime::RuntimeTunerSession>> {
1246 if task.operation_id != context.operation_id
1247 || task.controller_id != context.controller_id
1248 || task.target_node_id != context.target_node_id
1249 || context.study.controller_id != self.id.as_str()
1250 {
1251 return Err(crate::DagMlError::RuntimeValidation(
1252 "Methods HPO tuner task/context identity mismatch".to_string(),
1253 ));
1254 }
1255 let study = if let Some(checkpoint) = &context.resume_checkpoint {
1256 MethodsHpoStudy::restore(context.study.clone(), checkpoint)
1257 } else {
1258 MethodsHpoStudy::create(context.study.clone())
1259 }
1260 .map_err(|error| {
1261 crate::DagMlError::RuntimeValidation(format!(
1262 "cannot create controller-owned native Methods HPO study: {error}"
1263 ))
1264 })?;
1265 if context.resume_checkpoint.is_some() {
1266 let mut native = study.trials().map_err(|error| {
1267 crate::DagMlError::RuntimeValidation(format!(
1268 "cannot attest restored native Methods HPO ledger: {error}"
1269 ))
1270 })?;
1271 native.sort_by_key(|trial| trial.id);
1272 let native = canonical_hpo_terminal_ledger(native)?;
1273 let persisted = canonical_hpo_terminal_ledger(
1274 context
1275 .resume_terminal_trials
1276 .iter()
1277 .map(|snapshot| snapshot.trial.clone())
1278 .collect(),
1279 )?;
1280 if !hpo_terminal_trials_match(&native, &persisted) {
1281 return Err(crate::DagMlError::RuntimeValidation(
1282 "restored native Methods HPO ledger does not exactly match persisted terminal evidence"
1283 .to_string(),
1284 ));
1285 }
1286 }
1287 Ok(Box::new(MethodsHpoSession {
1288 study,
1289 context: context.clone(),
1290 controller_id: self.id.clone(),
1291 }))
1292 }
1293}
1294
1295#[cfg(feature = "methods-optimizer")]
1296struct MethodsHpoSession {
1297 study: MethodsHpoStudy,
1298 context: crate::runtime::RuntimeHpoExecutionContext,
1299 controller_id: crate::ControllerId,
1300}
1301
1302#[cfg(feature = "methods-optimizer")]
1303impl crate::runtime::RuntimeTunerSession for MethodsHpoSession {
1304 fn trial_history_len(&self) -> crate::Result<u32> {
1305 let count = self
1306 .study
1307 .trials()
1308 .map_err(|error| crate::DagMlError::RuntimeValidation(error.to_string()))?
1309 .len();
1310 u32::try_from(count).map_err(|_| {
1311 crate::DagMlError::RuntimeValidation(
1312 "native Methods HPO trial history exceeds u32 budget".to_string(),
1313 )
1314 })
1315 }
1316
1317 fn ask(&mut self) -> crate::Result<Option<crate::runtime::RuntimeHpoProposal>> {
1318 let trial = self
1319 .study
1320 .ask()
1321 .map_err(|error| crate::DagMlError::RuntimeValidation(error.to_string()))?;
1322 if self.context.study.search_space.parameters.len() != 1
1323 || self.context.parameter_paths.len() != 1
1324 || self.context.parameter_paths.get("n_components") != Some(&"n_components".to_string())
1325 || !matches!(self.context.study.search_space.parameters.first(), Some(HpoParameter::Int { name, low: 1, high: 3, step: 1, log: false }) if name == "n_components")
1326 {
1327 return Err(crate::DagMlError::RuntimeValidation(
1328 "Methods HPO v1 accepts only active integer n_components=1..3 mapped directly to the target model".to_string(),
1329 ));
1330 }
1331 let parameter = trial.parameters.get("n_components").ok_or_else(|| {
1332 crate::DagMlError::RuntimeValidation(
1333 "native Methods HPO trial omitted active n_components".to_string(),
1334 )
1335 })?;
1336 if !parameter.active
1337 || !parameter.integer
1338 || parameter.value.fract() != 0.0
1339 || !(1.0..=3.0).contains(¶meter.value)
1340 {
1341 return Err(crate::DagMlError::RuntimeValidation(
1342 "native Methods HPO emitted invalid n_components outside V1 integer bounds"
1343 .to_string(),
1344 ));
1345 }
1346 let mut variant = self.context.base_variant.clone();
1347 variant.choices.insert(
1348 "native_methods_hpo".to_string(),
1349 crate::generation::GenerationChoice {
1350 label: format!("trial:{}", trial.id),
1351 value: serde_json::json!({"trial_id": trial.id}),
1352 param_overrides: vec![crate::generation::GenerationParamOverride {
1353 node_id: self.context.target_node_id.clone(),
1354 params: BTreeMap::from([(
1355 "n_components".to_string(),
1356 serde_json::json!(parameter.value as i64),
1357 )]),
1358 }],
1359 active_subsequence: None,
1360 },
1361 );
1362 variant.variant_id = crate::VariantId::new(format!("hpo:trial:{}", trial.id))
1363 .map_err(|error| crate::DagMlError::RuntimeValidation(error.to_string()))?;
1364 variant.fingerprint = crate::campaign::stable_json_fingerprint(&(
1365 self.context.base_variant.fingerprint.as_str(),
1366 &variant.choices,
1367 trial.id,
1368 ))?;
1369 Ok(Some(crate::runtime::RuntimeHpoProposal {
1370 trial_id: trial.id,
1371 variant,
1372 }))
1373 }
1374
1375 fn report_intermediate(
1376 &mut self,
1377 value: crate::runtime::RuntimeHpoIntermediate,
1378 ) -> crate::Result<crate::runtime::RuntimeHpoIntermediateOutcome> {
1379 let pruned = self
1380 .study
1381 .report_intermediate(value.trial_id, value.step, value.score)
1382 .map_err(|error| crate::DagMlError::RuntimeValidation(error.to_string()))?;
1383 Ok(if pruned {
1384 crate::runtime::RuntimeHpoIntermediateOutcome::Pruned
1385 } else {
1386 crate::runtime::RuntimeHpoIntermediateOutcome::Continue
1387 })
1388 }
1389
1390 fn tell(
1391 &mut self,
1392 trial_id: i64,
1393 terminal: crate::runtime::RuntimeHpoTerminal,
1394 ) -> crate::Result<()> {
1395 let terminal = match terminal {
1396 crate::runtime::RuntimeHpoTerminal::Completed { score } => {
1397 HpoTerminal::Completed { score }
1398 }
1399 crate::runtime::RuntimeHpoTerminal::Failed { failure } => HpoTerminal::Failed {
1400 failure: HpoFailure {
1401 code: failure.code,
1402 message: failure.message,
1403 retryable: failure.retryable,
1404 },
1405 },
1406 };
1407 self.study
1408 .tell(trial_id, terminal)
1409 .map_err(|error| crate::DagMlError::RuntimeValidation(error.to_string()))?;
1410 Ok(())
1411 }
1412
1413 fn checkpoint(&self) -> crate::Result<N4moptCheckpointArtifact> {
1414 let checkpoint = self.study.save_checkpoint().map_err(|error| {
1415 crate::DagMlError::RuntimeValidation(format!(
1416 "cannot save native Methods HPO checkpoint: {error}"
1417 ))
1418 })?;
1419 checkpoint.validate().map_err(|error| {
1420 crate::DagMlError::RuntimeValidation(format!(
1421 "invalid native Methods HPO checkpoint: {error}"
1422 ))
1423 })?;
1424 if checkpoint.binding.controller_id != self.controller_id.as_str()
1425 || checkpoint.binding.controller_id != self.context.study.controller_id
1426 || checkpoint.binding.study_id != self.context.study.study_id
1427 || checkpoint.methods_abi != self.context.study.methods_abi
1428 {
1429 return Err(crate::DagMlError::RuntimeValidation(
1430 "native Methods HPO checkpoint binding/ABI does not match its scheduler context"
1431 .to_string(),
1432 ));
1433 }
1434 Ok(checkpoint)
1435 }
1436
1437 fn incumbent(
1438 &self,
1439 variants: &BTreeMap<i64, crate::VariantId>,
1440 ) -> crate::Result<Option<crate::runtime::RuntimeHpoIncumbent>> {
1441 let Some(best) = self.study.best().map_err(|error| {
1442 crate::DagMlError::RuntimeValidation(format!(
1443 "cannot read native Methods HPO incumbent: {error}"
1444 ))
1445 })?
1446 else {
1447 return Ok(None);
1448 };
1449 let score = if let Some(persisted) = self
1450 .context
1451 .resume_terminal_trials
1452 .iter()
1453 .find(|snapshot| snapshot.trial.id == best.trial.id)
1454 {
1455 let native = canonical_hpo_terminal_ledger(vec![best.trial.clone()])?;
1456 let prior = canonical_hpo_terminal_ledger(vec![persisted.trial.clone()])?;
1457 if !hpo_terminal_trials_match(&native, &prior) {
1458 return Err(crate::DagMlError::RuntimeValidation(
1459 "native Methods HPO incumbent does not match persisted terminal evidence"
1460 .to_string(),
1461 ));
1462 }
1463 persisted.trial.score.ok_or_else(|| {
1464 crate::DagMlError::RuntimeValidation(
1465 "persisted Methods HPO incumbent has no terminal score".to_string(),
1466 )
1467 })?
1468 } else {
1469 best.score
1470 };
1471 let variant_id = variants.get(&best.trial.id).cloned().ok_or_else(|| {
1472 crate::DagMlError::RuntimeValidation(
1473 "native Methods HPO best() returned a trial without scheduler variant identity"
1474 .to_string(),
1475 )
1476 })?;
1477 Ok(Some(crate::runtime::RuntimeHpoIncumbent {
1478 trial_id: best.trial.id,
1479 score,
1480 metric: self.context.selection.metric.name().to_string(),
1481 direction: self.context.selection.direction,
1482 variant_id,
1483 }))
1484 }
1485
1486 fn terminal_trial_snapshots(
1487 &self,
1488 variants: &BTreeMap<i64, crate::VariantId>,
1489 ) -> crate::Result<Vec<crate::runtime::RuntimeHpoTerminalSnapshot>> {
1490 let mut trials = self.study.trials().map_err(|error| {
1491 crate::DagMlError::RuntimeValidation(format!(
1492 "cannot read native Methods HPO terminal ledger: {error}"
1493 ))
1494 })?;
1495 trials.sort_by_key(|trial| trial.id);
1496 if trials.iter().any(|trial| {
1497 !matches!(
1498 trial.status,
1499 HpoTrialStatus::Completed | HpoTrialStatus::Pruned | HpoTrialStatus::Failed
1500 )
1501 }) {
1502 return Err(crate::DagMlError::RuntimeValidation(
1503 "native Methods HPO trial ledger contains a non-terminal trial".to_string(),
1504 ));
1505 }
1506 let persisted_by_id = self
1507 .context
1508 .resume_terminal_trials
1509 .iter()
1510 .map(|snapshot| (snapshot.trial.id, snapshot))
1511 .collect::<BTreeMap<_, _>>();
1512 trials
1513 .into_iter()
1514 .map(|trial| {
1515 if let Some(persisted) = persisted_by_id.get(&trial.id) {
1516 let native = canonical_hpo_terminal_ledger(vec![trial])?;
1517 let prior = canonical_hpo_terminal_ledger(vec![persisted.trial.clone()])?;
1518 if !hpo_terminal_trials_match(&native, &prior) {
1519 return Err(crate::DagMlError::RuntimeValidation(
1520 "restored native Methods HPO trial does not match persisted terminal evidence"
1521 .to_string(),
1522 ));
1523 }
1524 return Ok((*persisted).clone());
1525 }
1526 Ok(crate::runtime::RuntimeHpoTerminalSnapshot {
1527 variant_id: variants.get(&trial.id).cloned(),
1528 trial,
1529 })
1530 })
1531 .collect()
1532 }
1533}
1534
1535#[cfg(feature = "methods-optimizer")]
1536mod pls_controller {
1537 use std::collections::{BTreeMap, BTreeSet};
1538 use std::sync::atomic::{AtomicU64, Ordering};
1539 use std::sync::Mutex;
1540
1541 use super::*;
1542 use crate::runtime::{
1543 ArtifactBackend, ArtifactRef, HandleKind, HandleRef, LineageRecord, MethodsPlsData,
1544 MethodsPlsDataRequest, NodeResult, NodeTask, PredictionBlock, PredictionInputSpec,
1545 PredictionPartition, RegressionTargetBlock, RuntimeController, RuntimeDataProvider,
1546 };
1547 use crate::{
1548 ArtifactId, ControllerId, DagMlError, LineageId, Phase, PredictionLevel, PredictionUnitId,
1549 Result,
1550 };
1551 use n4m::{Config, Context, MatrixRef, Model};
1552
1553 pub struct MethodsPlsController {
1558 id: ControllerId,
1559 runtime: MethodsRuntime,
1560 next_handle: AtomicU64,
1561 exported_n4mm_by_artifact: Mutex<BTreeMap<ArtifactId, Vec<u8>>>,
1564 hydrated_n4mm_by_handle: Mutex<BTreeMap<u64, Vec<u8>>>,
1568 }
1569
1570 impl MethodsPlsController {
1571 pub fn new(runtime: MethodsRuntime) -> Self {
1572 Self {
1573 id: ControllerId::new(METHODS_PLS_CONTROLLER_ID)
1574 .expect("Methods PLS controller id is valid"),
1575 runtime,
1576 next_handle: AtomicU64::new(0),
1577 exported_n4mm_by_artifact: Mutex::new(BTreeMap::new()),
1578 hydrated_n4mm_by_handle: Mutex::new(BTreeMap::new()),
1579 }
1580 }
1581
1582 #[doc(hidden)]
1584 pub fn hydrated_payload_count(&self) -> Result<usize> {
1585 self.hydrated_n4mm_by_handle
1586 .lock()
1587 .map(|payloads| payloads.len())
1588 .map_err(|_| {
1589 DagMlError::RuntimeValidation(
1590 "portable Methods PLS hydrated N4MM lock poisoned".to_string(),
1591 )
1592 })
1593 }
1594
1595 fn handle(&self, kind: HandleKind) -> HandleRef {
1596 HandleRef {
1597 handle: self.next_handle.fetch_add(1, Ordering::SeqCst) + 1,
1598 kind,
1599 owner_controller: self.id.clone(),
1600 }
1601 }
1602
1603 fn request(
1604 task: &NodeTask,
1605 provider: &dyn RuntimeDataProvider,
1606 data_port: &str,
1607 ) -> Result<MethodsPlsDataRequest> {
1608 let bindings = task
1609 .node_plan
1610 .data_bindings
1611 .iter()
1612 .filter(|binding| binding.input_name == data_port)
1613 .collect::<Vec<_>>();
1614 let [binding] = bindings.as_slice() else {
1615 return Err(DagMlError::RuntimeValidation(format!(
1616 "portable Methods node `{}` requires exactly one `{data_port}` DataBinding",
1617 task.node_plan.node_id,
1618 )));
1619 };
1620 let identity = provider.training_data_identity(binding)?;
1621 if task.phase != Phase::Predict && identity.is_none() {
1622 return Err(DagMlError::RuntimeValidation(format!(
1623 "portable Methods PLS provider did not attest target-bound DataBinding `{}.{}` for {:?}",
1624 binding.node_id, binding.input_name, task.phase
1625 )));
1626 }
1627 let data_view_key = format!("data:{data_port}");
1628 let fit_view = task.data_views.get(data_port).or_else(|| task.data_views.get(&data_view_key)).cloned().ok_or_else(|| {
1629 DagMlError::RuntimeValidation(format!(
1630 "portable Methods node `{}` requires its scheduler-created `{data_port}` data view (available: {:?})",
1631 task.node_plan.node_id,
1632 task.data_views.keys().collect::<Vec<_>>(),
1633 ))
1634 })?;
1635 let prediction_view = if task.phase == Phase::FitCv {
1636 let validation_view_key = format!("data:{data_port}:validation");
1637 let validation_key = format!("{data_port}:validation");
1638 Some(task.data_views.get(&validation_key).or_else(|| task.data_views.get(&validation_view_key)).cloned().ok_or_else(|| {
1639 DagMlError::RuntimeValidation(format!(
1640 "portable Methods node `{}` requires its scheduler-created `{data_port}` validation view",
1641 task.node_plan.node_id,
1642 ))
1643 })?)
1644 } else {
1645 None
1646 };
1647 let request = MethodsPlsDataRequest {
1648 node_id: task.node_plan.node_id.clone(),
1649 phase: task.phase,
1650 variant_id: task.variant_id.clone(),
1651 fold_id: task.fold_id.clone(),
1652 binding: (*binding).clone(),
1653 identity,
1654 fit_view,
1655 prediction_view,
1656 };
1657 request.validate()?;
1658 Ok(request)
1659 }
1660
1661 fn params(task: &NodeTask) -> Result<MethodsPlsParams> {
1662 methods_pls_params(&task.node_plan.params).map_err(|error| {
1663 DagMlError::RuntimeValidation(format!(
1664 "portable Methods PLS node `{}` has invalid parameters: {error}",
1665 task.node_plan.node_id
1666 ))
1667 })
1668 }
1669
1670 fn validate_descriptor_for_task(
1671 task: &NodeTask,
1672 descriptor: Option<&crate::runtime::NativePredictorDescriptorV1>,
1673 ) -> Result<()> {
1674 validate_methods_pls_descriptor_against_params(&task.node_plan.params, descriptor)
1675 .map_err(|error| {
1676 DagMlError::RuntimeValidation(format!(
1677 "portable Methods PLS node `{}` descriptor mismatch: {error}",
1678 task.node_plan.node_id
1679 ))
1680 })
1681 }
1682
1683 fn native_error(operation: &str, error: n4m::Error) -> DagMlError {
1684 DagMlError::RuntimeValidation(format!(
1685 "portable Methods PLS {operation} failed: {error}"
1686 ))
1687 }
1688
1689 fn fit(task: &NodeTask, data: &MethodsPlsData) -> Result<(Context, Model)> {
1690 let context =
1691 Context::new().map_err(|error| Self::native_error("context_create", error))?;
1692 let mut config =
1693 Config::new().map_err(|error| Self::native_error("config_create", error))?;
1694 let params = Self::params(task)?;
1695 config
1696 .set_n_components(params.n_components)
1697 .map_err(|error| Self::native_error("config_set_n_components", error))?;
1698 if let Some(pipeline) = params.pipeline {
1699 config
1700 .set_snv_savgol_pipeline(pipeline.savgol_window, pipeline.savgol_poly_degree)
1701 .map_err(|error| Self::native_error("config_set_snv_savgol_pipeline", error))?;
1702 }
1703 let x = MatrixRef::row_major(&data.fit.x.values, data.fit.x.rows, data.fit.x.cols)
1704 .map_err(|error| Self::native_error("fit_x_matrix", error))?;
1705 let targets = data.fit.y.as_ref().ok_or_else(|| {
1706 DagMlError::RuntimeValidation(
1707 "portable Methods PLS fit requires targets".to_string(),
1708 )
1709 })?;
1710 let y = MatrixRef::row_major(&targets.values, targets.rows, targets.cols)
1711 .map_err(|error| Self::native_error("fit_y_matrix", error))?;
1712 let model = Model::fit(&context, &config, x, y)
1713 .map_err(|error| Self::native_error("fit", error))?;
1714 Ok((context, model))
1715 }
1716
1717 fn predict(
1718 context: &Context,
1719 model: &Model,
1720 data: &crate::runtime::MethodsPlsDataset,
1721 ) -> Result<Vec<Vec<f64>>> {
1722 let x = MatrixRef::row_major(&data.x.values, data.x.rows, data.x.cols)
1723 .map_err(|error| Self::native_error("predict_x_matrix", error))?;
1724 let prediction = model
1725 .predict(context, x)
1726 .map_err(|error| Self::native_error("predict", error))?;
1727 Ok(prediction
1728 .data
1729 .chunks(prediction.cols)
1730 .map(|row| row.to_vec())
1731 .collect())
1732 }
1733
1734 fn result(
1735 &self,
1736 task: &NodeTask,
1737 dataset: &crate::runtime::MethodsPlsDataset,
1738 values: Vec<Vec<f64>>,
1739 artifact: Option<(ArtifactRef, HandleRef)>,
1740 partition: PredictionPartition,
1741 ) -> Result<NodeResult> {
1742 let prediction = PredictionBlock {
1743 prediction_id: Some(format!(
1744 "methods-pls:{}:{}:{}",
1745 task.node_plan.node_id,
1746 task.phase.as_str(),
1747 task.fold_id
1748 .as_ref()
1749 .map(|id| id.as_str())
1750 .unwrap_or("full")
1751 )),
1752 producer_node: task.node_plan.node_id.clone(),
1753 producer_port: Some("oof".to_string()),
1754 partition,
1755 fold_id: (task.phase == Phase::FitCv)
1756 .then(|| task.fold_id.clone())
1757 .flatten(),
1758 sample_ids: dataset.sample_ids.clone(),
1759 values,
1760 target_names: dataset.target_names.clone(),
1761 };
1762 let regression_targets = if task.phase == Phase::FitCv {
1763 let targets = dataset.y.as_ref().ok_or_else(|| {
1764 DagMlError::RuntimeValidation(
1765 "portable Methods PLS FIT_CV requires validation targets".to_string(),
1766 )
1767 })?;
1768 vec![RegressionTargetBlock {
1769 level: PredictionLevel::Sample,
1770 unit_ids: dataset
1771 .sample_ids
1772 .iter()
1773 .cloned()
1774 .map(PredictionUnitId::Sample)
1775 .collect(),
1776 values: targets
1777 .values
1778 .chunks(targets.cols)
1779 .map(|row| row.to_vec())
1780 .collect(),
1781 target_names: dataset.target_names.clone(),
1782 }]
1783 } else {
1784 Vec::new()
1785 };
1786 let (artifacts, artifact_handles) = artifact
1787 .map(|(artifact, handle)| {
1788 (
1789 vec![artifact.clone()],
1790 BTreeMap::from([(artifact.id, handle)]),
1791 )
1792 })
1793 .unwrap_or_default();
1794 let artifact_refs = artifacts.clone();
1795 Ok(NodeResult {
1796 schema_version: None,
1797 node_id: task.node_plan.node_id.clone(),
1798 outputs: BTreeMap::from([("oof".to_string(), self.handle(HandleKind::Prediction))]),
1799 predictions: vec![prediction],
1800 observation_predictions: Vec::new(),
1801 aggregated_predictions: Vec::new(),
1802 explanations: Vec::new(),
1803 shape_deltas: Vec::new(),
1804 artifacts,
1805 artifact_handles,
1806 fit_influence_diagnostics: Vec::new(),
1807 regression_targets,
1808 lineage: LineageRecord {
1809 record_id: LineageId::new(format!(
1810 "lineage:methods-pls:{}:{}:{}:{}",
1811 task.node_plan.node_id,
1812 task.phase.as_str(),
1813 task.variant_id
1814 .as_ref()
1815 .map(|id| id.as_str())
1816 .unwrap_or("base"),
1817 task.fold_id
1818 .as_ref()
1819 .map(|id| id.as_str())
1820 .unwrap_or("full")
1821 ))
1822 .expect("valid native PLS lineage id"),
1823 run_id: task.run_id.clone(),
1824 node_id: task.node_plan.node_id.clone(),
1825 phase: task.phase,
1826 controller_id: self.id.clone(),
1827 controller_version: task.node_plan.controller_version.clone(),
1828 variant_id: task.variant_id.clone(),
1829 fold_id: task.fold_id.clone(),
1830 branch_path: task.branch_path.clone(),
1831 input_lineage: Vec::new(),
1832 artifact_refs,
1833 params_fingerprint: task.node_plan.params_fingerprint.clone(),
1834 data_model_shape_fingerprint: None,
1835 aggregation_policy_fingerprint: None,
1836 seed: task.seed,
1837 unsafe_flags: BTreeSet::new(),
1838 metrics: BTreeMap::new(),
1839 loss_attestations: Vec::new(),
1840 early_stopping_records: Vec::new(),
1841 },
1842 })
1843 }
1844 }
1845
1846 impl RuntimeController for MethodsPlsController {
1847 fn controller_id(&self) -> &ControllerId {
1848 &self.id
1849 }
1850
1851 fn export_artifact_payload(&self, artifact_id: &ArtifactId) -> Result<Option<Vec<u8>>> {
1852 Ok(self
1853 .exported_n4mm_by_artifact
1854 .lock()
1855 .map_err(|_| {
1856 DagMlError::RuntimeValidation(
1857 "portable Methods PLS N4MM sidecar lock poisoned".to_string(),
1858 )
1859 })?
1860 .remove(artifact_id))
1861 }
1862
1863 fn hydrate_artifact_payload(
1864 &self,
1865 request: &crate::runtime::ArtifactMaterializationRequest,
1866 payload: &[u8],
1867 ) -> Result<HandleRef> {
1868 self.runtime.ensure_n4mm_compatible(&request.artifact)?;
1869 if request.artifact.kind != "n4m_model"
1870 || request.artifact.backend != Some(ArtifactBackend::Raw)
1871 {
1872 return Err(DagMlError::RuntimeValidation(format!(
1873 "portable Methods PLS cannot hydrate non-N4MM artifact `{}`",
1874 request.artifact.id
1875 )));
1876 }
1877 if format!("{:x}", Sha256::digest(payload))
1878 != request
1879 .artifact
1880 .content_fingerprint
1881 .as_deref()
1882 .unwrap_or_default()
1883 || request.artifact.size_bytes != Some(payload.len() as u64)
1884 {
1885 return Err(DagMlError::RuntimeValidation(format!(
1886 "portable Methods PLS payload `{}` does not match its artifact reference",
1887 request.artifact.id
1888 )));
1889 }
1890 let inspected = inspect_methods_native_predictor_descriptor_v1(&self.id, payload)?;
1891 let expected = request.artifact.native_predictor_descriptor.as_ref();
1892 if expected.is_some_and(|expected| expected != &inspected)
1893 || (expected.is_none() && inspected.pipeline.is_some())
1894 {
1895 return Err(DagMlError::RuntimeValidation(format!(
1896 "portable Methods PLS payload `{}` does not match its inspected predictor descriptor",
1897 request.artifact.id
1898 )));
1899 }
1900 let context = Context::new()
1904 .map_err(|error| Self::native_error("hydrate_context_create", error))?;
1905 Model::import_n4mm(&context, payload)
1906 .map_err(|error| Self::native_error("hydrate_import_n4mm", error))?;
1907 let handle = self.handle(HandleKind::Model);
1908 self.hydrated_n4mm_by_handle
1909 .lock()
1910 .map_err(|_| {
1911 DagMlError::RuntimeValidation(
1912 "portable Methods PLS hydrated N4MM lock poisoned".to_string(),
1913 )
1914 })?
1915 .insert(handle.handle, payload.to_vec());
1916 Ok(handle)
1917 }
1918
1919 fn release_hydrated_artifact_payload(&self, handle: &HandleRef) -> Result<()> {
1920 if handle.kind != HandleKind::Model || handle.owner_controller != self.id {
1921 return Err(DagMlError::RuntimeValidation(format!(
1922 "portable Methods PLS cannot release foreign hydrated handle {}",
1923 handle.handle
1924 )));
1925 }
1926 self.hydrated_n4mm_by_handle
1931 .lock()
1932 .map_err(|_| {
1933 DagMlError::RuntimeValidation(
1934 "portable Methods PLS hydrated N4MM lock poisoned".to_string(),
1935 )
1936 })?
1937 .remove(&handle.handle);
1938 Ok(())
1939 }
1940
1941 fn invoke(&self, task: &NodeTask) -> Result<NodeResult> {
1942 Err(DagMlError::RuntimeValidation(format!(
1943 "portable Methods PLS node `{}` requires a RuntimeDataProvider numeric view",
1944 task.node_plan.node_id
1945 )))
1946 }
1947
1948 fn invoke_with_data_provider(
1949 &self,
1950 task: &NodeTask,
1951 provider: &dyn RuntimeDataProvider,
1952 ) -> Result<NodeResult> {
1953 if task.node_plan.kind != crate::graph::NodeKind::Model {
1954 return Err(DagMlError::RuntimeValidation(
1955 "portable Methods PLS controller only serves model nodes".to_string(),
1956 ));
1957 }
1958 let request = Self::request(task, provider, "x")?;
1959 provider.preflight_methods_pls(&request)?;
1960 let data = provider.methods_pls_data(&request)?;
1961 data.validate_for(&request)?;
1962 match task.phase {
1963 Phase::FitCv | Phase::Refit => {
1964 let (context, model) = Self::fit(task, &data)?;
1965 let prediction_data = data.prediction.as_ref().unwrap_or(&data.fit);
1966 let values = Self::predict(&context, &model, prediction_data)?;
1967 let artifact = if task.phase == Phase::Refit {
1968 let bytes = model
1969 .export_n4mm()
1970 .map_err(|error| Self::native_error("export_n4mm", error))?;
1971 let native_predictor_descriptor =
1972 inspect_methods_native_predictor_descriptor_v1(&self.id, &bytes)?;
1973 Self::validate_descriptor_for_task(
1974 task,
1975 Some(&native_predictor_descriptor),
1976 )?;
1977 let handle = self.handle(HandleKind::Model);
1978 let fingerprint = format!("{:x}", Sha256::digest(&bytes));
1979 let id = ArtifactId::new(format!(
1980 "artifact:methods-pls:{}:refit",
1981 task.node_plan.node_id
1982 ))
1983 .map_err(|error| DagMlError::RuntimeValidation(error.to_string()))?;
1984 self.exported_n4mm_by_artifact
1985 .lock()
1986 .map_err(|_| {
1987 DagMlError::RuntimeValidation(
1988 "portable Methods PLS N4MM sidecar lock poisoned".to_string(),
1989 )
1990 })?
1991 .insert(id.clone(), bytes.clone());
1992 Some((
1993 ArtifactRef {
1994 id,
1995 kind: "n4m_model".to_string(),
1996 controller_id: self.id.clone(),
1997 backend: Some(ArtifactBackend::Raw),
1998 uri: Some(format!(
2003 "methods/{}.n4mm",
2004 task.node_plan.node_id.as_str().replace(':', "_")
2005 )),
2006 content_fingerprint: Some(fingerprint),
2007 size_bytes: Some(bytes.len() as u64),
2008 plugin: None,
2009 plugin_version: None,
2010 abi_major: Some(METHODS_ABI_MAJOR),
2011 abi_min_minor: Some(
2012 if native_predictor_descriptor.pipeline.is_some() {
2013 METHODS_PIPELINE_N4MM_MIN_ABI_MINOR
2014 } else {
2015 METHODS_PLS_N4MM_MIN_ABI_MINOR
2016 },
2017 ),
2018 native_predictor_descriptor: Some(native_predictor_descriptor),
2019 },
2020 handle,
2021 ))
2022 } else {
2023 None
2024 };
2025 let partition = match task.phase {
2026 Phase::FitCv => PredictionPartition::Validation,
2027 Phase::Refit if data.prediction.is_some() => PredictionPartition::Test,
2032 Phase::Refit | Phase::Predict => PredictionPartition::Final,
2033 _ => unreachable!("match arm admits only FIT_CV/REFIT"),
2034 };
2035 self.result(task, prediction_data, values, artifact, partition)
2036 }
2037 Phase::Predict => {
2038 let artifact = task.artifact_inputs.values().find(|artifact| artifact.controller_id == self.id).ok_or_else(|| DagMlError::RuntimeValidation("portable Methods PLS PREDICT requires its retained N4MM artifact reference".to_string()))?;
2039 Self::validate_descriptor_for_task(
2040 task,
2041 artifact.artifact.native_predictor_descriptor.as_ref(),
2042 )?;
2043 let handle = task
2044 .input_handles
2045 .get(&crate::runtime::refit_artifact_input_key(&artifact.artifact.id))
2046 .ok_or_else(|| {
2047 DagMlError::RuntimeValidation(
2048 "portable Methods PLS PREDICT requires a hydrated N4MM runtime handle"
2049 .to_string(),
2050 )
2051 })?;
2052 let bytes = self.hydrated_n4mm_by_handle.lock().map_err(|_| DagMlError::RuntimeValidation("portable Methods PLS hydrated N4MM lock poisoned".to_string()))?.remove(&handle.handle).ok_or_else(|| DagMlError::RuntimeValidation("portable Methods PLS PREDICT requires N4MM bytes hydrated from the execution bundle in this controller instance".to_string()))?;
2056 let context = Context::new()
2057 .map_err(|error| Self::native_error("context_create", error))?;
2058 let model = Model::import_n4mm(&context, &bytes)
2059 .map_err(|error| Self::native_error("import_n4mm", error))?;
2060 let values = Self::predict(&context, &model, &data.fit)?;
2061 self.result(task, &data.fit, values, None, PredictionPartition::Final)
2062 }
2063 _ => Err(DagMlError::RuntimeValidation(
2064 "portable Methods PLS supports FIT_CV, REFIT, and PREDICT only".to_string(),
2065 )),
2066 }
2067 }
2068 }
2069
2070 pub struct MethodsRidgeController {
2079 id: ControllerId,
2080 runtime: MethodsRuntime,
2081 next_handle: AtomicU64,
2082 exported_n4mm_by_artifact: Mutex<BTreeMap<ArtifactId, Vec<u8>>>,
2083 hydrated_n4mm_by_handle: Mutex<BTreeMap<u64, Vec<u8>>>,
2084 }
2085
2086 type RidgePredictionInputs<'a> = BTreeMap<String, &'a PredictionInputSpec>;
2087
2088 struct RidgePredictionOutput {
2089 sample_ids: Vec<crate::SampleId>,
2090 target_names: Vec<String>,
2091 values: Vec<Vec<f64>>,
2092 partition: PredictionPartition,
2093 }
2094
2095 impl MethodsRidgeController {
2096 pub fn new(runtime: MethodsRuntime) -> Self {
2097 Self {
2098 id: ControllerId::new(METHODS_RIDGE_CONTROLLER_ID)
2099 .expect("Methods Ridge controller id is valid"),
2100 runtime,
2101 next_handle: AtomicU64::new(0),
2102 exported_n4mm_by_artifact: Mutex::new(BTreeMap::new()),
2103 hydrated_n4mm_by_handle: Mutex::new(BTreeMap::new()),
2104 }
2105 }
2106
2107 fn handle(&self, kind: HandleKind) -> HandleRef {
2108 HandleRef {
2109 handle: self.next_handle.fetch_add(1, Ordering::SeqCst) + 1,
2110 kind,
2111 owner_controller: self.id.clone(),
2112 }
2113 }
2114
2115 fn lambda(task: &NodeTask) -> Result<f64> {
2116 let lambda = task
2117 .node_plan
2118 .params
2119 .get("ridge_lambda")
2120 .and_then(serde_json::Value::as_f64)
2121 .ok_or_else(|| {
2122 DagMlError::RuntimeValidation(format!(
2123 "portable Methods Ridge node `{}` requires finite numeric `ridge_lambda`",
2124 task.node_plan.node_id
2125 ))
2126 })?;
2127 if !lambda.is_finite() || lambda < 0.0 {
2128 return Err(DagMlError::RuntimeValidation(
2129 "portable Methods Ridge `ridge_lambda` must be finite and non-negative"
2130 .to_string(),
2131 ));
2132 }
2133 Ok(lambda)
2134 }
2135
2136 fn feature_matrix(
2137 specs: &BTreeMap<String, &PredictionInputSpec>,
2138 sample_ids: &[crate::SampleId],
2139 expected_partition: PredictionPartition,
2140 label: &str,
2141 ) -> Result<crate::runtime::MethodsPlsMatrix> {
2142 if specs.len() < 2 {
2143 return Err(DagMlError::RuntimeValidation(format!(
2144 "portable Methods Ridge {label} requires OOF predictions from at least two base producers"
2145 )));
2146 }
2147 let mut cols = 0usize;
2148 for (key, spec) in specs {
2149 if spec.partition != expected_partition
2150 || spec.prediction_level != PredictionLevel::Sample
2151 || spec.sample_ids != sample_ids
2152 || spec.prediction_width == 0
2153 || spec.values.len() != sample_ids.len()
2154 || spec.values.iter().any(|row| {
2155 row.len() != spec.prediction_width
2156 || row.iter().any(|value| !value.is_finite())
2157 })
2158 {
2159 return Err(DagMlError::RuntimeValidation(format!(
2160 "portable Methods Ridge {label} input `{key}` is not an exact finite sample-level prediction matrix for the scheduler scope"
2161 )));
2162 }
2163 cols = cols.checked_add(spec.prediction_width).ok_or_else(|| {
2164 DagMlError::RuntimeValidation(
2165 "portable Methods Ridge feature width overflows usize".to_string(),
2166 )
2167 })?;
2168 }
2169 let capacity = sample_ids.len().checked_mul(cols).ok_or_else(|| {
2170 DagMlError::RuntimeValidation(
2171 "portable Methods Ridge feature matrix size overflows usize".to_string(),
2172 )
2173 })?;
2174 let mut values = Vec::with_capacity(capacity);
2175 for row in 0..sample_ids.len() {
2176 for spec in specs.values() {
2177 values.extend_from_slice(&spec.values[row]);
2178 }
2179 }
2180 let matrix = crate::runtime::MethodsPlsMatrix {
2181 values,
2182 rows: sample_ids.len(),
2183 cols,
2184 };
2185 matrix.validate(&format!("Ridge {label} OOF"))?;
2186 Ok(matrix)
2187 }
2188
2189 fn split_prediction_inputs<'a>(
2190 task: &'a NodeTask,
2191 suffix: &str,
2192 output_required: bool,
2193 ) -> Result<(RidgePredictionInputs<'a>, RidgePredictionInputs<'a>)> {
2194 let mut fit = BTreeMap::new();
2195 let mut output = BTreeMap::new();
2196 for (key, spec) in &task.prediction_inputs {
2197 if let Some(base) = key.strip_suffix(suffix) {
2198 if base.is_empty() || output.insert(base.to_string(), spec).is_some() {
2199 return Err(DagMlError::RuntimeValidation(format!(
2200 "portable Methods Ridge received duplicate or malformed output OOF input `{key}`"
2201 )));
2202 }
2203 } else if fit.insert(key.clone(), spec).is_some() {
2207 return Err(DagMlError::RuntimeValidation(format!(
2208 "portable Methods Ridge received unsupported prediction input `{key}`; expected base keys and `{suffix}` counterparts"
2209 )));
2210 }
2211 }
2212 if fit.is_empty()
2213 || (output_required
2214 && fit.keys().collect::<Vec<_>>() != output.keys().collect::<Vec<_>>())
2215 || (!output.is_empty()
2216 && fit.keys().collect::<Vec<_>>() != output.keys().collect::<Vec<_>>())
2217 {
2218 return Err(DagMlError::RuntimeValidation(format!(
2219 "portable Methods Ridge requires base OOF inputs and, when present, exactly paired `{suffix}` prediction inputs"
2220 )));
2221 }
2222 Ok((fit, output))
2223 }
2224
2225 fn predict_only_inputs(task: &NodeTask) -> Result<RidgePredictionInputs<'_>> {
2226 let mut inputs = BTreeMap::new();
2227 for (key, spec) in &task.prediction_inputs {
2228 let Some(base) = key.strip_suffix(":predict") else {
2229 return Err(DagMlError::RuntimeValidation(format!(
2230 "portable Methods Ridge PREDICT accepts only `:predict` OOF inputs, received `{key}`"
2231 )));
2232 };
2233 if base.is_empty() || inputs.insert(base.to_string(), spec).is_some() {
2234 return Err(DagMlError::RuntimeValidation(format!(
2235 "portable Methods Ridge PREDICT received duplicate or malformed OOF input `{key}`"
2236 )));
2237 }
2238 }
2239 if inputs.len() < 2 {
2240 return Err(DagMlError::RuntimeValidation(
2241 "portable Methods Ridge PREDICT requires at least two `:predict` OOF inputs"
2242 .to_string(),
2243 ));
2244 }
2245 Ok(inputs)
2246 }
2247
2248 fn fit(
2249 task: &NodeTask,
2250 features: &crate::runtime::MethodsPlsMatrix,
2251 targets: &crate::runtime::MethodsPlsMatrix,
2252 ) -> Result<(Context, Model)> {
2253 let context = Context::new().map_err(|error| {
2254 MethodsPlsController::native_error("ridge_context_create", error)
2255 })?;
2256 let config = Config::new().map_err(|error| {
2257 MethodsPlsController::native_error("ridge_config_create", error)
2258 })?;
2259 let x = MatrixRef::row_major(&features.values, features.rows, features.cols)
2260 .map_err(|error| MethodsPlsController::native_error("ridge_fit_features", error))?;
2261 let y = MatrixRef::row_major(&targets.values, targets.rows, targets.cols)
2262 .map_err(|error| MethodsPlsController::native_error("ridge_fit_targets", error))?;
2263 let model = Model::fit_ridge(&context, &config, x, y, Self::lambda(task)?)
2264 .map_err(|error| MethodsPlsController::native_error("ridge_fit", error))?;
2265 Ok((context, model))
2266 }
2267
2268 fn predict(
2269 context: &Context,
2270 model: &Model,
2271 features: &crate::runtime::MethodsPlsMatrix,
2272 ) -> Result<Vec<Vec<f64>>> {
2273 let x = MatrixRef::row_major(&features.values, features.rows, features.cols).map_err(
2274 |error| MethodsPlsController::native_error("ridge_predict_features", error),
2275 )?;
2276 let prediction = model
2277 .predict(context, x)
2278 .map_err(|error| MethodsPlsController::native_error("ridge_predict", error))?;
2279 Ok(prediction
2280 .data
2281 .chunks(prediction.cols)
2282 .map(|row| row.to_vec())
2283 .collect())
2284 }
2285
2286 fn result(
2287 &self,
2288 task: &NodeTask,
2289 output: RidgePredictionOutput,
2290 targets: Option<&crate::runtime::MethodsPlsMatrix>,
2291 artifact: Option<(ArtifactRef, HandleRef)>,
2292 ) -> Result<NodeResult> {
2293 let regression_targets = if task.phase == Phase::FitCv {
2294 let targets = targets.ok_or_else(|| {
2295 DagMlError::RuntimeValidation(
2296 "portable Methods Ridge FIT_CV requires validation targets".to_string(),
2297 )
2298 })?;
2299 vec![RegressionTargetBlock {
2300 level: PredictionLevel::Sample,
2301 unit_ids: output
2302 .sample_ids
2303 .iter()
2304 .cloned()
2305 .map(PredictionUnitId::Sample)
2306 .collect(),
2307 values: targets
2308 .values
2309 .chunks(targets.cols)
2310 .map(|row| row.to_vec())
2311 .collect(),
2312 target_names: output.target_names.clone(),
2313 }]
2314 } else {
2315 Vec::new()
2316 };
2317 let (artifacts, artifact_handles) = artifact
2318 .map(|(artifact, handle)| {
2319 (
2320 vec![artifact.clone()],
2321 BTreeMap::from([(artifact.id, handle)]),
2322 )
2323 })
2324 .unwrap_or_default();
2325 let artifact_refs = artifacts.clone();
2326 Ok(NodeResult {
2327 schema_version: None,
2328 node_id: task.node_plan.node_id.clone(),
2329 outputs: BTreeMap::from([("oof".to_string(), self.handle(HandleKind::Prediction))]),
2330 predictions: vec![PredictionBlock {
2331 prediction_id: Some(format!(
2332 "methods-ridge:{}:{}:{}",
2333 task.node_plan.node_id,
2334 task.phase.as_str(),
2335 task.fold_id
2336 .as_ref()
2337 .map(|id| id.as_str())
2338 .unwrap_or("full")
2339 )),
2340 producer_node: task.node_plan.node_id.clone(),
2341 producer_port: Some("oof".to_string()),
2342 partition: output.partition,
2343 fold_id: (task.phase == Phase::FitCv)
2344 .then(|| task.fold_id.clone())
2345 .flatten(),
2346 sample_ids: output.sample_ids,
2347 values: output.values,
2348 target_names: output.target_names,
2349 }],
2350 observation_predictions: Vec::new(),
2351 aggregated_predictions: Vec::new(),
2352 explanations: Vec::new(),
2353 shape_deltas: Vec::new(),
2354 artifacts,
2355 artifact_handles,
2356 fit_influence_diagnostics: Vec::new(),
2357 regression_targets,
2358 lineage: LineageRecord {
2359 record_id: LineageId::new(format!(
2360 "lineage:methods-ridge:{}:{}:{}:{}",
2361 task.node_plan.node_id,
2362 task.phase.as_str(),
2363 task.variant_id
2364 .as_ref()
2365 .map(|id| id.as_str())
2366 .unwrap_or("base"),
2367 task.fold_id
2368 .as_ref()
2369 .map(|id| id.as_str())
2370 .unwrap_or("full")
2371 ))
2372 .expect("valid native Ridge lineage id"),
2373 run_id: task.run_id.clone(),
2374 node_id: task.node_plan.node_id.clone(),
2375 phase: task.phase,
2376 controller_id: self.id.clone(),
2377 controller_version: task.node_plan.controller_version.clone(),
2378 variant_id: task.variant_id.clone(),
2379 fold_id: task.fold_id.clone(),
2380 branch_path: task.branch_path.clone(),
2381 input_lineage: Vec::new(),
2382 artifact_refs,
2383 params_fingerprint: task.node_plan.params_fingerprint.clone(),
2384 data_model_shape_fingerprint: None,
2385 aggregation_policy_fingerprint: None,
2386 seed: task.seed,
2387 unsafe_flags: BTreeSet::new(),
2388 metrics: BTreeMap::new(),
2389 loss_attestations: Vec::new(),
2390 early_stopping_records: Vec::new(),
2391 },
2392 })
2393 }
2394 }
2395
2396 impl RuntimeController for MethodsRidgeController {
2397 fn controller_id(&self) -> &ControllerId {
2398 &self.id
2399 }
2400
2401 fn export_artifact_payload(&self, artifact_id: &ArtifactId) -> Result<Option<Vec<u8>>> {
2402 Ok(self
2403 .exported_n4mm_by_artifact
2404 .lock()
2405 .map_err(|_| {
2406 DagMlError::RuntimeValidation(
2407 "portable Methods Ridge N4MM sidecar lock poisoned".to_string(),
2408 )
2409 })?
2410 .remove(artifact_id))
2411 }
2412
2413 fn hydrate_artifact_payload(
2414 &self,
2415 request: &crate::runtime::ArtifactMaterializationRequest,
2416 payload: &[u8],
2417 ) -> Result<HandleRef> {
2418 self.runtime.ensure_n4mm_compatible(&request.artifact)?;
2419 if request.artifact.kind != "n4m_model"
2420 || request.artifact.backend != Some(ArtifactBackend::Raw)
2421 || format!("{:x}", Sha256::digest(payload))
2422 != request
2423 .artifact
2424 .content_fingerprint
2425 .as_deref()
2426 .unwrap_or_default()
2427 || request.artifact.size_bytes != Some(payload.len() as u64)
2428 {
2429 return Err(DagMlError::RuntimeValidation(format!(
2430 "portable Methods Ridge payload `{}` does not match its N4MM artifact reference",
2431 request.artifact.id
2432 )));
2433 }
2434 let inspected = inspect_methods_native_predictor_descriptor_v1(&self.id, payload)?;
2435 if request
2436 .artifact
2437 .native_predictor_descriptor
2438 .as_ref()
2439 .is_some_and(|expected| expected != &inspected)
2440 {
2441 return Err(DagMlError::RuntimeValidation(format!(
2442 "portable Methods Ridge payload `{}` does not match its inspected predictor descriptor",
2443 request.artifact.id
2444 )));
2445 }
2446 let context = Context::new().map_err(|error| {
2447 MethodsPlsController::native_error("ridge_hydrate_context_create", error)
2448 })?;
2449 Model::import_n4mm(&context, payload).map_err(|error| {
2450 MethodsPlsController::native_error("ridge_hydrate_import_n4mm", error)
2451 })?;
2452 let handle = self.handle(HandleKind::Model);
2453 self.hydrated_n4mm_by_handle
2454 .lock()
2455 .map_err(|_| {
2456 DagMlError::RuntimeValidation(
2457 "portable Methods Ridge hydrated N4MM lock poisoned".to_string(),
2458 )
2459 })?
2460 .insert(handle.handle, payload.to_vec());
2461 Ok(handle)
2462 }
2463
2464 fn release_hydrated_artifact_payload(&self, handle: &HandleRef) -> Result<()> {
2465 if handle.kind != HandleKind::Model || handle.owner_controller != self.id {
2466 return Err(DagMlError::RuntimeValidation(format!(
2467 "portable Methods Ridge cannot release foreign hydrated handle {}",
2468 handle.handle
2469 )));
2470 }
2471 self.hydrated_n4mm_by_handle
2472 .lock()
2473 .map_err(|_| {
2474 DagMlError::RuntimeValidation(
2475 "portable Methods Ridge hydrated N4MM lock poisoned".to_string(),
2476 )
2477 })?
2478 .remove(&handle.handle);
2479 Ok(())
2480 }
2481
2482 fn invoke(&self, task: &NodeTask) -> Result<NodeResult> {
2483 Err(DagMlError::RuntimeValidation(format!(
2484 "portable Methods Ridge node `{}` requires a RuntimeDataProvider numeric view",
2485 task.node_plan.node_id
2486 )))
2487 }
2488
2489 fn invoke_with_data_provider(
2490 &self,
2491 task: &NodeTask,
2492 provider: &dyn RuntimeDataProvider,
2493 ) -> Result<NodeResult> {
2494 if task.node_plan.kind != crate::graph::NodeKind::Model {
2495 return Err(DagMlError::RuntimeValidation(
2496 "portable Methods Ridge controller only serves model nodes".to_string(),
2497 ));
2498 }
2499 let request = MethodsPlsController::request(task, provider, "x_original")?;
2503 provider.preflight_methods_pls(&request)?;
2504 let data = provider.methods_pls_data(&request)?;
2505 data.validate_for(&request)?;
2506 match task.phase {
2507 Phase::FitCv => {
2508 let (fit_specs, validation_specs) =
2509 Self::split_prediction_inputs(task, ":outer", true)?;
2510 let prediction = data.prediction.as_ref().ok_or_else(|| {
2511 DagMlError::RuntimeValidation(
2512 "portable Methods Ridge FIT_CV requires a validation data view"
2513 .to_string(),
2514 )
2515 })?;
2516 let fit_features = Self::feature_matrix(
2517 &fit_specs,
2518 &data.fit.sample_ids,
2519 PredictionPartition::Validation,
2520 "inner FIT_CV",
2521 )?;
2522 let validation_features = Self::feature_matrix(
2523 &validation_specs,
2524 &prediction.sample_ids,
2525 PredictionPartition::Validation,
2526 "outer FIT_CV",
2527 )?;
2528 let targets = data.fit.y.as_ref().ok_or_else(|| {
2529 DagMlError::RuntimeValidation(
2530 "portable Methods Ridge FIT_CV requires fitting targets".to_string(),
2531 )
2532 })?;
2533 let validation_targets = prediction.y.as_ref().ok_or_else(|| {
2534 DagMlError::RuntimeValidation(
2535 "portable Methods Ridge FIT_CV requires validation targets".to_string(),
2536 )
2537 })?;
2538 let (context, model) = Self::fit(task, &fit_features, targets)?;
2539 let values = Self::predict(&context, &model, &validation_features)?;
2540 Self::result(
2541 self,
2542 task,
2543 RidgePredictionOutput {
2544 sample_ids: prediction.sample_ids.clone(),
2545 target_names: prediction.target_names.clone(),
2546 values,
2547 partition: PredictionPartition::Validation,
2548 },
2549 Some(validation_targets),
2550 None,
2551 )
2552 }
2553 Phase::Refit => {
2554 let (fit_specs, refit_specs) =
2555 Self::split_prediction_inputs(task, ":refit", false)?;
2556 let fit_features = Self::feature_matrix(
2557 &fit_specs,
2558 &data.fit.sample_ids,
2559 PredictionPartition::Validation,
2560 "REFIT",
2561 )?;
2562 let targets = data.fit.y.as_ref().ok_or_else(|| {
2563 DagMlError::RuntimeValidation(
2564 "portable Methods Ridge REFIT requires targets".to_string(),
2565 )
2566 })?;
2567 let (context, model) = Self::fit(task, &fit_features, targets)?;
2568 let (output_ids, values, partition) = if refit_specs.is_empty() {
2569 (
2574 data.fit.sample_ids.clone(),
2575 Self::predict(&context, &model, &fit_features)?,
2576 PredictionPartition::Final,
2577 )
2578 } else {
2579 let output_ids = refit_specs
2580 .values()
2581 .next()
2582 .expect("paired inputs checked")
2583 .sample_ids
2584 .clone();
2585 let refit_features = Self::feature_matrix(
2586 &refit_specs,
2587 &output_ids,
2588 PredictionPartition::Test,
2589 "REFIT output",
2590 )?;
2591 (
2592 output_ids,
2593 Self::predict(&context, &model, &refit_features)?,
2594 PredictionPartition::Test,
2595 )
2596 };
2597 let bytes = model.export_n4mm().map_err(|error| {
2598 MethodsPlsController::native_error("ridge_export_n4mm", error)
2599 })?;
2600 let native_predictor_descriptor =
2601 inspect_methods_native_predictor_descriptor_v1(&self.id, &bytes)?;
2602 let id = ArtifactId::new(format!(
2603 "artifact:methods-ridge:{}:refit",
2604 task.node_plan.node_id
2605 ))
2606 .map_err(|error| DagMlError::RuntimeValidation(error.to_string()))?;
2607 let handle = self.handle(HandleKind::Model);
2608 self.exported_n4mm_by_artifact
2609 .lock()
2610 .map_err(|_| {
2611 DagMlError::RuntimeValidation(
2612 "portable Methods Ridge N4MM sidecar lock poisoned".to_string(),
2613 )
2614 })?
2615 .insert(id.clone(), bytes.clone());
2616 let artifact = ArtifactRef {
2617 id,
2618 kind: "n4m_model".to_string(),
2619 controller_id: self.id.clone(),
2620 backend: Some(ArtifactBackend::Raw),
2621 uri: Some(format!(
2622 "methods/{}.n4mm",
2623 task.node_plan.node_id.as_str().replace(':', "_")
2624 )),
2625 content_fingerprint: Some(format!("{:x}", Sha256::digest(&bytes))),
2626 size_bytes: Some(bytes.len() as u64),
2627 plugin: None,
2628 plugin_version: None,
2629 abi_major: Some(METHODS_ABI_MAJOR),
2630 abi_min_minor: Some(METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR),
2631 native_predictor_descriptor: Some(native_predictor_descriptor),
2632 };
2633 Self::result(
2634 self,
2635 task,
2636 RidgePredictionOutput {
2637 sample_ids: output_ids,
2638 target_names: data.fit.target_names.clone(),
2639 values,
2640 partition,
2641 },
2642 None,
2643 Some((artifact, handle)),
2644 )
2645 }
2646 Phase::Predict => {
2647 let predict_specs = Self::predict_only_inputs(task)?;
2648 let features = Self::feature_matrix(
2649 &predict_specs,
2650 &data.fit.sample_ids,
2651 PredictionPartition::Final,
2652 "PREDICT",
2653 )?;
2654 let artifact = task.artifact_inputs.values().find(|artifact| artifact.controller_id == self.id).ok_or_else(|| DagMlError::RuntimeValidation("portable Methods Ridge PREDICT requires its retained N4MM artifact reference".to_string()))?;
2655 let handle = task.input_handles.get(&crate::runtime::refit_artifact_input_key(&artifact.artifact.id)).ok_or_else(|| DagMlError::RuntimeValidation("portable Methods Ridge PREDICT requires a hydrated N4MM runtime handle".to_string()))?;
2656 let bytes = self.hydrated_n4mm_by_handle.lock().map_err(|_| DagMlError::RuntimeValidation("portable Methods Ridge hydrated N4MM lock poisoned".to_string()))?.remove(&handle.handle).ok_or_else(|| DagMlError::RuntimeValidation("portable Methods Ridge PREDICT requires N4MM bytes hydrated from the execution bundle in this controller instance".to_string()))?;
2657 let context = Context::new().map_err(|error| {
2658 MethodsPlsController::native_error("ridge_predict_context_create", error)
2659 })?;
2660 let model = Model::import_n4mm(&context, &bytes).map_err(|error| {
2661 MethodsPlsController::native_error("ridge_predict_import_n4mm", error)
2662 })?;
2663 let values = Self::predict(&context, &model, &features)?;
2664 Self::result(
2665 self,
2666 task,
2667 RidgePredictionOutput {
2668 sample_ids: data.fit.sample_ids.clone(),
2669 target_names: data.fit.target_names.clone(),
2670 values,
2671 partition: PredictionPartition::Final,
2672 },
2673 None,
2674 None,
2675 )
2676 }
2677 _ => Err(DagMlError::RuntimeValidation(
2678 "portable Methods Ridge supports FIT_CV, REFIT, and PREDICT only".to_string(),
2679 )),
2680 }
2681 }
2682 }
2683
2684 #[cfg(test)]
2685 mod ridge_tests {
2686 use std::collections::BTreeMap;
2687
2688 use super::*;
2689
2690 fn sample_ids() -> Vec<crate::SampleId> {
2691 vec![
2692 crate::SampleId::new("sample:1").unwrap(),
2693 crate::SampleId::new("sample:2").unwrap(),
2694 ]
2695 }
2696
2697 fn prediction(values: Vec<Vec<f64>>) -> PredictionInputSpec {
2698 PredictionInputSpec {
2699 producer_node: crate::NodeId::new("model:base").unwrap(),
2700 source_port: "oof".to_string(),
2701 target_port: "oof".to_string(),
2702 partition: PredictionPartition::Validation,
2703 prediction_level: PredictionLevel::Sample,
2704 fold_id: None,
2705 fold_ids: Vec::new(),
2706 unit_ids: Vec::new(),
2707 sample_ids: sample_ids(),
2708 values,
2709 prediction_width: 1,
2710 target_names: vec!["y".to_string()],
2711 }
2712 }
2713
2714 #[test]
2715 fn ridge_features_are_stably_ordered_and_identity_aligned() {
2716 let mut inputs = BTreeMap::new();
2717 inputs.insert(
2720 "model:z.oof".to_string(),
2721 prediction(vec![vec![30.0], vec![40.0]]),
2722 );
2723 inputs.insert(
2724 "model:a.oof".to_string(),
2725 prediction(vec![vec![10.0], vec![20.0]]),
2726 );
2727
2728 let refs = inputs
2729 .iter()
2730 .map(|(key, spec)| (key.clone(), spec))
2731 .collect();
2732 let matrix = MethodsRidgeController::feature_matrix(
2733 &refs,
2734 &sample_ids(),
2735 PredictionPartition::Validation,
2736 "test",
2737 )
2738 .unwrap();
2739 assert_eq!(matrix.rows, 2);
2740 assert_eq!(matrix.cols, 2);
2741 assert_eq!(matrix.values, vec![10.0, 30.0, 20.0, 40.0]);
2742
2743 let mut misaligned = inputs["model:a.oof"].clone();
2744 misaligned.sample_ids.reverse();
2745 let refs = BTreeMap::from([
2746 ("model:a.oof".to_string(), &misaligned),
2747 ("model:z.oof".to_string(), &inputs["model:z.oof"]),
2748 ]);
2749 assert!(MethodsRidgeController::feature_matrix(
2750 &refs,
2751 &sample_ids(),
2752 PredictionPartition::Validation,
2753 "test",
2754 )
2755 .is_err());
2756 }
2757 }
2758}
2759
2760#[cfg(feature = "methods-optimizer")]
2761pub use pls_controller::{MethodsPlsController, MethodsRidgeController};
2762
2763#[cfg(feature = "methods-optimizer")]
2771pub fn register_methods_runtime_controllers(
2772 registry: &mut crate::runtime::RuntimeControllerRegistry,
2773 hpo_controller_id: crate::ControllerId,
2774 runtime: MethodsRuntime,
2775) -> crate::Result<()> {
2776 let pls_controller_id = crate::ControllerId::new(METHODS_PLS_CONTROLLER_ID)
2777 .expect("the fixed Methods PLS controller id is valid");
2778 let ridge_controller_id = crate::ControllerId::new(METHODS_RIDGE_CONTROLLER_ID)
2779 .expect("the fixed Methods Ridge controller id is valid");
2780 if hpo_controller_id == pls_controller_id || hpo_controller_id == ridge_controller_id {
2781 return Err(crate::DagMlError::RuntimeValidation(
2782 "Methods HPO controller id must differ from the Methods PLS and Ridge controller ids"
2783 .to_string(),
2784 ));
2785 }
2786 for controller_id in [&pls_controller_id, &ridge_controller_id, &hpo_controller_id] {
2787 if registry.get(controller_id).is_some() {
2788 return Err(crate::DagMlError::RuntimeValidation(format!(
2789 "duplicate runtime controller `{controller_id}`"
2790 )));
2791 }
2792 }
2793 registry.register(Box::new(MethodsPlsController::new(runtime.clone())))?;
2794 registry.register(Box::new(MethodsRidgeController::new(runtime.clone())))?;
2795 registry.register(Box::new(MethodsHpoController::new(
2796 hpo_controller_id,
2797 runtime,
2798 )))?;
2799 Ok(())
2800}
2801
2802#[cfg(feature = "methods-optimizer")]
2803mod native {
2804 use super::*;
2805 use n4m::{
2806 Category, Direction, Error, ErrorKind, Optimizer, OptimizerOptions, Pruner, Sampler,
2807 SearchSpace, TrialError, TrialSnapshot, TrialStatus,
2808 };
2809
2810 pub(super) struct MethodsHpoStudy {
2811 manifest: MethodsHpoControllerManifest,
2812 methods_abi: String,
2813 _context: n4m::Context,
2814 optimizer: Optimizer,
2815 events: Vec<HpoEvent>,
2816 }
2817
2818 #[allow(dead_code)]
2822 struct MethodsHpoReporter<'a> {
2823 study: &'a mut MethodsHpoStudy,
2824 trial_id: i64,
2825 pruned: Option<HpoTrial>,
2826 }
2827
2828 impl HpoIntermediateReporter for MethodsHpoReporter<'_> {
2829 fn report(&mut self, step: i32, score: f64) -> HpoResult<HpoReportOutcome> {
2830 if self.pruned.is_some() {
2831 return Err(HpoError::InvalidTrial {
2832 reason: "cannot report after native pruning terminalized the trial".to_string(),
2833 });
2834 }
2835 if self.study.report_intermediate(self.trial_id, step, score)? {
2836 let snapshot = self.study.snapshot_for(self.trial_id)?;
2837 if snapshot.status != HpoTrialStatus::Pruned {
2838 return Err(HpoError::InvalidTrial {
2839 reason: "native pruner returned true without a PRUNED snapshot".to_string(),
2840 });
2841 }
2842 self.pruned = Some(snapshot.clone());
2843 return Ok(HpoReportOutcome::Pruned(snapshot));
2844 }
2845 Ok(HpoReportOutcome::Continue)
2846 }
2847 }
2848
2849 impl MethodsHpoStudy {
2850 pub(super) fn create(config: MethodsHpoStudyConfig) -> HpoResult<Self> {
2851 methods_optimizer_preflight()?;
2852 let methods_abi = config.methods_abi_identity()?;
2853 let fingerprint = config.search_space.fingerprint()?;
2854 let optimizer_fingerprint = config.optimizer.fingerprint()?;
2855 let manifest = MethodsHpoControllerManifest {
2856 schema_version: HPO_MANIFEST_SCHEMA_VERSION,
2857 binding: HpoStudyBinding {
2858 controller_id: config.controller_id,
2859 study_id: config.study_id,
2860 search_space_fingerprint: fingerprint,
2861 optimizer_fingerprint,
2862 },
2863 };
2864 manifest.validate()?;
2865 let context =
2866 n4m::Context::new().map_err(|error| native_error("context_create", error))?;
2867 let native_space = create_space(&config.search_space)?;
2868 let options = create_options(&config.optimizer);
2869 let optimizer = Optimizer::new(&context, &native_space, &options)
2870 .map_err(|error| native_error("optimizer_create", error))?;
2871 Ok(Self {
2872 manifest,
2873 methods_abi,
2874 _context: context,
2875 optimizer,
2876 events: Vec::new(),
2877 })
2878 }
2879 pub(super) fn restore(
2880 config: MethodsHpoStudyConfig,
2881 checkpoint: &N4moptCheckpointArtifact,
2882 ) -> HpoResult<Self> {
2883 methods_optimizer_preflight()?;
2884 let methods_abi = config.methods_abi_identity()?;
2885 let fingerprint = config.search_space.fingerprint()?;
2886 let optimizer_fingerprint = config.optimizer.fingerprint()?;
2887 let binding = HpoStudyBinding {
2888 controller_id: config.controller_id,
2889 study_id: config.study_id,
2890 search_space_fingerprint: fingerprint,
2891 optimizer_fingerprint,
2892 };
2893 checkpoint.validate()?;
2894 if checkpoint.binding != binding || checkpoint.methods_abi != methods_abi {
2895 return Err(HpoError::CheckpointBindingMismatch {
2896 reason: "study/search-space or Methods ABI differs from checkpoint".to_string(),
2897 });
2898 }
2899 let context =
2902 n4m::Context::new().map_err(|error| native_error("context_create", error))?;
2903 let optimizer = Optimizer::load_n4mopt(&context, &checkpoint.opaque_payload)
2904 .map_err(|error| native_error("load_n4mopt", error))?;
2905 Ok(Self {
2906 manifest: MethodsHpoControllerManifest {
2907 schema_version: HPO_MANIFEST_SCHEMA_VERSION,
2908 binding,
2909 },
2910 methods_abi,
2911 _context: context,
2912 optimizer,
2913 events: Vec::new(),
2914 })
2915 }
2916 #[allow(dead_code)]
2917 pub fn manifest(&self) -> &MethodsHpoControllerManifest {
2918 &self.manifest
2919 }
2920 #[allow(dead_code)]
2921 pub fn events(&self) -> &[HpoEvent] {
2922 &self.events
2923 }
2924 pub fn ask(&mut self) -> HpoResult<HpoTrial> {
2925 let id = self
2926 .optimizer
2927 .ask()
2928 .and_then(|trial| trial.id())
2929 .map_err(|error| native_error("ask", error))?;
2930 let trial = self.snapshot_for(id)?;
2931 self.events.push(HpoEvent::Asked { trial_id: trial.id });
2932 Ok(trial)
2933 }
2934 #[allow(dead_code)]
2935 pub fn ask_batch(&mut self, count: i32) -> HpoResult<HpoBatch> {
2936 match self.optimizer.ask_batch(count) {
2937 Ok(native_trials) => {
2938 let ids = native_trials
2939 .iter()
2940 .map(|trial| {
2941 trial
2942 .id()
2943 .map_err(|error| native_error("trial_get_id", error))
2944 })
2945 .collect::<HpoResult<Vec<_>>>()?;
2946 let trials = ids
2947 .into_iter()
2948 .map(|id| self.snapshot_for(id))
2949 .collect::<HpoResult<Vec<_>>>()?;
2950 for trial in &trials {
2951 self.events.push(HpoEvent::Asked { trial_id: trial.id });
2952 }
2953 Ok(HpoBatch {
2954 trials,
2955 native_error: None,
2956 })
2957 }
2958 Err(n4m::AskBatchError::Partial {
2959 error,
2960 trials: native_trials,
2961 }) => {
2962 let ids = native_trials
2963 .iter()
2964 .map(|trial| {
2965 trial
2966 .id()
2967 .map_err(|error| native_error("trial_get_id", error))
2968 })
2969 .collect::<HpoResult<Vec<_>>>()?;
2970 let trials = ids
2971 .into_iter()
2972 .map(|id| self.snapshot_for(id))
2973 .collect::<HpoResult<Vec<_>>>()?;
2974 for trial in &trials {
2975 self.events.push(HpoEvent::Asked { trial_id: trial.id });
2976 }
2977 Ok(HpoBatch {
2978 trials,
2979 native_error: Some(to_native_error(error)),
2980 })
2981 }
2982 Err(n4m::AskBatchError::Error(error)) => Err(native_error("ask_batch", error)),
2983 }
2984 }
2985 pub fn report_intermediate(
2986 &mut self,
2987 trial_id: i64,
2988 step: i32,
2989 score: f64,
2990 ) -> HpoResult<bool> {
2991 if !score.is_finite() {
2992 return Err(HpoError::InvalidTrial {
2993 reason: "intermediate score must be finite".to_string(),
2994 });
2995 }
2996 let should_prune = self
2997 .optimizer
2998 .tell_intermediate(trial_id, step, score)
2999 .map_err(|error| native_error("tell_intermediate", error))?;
3000 self.events.push(HpoEvent::Intermediate {
3001 trial_id,
3002 step,
3003 score,
3004 should_prune,
3005 });
3006 if should_prune {
3009 self.events.push(HpoEvent::Terminal {
3010 trial_id,
3011 status: HpoTrialStatus::Pruned,
3012 score: None,
3013 failure: None,
3014 });
3015 }
3016 Ok(should_prune)
3017 }
3018 pub fn tell(&mut self, trial_id: i64, terminal: HpoTerminal) -> HpoResult<HpoTrial> {
3022 let (status, score, failure) = match terminal {
3023 HpoTerminal::Completed { score } if score.is_finite() => {
3024 (TrialStatus::Completed, score, None)
3025 }
3026 HpoTerminal::Completed { .. } => {
3027 return Err(HpoError::InvalidTrial {
3028 reason: "terminal score must be finite".to_string(),
3029 })
3030 }
3031 HpoTerminal::Failed { failure } => (TrialStatus::Failed, 0.0, Some(failure)),
3032 HpoTerminal::Pruned { failure } => (TrialStatus::Pruned, 0.0, Some(failure)),
3035 HpoTerminal::Cancelled { failure } => (TrialStatus::Cancelled, 0.0, Some(failure)),
3036 };
3037 let native_failure = matches!(status, TrialStatus::Failed | TrialStatus::Cancelled)
3038 .then_some(failure.as_ref())
3039 .flatten()
3040 .map(|failure| TrialError {
3041 code: failure.code.clone(),
3042 message: failure.message.clone(),
3043 retryable: failure.retryable,
3044 });
3045 self.optimizer
3046 .tell_result(trial_id, status, score, native_failure.as_ref())
3047 .map_err(|error| native_error("tell_result", error))?;
3048 let snapshot = self.snapshot_for(trial_id)?;
3049 self.events.push(HpoEvent::Terminal {
3050 trial_id,
3051 status: snapshot.status,
3052 score: snapshot.score,
3053 failure: snapshot.failure.clone(),
3054 });
3055 Ok(snapshot)
3056 }
3057 #[allow(dead_code)]
3058 pub fn evaluate_one<E: HpoEvaluator>(
3059 &mut self,
3060 evaluator: &mut E,
3061 boundary: &mut HpoEvaluationBoundary<'_>,
3062 ) -> HpoResult<HpoTrial> {
3063 boundary.validate()?;
3064 let trial = self.ask()?;
3065 let (terminal, pruned) = {
3066 let mut reporter = MethodsHpoReporter {
3067 study: self,
3068 trial_id: trial.id,
3069 pruned: None,
3070 };
3071 let terminal = evaluator.evaluate_with_reporter(&trial, boundary, &mut reporter);
3072 (terminal, reporter.pruned.take())
3073 };
3074 let terminal = match terminal {
3075 Ok(terminal) => terminal,
3076 Err(error) => {
3077 if pruned.is_some() {
3078 return Err(error);
3081 }
3082 let failure = HpoFailure {
3083 code: "HPO_EVALUATION".to_string(),
3084 message: error.to_string(),
3085 retryable: true,
3086 };
3087 let _ = self.tell(trial.id, HpoTerminal::Failed { failure });
3090 return Err(error);
3091 }
3092 };
3093 if let Some(snapshot) = pruned {
3094 return Ok(snapshot);
3097 }
3098 self.tell(trial.id, terminal)
3099 }
3100 pub fn best(&self) -> HpoResult<Option<HpoBestTrial>> {
3101 let Some((trial, score)) = self
3102 .optimizer
3103 .best()
3104 .map_err(|error| native_error("best", error))?
3105 else {
3106 return Ok(None);
3107 };
3108 let id = trial
3109 .id()
3110 .map_err(|error| native_error("trial_get_id", error))?;
3111 self.snapshot_for(id)
3112 .map(|trial| Some(HpoBestTrial { trial, score }))
3113 }
3114 pub fn trials(&self) -> HpoResult<Vec<HpoTrial>> {
3115 self.optimizer
3116 .trials(0)
3117 .map_err(|error| native_error("trials", error))?
3118 .iter()
3119 .map(snapshot_trial)
3120 .collect()
3121 }
3122 pub fn save_checkpoint(&self) -> HpoResult<N4moptCheckpointArtifact> {
3123 let payload = self
3124 .optimizer
3125 .save_n4mopt()
3126 .map_err(|error| native_error("save_n4mopt", error))?;
3127 if payload.len() > MAX_N4MOPT_CHECKPOINT_BYTES {
3128 return Err(HpoError::InvalidCheckpoint {
3129 reason: "native checkpoint exceeds configured limit".to_string(),
3130 });
3131 }
3132 N4moptCheckpointArtifact::new(
3133 self.manifest.binding.clone(),
3134 self.methods_abi.clone(),
3135 payload,
3136 )
3137 }
3138 fn snapshot_for(&self, id: i64) -> HpoResult<HpoTrial> {
3139 self.optimizer
3140 .trials(id)
3141 .map_err(|error| native_error("trials", error))?
3142 .into_iter()
3143 .find(|trial| trial.id == id)
3144 .ok_or_else(|| HpoError::InvalidTrial {
3145 reason: format!("native trial history omitted committed trial `{id}`"),
3146 })
3147 .and_then(|trial| snapshot_trial(&trial))
3148 }
3149 }
3150
3151 fn create_space(space: &HpoSearchSpace) -> HpoResult<SearchSpace> {
3152 let mut result =
3153 SearchSpace::new().map_err(|error| native_error("search_space_create", error))?;
3154 for parameter in &space.parameters {
3155 let call = match parameter {
3156 HpoParameter::Int {
3157 name,
3158 low,
3159 high,
3160 step,
3161 log,
3162 } => result.add_int(name, *low, *high, *step, *log),
3163 HpoParameter::Float {
3164 name,
3165 low,
3166 high,
3167 step,
3168 log,
3169 } => result.add_float(name, *low, *high, *step, *log),
3170 HpoParameter::Categorical { name, values } => result
3171 .add_categorical(name, &values.iter().map(map_category).collect::<Vec<_>>()),
3172 HpoParameter::Ordinal { name, values } => result.add_ordinal(name, values),
3173 HpoParameter::SortedTuple {
3174 name,
3175 length,
3176 low,
3177 high,
3178 integer,
3179 } => result.add_sorted_tuple(name, *length, *low, *high, *integer),
3180 };
3181 call.map_err(|error| native_error("search_space_add", error))?;
3182 }
3183 Ok(result)
3184 }
3185 fn map_category(value: &HpoCategory) -> Category {
3186 match value {
3187 HpoCategory::String(value) => Category::Str(value.clone()),
3188 HpoCategory::Integer(value) => Category::Int(*value),
3189 HpoCategory::Float(value) => Category::Float(*value),
3190 HpoCategory::Boolean(value) => Category::Bool(*value),
3191 }
3192 }
3193 fn create_options(config: &HpoOptimizerConfig) -> OptimizerOptions {
3194 OptimizerOptions {
3195 sampler: match config.sampler {
3196 HpoSampler::Random => Sampler::Random,
3197 HpoSampler::Sobol => Sampler::Sobol,
3198 HpoSampler::Lhs => Sampler::Lhs,
3199 HpoSampler::Ternary => Sampler::Ternary,
3200 HpoSampler::Ga => Sampler::Ga,
3201 HpoSampler::Pso => Sampler::Pso,
3202 HpoSampler::Cmaes => Sampler::Cmaes,
3203 HpoSampler::Tpe => Sampler::Tpe,
3204 HpoSampler::GpEi => Sampler::GpEi,
3205 },
3206 pruner: match config.pruner {
3207 HpoPruner::None => Pruner::None,
3208 HpoPruner::Median => Pruner::Median,
3209 HpoPruner::Asha => Pruner::Asha,
3210 HpoPruner::Hyperband => Pruner::Hyperband,
3211 HpoPruner::Racing => Pruner::Racing,
3212 },
3213 direction: match config.direction {
3214 HpoDirection::Auto => Direction::Auto,
3215 HpoDirection::Minimize => Direction::Minimize,
3216 HpoDirection::Maximize => Direction::Maximize,
3217 },
3218 metric: match config.metric {
3219 HpoMetric::Rmse => n4m::Metric::Rmse,
3220 HpoMetric::Mse => n4m::Metric::Mse,
3221 HpoMetric::Mae => n4m::Metric::Mae,
3222 HpoMetric::R2 => n4m::Metric::R2,
3223 HpoMetric::Accuracy => n4m::Metric::Accuracy,
3224 HpoMetric::BalancedAccuracy => n4m::Metric::BalancedAccuracy,
3225 HpoMetric::F1 => n4m::Metric::F1,
3226 HpoMetric::Logloss => n4m::Metric::Logloss,
3227 },
3228 seed: config.seed,
3229 n_startup_trials: config.n_startup_trials,
3230 max_resource: config.max_resource,
3231 reduction_factor: config.reduction_factor,
3232 ..OptimizerOptions::default()
3233 }
3234 }
3235 fn map_status(value: TrialStatus) -> HpoTrialStatus {
3236 match value {
3237 TrialStatus::Running => HpoTrialStatus::Running,
3238 TrialStatus::Completed => HpoTrialStatus::Completed,
3239 TrialStatus::Pruned => HpoTrialStatus::Pruned,
3240 TrialStatus::Failed => HpoTrialStatus::Failed,
3241 TrialStatus::Cancelled => HpoTrialStatus::Cancelled,
3242 }
3243 }
3244 fn snapshot_trial(value: &TrialSnapshot) -> HpoResult<HpoTrial> {
3245 let mut parameters = BTreeMap::new();
3246 for (name, parameter) in &value.parameters {
3247 parameters.insert(
3248 name.clone(),
3249 HpoTrialParameter {
3250 name: name.clone(),
3251 value: parameter.value,
3252 native_kind: Some(map_parameter_kind(parameter.kind)),
3253 category_type: parameter.category_type.map(map_category_type),
3254 integer: parameter.integer,
3255 active: parameter.active,
3256 category_index: parameter.category_index,
3257 category_label: parameter.category_label.clone(),
3258 },
3259 );
3260 }
3261 Ok(HpoTrial {
3262 id: value.id,
3263 ask_sequence: value.ask_sequence,
3264 terminal_sequence: value.terminal_sequence,
3265 parameters,
3266 parameter_order: value.parameter_order.clone(),
3267 status: map_status(value.status),
3268 score: value.score,
3269 rung: value.rung,
3270 duration: value.duration,
3271 intermediates: value
3272 .intermediates
3273 .iter()
3274 .map(|item| HpoIntermediate {
3275 sequence: item.sequence,
3276 step: item.step,
3277 score: item.score,
3278 should_prune: item.should_prune,
3279 })
3280 .collect(),
3281 failure: value.error.as_ref().map(|item| HpoFailure {
3282 code: item.code.clone(),
3283 message: item.message.clone(),
3284 retryable: item.retryable,
3285 }),
3286 })
3287 }
3288 fn map_parameter_kind(value: n4m::ParameterKind) -> HpoNativeParameterKind {
3289 match value {
3290 n4m::ParameterKind::Int => HpoNativeParameterKind::Int,
3291 n4m::ParameterKind::Float => HpoNativeParameterKind::Float,
3292 n4m::ParameterKind::LogInt => HpoNativeParameterKind::LogInt,
3293 n4m::ParameterKind::LogFloat => HpoNativeParameterKind::LogFloat,
3294 n4m::ParameterKind::Categorical => HpoNativeParameterKind::Categorical,
3295 n4m::ParameterKind::Ordinal => HpoNativeParameterKind::Ordinal,
3296 n4m::ParameterKind::SortedTuple => HpoNativeParameterKind::SortedTuple,
3297 }
3298 }
3299 fn map_category_type(value: n4m::CategoryType) -> HpoCategoryType {
3300 match value {
3301 n4m::CategoryType::Str => HpoCategoryType::String,
3302 n4m::CategoryType::Int => HpoCategoryType::Integer,
3303 n4m::CategoryType::Float => HpoCategoryType::Float,
3304 n4m::CategoryType::Bool => HpoCategoryType::Boolean,
3305 }
3306 }
3307 fn native_error(operation: &str, error: Error) -> HpoError {
3308 HpoError::Native {
3309 operation: operation.to_string(),
3310 error: to_native_error(error),
3311 }
3312 }
3313 fn to_native_error(error: Error) -> HpoNativeError {
3314 HpoNativeError {
3315 status: error.status,
3316 kind: format!("{:?}", error.kind).to_lowercase(),
3317 retryable: matches!(
3318 error.kind,
3319 ErrorKind::OutOfMemory
3320 | ErrorKind::BackendUnavailable
3321 | ErrorKind::Cancelled
3322 | ErrorKind::Io
3323 ),
3324 message: error.message,
3325 }
3326 }
3327}
3328
3329#[cfg(feature = "methods-optimizer")]
3330use native::MethodsHpoStudy;
3331
3332#[cfg(test)]
3333mod tests {
3334 use super::*;
3335 #[cfg(feature = "methods-optimizer-local")]
3336 use crate::controller::{
3337 ArtifactPolicy, ControllerCapability, ControllerFitScope, ControllerManifest,
3338 ControllerRegistry, RngPolicy,
3339 };
3340 #[cfg(feature = "methods-optimizer-local")]
3341 use crate::graph::{GraphInterface, GraphSpec, NodeKind, NodeSpec, PortSchema};
3342 #[cfg(feature = "methods-optimizer-local")]
3343 use crate::metrics::RegressionMetricKind;
3344 #[cfg(feature = "methods-optimizer-local")]
3345 use crate::phase::Phase;
3346 #[cfg(feature = "methods-optimizer-local")]
3347 use crate::plan::{build_execution_plan, CampaignSpec, ExecutionPlan};
3348
3349 fn n4mm_ref(controller: &str, abi_min_minor: Option<u32>) -> crate::runtime::ArtifactRef {
3350 crate::runtime::ArtifactRef {
3351 id: crate::ArtifactId::new(format!("artifact:{controller}:abi-test")).unwrap(),
3352 kind: "n4m_model".to_string(),
3353 controller_id: crate::ControllerId::new(controller).unwrap(),
3354 backend: Some(crate::runtime::ArtifactBackend::Raw),
3355 uri: None,
3356 content_fingerprint: None,
3357 size_bytes: None,
3358 plugin: None,
3359 plugin_version: None,
3360 abi_major: abi_min_minor.map(|_| METHODS_ABI_MAJOR),
3361 abi_min_minor,
3362 native_predictor_descriptor: None,
3363 }
3364 }
3365
3366 #[test]
3367 fn methods_n4mm_abi_contract_is_capability_derived_and_fail_closed() {
3368 let historical_pls = n4mm_ref(METHODS_PLS_CONTROLLER_ID, None);
3369 assert_eq!(
3370 methods_n4mm_abi_requirement(&historical_pls).unwrap(),
3371 (METHODS_ABI_MAJOR, METHODS_PLS_N4MM_MIN_ABI_MINOR)
3372 );
3373 validate_methods_abi_compatibility(2, 2, 2, 0).unwrap();
3374 validate_methods_abi_compatibility(2, 3, 2, 0).unwrap();
3375
3376 let ridge = n4mm_ref(
3377 METHODS_RIDGE_CONTROLLER_ID,
3378 Some(METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR),
3379 );
3380 let requirement = methods_n4mm_abi_requirement(&ridge).unwrap();
3381 assert!(validate_methods_abi_compatibility(2, 2, requirement.0, requirement.1).is_err());
3382 validate_methods_abi_compatibility(2, 3, requirement.0, requirement.1).unwrap();
3383
3384 assert!(
3385 methods_n4mm_abi_requirement(&n4mm_ref(METHODS_RIDGE_CONTROLLER_ID, None))
3386 .unwrap_err()
3387 .to_string()
3388 .contains("requires an explicit ABI minimum")
3389 );
3390 assert!(methods_n4mm_abi_requirement(&n4mm_ref(
3391 METHODS_PLS_CONTROLLER_ID,
3392 Some(METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR),
3393 ))
3394 .is_err());
3395 }
3396 #[cfg(feature = "methods-optimizer-local")]
3397 use crate::runtime::{
3398 ArtifactBackend, ArtifactMaterializationRequest, ArtifactRef, RuntimeController,
3399 RuntimeControllerRegistry, RuntimeHpoExecutionContext, RuntimeHpoIntermediate,
3400 RuntimeHpoIntermediateOutcome, RuntimeHpoProvenance, RuntimeHpoSelectionTarget,
3401 RuntimeHpoTerminal,
3402 };
3403
3404 #[cfg(feature = "methods-optimizer-local")]
3405 fn native_runtime() -> MethodsRuntime {
3406 let library_path = std::env::var_os("N4M_LIBRARY_PATH")
3407 .expect("methods native tests require an explicit N4M_LIBRARY_PATH");
3408 MethodsRuntime::configure(library_path).expect("configure explicit Methods test runtime")
3409 }
3410
3411 #[test]
3412 fn default_build_refuses_before_any_native_or_host_work() {
3413 assert_eq!(
3414 methods_optimizer_preflight(),
3415 if cfg!(feature = "methods-optimizer") {
3416 Ok(())
3417 } else {
3418 Err(HpoError::MethodsOptimizerFeatureDisabled)
3419 }
3420 );
3421 }
3422
3423 #[cfg(feature = "methods-optimizer")]
3424 #[test]
3425 fn methods_runtime_refuses_relative_library_paths_before_native_loading() {
3426 assert!(matches!(
3427 MethodsRuntime::configure("libn4m.so"),
3428 Err(HpoError::RuntimeConfiguration { reason })
3429 if reason == "libn4m path must be absolute"
3430 ));
3431 }
3432
3433 #[cfg(feature = "methods-optimizer")]
3434 #[test]
3435 fn methods_pls_refuses_noncanonical_historical_sklearn_defaults() {
3436 let mut params: BTreeMap<String, serde_json::Value> = serde_json::from_str(include_str!(
3437 "../tests/fixtures/package/studio_pls_regression_v1_params.json"
3438 ))
3439 .unwrap();
3440 params.insert("scale".to_string(), serde_json::json!(false));
3441
3442 assert!(validate_methods_pls_node_params(¶ms)
3443 .unwrap_err()
3444 .to_string()
3445 .contains("historical sklearn parameters must keep canonical defaults"));
3446 }
3447
3448 #[cfg(feature = "methods-optimizer-local")]
3449 #[test]
3450 fn methods_runtime_registers_all_controllers_atomically() {
3451 let runtime = native_runtime();
3452 let hpo_id = crate::ControllerId::new("controller:tuner.methods").unwrap();
3453 let mut registry = RuntimeControllerRegistry::new();
3454
3455 register_methods_runtime_controllers(&mut registry, hpo_id.clone(), runtime.clone())
3456 .unwrap();
3457 let pls_id = crate::ControllerId::new(METHODS_PLS_CONTROLLER_ID).unwrap();
3458 let ridge_id = crate::ControllerId::new(METHODS_RIDGE_CONTROLLER_ID).unwrap();
3459 assert!(registry.get(&pls_id).is_some());
3460 assert!(registry.get(&ridge_id).is_some());
3461 assert!(registry.get(&hpo_id).is_some());
3462
3463 let error = register_methods_runtime_controllers(&mut registry, hpo_id.clone(), runtime)
3464 .unwrap_err();
3465 assert!(error.to_string().contains("duplicate runtime controller"));
3466 assert!(registry.get(&pls_id).is_some());
3467 assert!(registry.get(&ridge_id).is_some());
3468 assert!(registry.get(&hpo_id).is_some());
3469 }
3470
3471 #[cfg(feature = "methods-optimizer-local")]
3472 #[test]
3473 fn methods_pls_hydrated_payload_release_is_idempotent_and_handle_local() {
3474 let runtime = native_runtime();
3475 let context = n4m::Context::new().unwrap();
3476 let mut config = n4m::Config::new().unwrap();
3477 config.set_n_components(1).unwrap();
3478 let x_values = [1.0, 1.0, 2.0, 4.0, 3.0, 9.0, 4.0, 16.0];
3479 let y_values = [1.0, 2.0, 3.0, 4.0];
3480 let x = n4m::MatrixRef::row_major(&x_values, 4, 2).unwrap();
3481 let y = n4m::MatrixRef::row_major(&y_values, 4, 1).unwrap();
3482 let payload = n4m::Model::fit(&context, &config, x, y)
3483 .unwrap()
3484 .export_n4mm()
3485 .unwrap();
3486 let controller = MethodsPlsController::new(runtime);
3487 let controller_id = controller.controller_id().clone();
3488 let request = ArtifactMaterializationRequest {
3489 run_id: crate::RunId::new("run:methods-pls.release").unwrap(),
3490 bundle_id: crate::BundleId::new("bundle:methods-pls.release").unwrap(),
3491 node_id: crate::NodeId::new("model:methods-pls").unwrap(),
3492 phase: Phase::Predict,
3493 variant_id: None,
3494 controller_id: controller_id.clone(),
3495 artifact: ArtifactRef {
3496 id: crate::ArtifactId::new("artifact:methods-pls.release").unwrap(),
3497 kind: "n4m_model".to_string(),
3498 controller_id: controller_id.clone(),
3499 backend: Some(ArtifactBackend::Raw),
3500 uri: Some("methods/release.n4mm".to_string()),
3501 content_fingerprint: Some(format!("{:x}", Sha256::digest(&payload))),
3502 size_bytes: Some(payload.len() as u64),
3503 plugin: None,
3504 plugin_version: None,
3505 abi_major: Some(METHODS_ABI_MAJOR),
3506 abi_min_minor: Some(METHODS_PLS_N4MM_MIN_ABI_MINOR),
3507 native_predictor_descriptor: Some(
3508 inspect_methods_native_predictor_descriptor_v1(&controller_id, &payload)
3509 .unwrap(),
3510 ),
3511 },
3512 params_fingerprint: "params:methods-pls.release".to_string(),
3513 training_loss_fingerprint: None,
3514 };
3515
3516 let first = controller
3517 .hydrate_artifact_payload(&request, &payload)
3518 .unwrap();
3519 assert_eq!(controller.hydrated_payload_count().unwrap(), 1);
3520 controller
3521 .release_hydrated_artifact_payload(&first)
3522 .unwrap();
3523 controller
3524 .release_hydrated_artifact_payload(&first)
3525 .unwrap();
3526 assert_eq!(controller.hydrated_payload_count().unwrap(), 0);
3527
3528 let second = controller
3529 .hydrate_artifact_payload(&request, &payload)
3530 .unwrap();
3531 assert_ne!(first.handle, second.handle);
3532 controller
3533 .release_hydrated_artifact_payload(&second)
3534 .unwrap();
3535 assert_eq!(controller.hydrated_payload_count().unwrap(), 0);
3536 }
3537
3538 #[cfg(feature = "methods-optimizer-local")]
3539 #[test]
3540 fn native_predictor_descriptor_binds_pls_and_affine_bytes_fail_closed() {
3541 let runtime = native_runtime();
3542 let context = n4m::Context::new().unwrap();
3543 let mut config = n4m::Config::new().unwrap();
3544 config.set_n_components(1).unwrap();
3545 let x_values = [1.0, 1.0, 2.0, 4.0, 3.0, 9.0, 4.0, 16.0];
3546 let y_values = [1.0, 2.0, 3.0, 4.0];
3547 let x = n4m::MatrixRef::row_major(&x_values, 4, 2).unwrap();
3548 let y = n4m::MatrixRef::row_major(&y_values, 4, 1).unwrap();
3549 let pls_payload = n4m::Model::fit(&context, &config, x, y)
3550 .unwrap()
3551 .export_n4mm()
3552 .unwrap();
3553 let pls_id = crate::ControllerId::new(METHODS_PLS_CONTROLLER_ID).unwrap();
3554 let pls_descriptor =
3555 inspect_methods_native_predictor_descriptor_v1(&pls_id, &pls_payload).unwrap();
3556 assert_eq!(pls_descriptor.storage_algorithm, 0);
3557 assert_eq!(
3558 pls_descriptor.dimensions,
3559 crate::runtime::NativePredictorDimensionsV1 {
3560 training_samples: 4,
3561 n_features: 2,
3562 n_targets: 1,
3563 n_components: 1,
3564 }
3565 );
3566 assert_ne!(
3567 pls_descriptor.capabilities & n4m::SERIALIZED_MODEL_CAPABILITY_PREDICT,
3568 0
3569 );
3570
3571 let mut pipeline_config = n4m::Config::new().unwrap();
3572 pipeline_config.set_n_components(1).unwrap();
3573 pipeline_config.set_snv_savgol_pipeline(3, 2).unwrap();
3574 let pipeline_x_values = [
3575 1.0, 1.5, 2.0, 2.5, 2.0, 3.0, 4.0, 5.0, 3.0, 5.0, 8.0, 13.0, 4.0, 7.0, 11.0, 16.0,
3576 ];
3577 let pipeline_x = n4m::MatrixRef::row_major(&pipeline_x_values, 4, 4).unwrap();
3578 let pipeline_payload = n4m::Model::fit(&context, &pipeline_config, pipeline_x, y)
3579 .unwrap()
3580 .export_n4mm()
3581 .unwrap();
3582 let native_pipeline = n4m::inspect_n4mm(&pipeline_payload)
3583 .unwrap()
3584 .pipeline
3585 .unwrap();
3586 assert_eq!(
3587 native_pipeline.semantic_profile,
3588 n4m::PipelineSemanticProfile::Nirs4allSnvSavgolV1
3589 );
3590 assert_eq!(native_pipeline.snv_ddof, 0);
3591 assert_eq!(
3592 native_pipeline.savgol_mode,
3593 n4m::SerializedSavitzkyGolayMode::Interp
3594 );
3595 let pipeline_descriptor =
3596 inspect_methods_native_predictor_descriptor_v1(&pls_id, &pipeline_payload).unwrap();
3597 assert_eq!(pipeline_descriptor.format_version, 2);
3598 let inspected_pipeline = pipeline_descriptor.pipeline.as_ref().unwrap();
3599 assert_eq!(inspected_pipeline.savgol_window, 3);
3600 assert_eq!(inspected_pipeline.savgol_poly_degree, 2);
3601 assert_eq!(
3602 inspected_pipeline.native_fingerprint,
3603 format!("{:016x}", native_pipeline.fingerprint)
3604 );
3605 let params = BTreeMap::from([
3606 ("n_components".to_string(), serde_json::json!(1)),
3607 (
3608 "pipeline".to_string(),
3609 serde_json::json!({
3610 "schema_version": 1,
3611 "pipeline_type": "n4m.snv_savgol_smooth.v1",
3612 "savgol_window": 3,
3613 "savgol_poly_degree": 2
3614 }),
3615 ),
3616 ]);
3617 validate_methods_pls_descriptor_against_params(¶ms, Some(&pipeline_descriptor))
3618 .unwrap();
3619 let mut spoofed = params;
3620 spoofed.get_mut("pipeline").unwrap()["savgol_window"] = serde_json::json!(5);
3621 assert!(validate_methods_pls_descriptor_against_params(
3622 &spoofed,
3623 Some(&pipeline_descriptor)
3624 )
3625 .unwrap_err()
3626 .to_string()
3627 .contains("do not match"));
3628
3629 let coefficients = [2.0, 0.5, -1.0, 3.0];
3630 let intercept = [1.5, -2.0];
3631 let affine_payload =
3632 n4m::Model::import_linear_predictor(&context, 17, 2, 2, &coefficients, &intercept)
3633 .unwrap()
3634 .export_n4mm()
3635 .unwrap();
3636 let ridge_id = crate::ControllerId::new(METHODS_RIDGE_CONTROLLER_ID).unwrap();
3637 let ridge_descriptor =
3638 inspect_methods_native_predictor_descriptor_v1(&ridge_id, &affine_payload).unwrap();
3639 assert_eq!(ridge_descriptor.storage_algorithm, 11);
3640 assert_eq!(ridge_descriptor.dimensions.n_components, 0);
3641 assert_eq!(
3642 ridge_descriptor.capabilities
3643 & (n4m::SERIALIZED_MODEL_CAPABILITY_PREDICT
3644 | n4m::SERIALIZED_MODEL_CAPABILITY_AFFINE),
3645 n4m::SERIALIZED_MODEL_CAPABILITY_PREDICT | n4m::SERIALIZED_MODEL_CAPABILITY_AFFINE
3646 );
3647
3648 assert!(
3649 inspect_methods_native_predictor_descriptor_v1(&ridge_id, &pls_payload)
3650 .unwrap_err()
3651 .to_string()
3652 .contains("not product-supported")
3653 );
3654 assert!(
3655 inspect_methods_native_predictor_descriptor_v1(&pls_id, &affine_payload)
3656 .unwrap_err()
3657 .to_string()
3658 .contains("not product-supported")
3659 );
3660 let mut tampered = pls_payload.clone();
3661 let last = tampered.len() - 1;
3662 tampered[last] ^= 1;
3663 assert!(inspect_methods_native_predictor_descriptor_v1(&pls_id, &tampered).is_err());
3664
3665 let request_for =
3666 |controller_id: crate::ControllerId,
3667 payload: &[u8],
3668 descriptor: crate::runtime::NativePredictorDescriptorV1,
3669 abi_min_minor: u32| ArtifactMaterializationRequest {
3670 run_id: crate::RunId::new("run:descriptor-test").unwrap(),
3671 bundle_id: crate::BundleId::new("bundle:descriptor-test").unwrap(),
3672 node_id: crate::NodeId::new("model:descriptor-test").unwrap(),
3673 phase: Phase::Predict,
3674 variant_id: None,
3675 controller_id: controller_id.clone(),
3676 artifact: ArtifactRef {
3677 id: crate::ArtifactId::new("artifact:descriptor-test").unwrap(),
3678 kind: "n4m_model".to_string(),
3679 controller_id,
3680 backend: Some(ArtifactBackend::Raw),
3681 uri: Some("methods/descriptor-test.n4mm".to_string()),
3682 content_fingerprint: Some(format!("{:x}", Sha256::digest(payload))),
3683 size_bytes: Some(payload.len() as u64),
3684 plugin: None,
3685 plugin_version: None,
3686 abi_major: Some(METHODS_ABI_MAJOR),
3687 abi_min_minor: Some(abi_min_minor),
3688 native_predictor_descriptor: Some(descriptor),
3689 },
3690 params_fingerprint: "a".repeat(64),
3691 training_loss_fingerprint: None,
3692 };
3693
3694 let pls_controller = MethodsPlsController::new(runtime.clone());
3695 let good_pls = request_for(
3696 pls_id.clone(),
3697 &pls_payload,
3698 pls_descriptor.clone(),
3699 METHODS_PLS_N4MM_MIN_ABI_MINOR,
3700 );
3701 let handle = pls_controller
3702 .hydrate_artifact_payload(&good_pls, &pls_payload)
3703 .unwrap();
3704 pls_controller
3705 .release_hydrated_artifact_payload(&handle)
3706 .unwrap();
3707
3708 let mut wrong_dimensions = pls_descriptor.clone();
3709 wrong_dimensions.dimensions.n_features += 1;
3710 wrong_dimensions.descriptor_fingerprint = wrong_dimensions.compute_fingerprint().unwrap();
3711 let wrong_dimensions_request = request_for(
3712 pls_id,
3713 &pls_payload,
3714 wrong_dimensions,
3715 METHODS_PLS_N4MM_MIN_ABI_MINOR,
3716 );
3717 assert!(pls_controller
3718 .hydrate_artifact_payload(&wrong_dimensions_request, &pls_payload)
3719 .unwrap_err()
3720 .to_string()
3721 .contains("does not match its inspected predictor descriptor"));
3722
3723 let ridge_controller = MethodsRidgeController::new(runtime);
3724 let good_ridge = request_for(
3725 ridge_id,
3726 &affine_payload,
3727 ridge_descriptor,
3728 METHODS_IMPORTED_LINEAR_N4MM_MIN_ABI_MINOR,
3729 );
3730 let ridge_handle = ridge_controller
3731 .hydrate_artifact_payload(&good_ridge, &affine_payload)
3732 .unwrap();
3733 ridge_controller
3734 .release_hydrated_artifact_payload(&ridge_handle)
3735 .unwrap();
3736
3737 let mut future = pls_descriptor;
3738 future.schema_version = 2;
3739 future.descriptor_fingerprint = future.compute_fingerprint().unwrap();
3740 assert!(future
3741 .validate()
3742 .unwrap_err()
3743 .to_string()
3744 .contains("unsupported native predictor descriptor"));
3745 }
3746
3747 fn ledger_trial(score: f64) -> HpoTrial {
3748 HpoTrial {
3749 id: 7,
3750 ask_sequence: 3,
3751 terminal_sequence: Some(4),
3752 parameters: BTreeMap::new(),
3753 parameter_order: Vec::new(),
3754 status: HpoTrialStatus::Completed,
3755 score: Some(score),
3756 rung: 0,
3757 duration: 0.25,
3758 intermediates: vec![HpoIntermediate {
3759 sequence: 2,
3760 step: 0,
3761 score,
3762 should_prune: false,
3763 }],
3764 failure: None,
3765 }
3766 }
3767
3768 #[test]
3769 fn restored_ledger_accepts_only_one_ulp_score_drift_after_tcv1_projection() {
3770 let native = canonical_hpo_terminal_ledger(vec![ledger_trial(1.0)]).unwrap();
3771 let mut one_ulp = native.clone();
3772 one_ulp[0].score = Some(f64::from_bits(1.0_f64.to_bits() + 1));
3773 one_ulp[0].intermediates[0].score = f64::from_bits(1.0_f64.to_bits() + 1);
3774 let one_ulp = canonical_hpo_terminal_ledger(one_ulp).unwrap();
3775 assert!(hpo_terminal_trials_match(&native, &one_ulp));
3776
3777 let mut two_ulps = native.clone();
3778 two_ulps[0].score = Some(f64::from_bits(1.0_f64.to_bits() + 2));
3779 assert!(!hpo_terminal_trials_match(&native, &two_ulps));
3780
3781 let mut tampered = native.clone();
3782 tampered[0].score = Some(99.0);
3783 assert!(!hpo_terminal_trials_match(&native, &tampered));
3784 }
3785
3786 #[test]
3787 fn search_space_digest_is_canonical_and_order_sensitive() {
3788 let space = HpoSearchSpace {
3789 parameters: vec![HpoParameter::Int {
3790 name: "depth".into(),
3791 low: 1,
3792 high: 5,
3793 step: 1,
3794 log: false,
3795 }],
3796 };
3797 assert_eq!(space.fingerprint().unwrap(), space.fingerprint().unwrap());
3798 let swapped = HpoSearchSpace {
3799 parameters: vec![
3800 HpoParameter::Float {
3801 name: "rate".into(),
3802 low: 0.1,
3803 high: 1.0,
3804 step: 0.1,
3805 log: false,
3806 },
3807 space.parameters[0].clone(),
3808 ],
3809 };
3810 assert_ne!(space.fingerprint().unwrap(), swapped.fingerprint().unwrap());
3811 }
3812 #[test]
3813 fn checkpoint_rejects_oversize_before_native_decoder() {
3814 let binding = HpoStudyBinding {
3815 controller_id: "controller:hpo".into(),
3816 study_id: "study:one".into(),
3817 search_space_fingerprint: "a".into(),
3818 optimizer_fingerprint: "b".into(),
3819 };
3820 let checkpoint = N4moptCheckpointArtifact {
3821 schema_version: 1,
3822 artifact_kind: N4MOPT_ARTIFACT_KIND.into(),
3823 format: N4MOPT_FORMAT.into(),
3824 abi_major: METHODS_ABI_MAJOR,
3825 abi_min_minor: METHODS_N4MOPT_MIN_ABI_MINOR,
3826 binding,
3827 methods_abi: "n4m-abi-2.2".into(),
3828 opaque_payload: vec![0; MAX_N4MOPT_CHECKPOINT_BYTES + 1],
3829 payload_sha256: "x".into(),
3830 };
3831 assert!(matches!(
3832 checkpoint.validate(),
3833 Err(HpoError::InvalidCheckpoint { .. })
3834 ));
3835 }
3836
3837 #[test]
3838 fn historical_n4mopt_defaults_to_first_implemented_abi_and_new_writer_emits_it() {
3839 let payload = vec![1_u8, 2, 3];
3840 let historical = serde_json::json!({
3841 "schema_version": N4MOPT_CHECKPOINT_SCHEMA_VERSION,
3842 "artifact_kind": N4MOPT_ARTIFACT_KIND,
3843 "format": N4MOPT_FORMAT,
3844 "binding": {
3845 "controller_id": "controller:hpo",
3846 "study_id": "study:one",
3847 "search_space_fingerprint": "a",
3848 "optimizer_fingerprint": "b"
3849 },
3850 "methods_abi": "n4m-abi-2.2",
3851 "opaque_payload": payload,
3852 "payload_sha256": payload_sha256(&[1, 2, 3])
3853 });
3854 let checkpoint: N4moptCheckpointArtifact = serde_json::from_value(historical).unwrap();
3855 assert_eq!(checkpoint.abi_major, METHODS_ABI_MAJOR);
3856 assert_eq!(checkpoint.abi_min_minor, METHODS_N4MOPT_MIN_ABI_MINOR);
3857 checkpoint.validate().unwrap();
3858
3859 let emitted = serde_json::to_value(checkpoint).unwrap();
3860 assert_eq!(emitted["abi_major"], METHODS_ABI_MAJOR);
3861 assert_eq!(emitted["abi_min_minor"], METHODS_N4MOPT_MIN_ABI_MINOR);
3862 }
3863
3864 #[cfg(feature = "methods-optimizer-local")]
3865 fn native_config() -> MethodsHpoStudyConfig {
3866 MethodsHpoStudyConfig {
3867 controller_id: "controller:methods-hpo".into(),
3868 study_id: "study:native-lifecycle".into(),
3869 methods_abi: "n4m-abi-2.2".into(),
3870 search_space: HpoSearchSpace {
3871 parameters: vec![HpoParameter::Int {
3872 name: "n_components".into(),
3873 low: 1,
3874 high: 3,
3875 step: 1,
3876 log: false,
3877 }],
3878 },
3879 optimizer: HpoOptimizerConfig {
3880 sampler: HpoSampler::Random,
3881 pruner: HpoPruner::None,
3882 direction: HpoDirection::Minimize,
3883 metric: HpoMetric::Rmse,
3884 seed: 7,
3885 n_startup_trials: 1,
3886 max_resource: 0,
3887 reduction_factor: 0,
3888 },
3889 }
3890 }
3891
3892 #[cfg(feature = "methods-optimizer-local")]
3893 fn native_hpo_manifest(id: &str, kind: NodeKind) -> ControllerManifest {
3894 ControllerManifest {
3895 controller_id: crate::ControllerId::new(id).unwrap(),
3896 controller_version: "native-hpo-test".to_string(),
3897 operator_kind: kind,
3898 priority: 0,
3899 supported_phases: BTreeSet::from([Phase::FitCv]),
3900 input_ports: Vec::new(),
3901 output_ports: Vec::new(),
3902 data_requirements: None,
3903 capabilities: BTreeSet::from([ControllerCapability::Deterministic]),
3904 operator_selectors: Vec::new(),
3905 fit_scope: ControllerFitScope::FoldTrain,
3906 rng_policy: RngPolicy::UsesCoreSeed,
3907 artifact_policy: ArtifactPolicy::Serializable,
3908 }
3909 }
3910
3911 #[cfg(feature = "methods-optimizer-local")]
3912 fn native_hpo_node(id: &str, kind: NodeKind) -> NodeSpec {
3913 NodeSpec {
3914 id: crate::NodeId::new(id).unwrap(),
3915 kind,
3916 operator: None,
3917 params: BTreeMap::new(),
3918 ports: PortSchema {
3919 inputs: Vec::new(),
3920 outputs: Vec::new(),
3921 },
3922 metadata: BTreeMap::new(),
3923 seed_label: None,
3924 }
3925 }
3926
3927 #[cfg(feature = "methods-optimizer-local")]
3928 fn attested_native_hpo_context() -> (
3929 ExecutionPlan,
3930 RuntimeHpoExecutionContext,
3931 crate::runtime::RuntimeHpoCampaignTask,
3932 ) {
3933 let target_node_id = crate::NodeId::new("model:methods-pls").unwrap();
3934 let controller_id = crate::ControllerId::new("controller:methods-hpo").unwrap();
3935 let mut registry = ControllerRegistry::new();
3936 registry
3937 .register(native_hpo_manifest(
3938 "controller:methods-pls",
3939 NodeKind::Model,
3940 ))
3941 .unwrap();
3942 let plan = build_execution_plan(
3943 "plan:methods-hpo-checkpoint",
3944 GraphSpec {
3945 id: "graph:methods-hpo-checkpoint".to_string(),
3946 interface: GraphInterface::default(),
3947 nodes: vec![native_hpo_node("model:methods-pls", NodeKind::Model)],
3948 edges: Vec::new(),
3949 search_space_fingerprint: None,
3950 metadata: BTreeMap::new(),
3951 },
3952 CampaignSpec {
3953 inner_cv: None,
3954 id: "campaign:methods-hpo-checkpoint".to_string(),
3955 root_seed: Some(17),
3956 leakage_policy: Default::default(),
3957 aggregation_policy: Default::default(),
3958 split_invocation: None,
3959 generation: Default::default(),
3960 shape_plans: BTreeMap::new(),
3961 data_bindings: BTreeMap::new(),
3962 branch_view_plans: Vec::new(),
3963 metadata: BTreeMap::new(),
3964 },
3965 ®istry,
3966 )
3967 .unwrap();
3968 let context = RuntimeHpoExecutionContext {
3969 operation_id: "hpo:methods".to_string(),
3970 controller_id: controller_id.clone(),
3971 target_node_id: target_node_id.clone(),
3972 base_variant: plan.variants[0].clone(),
3973 trial_budget_total: 2,
3974 study: native_config(),
3975 parameter_paths: BTreeMap::from([(
3976 "n_components".to_string(),
3977 "n_components".to_string(),
3978 )]),
3979 resume_checkpoint: None,
3980 resume_variants: BTreeMap::new(),
3981 resume_terminal_trials: Vec::new(),
3982 selection: RuntimeHpoSelectionTarget {
3983 producer_node: target_node_id.clone(),
3984 producer_port: "prediction".to_string(),
3985 metric: RegressionMetricKind::Rmse,
3986 direction: HpoDirection::Minimize,
3987 },
3988 provenance: RuntimeHpoProvenance {
3989 graph_fingerprint: plan.graph_fingerprint.clone(),
3990 campaign_fingerprint: plan.campaign_fingerprint.clone(),
3991 controller_fingerprint: plan.controller_fingerprint.clone(),
3992 data_identities_fingerprint: "data-identities:methods-hpo".to_string(),
3993 fold_set_fingerprint: None,
3994 training_influence_fingerprint: "influence:methods-hpo".to_string(),
3995 relation_fingerprint: "relations:methods-hpo".to_string(),
3996 },
3997 };
3998 context.validate_for_plan(&plan).unwrap();
3999 let task = crate::runtime::RuntimeHpoCampaignTask {
4000 run_id: crate::RunId::new("run:methods-hpo-checkpoint").unwrap(),
4001 operation_id: "hpo:methods".to_string(),
4002 controller_id: controller_id.clone(),
4003 target_node_id,
4004 seed: Some(17),
4005 };
4006 assert_eq!(task.controller_id, controller_id);
4007 (plan, context, task)
4008 }
4009
4010 #[cfg(feature = "methods-optimizer-local")]
4011 fn proposal_components(proposal: &crate::runtime::RuntimeHpoProposal) -> i64 {
4012 let choice = proposal.variant.choices.get("native_methods_hpo").unwrap();
4013 let override_ = choice.param_overrides.first().unwrap();
4014 assert_eq!(override_.params.len(), 1);
4015 override_.params["n_components"].as_i64().unwrap()
4016 }
4017
4018 #[cfg(feature = "methods-optimizer-local")]
4019 fn assert_runtime_refusal(error: crate::DagMlError) {
4020 assert!(matches!(error, crate::DagMlError::RuntimeValidation(_)));
4021 }
4022
4023 #[cfg(feature = "methods-optimizer-local")]
4024 #[test]
4025 fn registered_methods_session_checkpoints_restores_and_refuses_tampering() {
4026 let runtime = native_runtime();
4027 let (plan, context, task) = attested_native_hpo_context();
4028 let controller_id = task.controller_id.clone();
4029 let mut controllers = RuntimeControllerRegistry::new();
4030 controllers
4031 .register(Box::new(MethodsHpoController::new(
4032 controller_id.clone(),
4033 runtime,
4034 )))
4035 .unwrap();
4036 let controller = controllers.get(&controller_id).unwrap();
4037
4038 let mut session = controller.create_tuner_session(&task, &context).unwrap();
4039 let first = session.ask().unwrap().unwrap();
4040 assert!((1..=3).contains(&proposal_components(&first)));
4041 assert_eq!(
4042 session
4043 .report_intermediate(RuntimeHpoIntermediate {
4044 trial_id: first.trial_id,
4045 step: 0,
4046 score: 1.5,
4047 })
4048 .unwrap(),
4049 RuntimeHpoIntermediateOutcome::Continue
4050 );
4051 session
4052 .tell(first.trial_id, RuntimeHpoTerminal::Completed { score: 1.0 })
4053 .unwrap();
4054 let checkpoint = session.checkpoint().unwrap();
4055 checkpoint.validate().unwrap();
4056 assert_eq!(checkpoint.binding.controller_id, controller_id.as_str());
4057 assert_eq!(checkpoint.binding.study_id, context.study.study_id);
4058 assert_eq!(checkpoint.methods_abi, context.study.methods_abi);
4059
4060 let checkpoint_trace = MethodsHpoStudy::restore(context.study.clone(), &checkpoint)
4063 .unwrap()
4064 .trials()
4065 .unwrap();
4066 assert_eq!(checkpoint_trace.len(), 1);
4067 assert_eq!(checkpoint_trace[0].id, first.trial_id);
4068 assert_eq!(checkpoint_trace[0].status, HpoTrialStatus::Completed);
4069 assert_eq!(checkpoint_trace[0].score, Some(1.0));
4070 assert_eq!(
4071 MethodsHpoStudy::restore(context.study.clone(), &checkpoint)
4072 .unwrap()
4073 .best()
4074 .unwrap()
4075 .unwrap()
4076 .trial
4077 .id,
4078 first.trial_id
4079 );
4080
4081 let mut expected = MethodsHpoStudy::restore(context.study.clone(), &checkpoint).unwrap();
4082 assert_eq!(expected.trials().unwrap(), checkpoint_trace);
4083 assert_eq!(expected.best().unwrap().unwrap().trial.id, first.trial_id);
4084 let expected_next = expected.ask().unwrap();
4085 let mut resumed_context = context.clone();
4086 resumed_context.resume_checkpoint = Some(checkpoint.clone());
4087 resumed_context.resume_terminal_trials = vec![crate::runtime::RuntimeHpoTerminalSnapshot {
4088 trial: checkpoint_trace[0].clone(),
4089 variant_id: Some(first.variant.variant_id.clone()),
4090 }];
4091 resumed_context.validate_for_plan(&plan).unwrap();
4092
4093 let mut tampered_ledger = resumed_context.clone();
4098 tampered_ledger.resume_terminal_trials[0].trial.score = Some(99.0);
4099 let error = match controller.create_tuner_session(&task, &tampered_ledger) {
4100 Err(error) => error,
4101 Ok(_) => panic!("tampered restored terminal ledger unexpectedly created a session"),
4102 };
4103 assert_runtime_refusal(error);
4104
4105 let mut resumed = controller
4106 .create_tuner_session(&task, &resumed_context)
4107 .unwrap();
4108 let resumed_next = resumed.ask().unwrap().unwrap();
4109 assert_eq!(resumed_next.trial_id, expected_next.id);
4110 assert_eq!(
4111 proposal_components(&resumed_next),
4112 expected_next.parameters["n_components"].value as i64
4113 );
4114 resumed
4115 .report_intermediate(RuntimeHpoIntermediate {
4116 trial_id: resumed_next.trial_id,
4117 step: 0,
4118 score: 0.5,
4119 })
4120 .unwrap();
4121 resumed
4122 .tell(
4123 resumed_next.trial_id,
4124 RuntimeHpoTerminal::Completed { score: 0.25 },
4125 )
4126 .unwrap();
4127 let resumed_checkpoint = resumed.checkpoint().unwrap();
4128 let resumed_trace = MethodsHpoStudy::restore(context.study.clone(), &resumed_checkpoint)
4129 .unwrap()
4130 .trials()
4131 .unwrap();
4132 assert_eq!(resumed_trace.len(), 2);
4133 assert_eq!(resumed_trace[1].id, resumed_next.trial_id);
4134 assert_eq!(resumed_trace[1].status, HpoTrialStatus::Completed);
4135 assert_eq!(resumed_trace[1].score, Some(0.25));
4136 assert_eq!(
4137 MethodsHpoStudy::restore(context.study.clone(), &resumed_checkpoint)
4138 .unwrap()
4139 .best()
4140 .unwrap()
4141 .unwrap()
4142 .trial
4143 .id,
4144 resumed_next.trial_id
4145 );
4146
4147 let mut wrong_abi = resumed_context.clone();
4148 wrong_abi.study.methods_abi = "n4m-abi-wrong".to_string();
4149 wrong_abi.validate_for_plan(&plan).unwrap();
4150 let error = match controller.create_tuner_session(&task, &wrong_abi) {
4151 Err(error) => error,
4152 Ok(_) => panic!("mismatched Methods ABI unexpectedly restored a session"),
4153 };
4154 assert_runtime_refusal(error);
4155
4156 let mut wrong_binding = resumed_context.clone();
4157 wrong_binding
4158 .resume_checkpoint
4159 .as_mut()
4160 .unwrap()
4161 .binding
4162 .study_id = "study:wrong-binding".to_string();
4163 wrong_binding.validate_for_plan(&plan).unwrap();
4164 let error = match controller.create_tuner_session(&task, &wrong_binding) {
4165 Err(error) => error,
4166 Ok(_) => panic!("mismatched checkpoint binding unexpectedly restored a session"),
4167 };
4168 assert_runtime_refusal(error);
4169
4170 let mut wrong_checksum = resumed_context;
4171 wrong_checksum
4172 .resume_checkpoint
4173 .as_mut()
4174 .unwrap()
4175 .opaque_payload[0] ^= 1;
4176 assert_runtime_refusal(wrong_checksum.validate_for_plan(&plan).unwrap_err());
4177 let error = match controller.create_tuner_session(&task, &wrong_checksum) {
4178 Err(error) => error,
4179 Ok(_) => panic!("bad checkpoint checksum unexpectedly restored a session"),
4180 };
4181 assert_runtime_refusal(error);
4182 }
4183
4184 #[cfg(feature = "methods-optimizer-local")]
4185 #[test]
4186 fn real_n4m_lifecycle_batch_trials_best_and_checkpoint() {
4187 let _runtime = native_runtime();
4188 let config = native_config();
4189 let mut study = MethodsHpoStudy::create(config.clone()).unwrap();
4190 let batch = study.ask_batch(2).unwrap();
4191 assert_eq!(batch.trials.len(), 2);
4192 assert!(batch.native_error.is_none());
4193 assert!(batch.trials.iter().all(|trial| trial.id >= 0));
4194 assert_eq!(batch.trials[0].parameter_order, vec!["n_components"]);
4195
4196 study
4197 .report_intermediate(batch.trials[0].id, 0, 2.0)
4198 .unwrap();
4199 study
4200 .tell(batch.trials[0].id, HpoTerminal::Completed { score: 1.0 })
4201 .unwrap();
4202 study
4203 .tell(
4204 batch.trials[1].id,
4205 HpoTerminal::Failed {
4206 failure: HpoFailure {
4207 code: "EVALUATION_FAILED".into(),
4208 message: "controlled test failure".into(),
4209 retryable: true,
4210 },
4211 },
4212 )
4213 .unwrap();
4214
4215 let trials = study.trials().unwrap();
4216 assert_eq!(trials.len(), 2);
4217 assert_eq!(trials[0].status, HpoTrialStatus::Completed);
4218 assert_eq!(trials[0].score, Some(1.0));
4219 assert_eq!(trials[1].status, HpoTrialStatus::Failed);
4220 assert!(trials[1].failure.as_ref().unwrap().retryable);
4221 assert_eq!(study.best().unwrap().unwrap().score, 1.0);
4222 assert!(study
4223 .events()
4224 .iter()
4225 .any(|event| matches!(event, HpoEvent::Intermediate { step: 0, .. })));
4226
4227 let checkpoint = study.save_checkpoint().unwrap();
4228 let restored = MethodsHpoStudy::restore(config, &checkpoint).unwrap();
4229 assert_eq!(restored.trials().unwrap().len(), 2);
4230 }
4231
4232 #[cfg(feature = "methods-optimizer-local")]
4233 #[test]
4234 fn real_tpe_pruner_failure_trace_and_checkpoint_resume_are_native() {
4235 let _runtime = native_runtime();
4236 let mut config = native_config();
4237 config.optimizer.sampler = HpoSampler::Tpe;
4238 config.optimizer.pruner = HpoPruner::Median;
4239 config.optimizer.n_startup_trials = 2;
4240 config.optimizer.seed = 51;
4241 let mut study = MethodsHpoStudy::create(config.clone()).unwrap();
4242
4243 let failed = study.ask().unwrap();
4247 let failed = study
4248 .tell(
4249 failed.id,
4250 HpoTerminal::Failed {
4251 failure: HpoFailure {
4252 code: "CV_PROVIDER_FAILURE".into(),
4253 message: "controlled fold materialization failure".into(),
4254 retryable: false,
4255 },
4256 },
4257 )
4258 .unwrap();
4259 assert_eq!(failed.status, HpoTrialStatus::Failed);
4260 assert_eq!(failed.failure.unwrap().code, "CV_PROVIDER_FAILURE");
4261
4262 let first = study.ask().unwrap();
4267 assert!(!study.report_intermediate(first.id, 0, 1.0).unwrap());
4268 let second = study.ask().unwrap();
4269 assert!(!study.report_intermediate(second.id, 0, 2.0).unwrap());
4270 let third = study.ask().unwrap();
4271 assert!(study.report_intermediate(third.id, 0, 9.0).unwrap());
4272
4273 let trials = study.trials().unwrap();
4274 let pruned = trials.iter().find(|trial| trial.id == third.id).unwrap();
4275 assert_eq!(pruned.status, HpoTrialStatus::Pruned);
4276 assert!(pruned.terminal_sequence.is_some());
4277 assert!(pruned
4278 .intermediates
4279 .iter()
4280 .any(|item| item.step == 0 && item.score == 9.0 && item.should_prune));
4281 assert!(study.events().iter().any(|event| {
4282 matches!(event, HpoEvent::Terminal { trial_id, status: HpoTrialStatus::Failed, .. } if *trial_id == failed.id)
4283 }));
4284 assert!(study.events().iter().any(|event| {
4285 matches!(event, HpoEvent::Terminal { trial_id, status: HpoTrialStatus::Pruned, .. } if *trial_id == third.id)
4286 }));
4287
4288 let checkpoint = study.save_checkpoint().unwrap();
4289 let checkpoint: N4moptCheckpointArtifact =
4293 serde_json::from_str(&serde_json::to_string(&checkpoint).unwrap()).unwrap();
4294 let mut resumed = MethodsHpoStudy::restore(config, &checkpoint).unwrap();
4295 for _ in 0..4 {
4296 let uninterrupted = study.ask().unwrap();
4297 let restored = resumed.ask().unwrap();
4298 assert_eq!(uninterrupted.id, restored.id);
4299 assert_eq!(uninterrupted.parameter_order, restored.parameter_order);
4300 assert_eq!(uninterrupted.parameters, restored.parameters);
4301 }
4302 }
4303
4304 #[cfg(feature = "methods-optimizer-local")]
4305 #[test]
4306 fn malformed_checkpoint_reaches_native_n4mopt_decoder_as_typed_error() {
4307 let _runtime = native_runtime();
4308 let config = native_config();
4309 let study = MethodsHpoStudy::create(config.clone()).unwrap();
4310 let mut checkpoint = study.save_checkpoint().unwrap();
4311 checkpoint.opaque_payload[0] ^= 1;
4312 checkpoint.payload_sha256 = payload_sha256(&checkpoint.opaque_payload);
4313 assert!(matches!(
4314 MethodsHpoStudy::restore(config, &checkpoint),
4315 Err(HpoError::Native { operation, .. }) if operation == "load_n4mopt"
4316 ));
4317 }
4318}