dag_ml_core/runtime/
artifact.rs

1// Auto-split from the former monolithic `runtime.rs` (pure refactor).
2use super::*;
3
4#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
5#[serde(rename_all = "snake_case")]
6pub enum HandleKind {
7    Data,
8    DataView,
9    Model,
10    Artifact,
11    Prediction,
12    Relation,
13}
14
15#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
16#[serde(deny_unknown_fields)]
17pub struct HandleRef {
18    pub handle: u64,
19    pub kind: HandleKind,
20    pub owner_controller: ControllerId,
21}
22
23#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
24#[serde(rename_all = "snake_case")]
25pub enum ArtifactBackend {
26    Joblib,
27    Torch,
28    Tensorflow,
29    Onnx,
30    Safetensors,
31    Json,
32    Raw,
33}
34
35pub const NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1: &str = "dagml.native_predictor_descriptor.v1";
36pub const NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1: u32 = 1;
37pub const NATIVE_PREDICTOR_FORMAT_N4MM: &str = "N4MM";
38pub const NATIVE_PREDICTOR_METHODS_PLS_OWNER: &str = "controller:methods.pls";
39pub const NATIVE_PREDICTOR_METHODS_RIDGE_OWNER: &str = "controller:methods.ridge";
40pub const NATIVE_PREDICTOR_ALGORITHM_PLS: i32 = 0;
41pub const NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR: i32 = 11;
42pub const NATIVE_PREDICTOR_CAPABILITY_PREDICT: u64 = 1 << 0;
43pub const NATIVE_PREDICTOR_CAPABILITY_TRANSFORM: u64 = 1 << 1;
44pub const NATIVE_PREDICTOR_CAPABILITY_AFFINE: u64 = 1 << 2;
45pub const NATIVE_PREDICTOR_CAPABILITY_PIPELINE: u64 = 1 << 3;
46pub const NATIVE_PREDICTOR_CAPABILITIES_V1: u64 = NATIVE_PREDICTOR_CAPABILITY_PREDICT
47    | NATIVE_PREDICTOR_CAPABILITY_TRANSFORM
48    | NATIVE_PREDICTOR_CAPABILITY_AFFINE
49    | NATIVE_PREDICTOR_CAPABILITY_PIPELINE;
50pub const NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1: &str = "n4m.snv_savgol_smooth.v1";
51pub const NATIVE_PREDICTOR_PIPELINE_FINGERPRINT_FNV1A64_V1: &str = "fnv1a64.v1";
52
53/// Writer ABI recorded inside a native predictor payload.
54#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
55#[serde(deny_unknown_fields)]
56pub struct NativePredictorWriterAbiV1 {
57    pub major: u32,
58    pub minor: u32,
59    pub patch: u32,
60}
61
62/// Dimensions authoritatively inspected from the native predictor payload.
63#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
64#[serde(deny_unknown_fields)]
65pub struct NativePredictorDimensionsV1 {
66    pub training_samples: i64,
67    pub n_features: i32,
68    pub n_targets: i32,
69    pub n_components: i32,
70}
71
72/// Opaque, Methods-attested description of the preprocessing embedded in N4MM.
73///
74/// DAG-ML does not decode operator numbers or reproduce their kernels. The
75/// public `n4m` inspector has already reduced the only supported format-2
76/// pipeline to this typed contract. Window and polynomial degree are retained
77/// solely so replay can cross-check the signed controller parameters against
78/// the imported native payload.
79#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
80#[serde(deny_unknown_fields)]
81pub struct NativePredictorPipelineV1 {
82    pub pipeline_type: String,
83    pub schema_version: u32,
84    pub operator_count: u32,
85    pub raw_n_features: i32,
86    pub model_n_features: i32,
87    pub fingerprint_algorithm: String,
88    pub native_fingerprint: String,
89    pub savgol_window: i32,
90    pub savgol_poly_degree: i32,
91}
92
93impl NativePredictorPipelineV1 {
94    pub fn validate(&self) -> Result<()> {
95        if self.pipeline_type != NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1
96            || self.schema_version != 1
97            || self.operator_count != 2
98        {
99            return Err(DagMlError::RuntimeValidation(
100                "native predictor pipeline has an unsupported type or schema".to_string(),
101            ));
102        }
103        if self.raw_n_features <= 0 || self.model_n_features <= 0 {
104            return Err(DagMlError::RuntimeValidation(
105                "native predictor pipeline has invalid feature dimensions".to_string(),
106            ));
107        }
108        if self.fingerprint_algorithm != NATIVE_PREDICTOR_PIPELINE_FINGERPRINT_FNV1A64_V1
109            || self.native_fingerprint.len() != 16
110            || !self
111                .native_fingerprint
112                .bytes()
113                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
114        {
115            return Err(DagMlError::RuntimeValidation(
116                "native predictor pipeline has an invalid native fingerprint".to_string(),
117            ));
118        }
119        if !(3..=501).contains(&self.savgol_window)
120            || self.savgol_window % 2 == 0
121            || self.savgol_poly_degree < 0
122            || self.savgol_poly_degree >= self.savgol_window
123        {
124            return Err(DagMlError::RuntimeValidation(
125                "native predictor pipeline has invalid Savitzky-Golay parameters".to_string(),
126            ));
127        }
128        Ok(())
129    }
130}
131
132/// Content-bound descriptor for a predictor owned by a native controller.
133///
134/// This contract deliberately records Methods' numeric algorithm and
135/// capability mask without translating them into host claims. For N4MM, the
136/// values are produced only from `n4m`'s model/pipeline serialization
137/// inspectors and the descriptor fingerprint is TCV1 over every field except
138/// itself.
139#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
140#[serde(deny_unknown_fields)]
141pub struct NativePredictorDescriptorV1 {
142    pub descriptor_type: String,
143    pub schema_version: u32,
144    pub artifact_sha256: String,
145    pub owner_controller: ControllerId,
146    pub format: String,
147    pub format_version: u32,
148    pub writer_abi: NativePredictorWriterAbiV1,
149    pub storage_algorithm: i32,
150    pub capabilities: u64,
151    pub dimensions: NativePredictorDimensionsV1,
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub pipeline: Option<NativePredictorPipelineV1>,
154    pub descriptor_fingerprint: String,
155}
156
157impl NativePredictorDescriptorV1 {
158    pub fn compute_fingerprint(&self) -> Result<String> {
159        let json = serde_json::to_string(self)?;
160        crate::canonical::parse_typed_json(&json)
161            .and_then(|value| value.fingerprint_without("descriptor_fingerprint"))
162            .map_err(|error| {
163                DagMlError::RuntimeValidation(format!(
164                    "native predictor descriptor is outside TCV1: {error}"
165                ))
166            })
167    }
168
169    pub fn validate(&self) -> Result<()> {
170        if self.descriptor_type != NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1
171            || self.schema_version != NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1
172        {
173            return Err(DagMlError::RuntimeValidation(format!(
174                "unsupported native predictor descriptor `{}` schema_version {}; expected `{}` schema_version {}",
175                self.descriptor_type,
176                self.schema_version,
177                NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1,
178                NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1
179            )));
180        }
181        validate_runtime_fingerprint("native predictor artifact", &self.artifact_sha256)?;
182        validate_runtime_fingerprint("native predictor descriptor", &self.descriptor_fingerprint)?;
183        if self.format != NATIVE_PREDICTOR_FORMAT_N4MM || self.format_version == 0 {
184            return Err(DagMlError::RuntimeValidation(format!(
185                "native predictor descriptor has unsupported format `{}:{}`",
186                self.format, self.format_version
187            )));
188        }
189        if self.writer_abi.major == 0 {
190            return Err(DagMlError::RuntimeValidation(
191                "native predictor descriptor writer ABI major must be non-zero".to_string(),
192            ));
193        }
194        if self.dimensions.training_samples <= 0
195            || self.dimensions.n_features <= 0
196            || self.dimensions.n_targets <= 0
197            || self.dimensions.n_components < 0
198        {
199            return Err(DagMlError::RuntimeValidation(
200                "native predictor descriptor has invalid model dimensions".to_string(),
201            ));
202        }
203        if self.capabilities & !NATIVE_PREDICTOR_CAPABILITIES_V1 != 0 {
204            return Err(DagMlError::RuntimeValidation(format!(
205                "native predictor descriptor uses unsupported capability bits {:#x}",
206                self.capabilities & !NATIVE_PREDICTOR_CAPABILITIES_V1
207            )));
208        }
209        let required_predict = self.capabilities & NATIVE_PREDICTOR_CAPABILITY_PREDICT != 0;
210        let has_pipeline_capability = self.capabilities & NATIVE_PREDICTOR_CAPABILITY_PIPELINE != 0;
211        let product_supported = match self.owner_controller.as_str() {
212            NATIVE_PREDICTOR_METHODS_PLS_OWNER => {
213                self.storage_algorithm == NATIVE_PREDICTOR_ALGORITHM_PLS
214                    && required_predict
215                    && self.dimensions.n_components > 0
216                    && match &self.pipeline {
217                        None => self.format_version == 1 && !has_pipeline_capability,
218                        Some(pipeline) => {
219                            self.format_version == 2
220                                && has_pipeline_capability
221                                && pipeline.validate().is_ok()
222                                && pipeline.raw_n_features == self.dimensions.n_features
223                                && pipeline.model_n_features == self.dimensions.n_features
224                        }
225                    }
226            }
227            NATIVE_PREDICTOR_METHODS_RIDGE_OWNER => {
228                self.storage_algorithm == NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR
229                    && required_predict
230                    && self.capabilities & NATIVE_PREDICTOR_CAPABILITY_AFFINE != 0
231                    && self.dimensions.n_components == 0
232                    && self.format_version == 1
233                    && !has_pipeline_capability
234                    && self.pipeline.is_none()
235            }
236            _ => false,
237        };
238        if !product_supported {
239            return Err(DagMlError::RuntimeValidation(format!(
240                "native predictor storage algorithm {} with capabilities {:#x} and {} component(s) is not product-supported by `{}`",
241                self.storage_algorithm,
242                self.capabilities,
243                self.dimensions.n_components,
244                self.owner_controller
245            )));
246        }
247        if self.descriptor_fingerprint != self.compute_fingerprint()? {
248            return Err(DagMlError::RuntimeValidation(
249                "native predictor descriptor fingerprint does not match TCV1 content".to_string(),
250            ));
251        }
252        Ok(())
253    }
254}
255
256#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
257pub struct ArtifactRef {
258    pub id: ArtifactId,
259    pub kind: String,
260    pub controller_id: ControllerId,
261    #[serde(default, skip_serializing_if = "Option::is_none")]
262    pub backend: Option<ArtifactBackend>,
263    #[serde(default, skip_serializing_if = "Option::is_none")]
264    pub uri: Option<String>,
265    #[serde(default, skip_serializing_if = "Option::is_none")]
266    pub content_fingerprint: Option<String>,
267    pub size_bytes: Option<u64>,
268    #[serde(default, skip_serializing_if = "Option::is_none")]
269    pub plugin: Option<String>,
270    #[serde(default, skip_serializing_if = "Option::is_none")]
271    pub plugin_version: Option<String>,
272    /// Native ABI family required to consume this payload.  Both ABI fields
273    /// are absent on historical non-native references; new native Methods
274    /// writers always emit the pair.
275    #[serde(default, skip_serializing_if = "Option::is_none")]
276    pub abi_major: Option<u32>,
277    /// Minimum compatible minor within [`Self::abi_major`].  This is derived
278    /// from the payload capability, never copied from the writer's runtime.
279    #[serde(default, skip_serializing_if = "Option::is_none")]
280    pub abi_min_minor: Option<u32>,
281    /// Content-derived native predictor metadata. Historical and host-owned
282    /// artifacts omit it; new native Methods writers always emit it.
283    #[serde(default, skip_serializing_if = "Option::is_none")]
284    pub native_predictor_descriptor: Option<NativePredictorDescriptorV1>,
285}
286
287impl ArtifactRef {
288    pub fn validate(&self) -> Result<()> {
289        if self.kind.trim().is_empty() {
290            return Err(DagMlError::RuntimeValidation(format!(
291                "artifact `{}` has empty kind",
292                self.id
293            )));
294        }
295        validate_artifact_optional_text("uri", &self.uri, &self.id)?;
296        validate_artifact_optional_text("plugin", &self.plugin, &self.id)?;
297        validate_artifact_optional_text("plugin_version", &self.plugin_version, &self.id)?;
298        if self.plugin_version.is_some() && self.plugin.is_none() {
299            return Err(DagMlError::RuntimeValidation(format!(
300                "artifact `{}` has plugin_version without plugin",
301                self.id
302            )));
303        }
304        if self.abi_major.is_some() != self.abi_min_minor.is_some() || self.abi_major == Some(0) {
305            return Err(DagMlError::RuntimeValidation(format!(
306                "artifact `{}` must declare a non-zero abi_major together with abi_min_minor",
307                self.id
308            )));
309        }
310        if let Some(content_fingerprint) = &self.content_fingerprint {
311            validate_runtime_fingerprint("artifact content", content_fingerprint)?;
312        }
313        if let Some(descriptor) = &self.native_predictor_descriptor {
314            descriptor.validate()?;
315            if descriptor.owner_controller != self.controller_id
316                || self.backend != Some(ArtifactBackend::Raw)
317                || self.kind != "n4m_model"
318                || self.content_fingerprint.as_deref() != Some(descriptor.artifact_sha256.as_str())
319                || self.abi_major != Some(descriptor.writer_abi.major)
320            {
321                return Err(DagMlError::RuntimeValidation(format!(
322                    "artifact `{}` does not match its native predictor descriptor",
323                    self.id
324                )));
325            }
326        }
327        if self.uri.is_some() && self.backend.is_none() {
328            return Err(DagMlError::RuntimeValidation(format!(
329                "artifact `{}` has uri without backend",
330                self.id
331            )));
332        }
333        if self.uri.is_some() && self.content_fingerprint.is_none() {
334            return Err(DagMlError::RuntimeValidation(format!(
335                "artifact `{}` has uri without content_fingerprint",
336                self.id
337            )));
338        }
339        Ok(())
340    }
341
342    /// Validate that the artifact carries portable metadata: a backend, a safe
343    /// relative URI and a content fingerprint. Legacy artifacts that only carry
344    /// inline metadata stay readable through [`ArtifactRef::validate`] but are
345    /// refused here so persisted manifests can be moved with their payloads.
346    pub fn validate_portable(&self) -> Result<()> {
347        self.validate()?;
348        let Some(uri) = self.uri.as_deref() else {
349            return Err(DagMlError::RuntimeValidation(format!(
350                "artifact `{}` is not portable: requires backend, uri and content_fingerprint",
351                self.id
352            )));
353        };
354        // `validate` already guarantees that a present URI implies a backend and
355        // a 64-hex content fingerprint, so confirming the URI is enough here.
356        validate_relative_artifact_uri(&self.id, uri)
357    }
358}
359
360pub fn refit_artifact_input_key(artifact_id: &ArtifactId) -> String {
361    format!("artifact:{artifact_id}")
362}
363
364#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
365pub struct ArtifactMaterializationRequest {
366    pub run_id: RunId,
367    pub bundle_id: BundleId,
368    pub node_id: NodeId,
369    pub phase: Phase,
370    pub variant_id: Option<VariantId>,
371    pub controller_id: ControllerId,
372    pub artifact: ArtifactRef,
373    pub params_fingerprint: String,
374    #[serde(default, skip_serializing_if = "Option::is_none")]
375    pub training_loss_fingerprint: Option<String>,
376}
377
378#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
379pub struct ArtifactHandleRecord {
380    pub handle: HandleRef,
381    pub node_id: NodeId,
382    pub controller_id: ControllerId,
383    pub artifact: ArtifactRef,
384    pub params_fingerprint: String,
385    #[serde(default, skip_serializing_if = "Option::is_none")]
386    pub training_loss_fingerprint: Option<String>,
387}
388
389impl ArtifactHandleRecord {
390    pub fn validate(&self) -> Result<()> {
391        self.artifact.validate()?;
392        if !matches!(self.handle.kind, HandleKind::Model | HandleKind::Artifact) {
393            return Err(DagMlError::RuntimeValidation(format!(
394                "artifact `{}` is registered with non-artifact/model handle kind {:?}",
395                self.artifact.id, self.handle.kind
396            )));
397        }
398        if self.handle.owner_controller != self.controller_id {
399            return Err(DagMlError::RuntimeValidation(format!(
400                "artifact `{}` handle owner `{}` does not match controller `{}`",
401                self.artifact.id, self.handle.owner_controller, self.controller_id
402            )));
403        }
404        if self.artifact.controller_id != self.controller_id {
405            return Err(DagMlError::RuntimeValidation(format!(
406                "artifact `{}` controller `{}` does not match record controller `{}`",
407                self.artifact.id, self.artifact.controller_id, self.controller_id
408            )));
409        }
410        if self.params_fingerprint.trim().is_empty() {
411            return Err(DagMlError::RuntimeValidation(format!(
412                "artifact `{}` has empty params fingerprint",
413                self.artifact.id
414            )));
415        }
416        if let Some(fingerprint) = &self.training_loss_fingerprint {
417            validate_runtime_fingerprint("artifact training loss", fingerprint)?;
418        }
419        Ok(())
420    }
421}
422
423pub trait RuntimeArtifactStore {
424    fn materialize(&self, request: &ArtifactMaterializationRequest) -> Result<HandleRef>;
425}
426
427#[derive(Clone, Debug, Default)]
428pub struct InMemoryArtifactStore {
429    records: BTreeMap<ArtifactId, ArtifactHandleRecord>,
430    refit_artifacts: BTreeMap<ArtifactId, RefitArtifactRecord>,
431}
432
433impl InMemoryArtifactStore {
434    pub fn new() -> Self {
435        Self::default()
436    }
437
438    pub fn register(&mut self, artifact: &RefitArtifactRecord, handle: HandleRef) -> Result<()> {
439        artifact.validate()?;
440        let record = ArtifactHandleRecord {
441            handle,
442            node_id: artifact.node_id.clone(),
443            controller_id: artifact.controller_id.clone(),
444            artifact: artifact.artifact.clone(),
445            params_fingerprint: artifact.params_fingerprint.clone(),
446            training_loss_fingerprint: artifact.training_loss_fingerprint.clone(),
447        };
448        record.validate()?;
449        if self.records.contains_key(&record.artifact.id)
450            || self.refit_artifacts.contains_key(&record.artifact.id)
451        {
452            return Err(DagMlError::RuntimeValidation(format!(
453                "duplicate artifact handle for `{}`",
454                artifact.artifact.id
455            )));
456        }
457        let previous_record = self.records.insert(record.artifact.id.clone(), record);
458        debug_assert!(previous_record.is_none());
459        let previous_artifact = self
460            .refit_artifacts
461            .insert(artifact.artifact.id.clone(), artifact.clone());
462        debug_assert!(previous_artifact.is_none());
463        Ok(())
464    }
465
466    pub fn capture_refit_artifacts(
467        &mut self,
468        task: &NodeTask,
469        result: &NodeResult,
470    ) -> Result<Vec<RefitArtifactRecord>> {
471        if task.phase != Phase::Refit {
472            return Err(DagMlError::RuntimeValidation(format!(
473                "cannot capture refit artifacts from phase {:?}",
474                task.phase
475            )));
476        }
477        let mut records = Vec::new();
478        for artifact in &result.artifacts {
479            let handle = result.artifact_handles.get(&artifact.id).ok_or_else(|| {
480                DagMlError::RuntimeValidation(format!(
481                    "node `{}` emitted artifact `{}` without artifact handle",
482                    task.node_plan.node_id, artifact.id
483                ))
484            })?;
485            let record = RefitArtifactRecord {
486                node_id: task.node_plan.node_id.clone(),
487                controller_id: task.node_plan.controller_id.clone(),
488                artifact: artifact.clone(),
489                params_fingerprint: task.node_plan.params_fingerprint.clone(),
490                training_loss_fingerprint: task
491                    .node_plan
492                    .training_loss_fingerprint(Phase::Refit)?,
493                data_requirement_keys: task
494                    .node_plan
495                    .data_bindings
496                    .iter()
497                    .map(|binding| {
498                        data_binding_requirement_key(&binding.node_id, &binding.input_name)
499                    })
500                    .collect(),
501                // Only the Validation-OOF meta-feature inputs become replay
502                // prediction-cache requirements. The off-fold (REFIT/PREDICT)
503                // test/predict base predictions are recomputed each phase, not
504                // replayed from cache, and they share the same producer/port as the
505                // Validation OOF input — so including them would duplicate the
506                // requirement key. They are excluded here (partition != Validation).
507                prediction_requirement_keys: task
508                    .prediction_inputs
509                    .values()
510                    .filter(|spec| spec.partition == PredictionPartition::Validation)
511                    .map(|spec| {
512                        bundle_prediction_requirement_key(
513                            &spec.producer_node,
514                            &spec.source_port,
515                            &task.node_plan.node_id,
516                            &spec.target_port,
517                        )
518                    })
519                    .collect(),
520            };
521            self.register(&record, handle.clone())?;
522            records.push(record);
523        }
524        Ok(records)
525    }
526
527    pub fn get(&self, artifact_id: &ArtifactId) -> Option<&ArtifactHandleRecord> {
528        self.records.get(artifact_id)
529    }
530
531    pub fn len(&self) -> usize {
532        self.records.len()
533    }
534
535    pub fn is_empty(&self) -> bool {
536        self.records.is_empty()
537    }
538
539    pub fn refit_artifacts(&self) -> Vec<RefitArtifactRecord> {
540        self.refit_artifacts.values().cloned().collect()
541    }
542}
543
544impl RuntimeArtifactStore for InMemoryArtifactStore {
545    fn materialize(&self, request: &ArtifactMaterializationRequest) -> Result<HandleRef> {
546        let record = self.records.get(&request.artifact.id).ok_or_else(|| {
547            DagMlError::RuntimeValidation(format!(
548                "artifact store is missing refit artifact `{}` for bundle `{}`",
549                request.artifact.id, request.bundle_id
550            ))
551        })?;
552        if record.node_id != request.node_id {
553            return Err(DagMlError::RuntimeValidation(format!(
554                "artifact `{}` is registered for node `{}` but requested for `{}`",
555                request.artifact.id, record.node_id, request.node_id
556            )));
557        }
558        if record.controller_id != request.controller_id {
559            return Err(DagMlError::RuntimeValidation(format!(
560                "artifact `{}` is registered for controller `{}` but requested for `{}`",
561                request.artifact.id, record.controller_id, request.controller_id
562            )));
563        }
564        if record.artifact != request.artifact {
565            return Err(DagMlError::RuntimeValidation(format!(
566                "artifact `{}` metadata does not match bundle record",
567                request.artifact.id
568            )));
569        }
570        if record.params_fingerprint != request.params_fingerprint {
571            return Err(DagMlError::RuntimeValidation(format!(
572                "artifact `{}` params fingerprint does not match bundle record",
573                request.artifact.id
574            )));
575        }
576        if record.training_loss_fingerprint != request.training_loss_fingerprint {
577            return Err(DagMlError::RuntimeValidation(format!(
578                "artifact `{}` training loss fingerprint does not match bundle record",
579                request.artifact.id
580            )));
581        }
582        record.validate()?;
583        Ok(record.handle.clone())
584    }
585}
586
587pub const FILE_ARTIFACT_MANIFEST_SCHEMA_VERSION: u32 = 1;
588pub const FILE_ARTIFACT_MANIFEST_FILE: &str = "artifact_manifest.json";
589
590pub(crate) fn default_file_artifact_manifest_schema_version() -> u32 {
591    FILE_ARTIFACT_MANIFEST_SCHEMA_VERSION
592}
593
594/// One persisted artifact entry. Mirrors the bundle [`RefitArtifactRecord`]
595/// identity (node, controller, artifact, parameters and training loss) while requiring
596/// the [`ArtifactRef`] to be portable so the manifest stays movable with its
597/// payloads.
598#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
599pub struct FileArtifactManifestEntry {
600    pub node_id: NodeId,
601    pub controller_id: ControllerId,
602    pub artifact: ArtifactRef,
603    pub params_fingerprint: String,
604    #[serde(default, skip_serializing_if = "Option::is_none")]
605    pub training_loss_fingerprint: Option<String>,
606}
607
608impl FileArtifactManifestEntry {
609    fn from_refit_record(record: &RefitArtifactRecord) -> Result<Self> {
610        let entry = Self {
611            node_id: record.node_id.clone(),
612            controller_id: record.controller_id.clone(),
613            artifact: record.artifact.clone(),
614            params_fingerprint: record.params_fingerprint.clone(),
615            training_loss_fingerprint: record.training_loss_fingerprint.clone(),
616        };
617        entry.validate()?;
618        Ok(entry)
619    }
620
621    pub fn validate(&self) -> Result<()> {
622        self.artifact.validate_portable()?;
623        if self.artifact.controller_id != self.controller_id {
624            return Err(DagMlError::RuntimeValidation(format!(
625                "artifact manifest entry `{}` controller `{}` does not match artifact controller `{}`",
626                self.artifact.id, self.controller_id, self.artifact.controller_id
627            )));
628        }
629        validate_runtime_fingerprint("artifact manifest params", &self.params_fingerprint)?;
630        if let Some(fingerprint) = &self.training_loss_fingerprint {
631            validate_runtime_fingerprint("artifact manifest training loss", fingerprint)?;
632        }
633        Ok(())
634    }
635
636    fn matches_refit_record(&self, record: &RefitArtifactRecord) -> bool {
637        self.node_id == record.node_id
638            && self.controller_id == record.controller_id
639            && self.artifact == record.artifact
640            && self.params_fingerprint == record.params_fingerprint
641            && self.training_loss_fingerprint == record.training_loss_fingerprint
642    }
643}
644
645/// Versioned, file-backed artifact manifest. This is a manifest/portability
646/// layer only: it records portable [`ArtifactRef`] metadata for a bundle's
647/// refit artifacts. It does not deserialize ML objects or materialize artifact
648/// payloads; payload stores remain future work.
649#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
650pub struct FileArtifactManifest {
651    pub bundle_id: BundleId,
652    #[serde(default = "default_file_artifact_manifest_schema_version")]
653    pub schema_version: u32,
654    #[serde(default)]
655    pub artifacts: Vec<FileArtifactManifestEntry>,
656}
657
658impl FileArtifactManifest {
659    pub fn validate(&self) -> Result<()> {
660        if self.schema_version != FILE_ARTIFACT_MANIFEST_SCHEMA_VERSION {
661            return Err(DagMlError::RuntimeValidation(format!(
662                "file artifact manifest for bundle `{}` uses unsupported schema_version {}, expected {}",
663                self.bundle_id, self.schema_version, FILE_ARTIFACT_MANIFEST_SCHEMA_VERSION
664            )));
665        }
666        let mut artifact_ids = BTreeSet::new();
667        let mut uris = BTreeSet::new();
668        for entry in &self.artifacts {
669            entry.validate()?;
670            if !artifact_ids.insert(entry.artifact.id.as_str()) {
671                return Err(DagMlError::RuntimeValidation(format!(
672                    "file artifact manifest for bundle `{}` has duplicate artifact id `{}`",
673                    self.bundle_id, entry.artifact.id
674                )));
675            }
676            // `entry.validate` guarantees a portable URI is present.
677            if let Some(uri) = entry.artifact.uri.as_deref() {
678                if !uris.insert(uri) {
679                    return Err(DagMlError::RuntimeValidation(format!(
680                        "file artifact manifest for bundle `{}` has duplicate artifact uri `{}`",
681                        self.bundle_id, uri
682                    )));
683                }
684            }
685        }
686        Ok(())
687    }
688
689    pub fn validate_against_bundle(&self, bundle: &ExecutionBundle) -> Result<()> {
690        self.validate()?;
691        bundle.validate()?;
692        if self.bundle_id != bundle.bundle_id {
693            return Err(DagMlError::RuntimeValidation(format!(
694                "file artifact manifest bundle `{}` does not match bundle `{}`",
695                self.bundle_id, bundle.bundle_id
696            )));
697        }
698        if self.artifacts.len() != bundle.refit_artifacts.len() {
699            return Err(DagMlError::RuntimeValidation(format!(
700                "file artifact manifest for bundle `{}` has {} artifact(s) for {} bundle refit artifact(s)",
701                self.bundle_id,
702                self.artifacts.len(),
703                bundle.refit_artifacts.len()
704            )));
705        }
706        let entries_by_id = self
707            .artifacts
708            .iter()
709            .map(|entry| (entry.artifact.id.as_str(), entry))
710            .collect::<BTreeMap<_, _>>();
711        for record in &bundle.refit_artifacts {
712            let entry = entries_by_id
713                .get(record.artifact.id.as_str())
714                .ok_or_else(|| {
715                    DagMlError::RuntimeValidation(format!(
716                        "file artifact manifest for bundle `{}` is missing refit artifact `{}`",
717                        self.bundle_id, record.artifact.id
718                    ))
719                })?;
720            if !entry.matches_refit_record(record) {
721                return Err(DagMlError::RuntimeValidation(format!(
722                    "file artifact manifest entry `{}` does not match bundle refit artifact",
723                    entry.artifact.id
724                )));
725            }
726        }
727        Ok(())
728    }
729}
730
731/// File-backed artifact manifest store rooted at a directory.
732///
733/// This is a portability/manifest layer: [`FileArtifactManifestStore::write`]
734/// serializes portable artifact references from a validated bundle and
735/// [`FileArtifactManifestStore::open`] reloads and revalidates them against the
736/// bundle. It never reads, writes or deserializes artifact payloads.
737#[derive(Clone, Debug)]
738pub struct FileArtifactManifestStore {
739    root: PathBuf,
740    manifest: FileArtifactManifest,
741}
742
743impl FileArtifactManifestStore {
744    pub fn write(root: impl AsRef<Path>, bundle: &ExecutionBundle) -> Result<FileArtifactManifest> {
745        bundle.validate()?;
746        let root = root.as_ref();
747        fs::create_dir_all(root).map_err(|err| {
748            DagMlError::RuntimeValidation(format!(
749                "failed to create artifact manifest store `{}`: {err}",
750                root.display()
751            ))
752        })?;
753        let mut entries = Vec::with_capacity(bundle.refit_artifacts.len());
754        for record in &bundle.refit_artifacts {
755            entries.push(FileArtifactManifestEntry::from_refit_record(record)?);
756        }
757        entries.sort_by(|left, right| left.artifact.id.cmp(&right.artifact.id));
758        let manifest = FileArtifactManifest {
759            bundle_id: bundle.bundle_id.clone(),
760            schema_version: FILE_ARTIFACT_MANIFEST_SCHEMA_VERSION,
761            artifacts: entries,
762        };
763        manifest.validate_against_bundle(bundle)?;
764        write_runtime_json(
765            &root.join(FILE_ARTIFACT_MANIFEST_FILE),
766            &manifest,
767            "artifact manifest",
768        )?;
769        Ok(manifest)
770    }
771
772    pub fn open(root: impl Into<PathBuf>, bundle: &ExecutionBundle) -> Result<Self> {
773        bundle.validate()?;
774        let root = root.into();
775        let manifest: FileArtifactManifest =
776            read_runtime_json(&root.join(FILE_ARTIFACT_MANIFEST_FILE), "artifact manifest")?;
777        manifest.validate_against_bundle(bundle)?;
778        Ok(Self { root, manifest })
779    }
780
781    pub fn root(&self) -> &Path {
782        &self.root
783    }
784
785    pub fn manifest(&self) -> &FileArtifactManifest {
786        &self.manifest
787    }
788}
789
790#[derive(Clone, Debug, Eq, PartialEq)]
791pub struct ArtifactPayloadMaterializationRecord {
792    pub run_id: RunId,
793    pub bundle_id: BundleId,
794    pub node_id: NodeId,
795    pub phase: Phase,
796    pub variant_id: Option<VariantId>,
797    pub artifact_id: ArtifactId,
798    pub training_loss_fingerprint: Option<String>,
799    pub payload_uri: String,
800    pub content_fingerprint: String,
801    pub size_bytes: u64,
802    pub handle: HandleRef,
803}
804
805#[derive(Clone, Debug, Eq, PartialEq)]
806pub(crate) struct ArtifactPayloadMetadata {
807    pub(crate) uri: String,
808    pub(crate) content_fingerprint: String,
809    pub(crate) size_bytes: u64,
810}
811
812#[derive(Clone, Debug)]
813pub struct FileArtifactPayloadStore {
814    root: PathBuf,
815    manifest: FileArtifactManifest,
816    records_by_artifact_id: BTreeMap<ArtifactId, RefitArtifactRecord>,
817    materialization_records: RefCell<Vec<ArtifactPayloadMaterializationRecord>>,
818}
819
820impl FileArtifactPayloadStore {
821    pub fn write_from_source(
822        output_root: impl AsRef<Path>,
823        source_root: impl AsRef<Path>,
824        bundle: &ExecutionBundle,
825    ) -> Result<Self> {
826        bundle.validate()?;
827        let output_root = output_root.as_ref();
828        let source_root = source_root.as_ref();
829        fs::create_dir_all(output_root).map_err(|err| {
830            DagMlError::RuntimeValidation(format!(
831                "failed to create artifact payload store `{}`: {err}",
832                output_root.display()
833            ))
834        })?;
835        for record in &bundle.refit_artifacts {
836            record.artifact.validate_portable()?;
837            validate_artifact_payload_file(source_root, &record.artifact)?;
838            let source_path = artifact_payload_path(source_root, &record.artifact)?;
839            let output_path = artifact_payload_path(output_root, &record.artifact)?;
840            if let Some(parent) = output_path.parent() {
841                fs::create_dir_all(parent).map_err(|err| {
842                    DagMlError::RuntimeValidation(format!(
843                        "failed to create artifact payload directory `{}`: {err}",
844                        parent.display()
845                    ))
846                })?;
847            }
848            if source_path != output_path {
849                fs::copy(&source_path, &output_path).map_err(|err| {
850                    DagMlError::RuntimeValidation(format!(
851                        "failed to copy artifact payload `{}` from {} to {}: {err}",
852                        record.artifact.id,
853                        source_path.display(),
854                        output_path.display()
855                    ))
856                })?;
857            }
858        }
859        FileArtifactManifestStore::write(output_root, bundle)?;
860        Self::open(output_root.to_path_buf(), bundle)
861    }
862
863    pub fn open(root: impl Into<PathBuf>, bundle: &ExecutionBundle) -> Result<Self> {
864        bundle.validate()?;
865        let root = root.into();
866        let manifest_store = FileArtifactManifestStore::open(root.clone(), bundle)?;
867        let records_by_artifact_id = bundle
868            .refit_artifacts
869            .iter()
870            .cloned()
871            .map(|record| (record.artifact.id.clone(), record))
872            .collect::<BTreeMap<_, _>>();
873        let store = Self {
874            root,
875            manifest: manifest_store.manifest().clone(),
876            records_by_artifact_id,
877            materialization_records: RefCell::new(Vec::new()),
878        };
879        store.validate_payloads()?;
880        Ok(store)
881    }
882
883    pub fn root(&self) -> &Path {
884        &self.root
885    }
886
887    pub fn manifest(&self) -> &FileArtifactManifest {
888        &self.manifest
889    }
890
891    pub fn payload_count(&self) -> usize {
892        self.manifest.artifacts.len()
893    }
894
895    pub fn materialization_records(&self) -> Vec<ArtifactPayloadMaterializationRecord> {
896        self.materialization_records.borrow().clone()
897    }
898
899    pub fn validate_payloads(&self) -> Result<()> {
900        self.manifest.validate()?;
901        for entry in &self.manifest.artifacts {
902            let record = self
903                .records_by_artifact_id
904                .get(&entry.artifact.id)
905                .ok_or_else(|| {
906                    DagMlError::RuntimeValidation(format!(
907                        "artifact payload store for bundle `{}` has no bundle record for `{}`",
908                        self.manifest.bundle_id, entry.artifact.id
909                    ))
910                })?;
911            if !entry.matches_refit_record(record) {
912                return Err(DagMlError::RuntimeValidation(format!(
913                    "artifact payload store entry `{}` does not match bundle refit artifact",
914                    entry.artifact.id
915                )));
916            }
917            validate_artifact_payload_file(&self.root, &entry.artifact)?;
918        }
919        Ok(())
920    }
921}
922
923impl RuntimeArtifactStore for FileArtifactPayloadStore {
924    fn materialize(&self, request: &ArtifactMaterializationRequest) -> Result<HandleRef> {
925        request.artifact.validate_portable()?;
926        let record = self
927            .records_by_artifact_id
928            .get(&request.artifact.id)
929            .ok_or_else(|| {
930                DagMlError::RuntimeValidation(format!(
931                    "artifact payload store is missing refit artifact `{}` for bundle `{}`",
932                    request.artifact.id, request.bundle_id
933                ))
934            })?;
935        if record.node_id != request.node_id {
936            return Err(DagMlError::RuntimeValidation(format!(
937                "artifact `{}` is registered for node `{}` but requested for `{}`",
938                request.artifact.id, record.node_id, request.node_id
939            )));
940        }
941        if record.controller_id != request.controller_id {
942            return Err(DagMlError::RuntimeValidation(format!(
943                "artifact `{}` is registered for controller `{}` but requested for `{}`",
944                request.artifact.id, record.controller_id, request.controller_id
945            )));
946        }
947        if record.artifact != request.artifact {
948            return Err(DagMlError::RuntimeValidation(format!(
949                "artifact `{}` metadata does not match bundle record",
950                request.artifact.id
951            )));
952        }
953        if record.params_fingerprint != request.params_fingerprint {
954            return Err(DagMlError::RuntimeValidation(format!(
955                "artifact `{}` params fingerprint does not match bundle record",
956                request.artifact.id
957            )));
958        }
959        if record.training_loss_fingerprint != request.training_loss_fingerprint {
960            return Err(DagMlError::RuntimeValidation(format!(
961                "artifact `{}` training loss fingerprint does not match bundle record",
962                request.artifact.id
963            )));
964        }
965        let metadata = validate_artifact_payload_file(&self.root, &request.artifact)?;
966        let fingerprint = stable_json_fingerprint(&(
967            &request.run_id,
968            &request.bundle_id,
969            &request.node_id,
970            request.phase,
971            &request.variant_id,
972            &request.artifact.id,
973            &metadata.content_fingerprint,
974            &request.params_fingerprint,
975            &request.training_loss_fingerprint,
976        ))?;
977        let handle = HandleRef {
978            handle: u64::from_str_radix(&fingerprint[..16], 16)
979                .expect("sha256 hex prefix should fit into u64"),
980            kind: HandleKind::Artifact,
981            owner_controller: request.controller_id.clone(),
982        };
983        self.materialization_records
984            .borrow_mut()
985            .push(ArtifactPayloadMaterializationRecord {
986                run_id: request.run_id.clone(),
987                bundle_id: request.bundle_id.clone(),
988                node_id: request.node_id.clone(),
989                phase: request.phase,
990                variant_id: request.variant_id.clone(),
991                artifact_id: request.artifact.id.clone(),
992                training_loss_fingerprint: request.training_loss_fingerprint.clone(),
993                payload_uri: metadata.uri,
994                content_fingerprint: metadata.content_fingerprint,
995                size_bytes: metadata.size_bytes,
996                handle: handle.clone(),
997            });
998        Ok(handle)
999    }
1000}
1001
1002#[cfg(test)]
1003mod native_predictor_descriptor_tests {
1004    use super::*;
1005
1006    fn signed_descriptor(
1007        owner: &str,
1008        algorithm: i32,
1009        capabilities: u64,
1010        n_components: i32,
1011    ) -> NativePredictorDescriptorV1 {
1012        let mut descriptor = NativePredictorDescriptorV1 {
1013            descriptor_type: NATIVE_PREDICTOR_DESCRIPTOR_TYPE_V1.to_string(),
1014            schema_version: NATIVE_PREDICTOR_DESCRIPTOR_SCHEMA_VERSION_V1,
1015            artifact_sha256: "a".repeat(64),
1016            owner_controller: ControllerId::new(owner).unwrap(),
1017            format: NATIVE_PREDICTOR_FORMAT_N4MM.to_string(),
1018            format_version: 1,
1019            writer_abi: NativePredictorWriterAbiV1 {
1020                major: 2,
1021                minor: 4,
1022                patch: 0,
1023            },
1024            storage_algorithm: algorithm,
1025            capabilities,
1026            dimensions: NativePredictorDimensionsV1 {
1027                training_samples: 8,
1028                n_features: 3,
1029                n_targets: 1,
1030                n_components,
1031            },
1032            pipeline: None,
1033            descriptor_fingerprint: String::new(),
1034        };
1035        descriptor.descriptor_fingerprint = descriptor.compute_fingerprint().unwrap();
1036        descriptor
1037    }
1038
1039    #[test]
1040    fn pure_descriptor_validation_rejects_controller_algorithm_capability_and_dimension_spoofs() {
1041        signed_descriptor(
1042            NATIVE_PREDICTOR_METHODS_PLS_OWNER,
1043            NATIVE_PREDICTOR_ALGORITHM_PLS,
1044            NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_TRANSFORM,
1045            1,
1046        )
1047        .validate()
1048        .unwrap();
1049        signed_descriptor(
1050            NATIVE_PREDICTOR_METHODS_RIDGE_OWNER,
1051            NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR,
1052            NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_AFFINE,
1053            0,
1054        )
1055        .validate()
1056        .unwrap();
1057
1058        let invalid = [
1059            signed_descriptor(
1060                NATIVE_PREDICTOR_METHODS_RIDGE_OWNER,
1061                NATIVE_PREDICTOR_ALGORITHM_PLS,
1062                NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_TRANSFORM,
1063                1,
1064            ),
1065            signed_descriptor(
1066                NATIVE_PREDICTOR_METHODS_PLS_OWNER,
1067                NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR,
1068                NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_AFFINE,
1069                0,
1070            ),
1071            signed_descriptor(
1072                NATIVE_PREDICTOR_METHODS_PLS_OWNER,
1073                NATIVE_PREDICTOR_ALGORITHM_PLS,
1074                NATIVE_PREDICTOR_CAPABILITY_TRANSFORM,
1075                1,
1076            ),
1077            signed_descriptor(
1078                NATIVE_PREDICTOR_METHODS_RIDGE_OWNER,
1079                NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR,
1080                NATIVE_PREDICTOR_CAPABILITY_PREDICT,
1081                0,
1082            ),
1083            signed_descriptor(
1084                NATIVE_PREDICTOR_METHODS_PLS_OWNER,
1085                NATIVE_PREDICTOR_ALGORITHM_PLS,
1086                NATIVE_PREDICTOR_CAPABILITY_PREDICT,
1087                0,
1088            ),
1089            signed_descriptor(
1090                NATIVE_PREDICTOR_METHODS_RIDGE_OWNER,
1091                NATIVE_PREDICTOR_ALGORITHM_IMPORTED_LINEAR,
1092                NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_AFFINE,
1093                1,
1094            ),
1095        ];
1096        for descriptor in invalid {
1097            assert!(descriptor
1098                .validate()
1099                .unwrap_err()
1100                .to_string()
1101                .contains("not product-supported"));
1102        }
1103    }
1104
1105    #[test]
1106    fn pipeline_descriptor_is_additive_and_keeps_historical_json_shape() {
1107        let historical = signed_descriptor(
1108            NATIVE_PREDICTOR_METHODS_PLS_OWNER,
1109            NATIVE_PREDICTOR_ALGORITHM_PLS,
1110            NATIVE_PREDICTOR_CAPABILITY_PREDICT | NATIVE_PREDICTOR_CAPABILITY_TRANSFORM,
1111            1,
1112        );
1113        assert!(serde_json::to_value(&historical)
1114            .unwrap()
1115            .get("pipeline")
1116            .is_none());
1117        assert_eq!(
1118            historical.descriptor_fingerprint,
1119            historical.compute_fingerprint().unwrap()
1120        );
1121
1122        let mut pipeline = historical;
1123        pipeline.format_version = 2;
1124        pipeline.capabilities |= NATIVE_PREDICTOR_CAPABILITY_PIPELINE;
1125        pipeline.pipeline = Some(NativePredictorPipelineV1 {
1126            pipeline_type: NATIVE_PREDICTOR_PIPELINE_TYPE_SNV_SAVGOL_V1.to_string(),
1127            schema_version: 1,
1128            operator_count: 2,
1129            raw_n_features: 3,
1130            model_n_features: 3,
1131            fingerprint_algorithm: NATIVE_PREDICTOR_PIPELINE_FINGERPRINT_FNV1A64_V1.to_string(),
1132            native_fingerprint: "0123456789abcdef".to_string(),
1133            savgol_window: 3,
1134            savgol_poly_degree: 2,
1135        });
1136        pipeline.descriptor_fingerprint = pipeline.compute_fingerprint().unwrap();
1137        pipeline.validate().unwrap();
1138
1139        let mut missing_capability = pipeline.clone();
1140        missing_capability.capabilities &= !NATIVE_PREDICTOR_CAPABILITY_PIPELINE;
1141        missing_capability.descriptor_fingerprint =
1142            missing_capability.compute_fingerprint().unwrap();
1143        assert!(missing_capability.validate().is_err());
1144
1145        let mut wrong_format = pipeline;
1146        wrong_format.format_version = 1;
1147        wrong_format.descriptor_fingerprint = wrong_format.compute_fingerprint().unwrap();
1148        assert!(wrong_format.validate().is_err());
1149    }
1150}
1151
1152#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1153#[serde(deny_unknown_fields)]
1154pub struct LineageRecord {
1155    pub record_id: LineageId,
1156    pub run_id: RunId,
1157    pub node_id: NodeId,
1158    pub phase: Phase,
1159    pub controller_id: ControllerId,
1160    pub controller_version: String,
1161    pub variant_id: Option<VariantId>,
1162    pub fold_id: Option<FoldId>,
1163    #[serde(default)]
1164    pub branch_path: Vec<BranchId>,
1165    #[serde(default)]
1166    pub input_lineage: Vec<LineageId>,
1167    #[serde(default)]
1168    pub artifact_refs: Vec<ArtifactRef>,
1169    pub params_fingerprint: String,
1170    pub data_model_shape_fingerprint: Option<String>,
1171    pub aggregation_policy_fingerprint: Option<String>,
1172    pub seed: Option<u64>,
1173    #[serde(default)]
1174    pub unsafe_flags: BTreeSet<String>,
1175    #[serde(default)]
1176    pub metrics: BTreeMap<String, f64>,
1177    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1178    pub loss_attestations: Vec<LossExecutionAttestation>,
1179    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1180    pub early_stopping_records: Vec<EarlyStoppingRecord>,
1181}
1182
1183impl LineageRecord {
1184    pub fn validate(&self) -> Result<()> {
1185        if self.params_fingerprint.trim().is_empty() {
1186            return Err(DagMlError::RuntimeValidation(format!(
1187                "lineage `{}` has empty params fingerprint",
1188                self.record_id
1189            )));
1190        }
1191        for artifact in &self.artifact_refs {
1192            artifact.validate()?;
1193        }
1194        for attestation in &self.loss_attestations {
1195            attestation.validate()?;
1196            if attestation.node_id != self.node_id || attestation.phase != self.phase {
1197                return Err(DagMlError::RuntimeValidation(format!(
1198                    "lineage `{}` contains a loss attestation outside its node/phase scope",
1199                    self.record_id
1200                )));
1201            }
1202        }
1203        let mut early_stopping_roles = BTreeSet::new();
1204        for record in &self.early_stopping_records {
1205            record.validate_against(&self.node_id, self.phase, self.fold_id.as_ref())?;
1206            if !early_stopping_roles.insert(record.metric_role.role_id.as_str()) {
1207                return Err(DagMlError::RuntimeValidation(format!(
1208                    "lineage `{}` contains duplicate early-stopping role `{}`",
1209                    self.record_id, record.metric_role.role_id
1210                )));
1211            }
1212        }
1213        Ok(())
1214    }
1215}
1216
1217#[derive(Clone, Debug, Default)]
1218pub struct InMemoryLineageRecorder {
1219    records: BTreeMap<LineageId, LineageRecord>,
1220}
1221
1222impl InMemoryLineageRecorder {
1223    pub fn new() -> Self {
1224        Self::default()
1225    }
1226
1227    pub fn record(&mut self, record: LineageRecord) -> Result<()> {
1228        record.validate()?;
1229        if self
1230            .records
1231            .insert(record.record_id.clone(), record)
1232            .is_some()
1233        {
1234            return Err(DagMlError::RuntimeValidation(
1235                "duplicate lineage record id".to_string(),
1236            ));
1237        }
1238        Ok(())
1239    }
1240
1241    pub fn get(&self, id: &LineageId) -> Option<&LineageRecord> {
1242        self.records.get(id)
1243    }
1244
1245    pub fn len(&self) -> usize {
1246        self.records.len()
1247    }
1248
1249    pub fn is_empty(&self) -> bool {
1250        self.records.is_empty()
1251    }
1252
1253    pub fn records(&self) -> impl Iterator<Item = &LineageRecord> {
1254        self.records.values()
1255    }
1256}
dag_ml_core/runtime/artifact.rs

dag_ml_core/runtime/
artifact.rs
