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