1use std::collections::{BTreeMap, BTreeSet};
10
11use serde::{Deserialize, Serialize};
12
13use crate::bundle::{bundle_prediction_requirement_key, ExecutionBundle, RefitArtifactRecord};
14use crate::canonical::parse_typed_json;
15use crate::conformal_runtime::ConformalCalibration;
16use crate::controller::{
17 ArtifactPolicy, ControllerCapability, ControllerFitScope, ControllerManifest,
18 ControllerRegistry,
19};
20use crate::criteria::TrainingLossRoleReference;
21use crate::data::{data_binding_requirement_key, DataBinding, ExternalDataPlanEnvelope};
22use crate::error::{DagMlError, Result};
23use crate::fold::fold_set_fingerprint;
24use crate::graph::{GraphSpec, NodeKind, PortKind};
25use crate::ids::{
26 ArtifactId, BundleId, ControllerId, FoldId, GroupId, NodeId, SampleId, VariantId,
27};
28use crate::phase::Phase;
29use crate::plan::{build_execution_plan, CampaignSpec, ExecutionPlan};
30use crate::policy::PredictionLevel;
31use crate::relation::{EntityUnitLevel, SampleRelationSet};
32use crate::replay::{replay_request_from_outcome, TrainingReplayOutcome};
33use crate::selection::{RefitStrategy, SelectionPolicy};
34
35pub const TRAINING_REQUEST_SCHEMA_VERSION: u32 = 1;
36pub const TRAINING_REQUEST_SCHEMA_ID: &str =
37 "https://github.com/GBeurier/dag-ml/schemas/training_request.v1.schema.json";
38pub const CACHE_NAMESPACE_SCHEMA_VERSION: u32 = 1;
39pub const CACHE_NAMESPACE_SCHEMA_ID: &str =
40 "https://github.com/GBeurier/dag-ml/schemas/cache_namespace.v1.schema.json";
41pub const PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION: u32 = 2;
43pub const LEGACY_PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION: u32 = 1;
44pub const MIN_READABLE_PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION: u32 = 1;
45pub const PORTABLE_PREDICTOR_PACKAGE_V3_SCHEMA_VERSION: u32 = 3;
48pub const PORTABLE_REFIT_RECIPE_SCHEMA_VERSION: u32 = 1;
49pub const PORTABLE_REFIT_PROVENANCE_SCHEMA_VERSION: u32 = 1;
50pub const PORTABLE_PREDICTOR_PACKAGE_SCHEMA_ID: &str =
51 "https://github.com/GBeurier/dag-ml/schemas/portable_predictor_package.v1.schema.json";
52pub const OUTPUT_BINDING_SCHEMA_VERSION: u32 = 1;
53pub const TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION: u32 = 1;
54pub const PARAMETER_PATCH_SCHEMA_VERSION: u32 = 1;
55pub const PARAMETER_PROJECTION_SCHEMA_VERSION: u32 = 1;
56
57type InfluenceCoordinate = (TrainingInfluenceKind, String, Option<NodeId>);
58type ExpectedInfluenceCoordinates = BTreeMap<InfluenceCoordinate, BTreeSet<SampleId>>;
59type InfluenceCapabilitySlot = (NodeId, TrainingInfluenceKind, Phase, Option<FoldId>);
60
61fn deserialize_required_nullable<'de, D, T>(
72 deserializer: D,
73) -> std::result::Result<Option<T>, D::Error>
74where
75 D: serde::Deserializer<'de>,
76 T: Deserialize<'de>,
77{
78 Option::<T>::deserialize(deserializer)
79}
80
81#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
82#[serde(rename_all = "snake_case")]
83pub enum PredictionKind {
84 RegressionPoint,
85 ClassLabel,
86 ClassProbability,
87 DecisionScore,
88}
89
90#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
91#[serde(rename_all = "snake_case")]
92pub enum PredictionSource {
93 FinalRefit,
94 CvEnsemble,
95 FoldMember,
96}
97
98#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
99#[serde(rename_all = "snake_case")]
100pub enum OutputOrder {
101 TargetOrder,
102 TargetMajorClassMinor,
103}
104
105#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
107#[serde(deny_unknown_fields)]
108pub struct TrainingOutputRequest {
109 pub output_id: String,
110 pub node_id: NodeId,
111 #[serde(default, skip_serializing_if = "Option::is_none")]
112 pub port_name: Option<String>,
113 pub prediction_level: PredictionLevel,
114 #[serde(deserialize_with = "deserialize_required_nullable")]
115 pub unit_level: Option<EntityUnitLevel>,
116 pub prediction_kind: PredictionKind,
117 pub target_names: Vec<String>,
118 pub target_units: Vec<Option<String>>,
119 pub class_labels: Vec<Vec<String>>,
120 pub output_order: OutputOrder,
121 pub target_space: String,
122}
123
124#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
125#[serde(deny_unknown_fields)]
126pub struct ResolvedTrainingOutput {
127 pub output_id: String,
128 pub node_id: NodeId,
129 pub port_name: String,
130 pub prediction_level: PredictionLevel,
131 #[serde(default)]
132 pub unit_level: Option<EntityUnitLevel>,
133 pub prediction_kind: PredictionKind,
134 pub target_names: Vec<String>,
135 pub target_units: Vec<Option<String>>,
136 pub class_labels: Vec<Vec<String>>,
137 pub output_order: OutputOrder,
138 pub target_space: String,
139}
140
141impl TrainingOutputRequest {
142 pub fn validate(&self) -> Result<()> {
143 validate_identifier_text("training output_id", &self.output_id)?;
144 validate_output_unit_level(self.prediction_level, self.unit_level)?;
145 validate_output_shape(
146 self.prediction_kind,
147 self.output_order,
148 &self.target_names,
149 &self.target_units,
150 &self.class_labels,
151 &self.target_space,
152 )?;
153 if self
154 .port_name
155 .as_ref()
156 .is_some_and(|name| name.trim().is_empty())
157 {
158 return contract_error(format!(
159 "training output `{}` has an empty port_name",
160 self.output_id
161 ));
162 }
163 Ok(())
164 }
165
166 pub fn resolve(&self, graph: &GraphSpec) -> Result<ResolvedTrainingOutput> {
169 self.validate()?;
170 let node = graph
171 .nodes
172 .iter()
173 .find(|node| node.id == self.node_id)
174 .ok_or_else(|| {
175 DagMlError::CampaignValidation(format!(
176 "training output `{}` references unknown node `{}`",
177 self.output_id, self.node_id
178 ))
179 })?;
180 let prediction_ports = node
181 .ports
182 .outputs
183 .iter()
184 .filter(|port| port.kind == PortKind::Prediction)
185 .collect::<Vec<_>>();
186 let port_name = match self.port_name.as_deref() {
187 Some(requested) => {
188 let port = node
189 .ports
190 .outputs
191 .iter()
192 .find(|port| port.name == requested)
193 .ok_or_else(|| {
194 DagMlError::CampaignValidation(format!(
195 "training output `{}` references absent port `{}.{requested}`",
196 self.output_id, self.node_id
197 ))
198 })?;
199 if port.kind != PortKind::Prediction {
200 return contract_error(format!(
201 "training output `{}` port `{}.{requested}` is not a prediction port",
202 self.output_id, self.node_id
203 ));
204 }
205 requested.to_string()
206 }
207 None => match prediction_ports.as_slice() {
208 [] => {
209 return contract_error(format!(
210 "training output `{}` node `{}` exposes no prediction output",
211 self.output_id, self.node_id
212 ));
213 }
214 [only] => only.name.clone(),
215 _ => {
216 return contract_error(format!(
217 "training output `{}` node `{}` exposes multiple prediction outputs; port_name is required",
218 self.output_id, self.node_id
219 ));
220 }
221 },
222 };
223 Ok(ResolvedTrainingOutput {
224 output_id: self.output_id.clone(),
225 node_id: self.node_id.clone(),
226 port_name,
227 prediction_level: self.prediction_level,
228 unit_level: self.unit_level,
229 prediction_kind: self.prediction_kind,
230 target_names: self.target_names.clone(),
231 target_units: self.target_units.clone(),
232 class_labels: self.class_labels.clone(),
233 output_order: self.output_order,
234 target_space: self.target_space.clone(),
235 })
236 }
237}
238
239#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
240#[serde(rename_all = "snake_case")]
241pub enum TrainingSchedulerKind {
242 Sequential,
243 Parallel,
244}
245
246#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
247#[serde(rename_all = "snake_case")]
248pub enum TrainingSchedulerBackend {
249 Threads,
250 Processes,
251}
252
253#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
254#[serde(deny_unknown_fields)]
255pub struct TrainingSchedulerOptions {
256 pub kind: TrainingSchedulerKind,
257 #[serde(default)]
258 pub backend: Option<TrainingSchedulerBackend>,
259 pub workers: u32,
260}
261
262impl TrainingSchedulerOptions {
263 fn validate(&self) -> Result<()> {
264 match (self.kind, self.backend, self.workers) {
265 (TrainingSchedulerKind::Sequential, None, 1) => Ok(()),
266 (TrainingSchedulerKind::Sequential, Some(_), _) => contract_error(
267 "sequential training scheduler forbids a parallel backend".to_string(),
268 ),
269 (TrainingSchedulerKind::Sequential, None, _) => {
270 contract_error("sequential training scheduler requires workers=1".to_string())
271 }
272 (TrainingSchedulerKind::Parallel, None, _) => contract_error(
273 "parallel training scheduler requires an explicit backend".to_string(),
274 ),
275 (TrainingSchedulerKind::Parallel, Some(_), 0 | 1) => {
276 contract_error("parallel training scheduler requires workers>=2".to_string())
277 }
278 (TrainingSchedulerKind::Parallel, Some(_), _) => Ok(()),
279 }
280 }
281}
282
283#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
284#[serde(deny_unknown_fields)]
285pub struct TrainingResourceLimits {
286 pub cpu_threads: u32,
287 #[serde(default, skip_serializing_if = "Option::is_none")]
288 pub memory_bytes: Option<u64>,
289 pub gpu_devices: Vec<String>,
292 #[serde(default, skip_serializing_if = "Option::is_none")]
293 pub wall_time_ms: Option<u64>,
294}
295
296impl TrainingResourceLimits {
297 fn validate(&self, scheduler: &TrainingSchedulerOptions) -> Result<()> {
298 if self.cpu_threads == 0 {
299 return contract_error("training resources require cpu_threads>=1".to_string());
300 }
301 if scheduler.workers > self.cpu_threads {
302 return contract_error(format!(
303 "training scheduler workers={} exceeds cpu_threads={}",
304 scheduler.workers, self.cpu_threads
305 ));
306 }
307 if self.memory_bytes == Some(0) {
308 return contract_error("training memory_bytes must be positive".to_string());
309 }
310 if self.wall_time_ms == Some(0) {
311 return contract_error("training wall_time_ms must be positive".to_string());
312 }
313 validate_sorted_unique_text("training gpu_devices", &self.gpu_devices, false)
314 }
315}
316
317#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
318#[serde(rename_all = "snake_case")]
319pub enum CvArtifactRetention {
320 Discard,
321 MetadataOnly,
322 Retain,
323}
324
325#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
326#[serde(rename_all = "snake_case")]
327pub enum PredictionCacheRetention {
328 Discard,
329 Retain,
330}
331
332#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
333#[serde(rename_all = "snake_case")]
334pub enum FittedArtifactMode {
335 PortableRequired,
336 AllowHostSidecar,
337}
338
339#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
340#[serde(deny_unknown_fields)]
341pub struct TrainingArtifactOptions {
342 pub cv_artifacts: CvArtifactRetention,
343 pub prediction_caches: PredictionCacheRetention,
344 pub fitted_artifacts: FittedArtifactMode,
345}
346
347#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
348#[serde(deny_unknown_fields)]
349pub struct TrainingOptions {
350 pub refit: bool,
351 #[serde(deserialize_with = "deserialize_required_nullable")]
352 pub refit_strategy: Option<RefitStrategy>,
353 pub seed: u64,
354 pub selection: SelectionPolicy,
355 pub selection_output_id: String,
356 pub outputs: Vec<TrainingOutputRequest>,
357 pub scheduler: TrainingSchedulerOptions,
358 pub resources: TrainingResourceLimits,
359 pub artifacts: TrainingArtifactOptions,
360}
361
362impl TrainingOptions {
363 fn validate(&self, graph: &GraphSpec) -> Result<Vec<ResolvedTrainingOutput>> {
364 match (self.refit, self.refit_strategy) {
365 (true, None) => {
366 return contract_error(
367 "training refit=true requires an explicit refit_strategy".to_string(),
368 );
369 }
370 (false, Some(_)) => {
371 return contract_error("training refit=false forbids refit_strategy".to_string());
372 }
373 _ => {}
374 }
375 self.selection.validate()?;
376 validate_identifier_text("training selection_output_id", &self.selection_output_id)?;
377 self.scheduler.validate()?;
378 self.resources.validate(&self.scheduler)?;
379 if !self.refit && self.artifacts.prediction_caches != PredictionCacheRetention::Retain {
380 return contract_error(
381 "training refit=false requires retained prediction caches for REFIT replay"
382 .to_string(),
383 );
384 }
385 if self.outputs.is_empty() {
386 return contract_error("training options require at least one output".to_string());
387 }
388 let mut previous_id: Option<&str> = None;
389 let mut coordinates = BTreeSet::new();
390 let mut resolved = Vec::with_capacity(self.outputs.len());
391 for output in &self.outputs {
392 if previous_id.is_some_and(|previous| previous >= output.output_id.as_str()) {
393 return contract_error(
394 "training outputs must be strictly sorted by output_id".to_string(),
395 );
396 }
397 previous_id = Some(output.output_id.as_str());
398 let output = output.resolve(graph)?;
399 if !coordinates.insert((output.node_id.clone(), output.port_name.clone())) {
400 return contract_error(format!(
401 "training outputs bind `{}.{}` more than once",
402 output.node_id, output.port_name
403 ));
404 }
405 resolved.push(output);
406 }
407 if !resolved
408 .iter()
409 .any(|output| output.output_id == self.selection_output_id)
410 {
411 return contract_error(format!(
412 "training selection_output_id `{}` does not identify a declared output",
413 self.selection_output_id
414 ));
415 }
416 Ok(resolved)
417 }
418}
419
420#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
422#[serde(deny_unknown_fields)]
423pub struct TrainingDataIdentity {
424 pub requirement_key: String,
425 pub schema_fingerprint: String,
426 pub plan_fingerprint: String,
427 pub relation_fingerprint: String,
428 pub data_content_fingerprint: String,
429 pub target_content_fingerprint: String,
430 pub identity_fingerprint: String,
431}
432
433impl TrainingDataIdentity {
434 pub fn from_binding_envelope(
440 binding: &DataBinding,
441 envelope: &ExternalDataPlanEnvelope,
442 ) -> Result<Self> {
443 binding.validate_envelope(envelope)?;
444 let requirement_key = data_binding_requirement_key(&binding.node_id, &binding.input_name);
445 let relation_fingerprint = envelope.relation_fingerprint.clone().ok_or_else(|| {
446 DagMlError::CampaignValidation(format!(
447 "external data envelope for `{requirement_key}` cannot attest training without a relation fingerprint"
448 ))
449 })?;
450 let data_content_fingerprint =
451 envelope.data_content_fingerprint.clone().ok_or_else(|| {
452 DagMlError::CampaignValidation(format!(
453 "external data envelope for `{requirement_key}` cannot attest training without a data content fingerprint"
454 ))
455 })?;
456 let target_content_fingerprint =
457 envelope.target_content_fingerprint.clone().ok_or_else(|| {
458 DagMlError::CampaignValidation(format!(
459 "external data envelope for `{requirement_key}` cannot attest training without a target content fingerprint"
460 ))
461 })?;
462 let mut identity = Self {
463 requirement_key,
464 schema_fingerprint: envelope.schema_fingerprint.clone(),
465 plan_fingerprint: envelope.plan_fingerprint.clone(),
466 relation_fingerprint,
467 data_content_fingerprint,
468 target_content_fingerprint,
469 identity_fingerprint: zero_fingerprint(),
470 };
471 identity.identity_fingerprint = identity.compute_fingerprint()?;
472 identity.validate()?;
473 Ok(identity)
474 }
475
476 pub fn compute_fingerprint(&self) -> Result<String> {
477 tcv1_fingerprint_without(self, "identity_fingerprint", "training data identity")
478 }
479
480 pub fn validate(&self) -> Result<()> {
481 validate_non_empty("training data requirement_key", &self.requirement_key)?;
482 for (label, value) in [
483 ("training data schema", &self.schema_fingerprint),
484 ("training data plan", &self.plan_fingerprint),
485 ("training data relation", &self.relation_fingerprint),
486 ("training data content", &self.data_content_fingerprint),
487 ("training target content", &self.target_content_fingerprint),
488 ("training data identity", &self.identity_fingerprint),
489 ] {
490 validate_sha256(label, value)?;
491 }
492 if self.identity_fingerprint != self.compute_fingerprint()? {
493 return contract_error(format!(
494 "training data identity `{}` fingerprint does not match TCV1 content",
495 self.requirement_key
496 ));
497 }
498 Ok(())
499 }
500}
501
502#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
503#[serde(rename_all = "snake_case")]
504pub enum ParameterNamespace {
505 Operator,
506 Fit,
507 Control,
508 Structural,
509}
510
511impl ParameterNamespace {
512 pub const fn plan_root(self) -> &'static str {
515 match self {
516 Self::Operator => "params",
517 Self::Fit => "fit_params",
518 Self::Control => "control_params",
519 Self::Structural => "structural_params",
520 }
521 }
522}
523
524#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
525#[serde(deny_unknown_fields)]
526pub struct ParameterPatch {
527 pub schema_version: u32,
528 pub node_id: NodeId,
529 pub namespace: ParameterNamespace,
530 pub path: Vec<String>,
531 pub value: serde_json::Value,
532}
533
534impl ParameterPatch {
535 pub fn validate(&self) -> Result<()> {
536 if self.schema_version != PARAMETER_PATCH_SCHEMA_VERSION {
537 return unsupported_version(
538 "parameter patch",
539 self.schema_version,
540 PARAMETER_PATCH_SCHEMA_VERSION,
541 );
542 }
543 if self.path.is_empty() {
544 return contract_error(format!(
545 "parameter patch for `{}` has an empty path",
546 self.node_id
547 ));
548 }
549 for segment in &self.path {
550 if segment.trim().is_empty() || segment == "-" {
551 return contract_error(format!(
552 "parameter patch for `{}` has an invalid path segment `{segment}`",
553 self.node_id
554 ));
555 }
556 }
557 Ok(())
558 }
559}
560
561#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
562#[serde(deny_unknown_fields)]
563pub struct NodePatchPolicy {
564 pub node_id: NodeId,
565 pub allowed_namespaces: BTreeSet<ParameterNamespace>,
566}
567
568impl NodePatchPolicy {
569 fn validate(&self) -> Result<()> {
570 if self.allowed_namespaces.is_empty() {
571 return contract_error(format!(
572 "node patch policy `{}` allows no namespaces",
573 self.node_id
574 ));
575 }
576 Ok(())
577 }
578}
579
580#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
581#[serde(deny_unknown_fields)]
582pub struct NamespacedNodeParameters {
583 #[serde(default)]
584 pub params: BTreeMap<String, serde_json::Value>,
585 #[serde(default)]
586 pub fit_params: BTreeMap<String, serde_json::Value>,
587 #[serde(default)]
588 pub control_params: BTreeMap<String, serde_json::Value>,
589 #[serde(default)]
590 pub structural_params: BTreeMap<String, serde_json::Value>,
591}
592
593impl NamespacedNodeParameters {
594 fn namespace_mut(
595 &mut self,
596 namespace: ParameterNamespace,
597 ) -> &mut BTreeMap<String, serde_json::Value> {
598 match namespace {
599 ParameterNamespace::Operator => &mut self.params,
600 ParameterNamespace::Fit => &mut self.fit_params,
601 ParameterNamespace::Control => &mut self.control_params,
602 ParameterNamespace::Structural => &mut self.structural_params,
603 }
604 }
605}
606
607#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
608#[serde(deny_unknown_fields)]
609pub struct ParameterProjection {
610 pub schema_version: u32,
611 pub nodes: BTreeMap<NodeId, NamespacedNodeParameters>,
612 pub requires_recompile: bool,
613 pub structural_patch_count: u32,
614 pub patches_fingerprint: String,
615 pub projection_fingerprint: String,
616}
617
618impl ParameterProjection {
619 pub fn from_json(json: &str) -> Result<Self> {
620 let raw_fingerprint = strict_tcv1_fingerprint_without(
621 json,
622 "projection_fingerprint",
623 "parameter projection",
624 )?;
625 let projection: Self = serde_json::from_str(json)?;
626 if projection.projection_fingerprint != raw_fingerprint {
627 return contract_error(
628 "parameter projection fingerprint does not match original TCV1 JSON".to_string(),
629 );
630 }
631 projection.validate()?;
632 Ok(projection)
633 }
634
635 pub fn compute_fingerprint(&self) -> Result<String> {
636 tcv1_fingerprint_without(self, "projection_fingerprint", "parameter projection")
637 }
638
639 pub fn validate(&self) -> Result<()> {
640 if self.schema_version != PARAMETER_PROJECTION_SCHEMA_VERSION {
641 return unsupported_version(
642 "parameter projection",
643 self.schema_version,
644 PARAMETER_PROJECTION_SCHEMA_VERSION,
645 );
646 }
647 validate_sha256("parameter patches", &self.patches_fingerprint)?;
648 validate_sha256("parameter projection", &self.projection_fingerprint)?;
649 if self.requires_recompile != (self.structural_patch_count > 0) {
650 return contract_error(
651 "parameter projection requires_recompile must equal structural_patch_count>0"
652 .to_string(),
653 );
654 }
655 if self.projection_fingerprint != self.compute_fingerprint()? {
656 return contract_error(
657 "parameter projection fingerprint does not match TCV1 content".to_string(),
658 );
659 }
660 Ok(())
661 }
662}
663
664pub fn project_parameter_patches(
668 plan: &ExecutionPlan,
669 patches: &[ParameterPatch],
670 policies: &[NodePatchPolicy],
671) -> Result<ParameterProjection> {
672 plan.validate()?;
673 validate_canonical_patches(patches)?;
674 let policy_map = validate_patch_policies(plan, policies)?;
675 let patched_nodes = patches
676 .iter()
677 .map(|patch| patch.node_id.clone())
678 .collect::<BTreeSet<_>>();
679 if policy_map.keys().cloned().collect::<BTreeSet<_>>() != patched_nodes {
680 return contract_error(
681 "node patch policies must exactly cover nodes targeted by patches".to_string(),
682 );
683 }
684 let mut nodes = plan
685 .node_plans
686 .iter()
687 .map(|(node_id, node_plan)| {
688 (
689 node_id.clone(),
690 NamespacedNodeParameters {
691 params: node_plan.params.clone(),
692 ..NamespacedNodeParameters::default()
693 },
694 )
695 })
696 .collect::<BTreeMap<_, _>>();
697 let mut structural_patch_count = 0_u32;
698 for patch in patches {
699 let policy = policy_map.get(&patch.node_id).ok_or_else(|| {
700 DagMlError::CampaignValidation(format!(
701 "parameter patch for `{}` has no node patch policy",
702 patch.node_id
703 ))
704 })?;
705 if !policy.contains(&patch.namespace) {
706 return contract_error(format!(
707 "parameter namespace `{:?}` is forbidden for node `{}`",
708 patch.namespace, patch.node_id
709 ));
710 }
711 let node = nodes.get_mut(&patch.node_id).ok_or_else(|| {
712 DagMlError::CampaignValidation(format!(
713 "parameter patch references unknown node `{}`",
714 patch.node_id
715 ))
716 })?;
717 deep_set_object_key(
718 node.namespace_mut(patch.namespace),
719 &patch.path,
720 patch.value.clone(),
721 &patch.node_id,
722 )?;
723 if patch.namespace == ParameterNamespace::Structural {
724 structural_patch_count = structural_patch_count.checked_add(1).ok_or_else(|| {
725 DagMlError::CampaignValidation(
726 "parameter projection has too many structural patches".to_string(),
727 )
728 })?;
729 }
730 }
731 let patches_fingerprint = tcv1_fingerprint(patches, "parameter patches")?;
732 let mut projection = ParameterProjection {
733 schema_version: PARAMETER_PROJECTION_SCHEMA_VERSION,
734 nodes,
735 requires_recompile: structural_patch_count > 0,
736 structural_patch_count,
737 patches_fingerprint,
738 projection_fingerprint: zero_fingerprint(),
739 };
740 projection.projection_fingerprint = projection.compute_fingerprint()?;
741 projection.validate()?;
742 Ok(projection)
743}
744
745#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
746#[serde(rename_all = "snake_case")]
747pub enum TrainingInfluenceKind {
748 TransformFit,
749 ModelFit,
750 HpoSelection,
751 EarlyStopping,
752 WeightingResampling,
753 TrainedMetaAggregation,
754}
755
756#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
757#[serde(deny_unknown_fields)]
758pub struct ControllerInfluenceRequirement {
759 pub node_id: NodeId,
760 pub kind: TrainingInfluenceKind,
761 pub scope_id: String,
762 pub phase: Phase,
763 #[serde(deserialize_with = "deserialize_required_nullable")]
764 pub fold_id: Option<FoldId>,
765 pub physical_sample_ids: Vec<SampleId>,
766}
767
768#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
769#[serde(deny_unknown_fields)]
770pub struct TrainingInfluenceEntry {
771 pub kind: TrainingInfluenceKind,
772 pub scope_id: String,
773 #[serde(deserialize_with = "deserialize_required_nullable")]
774 pub node_id: Option<NodeId>,
775 pub physical_sample_ids: Vec<SampleId>,
776 pub origin_sample_ids: Vec<SampleId>,
777 pub group_ids: Vec<GroupId>,
778}
779
780impl TrainingInfluenceEntry {
781 fn validate(&self) -> Result<()> {
782 validate_identifier_text("training influence scope_id", &self.scope_id)?;
783 validate_sorted_unique_ids(
784 "training influence physical_sample_ids",
785 &self.physical_sample_ids,
786 true,
787 )?;
788 validate_sorted_unique_ids(
789 "training influence origin_sample_ids",
790 &self.origin_sample_ids,
791 false,
792 )?;
793 validate_sorted_unique_ids("training influence group_ids", &self.group_ids, false)
794 }
795}
796
797#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
798#[serde(deny_unknown_fields)]
799pub struct TrainingInfluenceManifest {
800 pub schema_version: u32,
801 pub relation_fingerprint: String,
802 pub entries: Vec<TrainingInfluenceEntry>,
803 pub manifest_fingerprint: String,
804}
805
806impl TrainingInfluenceManifest {
807 pub fn compute_fingerprint(&self) -> Result<String> {
808 tcv1_fingerprint_without(self, "manifest_fingerprint", "training influence manifest")
809 }
810
811 pub fn derive_for_projection(
812 projection: &TrainingContractProjection,
813 request: &TrainingRequest,
814 relations: &SampleRelationSet,
815 ) -> Result<Self> {
816 projection.validate()?;
817 relations.validate()?;
818 let relation_fingerprint = relations.fingerprint()?;
819 if request
820 .data_identities
821 .iter()
822 .any(|identity| identity.relation_fingerprint != relation_fingerprint)
823 {
824 return contract_error(
825 "training data identities do not all bind the influence relation".to_string(),
826 );
827 }
828 let expected = expected_influence_coordinates(
829 request,
830 &projection.plan,
831 &projection.predictor_node_ids,
832 )?;
833 let mut entries = Vec::with_capacity(expected.len());
834 for ((kind, scope_id, node_id), samples) in expected {
835 let (origin_sample_ids, group_ids) =
836 influence_identity_closure_for_samples(&scope_id, &samples, relations)?;
837 entries.push(TrainingInfluenceEntry {
838 kind,
839 scope_id,
840 node_id,
841 physical_sample_ids: samples.into_iter().collect(),
842 origin_sample_ids,
843 group_ids,
844 });
845 }
846 let mut manifest = Self {
847 schema_version: TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
848 relation_fingerprint,
849 entries,
850 manifest_fingerprint: zero_fingerprint(),
851 };
852 manifest.manifest_fingerprint = manifest.compute_fingerprint()?;
853 manifest.validate_for_projection(projection, request, relations)?;
854 Ok(manifest)
855 }
856
857 pub fn validate(&self) -> Result<()> {
858 if self.schema_version != TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION {
859 return unsupported_version(
860 "training influence manifest",
861 self.schema_version,
862 TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
863 );
864 }
865 validate_sha256("training influence relation", &self.relation_fingerprint)?;
866 validate_sha256("training influence manifest", &self.manifest_fingerprint)?;
867 if self.entries.is_empty() {
868 return contract_error(
869 "training influence manifest requires at least one entry".to_string(),
870 );
871 }
872 let mut previous: Option<(TrainingInfluenceKind, &str, Option<&NodeId>)> = None;
873 for entry in &self.entries {
874 entry.validate()?;
875 let key = (entry.kind, entry.scope_id.as_str(), entry.node_id.as_ref());
876 if previous.as_ref().is_some_and(|previous| previous >= &key) {
877 return contract_error(
878 "training influence entries must be strictly canonically sorted".to_string(),
879 );
880 }
881 previous = Some(key);
882 }
883 if self.manifest_fingerprint != self.compute_fingerprint()? {
884 return contract_error(
885 "training influence manifest fingerprint does not match TCV1 content".to_string(),
886 );
887 }
888 Ok(())
889 }
890
891 pub fn validate_for_projection(
892 &self,
893 projection: &TrainingContractProjection,
894 request: &TrainingRequest,
895 relations: &SampleRelationSet,
896 ) -> Result<()> {
897 self.validate()?;
898 relations.validate()?;
899 let relation_fingerprint = relations.fingerprint()?;
900 if self.relation_fingerprint != relation_fingerprint {
901 return contract_error(
902 "training influence relation fingerprint does not match relation set".to_string(),
903 );
904 }
905 if request
906 .data_identities
907 .iter()
908 .any(|identity| identity.relation_fingerprint != relation_fingerprint)
909 {
910 return contract_error(
911 "training data identities do not all bind the influence relation".to_string(),
912 );
913 }
914
915 let expected = expected_influence_coordinates(
916 request,
917 &projection.plan,
918 &projection.predictor_node_ids,
919 )?;
920 let mut actual = BTreeSet::new();
921 for entry in &self.entries {
922 if let Some(node_id) = &entry.node_id {
923 if !projection.predictor_node_ids.contains(node_id) {
924 return contract_error(format!(
925 "training influence node `{node_id}` is outside predictor closure"
926 ));
927 }
928 }
929 let coordinate = (entry.kind, entry.scope_id.clone(), entry.node_id.clone());
930 let expected_samples = expected.get(&coordinate).ok_or_else(|| {
931 DagMlError::CampaignValidation(format!(
932 "training influence contains undeclared coordinate `{:?}/{}/{:?}`",
933 entry.kind, entry.scope_id, entry.node_id
934 ))
935 })?;
936 if entry
937 .physical_sample_ids
938 .iter()
939 .cloned()
940 .collect::<BTreeSet<_>>()
941 != *expected_samples
942 {
943 return contract_error(format!(
944 "training influence coordinate `{:?}/{}/{:?}` does not contain the exact scope samples",
945 entry.kind, entry.scope_id, entry.node_id
946 ));
947 }
948 validate_influence_identity_closure(entry, relations)?;
949 actual.insert(coordinate);
950 }
951 let expected_keys = expected.into_keys().collect::<BTreeSet<_>>();
952 if actual != expected_keys {
953 return contract_error(
954 "training influence entries do not exactly cover capability-derived phase scopes"
955 .to_string(),
956 );
957 }
958 Ok(())
959 }
960}
961
962#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
963#[serde(deny_unknown_fields)]
964pub struct TrainingRequest {
965 pub schema_version: u32,
966 pub request_id: String,
967 pub plan_id: String,
968 pub graph: GraphSpec,
969 pub campaign: CampaignSpec,
970 pub controller_manifests: Vec<ControllerManifest>,
971 pub data_identities: Vec<TrainingDataIdentity>,
972 pub parameter_patches: Vec<ParameterPatch>,
973 pub patch_policies: Vec<NodePatchPolicy>,
974 pub influence_requirements: Vec<ControllerInfluenceRequirement>,
975 #[serde(default, skip_serializing_if = "Vec::is_empty")]
980 pub training_losses: Vec<TrainingLossRoleReference>,
981 pub options: TrainingOptions,
982 pub request_fingerprint: String,
983}
984
985#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
986#[serde(deny_unknown_fields)]
987pub struct TrainingContractProjection {
988 pub request_id: String,
989 pub request_fingerprint: String,
990 pub plan: ExecutionPlan,
991 pub outputs: Vec<ResolvedTrainingOutput>,
992 pub predictor_node_ids: BTreeSet<NodeId>,
993 pub parameters: ParameterProjection,
994}
995
996impl TrainingContractProjection {
997 pub fn from_json(json: &str) -> Result<Self> {
998 parse_typed_json(json).map_err(|error| {
999 DagMlError::RuntimeValidation(format!(
1000 "training contract projection is outside strict TCV1 JSON: {error}"
1001 ))
1002 })?;
1003 let mut deserializer = serde_json::Deserializer::from_str(json);
1004 let mut ignored_paths = Vec::new();
1005 let projection: Self = serde_ignored::deserialize(&mut deserializer, |path| {
1006 ignored_paths.push(path.to_string());
1007 })?;
1008 if !ignored_paths.is_empty() {
1009 ignored_paths.sort();
1010 ignored_paths.dedup();
1011 return contract_error(format!(
1012 "training contract projection contains unknown field(s) at: {}",
1013 ignored_paths.join(", ")
1014 ));
1015 }
1016 projection.validate()?;
1017 Ok(projection)
1018 }
1019
1020 pub fn validate(&self) -> Result<()> {
1021 validate_identifier_text("training projection request_id", &self.request_id)?;
1022 validate_sha256(
1023 "training projection request_fingerprint",
1024 &self.request_fingerprint,
1025 )?;
1026 self.plan.validate()?;
1027 self.parameters.validate()?;
1028 if self.parameters.nodes.keys().collect::<BTreeSet<_>>()
1029 != self.plan.node_plans.keys().collect::<BTreeSet<_>>()
1030 {
1031 return contract_error(
1032 "training projection parameter nodes do not exactly match execution plan"
1033 .to_string(),
1034 );
1035 }
1036 if self.outputs.is_empty() {
1037 return contract_error("training projection requires at least one output".to_string());
1038 }
1039 let mut previous_id: Option<&str> = None;
1040 let mut coordinates = BTreeSet::new();
1041 for output in &self.outputs {
1042 if previous_id.is_some_and(|previous| previous >= output.output_id.as_str()) {
1043 return contract_error(
1044 "training projection outputs must be strictly sorted by output_id".to_string(),
1045 );
1046 }
1047 previous_id = Some(output.output_id.as_str());
1048 let requested = TrainingOutputRequest {
1049 output_id: output.output_id.clone(),
1050 node_id: output.node_id.clone(),
1051 port_name: Some(output.port_name.clone()),
1052 prediction_level: output.prediction_level,
1053 unit_level: output.unit_level,
1054 prediction_kind: output.prediction_kind,
1055 target_names: output.target_names.clone(),
1056 target_units: output.target_units.clone(),
1057 class_labels: output.class_labels.clone(),
1058 output_order: output.output_order,
1059 target_space: output.target_space.clone(),
1060 };
1061 if requested.resolve(&self.plan.graph_plan.graph)? != *output {
1062 return contract_error(
1063 "training projection contains a non-canonical resolved output".to_string(),
1064 );
1065 }
1066 if !coordinates.insert((output.node_id.clone(), output.port_name.clone())) {
1067 return contract_error(
1068 "training projection contains duplicate output coordinates".to_string(),
1069 );
1070 }
1071 }
1072 let expected_closure = predictor_closure(
1073 &self.plan,
1074 self.outputs.iter().map(|output| &output.node_id),
1075 )?;
1076 if self.predictor_node_ids != expected_closure {
1077 return contract_error(
1078 "training projection predictor_node_ids do not match output closure".to_string(),
1079 );
1080 }
1081 Ok(())
1082 }
1083}
1084
1085impl TrainingRequest {
1086 pub fn from_json(json: &str) -> Result<Self> {
1087 let raw_fingerprint =
1088 strict_tcv1_fingerprint_without(json, "request_fingerprint", "training request")?;
1089 let request: Self = serde_json::from_str(json)?;
1090 if request.request_fingerprint != raw_fingerprint {
1091 return contract_error(
1092 "training request fingerprint does not match original TCV1 JSON".to_string(),
1093 );
1094 }
1095 request.validate()?;
1096 Ok(request)
1097 }
1098
1099 pub fn compute_fingerprint(&self) -> Result<String> {
1100 tcv1_fingerprint_without(self, "request_fingerprint", "training request")
1101 }
1102
1103 pub fn validate(&self) -> Result<()> {
1104 self.project().map(|_| ())
1105 }
1106
1107 pub fn project(&self) -> Result<TrainingContractProjection> {
1108 if self.schema_version != TRAINING_REQUEST_SCHEMA_VERSION {
1109 return unsupported_version(
1110 "training request",
1111 self.schema_version,
1112 TRAINING_REQUEST_SCHEMA_VERSION,
1113 );
1114 }
1115 validate_identifier_text("training request_id", &self.request_id)?;
1116 validate_non_empty("training plan_id", &self.plan_id)?;
1117 self.graph.validate()?;
1118 self.campaign.validate()?;
1119 if self.campaign.root_seed != Some(self.options.seed) {
1120 return contract_error(
1121 "training options seed must exactly match campaign.root_seed".to_string(),
1122 );
1123 }
1124 validate_sha256("training request", &self.request_fingerprint)?;
1125 if self.request_fingerprint != self.compute_fingerprint()? {
1126 return contract_error(
1127 "training request fingerprint does not match TCV1 content".to_string(),
1128 );
1129 }
1130 let outputs = self.options.validate(&self.graph)?;
1131 let mut registry = ControllerRegistry::new();
1132 let mut previous_controller: Option<&str> = None;
1133 for manifest in &self.controller_manifests {
1134 if previous_controller
1135 .is_some_and(|previous| previous >= manifest.controller_id.as_str())
1136 {
1137 return contract_error(
1138 "training controller_manifests must be strictly sorted by controller_id"
1139 .to_string(),
1140 );
1141 }
1142 previous_controller = Some(manifest.controller_id.as_str());
1143 registry.register(manifest.clone())?;
1144 }
1145 let plan = build_execution_plan(
1146 self.plan_id.clone(),
1147 self.graph.clone(),
1148 self.campaign.clone(),
1149 ®istry,
1150 )?
1151 .with_training_losses(self.training_losses.clone())?;
1152 validate_output_controllers(&plan, &outputs)?;
1153 validate_selection_output(&plan, &self.options, &outputs)?;
1154 validate_training_data_identities(self, &plan)?;
1155 let parameters =
1156 project_parameter_patches(&plan, &self.parameter_patches, &self.patch_policies)?;
1157 let predictor_node_ids =
1158 predictor_closure(&plan, outputs.iter().map(|output| &output.node_id))?;
1159 validate_scheduler_capabilities(&self.options.scheduler, &plan, &predictor_node_ids)?;
1160 validate_artifact_mode(&self.options.artifacts, &plan, &predictor_node_ids)?;
1161 validate_influence_requirements(self, &plan, &predictor_node_ids)?;
1162 let projection = TrainingContractProjection {
1163 request_id: self.request_id.clone(),
1164 request_fingerprint: self.request_fingerprint.clone(),
1165 plan,
1166 outputs,
1167 predictor_node_ids,
1168 parameters,
1169 };
1170 projection.validate()?;
1171 Ok(projection)
1172 }
1173}
1174
1175#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1178#[serde(deny_unknown_fields)]
1179pub struct CacheNamespace {
1180 pub schema_version: u32,
1181 pub prediction_requirement_key: String,
1182 pub data_requirement_key: String,
1183 pub producer_node_id: NodeId,
1184 pub source_port_name: String,
1185 pub consumer_node_id: NodeId,
1186 pub target_port_name: String,
1187 pub phase: Phase,
1188 pub params_fingerprint: String,
1189 pub data_identity_fingerprint: String,
1190 pub fold_id: FoldId,
1191 pub trial_id: String,
1192 pub seed: u64,
1193 pub namespace_fingerprint: String,
1194}
1195
1196impl CacheNamespace {
1197 #[allow(clippy::too_many_arguments)]
1198 pub fn new(
1199 prediction_requirement_key: String,
1200 data_requirement_key: String,
1201 producer_node_id: NodeId,
1202 source_port_name: String,
1203 consumer_node_id: NodeId,
1204 target_port_name: String,
1205 params_fingerprint: String,
1206 data_identity_fingerprint: String,
1207 fold_id: FoldId,
1208 trial_id: String,
1209 seed: u64,
1210 ) -> Result<Self> {
1211 let mut namespace = Self {
1212 schema_version: CACHE_NAMESPACE_SCHEMA_VERSION,
1213 prediction_requirement_key,
1214 data_requirement_key,
1215 producer_node_id,
1216 source_port_name,
1217 consumer_node_id,
1218 target_port_name,
1219 phase: Phase::FitCv,
1220 params_fingerprint,
1221 data_identity_fingerprint,
1222 fold_id,
1223 trial_id,
1224 seed,
1225 namespace_fingerprint: zero_fingerprint(),
1226 };
1227 namespace.namespace_fingerprint = namespace.compute_fingerprint()?;
1228 namespace.validate()?;
1229 Ok(namespace)
1230 }
1231
1232 pub fn from_json(json: &str) -> Result<Self> {
1233 let raw_fingerprint =
1234 strict_tcv1_fingerprint_without(json, "namespace_fingerprint", "cache namespace")?;
1235 let namespace: Self = serde_json::from_str(json)?;
1236 if namespace.namespace_fingerprint != raw_fingerprint {
1237 return contract_error(
1238 "cache namespace fingerprint does not match original TCV1 JSON".to_string(),
1239 );
1240 }
1241 namespace.validate()?;
1242 Ok(namespace)
1243 }
1244
1245 pub fn compute_fingerprint(&self) -> Result<String> {
1246 tcv1_fingerprint_without(self, "namespace_fingerprint", "cache namespace")
1247 }
1248
1249 pub fn validate(&self) -> Result<()> {
1250 if self.schema_version != CACHE_NAMESPACE_SCHEMA_VERSION {
1251 return unsupported_version(
1252 "cache namespace",
1253 self.schema_version,
1254 CACHE_NAMESPACE_SCHEMA_VERSION,
1255 );
1256 }
1257 validate_non_empty(
1258 "cache namespace prediction_requirement_key",
1259 &self.prediction_requirement_key,
1260 )?;
1261 validate_non_empty(
1262 "cache namespace data_requirement_key",
1263 &self.data_requirement_key,
1264 )?;
1265 validate_non_empty("cache namespace source_port_name", &self.source_port_name)?;
1266 validate_non_empty("cache namespace target_port_name", &self.target_port_name)?;
1267 if self.phase != Phase::FitCv {
1268 return contract_error(
1269 "cache namespace V1 is fold-scoped and permits only FIT_CV".to_string(),
1270 );
1271 }
1272 let expected_requirement_key = bundle_prediction_requirement_key(
1273 &self.producer_node_id,
1274 &self.source_port_name,
1275 &self.consumer_node_id,
1276 &self.target_port_name,
1277 );
1278 if self.prediction_requirement_key != expected_requirement_key {
1279 return contract_error(
1280 "cache namespace requirement_key does not match producer/source/consumer/target coordinates"
1281 .to_string(),
1282 );
1283 }
1284 validate_identifier_text("cache namespace trial_id", &self.trial_id)?;
1285 for (label, fingerprint) in [
1286 ("cache params", &self.params_fingerprint),
1287 ("cache data identity", &self.data_identity_fingerprint),
1288 ("cache namespace", &self.namespace_fingerprint),
1289 ] {
1290 validate_sha256(label, fingerprint)?;
1291 }
1292 if self.namespace_fingerprint != self.compute_fingerprint()? {
1293 return contract_error(
1294 "cache namespace fingerprint does not match TCV1 content".to_string(),
1295 );
1296 }
1297 Ok(())
1298 }
1299
1300 pub fn validate_for_identity(&self, identity: &TrainingDataIdentity) -> Result<()> {
1301 self.validate()?;
1302 identity.validate()?;
1303 if self.data_requirement_key != identity.requirement_key
1304 || self.data_identity_fingerprint != identity.identity_fingerprint
1305 {
1306 return contract_error(
1307 "cache namespace does not bind the complete training data identity".to_string(),
1308 );
1309 }
1310 Ok(())
1311 }
1312}
1313
1314#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1316#[serde(deny_unknown_fields)]
1317pub struct OutputBinding {
1318 pub schema_version: u32,
1319 pub binding_id: String,
1320 pub node_id: NodeId,
1321 pub port_name: String,
1322 pub prediction_level: PredictionLevel,
1323 #[serde(default)]
1324 pub unit_level: Option<EntityUnitLevel>,
1325 pub prediction_kind: PredictionKind,
1326 pub prediction_source: PredictionSource,
1327 #[serde(default)]
1328 pub refit_strategy: Option<RefitStrategy>,
1329 pub aggregation_fingerprint: String,
1330 pub target_names: Vec<String>,
1331 pub target_units: Vec<Option<String>>,
1332 pub class_labels: Vec<Vec<String>>,
1333 pub output_order: OutputOrder,
1334 pub target_space: String,
1335 pub binding_fingerprint: String,
1336}
1337
1338impl OutputBinding {
1339 pub fn compute_fingerprint(&self) -> Result<String> {
1340 tcv1_fingerprint_without(self, "binding_fingerprint", "output binding")
1341 }
1342
1343 pub fn validate(&self, graph: &GraphSpec) -> Result<()> {
1344 if self.schema_version != OUTPUT_BINDING_SCHEMA_VERSION {
1345 return unsupported_version(
1346 "output binding",
1347 self.schema_version,
1348 OUTPUT_BINDING_SCHEMA_VERSION,
1349 );
1350 }
1351 validate_identifier_text("output binding_id", &self.binding_id)?;
1352 validate_non_empty("output binding port_name", &self.port_name)?;
1353 validate_sha256("output aggregation", &self.aggregation_fingerprint)?;
1354 validate_sha256("output binding", &self.binding_fingerprint)?;
1355 validate_output_unit_level(self.prediction_level, self.unit_level)?;
1356 validate_output_shape(
1357 self.prediction_kind,
1358 self.output_order,
1359 &self.target_names,
1360 &self.target_units,
1361 &self.class_labels,
1362 &self.target_space,
1363 )?;
1364 match (self.prediction_source, self.refit_strategy) {
1365 (PredictionSource::FinalRefit, None) => {
1366 return contract_error(
1367 "final_refit output binding requires refit_strategy".to_string(),
1368 );
1369 }
1370 (PredictionSource::CvEnsemble | PredictionSource::FoldMember, Some(_)) => {
1371 return contract_error(
1372 "non-final output binding forbids refit_strategy".to_string(),
1373 );
1374 }
1375 _ => {}
1376 }
1377 let request = TrainingOutputRequest {
1378 output_id: self.binding_id.clone(),
1379 node_id: self.node_id.clone(),
1380 port_name: Some(self.port_name.clone()),
1381 prediction_level: self.prediction_level,
1382 unit_level: self.unit_level,
1383 prediction_kind: self.prediction_kind,
1384 target_names: self.target_names.clone(),
1385 target_units: self.target_units.clone(),
1386 class_labels: self.class_labels.clone(),
1387 output_order: self.output_order,
1388 target_space: self.target_space.clone(),
1389 };
1390 request.resolve(graph)?;
1391 if self.binding_fingerprint != self.compute_fingerprint()? {
1392 return contract_error(
1393 "output binding fingerprint does not match TCV1 content".to_string(),
1394 );
1395 }
1396 Ok(())
1397 }
1398}
1399
1400#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1401#[serde(deny_unknown_fields)]
1402pub struct PredictorTemplate {
1403 pub graph: GraphSpec,
1404 pub campaign: CampaignSpec,
1405 pub controller_manifests: BTreeMap<crate::ids::ControllerId, ControllerManifest>,
1406 pub template_fingerprint: String,
1407}
1408
1409impl PredictorTemplate {
1410 pub fn compute_fingerprint(&self) -> Result<String> {
1411 tcv1_fingerprint_without(self, "template_fingerprint", "predictor template")
1412 }
1413
1414 pub fn validate(&self) -> Result<()> {
1415 self.graph.validate()?;
1416 self.campaign.validate()?;
1417 for (controller_id, manifest) in &self.controller_manifests {
1418 if controller_id != &manifest.controller_id {
1419 return contract_error(format!(
1420 "predictor template controller key `{controller_id}` does not match manifest `{}`",
1421 manifest.controller_id
1422 ));
1423 }
1424 manifest.validate()?;
1425 }
1426 validate_sha256("predictor template", &self.template_fingerprint)?;
1427 if self.template_fingerprint != self.compute_fingerprint()? {
1428 return contract_error(
1429 "predictor template fingerprint does not match TCV1 content".to_string(),
1430 );
1431 }
1432 Ok(())
1433 }
1434}
1435
1436#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1437#[serde(deny_unknown_fields)]
1438pub struct TrainingOutcomeRef {
1439 pub outcome_id: String,
1440 pub outcome_fingerprint: String,
1441 #[serde(default, skip_serializing_if = "Option::is_none")]
1445 pub pre_conformal_outcome_fingerprint: Option<String>,
1446 pub training_request_fingerprint: String,
1447 pub effective_plan_fingerprint: String,
1448 pub execution_bundle_id: BundleId,
1449 pub execution_bundle_fingerprint: String,
1450 pub output_binding_fingerprints: Vec<String>,
1451 pub training_influence_fingerprint: String,
1452 pub data_identities_fingerprint: String,
1453}
1454
1455impl TrainingOutcomeRef {
1456 pub(crate) fn validate(&self) -> Result<()> {
1457 validate_identifier_text("training outcome_id", &self.outcome_id)?;
1458 for (label, fingerprint) in [
1459 ("training outcome", &self.outcome_fingerprint),
1460 (
1461 "training outcome request",
1462 &self.training_request_fingerprint,
1463 ),
1464 (
1465 "training outcome effective plan",
1466 &self.effective_plan_fingerprint,
1467 ),
1468 (
1469 "training outcome influence",
1470 &self.training_influence_fingerprint,
1471 ),
1472 (
1473 "training outcome execution bundle",
1474 &self.execution_bundle_fingerprint,
1475 ),
1476 (
1477 "training outcome data identities",
1478 &self.data_identities_fingerprint,
1479 ),
1480 ] {
1481 validate_sha256(label, fingerprint)?;
1482 }
1483 if let Some(fingerprint) = &self.pre_conformal_outcome_fingerprint {
1484 validate_sha256("pre-conformal training outcome", fingerprint)?;
1485 }
1486 if self.output_binding_fingerprints.is_empty() {
1487 return contract_error(
1488 "training outcome reference requires output binding fingerprints".to_string(),
1489 );
1490 }
1491 for fingerprint in &self.output_binding_fingerprints {
1492 validate_sha256("training outcome output binding", fingerprint)?;
1493 }
1494 Ok(())
1495 }
1496}
1497
1498#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1499#[serde(rename_all = "snake_case")]
1500pub enum ArtifactLoadMode {
1501 NativePortable,
1502 HostSidecar,
1503}
1504
1505#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1506#[serde(deny_unknown_fields)]
1507pub struct PackageArtifactBinding {
1508 pub artifact_id: ArtifactId,
1509 pub load_mode: ArtifactLoadMode,
1510}
1511
1512#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1516#[serde(rename_all = "snake_case")]
1517pub enum PortableRefitMode {
1518 Full,
1519 Transfer,
1520 Finetune,
1521}
1522
1523#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1525#[serde(deny_unknown_fields)]
1526pub struct PortableRefitController {
1527 pub node_id: NodeId,
1528 pub controller_id: ControllerId,
1529 pub controller_version: String,
1530 pub manifest_fingerprint: String,
1531 pub capabilities: BTreeSet<ControllerCapability>,
1532}
1533
1534impl PortableRefitController {
1535 fn validate(&self) -> Result<()> {
1536 validate_identifier_text("portable refit controller node_id", self.node_id.as_str())?;
1537 validate_identifier_text(
1538 "portable refit controller controller_id",
1539 self.controller_id.as_str(),
1540 )?;
1541 validate_non_empty(
1542 "portable refit controller controller_version",
1543 &self.controller_version,
1544 )?;
1545 validate_sha256(
1546 "portable refit controller manifest",
1547 &self.manifest_fingerprint,
1548 )?;
1549 if !self
1550 .capabilities
1551 .contains(&ControllerCapability::SupportsPortableFullRefit)
1552 {
1553 return contract_error(format!(
1554 "portable refit controller `{}` is missing supports_portable_full_refit",
1555 self.controller_id
1556 ));
1557 }
1558 Ok(())
1559 }
1560}
1561
1562#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1567#[serde(deny_unknown_fields)]
1568pub struct PortableRefitRecipe {
1569 pub schema_version: u32,
1570 pub recipe_id: String,
1571 pub mode: PortableRefitMode,
1572 pub parent_package_fingerprint: String,
1573 pub parent_outcome: TrainingOutcomeRef,
1574 pub effective_plan_fingerprint: String,
1575 pub selected_variant_id: VariantId,
1576 pub selected_variant_fingerprint: String,
1577 pub selected_parameter_projection_fingerprint: String,
1578 pub target_binding_fingerprints: Vec<String>,
1579 pub target_schema_fingerprint: String,
1580 pub controllers: Vec<PortableRefitController>,
1581 pub recipe_fingerprint: String,
1582}
1583
1584impl PortableRefitRecipe {
1585 pub fn compute_fingerprint(&self) -> Result<String> {
1586 tcv1_fingerprint_without(self, "recipe_fingerprint", "portable refit recipe")
1587 }
1588
1589 pub fn from_json(json: &str) -> Result<Self> {
1590 let raw_fingerprint =
1591 strict_tcv1_fingerprint_without(json, "recipe_fingerprint", "portable refit recipe")?;
1592 let recipe: Self = serde_json::from_str(json)?;
1593 if recipe.recipe_fingerprint != raw_fingerprint {
1594 return contract_error(
1595 "portable refit recipe fingerprint does not match original TCV1 JSON".to_string(),
1596 );
1597 }
1598 recipe.validate()?;
1599 Ok(recipe)
1600 }
1601
1602 pub fn validate(&self) -> Result<()> {
1603 if self.schema_version != PORTABLE_REFIT_RECIPE_SCHEMA_VERSION {
1604 return unsupported_version(
1605 "portable refit recipe",
1606 self.schema_version,
1607 PORTABLE_REFIT_RECIPE_SCHEMA_VERSION,
1608 );
1609 }
1610 if self.mode != PortableRefitMode::Full {
1611 return contract_error(
1612 "portable refit recipe currently supports only full mode; transfer and finetune require separate controller contracts".to_string(),
1613 );
1614 }
1615 validate_identifier_text("portable refit recipe_id", &self.recipe_id)?;
1616 validate_sha256(
1617 "portable refit parent package",
1618 &self.parent_package_fingerprint,
1619 )?;
1620 self.parent_outcome.validate()?;
1621 for (label, fingerprint) in [
1622 (
1623 "portable refit effective plan",
1624 &self.effective_plan_fingerprint,
1625 ),
1626 (
1627 "portable refit selected variant",
1628 &self.selected_variant_fingerprint,
1629 ),
1630 (
1631 "portable refit selected parameter projection",
1632 &self.selected_parameter_projection_fingerprint,
1633 ),
1634 (
1635 "portable refit target schema",
1636 &self.target_schema_fingerprint,
1637 ),
1638 ("portable refit recipe", &self.recipe_fingerprint),
1639 ] {
1640 validate_sha256(label, fingerprint)?;
1641 }
1642 validate_identifier_text(
1643 "portable refit selected_variant_id",
1644 self.selected_variant_id.as_str(),
1645 )?;
1646 if self.target_binding_fingerprints.is_empty()
1647 || !self
1648 .target_binding_fingerprints
1649 .windows(2)
1650 .all(|pair| pair[0] < pair[1])
1651 {
1652 return contract_error(
1653 "portable refit target binding fingerprints must be non-empty, sorted, and unique"
1654 .to_string(),
1655 );
1656 }
1657 for fingerprint in &self.target_binding_fingerprints {
1658 validate_sha256("portable refit target binding", fingerprint)?;
1659 }
1660 if self.controllers.is_empty()
1661 || !self
1662 .controllers
1663 .windows(2)
1664 .all(|pair| pair[0].node_id < pair[1].node_id)
1665 {
1666 return contract_error(
1667 "portable refit controllers must be non-empty and strictly sorted by node_id"
1668 .to_string(),
1669 );
1670 }
1671 for controller in &self.controllers {
1672 controller.validate()?;
1673 }
1674 if self.recipe_fingerprint != self.compute_fingerprint()? {
1675 return contract_error(
1676 "portable refit recipe fingerprint does not match TCV1 content".to_string(),
1677 );
1678 }
1679 Ok(())
1680 }
1681
1682 pub fn derive_from_package(
1687 package: &PortablePredictorPackage,
1688 recipe_id: impl Into<String>,
1689 ) -> Result<Self> {
1690 package.validate()?;
1691 if package.fitted_artifact_mode != FittedArtifactMode::PortableRequired
1692 || package
1693 .artifact_bindings
1694 .iter()
1695 .any(|binding| binding.load_mode != ArtifactLoadMode::NativePortable)
1696 {
1697 return contract_error(
1698 "portable full refit requires a package with only native_portable artifacts"
1699 .to_string(),
1700 );
1701 }
1702 let selected_variant_id = package
1703 .execution_bundle
1704 .selected_variant_id
1705 .clone()
1706 .ok_or_else(|| {
1707 DagMlError::RuntimeValidation(
1708 "portable full refit requires a selected package variant".to_string(),
1709 )
1710 })?;
1711 let selected_variant = package
1712 .effective_plan
1713 .variants
1714 .iter()
1715 .find(|variant| variant.variant_id == selected_variant_id)
1716 .ok_or_else(|| {
1717 DagMlError::RuntimeValidation(
1718 "portable full refit selected variant is absent from the effective plan"
1719 .to_string(),
1720 )
1721 })?;
1722 let selected_parameters = package
1723 .effective_plan
1724 .node_plans
1725 .iter()
1726 .map(|(node_id, node)| (node_id.clone(), node.params.clone()))
1727 .collect::<BTreeMap<_, _>>();
1728 let selected_parameter_projection_fingerprint = tcv1_fingerprint(
1729 &(selected_variant.fingerprint.clone(), selected_parameters),
1730 "portable refit selected parameter projection",
1731 )?;
1732 let mut target_binding_fingerprints = package
1733 .output_bindings
1734 .iter()
1735 .map(|binding| binding.binding_fingerprint.clone())
1736 .collect::<Vec<_>>();
1737 target_binding_fingerprints.sort();
1738 if target_binding_fingerprints
1739 .windows(2)
1740 .any(|pair| pair[0] == pair[1])
1741 {
1742 return contract_error(
1743 "portable full refit package has duplicate output binding fingerprints".to_string(),
1744 );
1745 }
1746 let target_schema_fingerprint =
1747 tcv1_fingerprint(&package.output_bindings, "portable refit target schema")?;
1748 let mut controllers = Vec::with_capacity(package.predictor_node_ids.len());
1749 for node_id in &package.predictor_node_ids {
1750 let node = package.effective_plan.node_plans.get(node_id).ok_or_else(|| {
1751 DagMlError::RuntimeValidation(format!(
1752 "portable full refit predictor node `{node_id}` is absent from the effective plan"
1753 ))
1754 })?;
1755 let manifest = package
1756 .effective_plan
1757 .controller_manifests
1758 .get(&node.controller_id)
1759 .ok_or_else(|| {
1760 DagMlError::RuntimeValidation(format!(
1761 "portable full refit node `{node_id}` has no controller manifest"
1762 ))
1763 })?;
1764 if manifest.controller_id != node.controller_id
1765 || manifest.controller_version != node.controller_version
1766 || manifest.capabilities != node.controller_capabilities
1767 {
1768 return contract_error(format!(
1769 "portable full refit node `{node_id}` does not exactly match its controller manifest"
1770 ));
1771 }
1772 controllers.push(PortableRefitController {
1773 node_id: node_id.clone(),
1774 controller_id: node.controller_id.clone(),
1775 controller_version: node.controller_version.clone(),
1776 manifest_fingerprint: tcv1_fingerprint(
1777 manifest,
1778 "portable refit controller manifest",
1779 )?,
1780 capabilities: node.controller_capabilities.clone(),
1781 });
1782 }
1783 controllers.sort_by(|left, right| left.node_id.cmp(&right.node_id));
1784 let mut recipe = Self {
1785 schema_version: PORTABLE_REFIT_RECIPE_SCHEMA_VERSION,
1786 recipe_id: recipe_id.into(),
1787 mode: PortableRefitMode::Full,
1788 parent_package_fingerprint: package.package_fingerprint.clone(),
1789 parent_outcome: package.training_outcome.clone(),
1790 effective_plan_fingerprint: package.training_outcome.effective_plan_fingerprint.clone(),
1791 selected_variant_id,
1792 selected_variant_fingerprint: selected_variant.fingerprint.clone(),
1793 selected_parameter_projection_fingerprint,
1794 target_binding_fingerprints,
1795 target_schema_fingerprint,
1796 controllers,
1797 recipe_fingerprint: zero_fingerprint(),
1798 };
1799 recipe.recipe_fingerprint = recipe.compute_fingerprint()?;
1800 recipe.validate()?;
1801 Ok(recipe)
1802 }
1803
1804 pub fn validate_against_source_package(
1809 &self,
1810 package: &PortablePredictorPackage,
1811 ) -> Result<()> {
1812 self.validate()?;
1813 let expected = Self::derive_from_package(package, self.recipe_id.clone())?;
1814 if self != &expected {
1815 return contract_error(
1816 "portable refit recipe does not exactly match its source package".to_string(),
1817 );
1818 }
1819 Ok(())
1820 }
1821}
1822
1823#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1828#[serde(deny_unknown_fields)]
1829pub struct PortableRefitProvenance {
1830 pub schema_version: u32,
1831 pub parent_recipe_id: String,
1832 pub parent_recipe_fingerprint: String,
1833 pub parent_package_fingerprint: String,
1834 pub parent_outcome: TrainingOutcomeRef,
1835 pub target_training_request_fingerprint: String,
1836 pub target_data_identities_fingerprint: String,
1837 pub target_training_influence_fingerprint: String,
1838 pub target_relation_fingerprint: String,
1839 pub provenance_fingerprint: String,
1840}
1841
1842impl PortableRefitProvenance {
1843 pub fn compute_fingerprint(&self) -> Result<String> {
1844 tcv1_fingerprint_without(self, "provenance_fingerprint", "portable refit provenance")
1845 }
1846
1847 pub fn from_target_cohort(
1848 recipe: &PortableRefitRecipe,
1849 target_training_request_fingerprint: String,
1850 data_identities: &[TrainingDataIdentity],
1851 training_influence: &TrainingInfluenceManifest,
1852 ) -> Result<Self> {
1853 recipe.validate()?;
1854 if data_identities.is_empty()
1855 || !data_identities
1856 .windows(2)
1857 .all(|pair| pair[0].requirement_key < pair[1].requirement_key)
1858 {
1859 return contract_error(
1860 "portable refit target data identities must be non-empty and strictly sorted"
1861 .to_string(),
1862 );
1863 }
1864 for identity in data_identities {
1865 identity.validate()?;
1866 }
1867 training_influence.validate()?;
1868 let target_data_identities_fingerprint =
1869 tcv1_fingerprint(data_identities, "portable refit target data identities")?;
1870 let target_relation_fingerprint = training_influence.relation_fingerprint.clone();
1871 if data_identities
1872 .iter()
1873 .any(|identity| identity.relation_fingerprint != target_relation_fingerprint)
1874 {
1875 return contract_error(
1876 "portable refit target data identities do not exactly bind target influence relations"
1877 .to_string(),
1878 );
1879 }
1880 if target_training_request_fingerprint == recipe.parent_outcome.training_request_fingerprint
1881 || target_data_identities_fingerprint
1882 == recipe.parent_outcome.data_identities_fingerprint
1883 {
1884 return contract_error(
1885 "portable refit target cohort must not reuse the parent training request or data identities"
1886 .to_string(),
1887 );
1888 }
1889 let mut provenance = Self {
1890 schema_version: PORTABLE_REFIT_PROVENANCE_SCHEMA_VERSION,
1891 parent_recipe_id: recipe.recipe_id.clone(),
1892 parent_recipe_fingerprint: recipe.recipe_fingerprint.clone(),
1893 parent_package_fingerprint: recipe.parent_package_fingerprint.clone(),
1894 parent_outcome: recipe.parent_outcome.clone(),
1895 target_training_request_fingerprint,
1896 target_data_identities_fingerprint,
1897 target_training_influence_fingerprint: training_influence.manifest_fingerprint.clone(),
1898 target_relation_fingerprint,
1899 provenance_fingerprint: zero_fingerprint(),
1900 };
1901 provenance.provenance_fingerprint = provenance.compute_fingerprint()?;
1902 provenance.validate_against_recipe(recipe)?;
1903 Ok(provenance)
1904 }
1905
1906 pub fn from_json_for_recipe(json: &str, recipe: &PortableRefitRecipe) -> Result<Self> {
1910 let raw_fingerprint = strict_tcv1_fingerprint_without(
1911 json,
1912 "provenance_fingerprint",
1913 "portable refit provenance",
1914 )?;
1915 let provenance: Self = serde_json::from_str(json)?;
1916 if provenance.provenance_fingerprint != raw_fingerprint {
1917 return contract_error(
1918 "portable refit provenance fingerprint does not match original TCV1 JSON"
1919 .to_string(),
1920 );
1921 }
1922 provenance.validate_against_recipe(recipe)?;
1923 Ok(provenance)
1924 }
1925
1926 pub fn validate_against_recipe(&self, recipe: &PortableRefitRecipe) -> Result<()> {
1927 if self.schema_version != PORTABLE_REFIT_PROVENANCE_SCHEMA_VERSION {
1928 return unsupported_version(
1929 "portable refit provenance",
1930 self.schema_version,
1931 PORTABLE_REFIT_PROVENANCE_SCHEMA_VERSION,
1932 );
1933 }
1934 recipe.validate()?;
1935 self.parent_outcome.validate()?;
1936 validate_identifier_text("portable refit parent recipe_id", &self.parent_recipe_id)?;
1937 for (label, fingerprint) in [
1938 (
1939 "portable refit parent recipe",
1940 &self.parent_recipe_fingerprint,
1941 ),
1942 (
1943 "portable refit parent package",
1944 &self.parent_package_fingerprint,
1945 ),
1946 (
1947 "portable refit target training request",
1948 &self.target_training_request_fingerprint,
1949 ),
1950 (
1951 "portable refit target data identities",
1952 &self.target_data_identities_fingerprint,
1953 ),
1954 (
1955 "portable refit target training influence",
1956 &self.target_training_influence_fingerprint,
1957 ),
1958 (
1959 "portable refit target relation",
1960 &self.target_relation_fingerprint,
1961 ),
1962 ("portable refit provenance", &self.provenance_fingerprint),
1963 ] {
1964 validate_sha256(label, fingerprint)?;
1965 }
1966 if self.parent_recipe_id != recipe.recipe_id
1967 || self.parent_recipe_fingerprint != recipe.recipe_fingerprint
1968 || self.parent_package_fingerprint != recipe.parent_package_fingerprint
1969 || self.parent_outcome != recipe.parent_outcome
1970 {
1971 return contract_error(
1972 "portable refit provenance does not exactly bind its parent recipe".to_string(),
1973 );
1974 }
1975 if self.target_training_request_fingerprint
1976 == recipe.parent_outcome.training_request_fingerprint
1977 || self.target_data_identities_fingerprint
1978 == recipe.parent_outcome.data_identities_fingerprint
1979 {
1980 return contract_error(
1981 "portable refit provenance reuses parent training evidence".to_string(),
1982 );
1983 }
1984 if self.provenance_fingerprint != self.compute_fingerprint()? {
1985 return contract_error(
1986 "portable refit provenance fingerprint does not match TCV1 content".to_string(),
1987 );
1988 }
1989 Ok(())
1990 }
1991}
1992
1993#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1997#[serde(deny_unknown_fields)]
1998pub struct PortablePredictorPackage {
1999 pub schema_version: u32,
2000 pub package_id: String,
2001 pub template: PredictorTemplate,
2002 pub training_request_fingerprint: String,
2003 pub training_outcome: TrainingOutcomeRef,
2004 pub effective_plan: ExecutionPlan,
2005 pub execution_bundle: ExecutionBundle,
2006 #[serde(default, skip_serializing_if = "Option::is_none")]
2009 pub conformal_calibration: Option<ConformalCalibration>,
2010 #[serde(default, skip_serializing_if = "Option::is_none")]
2012 pub conformal_calibration_replay: Option<TrainingReplayOutcome>,
2013 pub output_bindings: Vec<OutputBinding>,
2014 pub predictor_node_ids: Vec<NodeId>,
2015 pub training_influence: TrainingInfluenceManifest,
2016 pub data_identities: Vec<TrainingDataIdentity>,
2017 pub fitted_artifact_mode: FittedArtifactMode,
2018 pub artifact_bindings: Vec<PackageArtifactBinding>,
2019 pub package_fingerprint: String,
2020}
2021
2022impl PortablePredictorPackage {
2023 pub fn compute_fingerprint(&self) -> Result<String> {
2024 tcv1_fingerprint_without(self, "package_fingerprint", "portable predictor package")
2025 }
2026
2027 pub fn from_json(json: &str) -> Result<Self> {
2028 let raw_fingerprint = strict_tcv1_fingerprint_without(
2029 json,
2030 "package_fingerprint",
2031 "portable predictor package",
2032 )?;
2033 let package: Self = serde_json::from_str(json)?;
2034 if package.package_fingerprint != raw_fingerprint {
2035 return contract_error(
2036 "portable predictor package fingerprint does not match original TCV1 JSON"
2037 .to_string(),
2038 );
2039 }
2040 package.validate()?;
2041 Ok(package)
2042 }
2043
2044 pub fn validate(&self) -> Result<()> {
2045 if self.schema_version < MIN_READABLE_PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION
2046 || self.schema_version > PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION
2047 {
2048 return unsupported_version(
2049 "portable predictor package",
2050 self.schema_version,
2051 PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION,
2052 );
2053 }
2054 if self.schema_version == LEGACY_PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION
2055 && (self.conformal_calibration.is_some() || self.conformal_calibration_replay.is_some())
2056 {
2057 return contract_error(
2058 "portable predictor package V1 cannot carry conformal state; migrate to a published V2 package schema".to_string(),
2059 );
2060 }
2061 validate_identifier_text("portable predictor package_id", &self.package_id)?;
2062 validate_sha256(
2063 "portable predictor training request",
2064 &self.training_request_fingerprint,
2065 )?;
2066 validate_sha256("portable predictor package", &self.package_fingerprint)?;
2067 self.training_outcome.validate()?;
2068 if self.training_request_fingerprint != self.training_outcome.training_request_fingerprint {
2069 return contract_error(
2070 "portable predictor request fingerprint is not cross-linked by outcome reference"
2071 .to_string(),
2072 );
2073 }
2074 self.template.validate()?;
2075 self.effective_plan.validate()?;
2076 if self.template.graph != self.effective_plan.graph_plan.graph
2077 || self.template.campaign != self.effective_plan.campaign
2078 || self.template.controller_manifests != self.effective_plan.controller_manifests
2079 {
2080 return contract_error(
2081 "portable predictor template does not exactly match effective plan".to_string(),
2082 );
2083 }
2084 self.execution_bundle
2085 .validate_against_plan(&self.effective_plan)?;
2086 if self.schema_version == PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION
2087 && self.execution_bundle.schema_version
2088 != crate::bundle::EXECUTION_BUNDLE_SCHEMA_VERSION
2089 {
2090 return contract_error(
2091 "portable predictor package V2 requires an execution bundle V2".to_string(),
2092 );
2093 }
2094 match (
2095 &self.conformal_calibration,
2096 &self.conformal_calibration_replay,
2097 &self.execution_bundle.conformal_calibration,
2098 ) {
2099 (Some(calibration), Some(replay), Some(reference)) => {
2100 reference.validate_against(calibration)?;
2101 calibration.validate()?;
2102 replay.validate()?;
2103 let binding = self
2104 .output_bindings
2105 .iter()
2106 .find(|binding| binding.binding_id == calibration.binding_id)
2107 .ok_or_else(|| {
2108 DagMlError::RuntimeValidation(
2109 "portable predictor conformal binding is absent".to_string(),
2110 )
2111 })?;
2112 let fold_set = self.effective_plan.fold_set.as_ref().ok_or_else(|| {
2113 DagMlError::RuntimeValidation(
2114 "portable predictor conformal calibration requires FoldSet".to_string(),
2115 )
2116 })?;
2117 let context = &calibration.context;
2118 let mut pre_conformal_bundle = self.execution_bundle.clone();
2119 pre_conformal_bundle.conformal_calibration = None;
2120 let pre_conformal_outcome_fingerprint = self
2121 .training_outcome
2122 .pre_conformal_outcome_fingerprint
2123 .as_ref()
2124 .ok_or_else(|| {
2125 DagMlError::RuntimeValidation(
2126 "portable predictor conformal state has no pre-conformal source anchor"
2127 .to_string(),
2128 )
2129 })?;
2130 let expected_replay_source = TrainingOutcomeRef {
2131 outcome_id: self.training_outcome.outcome_id.clone(),
2132 outcome_fingerprint: pre_conformal_outcome_fingerprint.clone(),
2133 pre_conformal_outcome_fingerprint: None,
2134 training_request_fingerprint: self
2135 .training_outcome
2136 .training_request_fingerprint
2137 .clone(),
2138 effective_plan_fingerprint: self
2139 .training_outcome
2140 .effective_plan_fingerprint
2141 .clone(),
2142 execution_bundle_id: self.training_outcome.execution_bundle_id.clone(),
2143 execution_bundle_fingerprint: tcv1_fingerprint(
2144 &pre_conformal_bundle,
2145 "portable predictor pre-conformal execution bundle",
2146 )?,
2147 output_binding_fingerprints: self
2148 .training_outcome
2149 .output_binding_fingerprints
2150 .clone(),
2151 training_influence_fingerprint: self
2152 .training_outcome
2153 .training_influence_fingerprint
2154 .clone(),
2155 data_identities_fingerprint: self
2156 .training_outcome
2157 .data_identities_fingerprint
2158 .clone(),
2159 };
2160 let replay_request = replay_request_from_outcome(replay);
2161 replay_request.validate()?;
2162 if context.predictor_binding_fingerprint != binding.binding_fingerprint
2163 || self
2164 .training_outcome
2165 .pre_conformal_outcome_fingerprint
2166 .as_deref()
2167 != Some(context.source_training_outcome_fingerprint.as_str())
2168 || context.data_identities_fingerprint
2169 != self.training_outcome.data_identities_fingerprint
2170 || context.training_influence_fingerprint
2171 != self.training_influence.manifest_fingerprint
2172 || context.fold_set_fingerprint != fold_set_fingerprint(fold_set)?
2173 || context.calibration_replay_outcome_fingerprint != replay.outcome_fingerprint
2174 || replay.source_training_outcome != expected_replay_source
2175 || replay_request.source_outcome_fingerprint
2176 != *pre_conformal_outcome_fingerprint
2177 || replay.replay_request_id != replay_request.request_id
2178 || replay.phase != Phase::Predict
2179 || replay.bundle_id != self.execution_bundle.bundle_id
2180 || replay.plan_id != self.effective_plan.id
2181 {
2182 return contract_error("portable predictor conformal context does not exactly cross-link package provenance".to_string());
2183 }
2184 if context.relation_fingerprint == self.training_influence.relation_fingerprint {
2185 return contract_error(
2186 "portable predictor calibration relation authority must be distinct from development relations"
2187 .to_string(),
2188 );
2189 }
2190 let package_bindings = self
2191 .output_bindings
2192 .iter()
2193 .map(|binding| (binding.binding_id.as_str(), binding))
2194 .collect::<BTreeMap<_, _>>();
2195 for replay_output in &replay.outputs {
2196 let package_binding = package_bindings
2197 .get(replay_output.binding.binding_id.as_str())
2198 .ok_or_else(|| {
2199 DagMlError::RuntimeValidation(
2200 "portable predictor calibration replay contains an output binding absent from the package"
2201 .to_string(),
2202 )
2203 })?;
2204 if &replay_output.binding != *package_binding {
2205 return contract_error(
2206 "portable predictor calibration replay output binding does not exactly match the package"
2207 .to_string(),
2208 );
2209 }
2210 replay_output.validate(&self.effective_plan)?;
2211 }
2212 let replay_output = replay
2213 .outputs
2214 .iter()
2215 .find(|output| output.binding.binding_id == calibration.binding_id)
2216 .ok_or_else(|| {
2217 DagMlError::RuntimeValidation(
2218 "portable predictor calibration replay is missing its selected binding"
2219 .to_string(),
2220 )
2221 })?;
2222 let [point] = replay_output.predictions.as_slice() else {
2223 return contract_error(
2224 "portable predictor calibration replay requires exactly one selected point block"
2225 .to_string(),
2226 );
2227 };
2228 if replay_output.binding != *binding
2229 || point.sample_ids != calibration.sample_ids
2230 || point.sample_ids != context.calibration_cohort.physical_sample_ids
2231 || !replay.conformal_intervals.is_empty()
2232 || replay.input_data_identities.iter().any(|identity| {
2233 identity.relation_fingerprint != context.relation_fingerprint
2234 })
2235 {
2236 return contract_error(
2237 "portable predictor calibration replay evidence does not match its selected binding, samples, or relation authority"
2238 .to_string(),
2239 );
2240 }
2241 let training_ids = self
2242 .training_influence
2243 .entries
2244 .iter()
2245 .flat_map(|entry| {
2246 entry
2247 .physical_sample_ids
2248 .iter()
2249 .chain(entry.origin_sample_ids.iter())
2250 })
2251 .collect::<BTreeSet<_>>();
2252 if context
2253 .calibration_cohort
2254 .physical_sample_ids
2255 .iter()
2256 .chain(context.calibration_cohort.origin_sample_ids.iter())
2257 .any(|sample_id| training_ids.contains(sample_id))
2258 {
2259 return contract_error(
2260 "portable predictor calibration cohort overlaps training influence closure"
2261 .to_string(),
2262 );
2263 }
2264 }
2265 (None, None, None) => {
2266 if self
2267 .training_outcome
2268 .pre_conformal_outcome_fingerprint
2269 .is_some()
2270 {
2271 return contract_error(
2272 "portable predictor pre-conformal source requires conformal state"
2273 .to_string(),
2274 );
2275 }
2276 }
2277 _ => {
2278 return contract_error(
2279 "portable predictor package and execution bundle conformal state disagree"
2280 .to_string(),
2281 )
2282 }
2283 }
2284 let effective_plan_fingerprint =
2285 tcv1_fingerprint(&self.effective_plan, "portable predictor effective plan")?;
2286 if effective_plan_fingerprint != self.training_outcome.effective_plan_fingerprint {
2287 return contract_error(
2288 "portable predictor effective plan fingerprint is not cross-linked by outcome reference"
2289 .to_string(),
2290 );
2291 }
2292 if self.execution_bundle.bundle_id != self.training_outcome.execution_bundle_id {
2293 return contract_error(
2294 "portable predictor bundle id is not cross-linked by outcome reference".to_string(),
2295 );
2296 }
2297 let execution_bundle_fingerprint = tcv1_fingerprint(
2298 &self.execution_bundle,
2299 "portable predictor execution bundle",
2300 )?;
2301 if execution_bundle_fingerprint != self.training_outcome.execution_bundle_fingerprint {
2302 return contract_error(
2303 "portable predictor execution bundle content is not cross-linked by outcome reference"
2304 .to_string(),
2305 );
2306 }
2307 self.training_influence.validate()?;
2308 if self.training_influence.manifest_fingerprint
2309 != self.training_outcome.training_influence_fingerprint
2310 {
2311 return contract_error(
2312 "portable predictor influence is not cross-linked by outcome reference".to_string(),
2313 );
2314 }
2315 if self.output_bindings.is_empty() {
2316 return contract_error(
2317 "portable predictor package requires at least one output binding".to_string(),
2318 );
2319 }
2320 let mut previous_binding: Option<&str> = None;
2321 let mut output_nodes = Vec::new();
2322 let mut coordinates = BTreeSet::new();
2323 for binding in &self.output_bindings {
2324 if previous_binding.is_some_and(|previous| previous >= binding.binding_id.as_str()) {
2325 return contract_error(
2326 "portable predictor output bindings must be sorted by binding_id".to_string(),
2327 );
2328 }
2329 previous_binding = Some(binding.binding_id.as_str());
2330 binding.validate(&self.effective_plan.graph_plan.graph)?;
2331 if !coordinates.insert((binding.node_id.clone(), binding.port_name.clone())) {
2332 return contract_error(format!(
2333 "portable predictor binds `{}.{}` more than once",
2334 binding.node_id, binding.port_name
2335 ));
2336 }
2337 if binding.prediction_source == PredictionSource::FinalRefit
2338 && self.execution_bundle.refit_artifacts.is_empty()
2339 {
2340 return contract_error(
2341 "final_refit output binding requires refit artifacts".to_string(),
2342 );
2343 }
2344 output_nodes.push(&binding.node_id);
2345 }
2346 if self
2347 .output_bindings
2348 .iter()
2349 .map(|binding| binding.binding_fingerprint.clone())
2350 .collect::<Vec<_>>()
2351 != self.training_outcome.output_binding_fingerprints
2352 {
2353 return contract_error(
2354 "portable predictor output bindings are not cross-linked by outcome reference"
2355 .to_string(),
2356 );
2357 }
2358 let expected_closure = predictor_closure(&self.effective_plan, output_nodes)?;
2359 validate_sorted_unique_ids(
2360 "portable predictor_node_ids",
2361 &self.predictor_node_ids,
2362 true,
2363 )?;
2364 if self
2365 .predictor_node_ids
2366 .iter()
2367 .cloned()
2368 .collect::<BTreeSet<_>>()
2369 != expected_closure
2370 {
2371 return contract_error(
2372 "portable predictor_node_ids do not exactly match output closure".to_string(),
2373 );
2374 }
2375 if self.training_influence.entries.iter().any(|entry| {
2376 entry
2377 .node_id
2378 .as_ref()
2379 .is_some_and(|node_id| !expected_closure.contains(node_id))
2380 }) {
2381 return contract_error(
2382 "portable predictor influence references a node outside predictor closure"
2383 .to_string(),
2384 );
2385 }
2386 validate_package_base_influence(
2387 &self.training_influence,
2388 &self.effective_plan,
2389 &expected_closure,
2390 )?;
2391 validate_package_data_identities(self)?;
2392 let data_identities_fingerprint =
2393 tcv1_fingerprint(&self.data_identities, "portable predictor data identities")?;
2394 if data_identities_fingerprint != self.training_outcome.data_identities_fingerprint {
2395 return contract_error(
2396 "portable predictor data identity content is not cross-linked by outcome reference"
2397 .to_string(),
2398 );
2399 }
2400 if self.data_identities.iter().any(|identity| {
2401 identity.relation_fingerprint != self.training_influence.relation_fingerprint
2402 }) {
2403 return contract_error(
2404 "portable predictor data identities and training influence bind different relations"
2405 .to_string(),
2406 );
2407 }
2408 validate_package_artifact_bindings(self)?;
2409 if !crate::training_runtime::closure_predict_replayable(
2415 &self.effective_plan,
2416 &expected_closure,
2417 &self.execution_bundle,
2418 )? {
2419 return contract_error(
2420 "portable predictor package is not PREDICT-replayable: its predictor closure does not support PREDICT with self-contained retained artifacts".to_string(),
2421 );
2422 }
2423 let value = serde_json::to_value(self)?;
2424 if contains_runtime_handle(&value) {
2425 return contract_error(
2426 "portable predictor package must not contain runtime handles".to_string(),
2427 );
2428 }
2429 if self.package_fingerprint != self.compute_fingerprint()? {
2430 return contract_error(
2431 "portable predictor package fingerprint does not match TCV1 content".to_string(),
2432 );
2433 }
2434 Ok(())
2435 }
2436
2437 pub fn load_with<H>(
2438 self,
2439 mut resolver: impl FnMut(&RefitArtifactRecord) -> Result<H>,
2440 ) -> Result<LoadedPredictor<H>> {
2441 self.validate()?;
2442 let mut artifacts = BTreeMap::new();
2443 let sidecar_ids = self
2444 .artifact_bindings
2445 .iter()
2446 .filter(|binding| binding.load_mode == ArtifactLoadMode::HostSidecar)
2447 .map(|binding| &binding.artifact_id)
2448 .collect::<BTreeSet<_>>();
2449 for record in self
2450 .execution_bundle
2451 .refit_artifacts
2452 .iter()
2453 .filter(|record| sidecar_ids.contains(&record.artifact.id))
2454 {
2455 let handle = resolver(record)?;
2456 artifacts.insert(record.artifact.id.clone(), handle);
2457 }
2458 LoadedPredictor::new(self, artifacts)
2459 }
2460}
2461
2462pub struct LoadedPredictor<H> {
2465 package: PortablePredictorPackage,
2466 artifacts: BTreeMap<ArtifactId, H>,
2467}
2468
2469impl<H> LoadedPredictor<H> {
2470 pub fn new(
2471 package: PortablePredictorPackage,
2472 artifacts: BTreeMap<ArtifactId, H>,
2473 ) -> Result<Self> {
2474 package.validate()?;
2475 let expected = package
2476 .artifact_bindings
2477 .iter()
2478 .filter(|binding| binding.load_mode == ArtifactLoadMode::HostSidecar)
2479 .map(|binding| binding.artifact_id.clone())
2480 .collect::<BTreeSet<_>>();
2481 let actual = artifacts.keys().cloned().collect::<BTreeSet<_>>();
2482 if actual != expected {
2483 return contract_error(
2484 "loaded predictor sidecar artifacts do not exactly match package references"
2485 .to_string(),
2486 );
2487 }
2488 Ok(Self { package, artifacts })
2489 }
2490
2491 pub fn package(&self) -> &PortablePredictorPackage {
2492 &self.package
2493 }
2494
2495 pub fn artifact(&self, artifact_id: &ArtifactId) -> Option<&H> {
2496 self.artifacts.get(artifact_id)
2497 }
2498
2499 pub fn into_parts(self) -> (PortablePredictorPackage, BTreeMap<ArtifactId, H>) {
2500 (self.package, self.artifacts)
2501 }
2502}
2503
2504fn validate_output_unit_level(
2505 prediction_level: PredictionLevel,
2506 unit_level: Option<EntityUnitLevel>,
2507) -> Result<()> {
2508 match prediction_level {
2509 PredictionLevel::Sample if unit_level != Some(EntityUnitLevel::PhysicalSample) => {
2510 contract_error("sample-level output requires unit_level=physical_sample".to_string())
2511 }
2512 PredictionLevel::Target | PredictionLevel::Group if unit_level.is_some() => {
2513 contract_error("target/group output requires unit_level=null".to_string())
2514 }
2515 _ => Ok(()),
2516 }
2517}
2518
2519fn validate_output_shape(
2520 prediction_kind: PredictionKind,
2521 output_order: OutputOrder,
2522 target_names: &[String],
2523 target_units: &[Option<String>],
2524 class_labels: &[Vec<String>],
2525 target_space: &str,
2526) -> Result<()> {
2527 validate_non_empty("output target_space", target_space)?;
2528 validate_unique_text("output target_names", target_names, true)?;
2529 if target_units.len() != target_names.len() || class_labels.len() != target_names.len() {
2530 return contract_error(
2531 "output target_units and class_labels must have one entry per target".to_string(),
2532 );
2533 }
2534 for unit in target_units.iter().flatten() {
2535 validate_non_empty("output target unit", unit)?;
2536 }
2537 let class_output = prediction_kind == PredictionKind::ClassProbability;
2538 for labels in class_labels {
2539 validate_unique_text("output class labels", labels, class_output)?;
2540 }
2541 if prediction_kind == PredictionKind::RegressionPoint
2542 && class_labels.iter().any(|labels| !labels.is_empty())
2543 {
2544 return contract_error("regression output class label arrays must be empty".to_string());
2545 }
2546 match (prediction_kind, output_order) {
2547 (PredictionKind::ClassProbability, OutputOrder::TargetMajorClassMinor) => Ok(()),
2548 (PredictionKind::ClassProbability, _) => contract_error(
2549 "class_probability output requires target_major_class_minor order".to_string(),
2550 ),
2551 (_, OutputOrder::TargetOrder) => Ok(()),
2552 _ => contract_error("non-probability output requires target_order".to_string()),
2553 }
2554}
2555
2556fn validate_canonical_patches(patches: &[ParameterPatch]) -> Result<()> {
2557 let mut previous: Option<(&NodeId, ParameterNamespace, &[String])> = None;
2558 for patch in patches {
2559 patch.validate()?;
2560 let key = (&patch.node_id, patch.namespace, patch.path.as_slice());
2561 if let Some(previous_key) = previous.as_ref() {
2562 if previous_key >= &key {
2563 return contract_error(
2564 "parameter patches must be strictly sorted by (node_id, namespace, path)"
2565 .to_string(),
2566 );
2567 }
2568 if previous_key.0 == key.0
2569 && previous_key.1 == key.1
2570 && (key.2.starts_with(previous_key.2) || previous_key.2.starts_with(key.2))
2571 {
2572 return contract_error(format!(
2573 "parameter patches for `{}` contain a conflicting parent/child path",
2574 patch.node_id
2575 ));
2576 }
2577 }
2578 previous = Some(key);
2579 }
2580 Ok(())
2581}
2582
2583fn validate_patch_policies(
2584 plan: &ExecutionPlan,
2585 policies: &[NodePatchPolicy],
2586) -> Result<BTreeMap<NodeId, BTreeSet<ParameterNamespace>>> {
2587 let mut previous: Option<&NodeId> = None;
2588 let mut result = BTreeMap::new();
2589 for policy in policies {
2590 policy.validate()?;
2591 if previous.is_some_and(|previous| previous >= &policy.node_id) {
2592 return contract_error(
2593 "node patch policies must be strictly sorted by node_id".to_string(),
2594 );
2595 }
2596 previous = Some(&policy.node_id);
2597 if !plan.node_plans.contains_key(&policy.node_id) {
2598 return contract_error(format!(
2599 "node patch policy references unknown node `{}`",
2600 policy.node_id
2601 ));
2602 }
2603 result.insert(policy.node_id.clone(), policy.allowed_namespaces.clone());
2604 }
2605 Ok(result)
2606}
2607
2608fn deep_set_object_key(
2609 root: &mut BTreeMap<String, serde_json::Value>,
2610 path: &[String],
2611 value: serde_json::Value,
2612 node_id: &NodeId,
2613) -> Result<()> {
2614 if path.len() == 1 {
2615 root.insert(path[0].clone(), value);
2616 return Ok(());
2617 }
2618 let first = root.get_mut(&path[0]).ok_or_else(|| {
2619 DagMlError::CampaignValidation(format!(
2620 "parameter patch for `{node_id}` is missing intermediate path `{}`",
2621 path[0]
2622 ))
2623 })?;
2624 let mut cursor = first;
2625 for segment in &path[1..path.len() - 1] {
2626 let object = cursor.as_object_mut().ok_or_else(|| {
2627 DagMlError::CampaignValidation(format!(
2628 "parameter patch for `{node_id}` crosses a scalar or array at `{segment}`"
2629 ))
2630 })?;
2631 cursor = object.get_mut(segment).ok_or_else(|| {
2632 DagMlError::CampaignValidation(format!(
2633 "parameter patch for `{node_id}` is missing intermediate path `{segment}`"
2634 ))
2635 })?;
2636 }
2637 let object = cursor.as_object_mut().ok_or_else(|| {
2638 DagMlError::CampaignValidation(format!(
2639 "parameter patch for `{node_id}` crosses a scalar or array before final key"
2640 ))
2641 })?;
2642 object.insert(path[path.len() - 1].clone(), value);
2643 Ok(())
2644}
2645
2646fn predictor_closure<'a>(
2647 plan: &ExecutionPlan,
2648 roots: impl IntoIterator<Item = &'a NodeId>,
2649) -> Result<BTreeSet<NodeId>> {
2650 let mut pending = roots.into_iter().cloned().collect::<Vec<_>>();
2651 let mut closure = BTreeSet::new();
2652 while let Some(node_id) = pending.pop() {
2653 if !closure.insert(node_id.clone()) {
2654 continue;
2655 }
2656 let node = plan.node_plans.get(&node_id).ok_or_else(|| {
2657 DagMlError::CampaignValidation(format!(
2658 "predictor closure references unknown node `{node_id}`"
2659 ))
2660 })?;
2661 pending.extend(node.input_nodes.iter().cloned());
2662 }
2663 Ok(closure)
2664}
2665
2666fn base_influence_kind(plan: &ExecutionPlan, node_id: &NodeId) -> Option<TrainingInfluenceKind> {
2667 let node_plan = &plan.node_plans[node_id];
2668 if matches!(
2669 node_plan.fit_scope,
2670 ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly
2671 ) {
2672 return None;
2673 }
2674 let oof_consumers = plan
2675 .graph_plan
2676 .graph
2677 .edges
2678 .iter()
2679 .filter(|edge| edge.contract.requires_oof)
2680 .map(|edge| edge.target.node_id.clone())
2681 .collect::<BTreeSet<_>>();
2682 Some(
2683 if oof_consumers.contains(node_id)
2684 || node_plan
2685 .controller_capabilities
2686 .contains(&ControllerCapability::TrainsAggregation)
2687 {
2688 TrainingInfluenceKind::TrainedMetaAggregation
2689 } else if node_plan.kind == NodeKind::Model {
2690 TrainingInfluenceKind::ModelFit
2691 } else if node_plan.kind == NodeKind::Tuner {
2692 TrainingInfluenceKind::HpoSelection
2693 } else {
2694 TrainingInfluenceKind::TransformFit
2695 },
2696 )
2697}
2698
2699fn capability_influence_kinds(
2700 plan: &ExecutionPlan,
2701 node_id: &NodeId,
2702) -> BTreeSet<TrainingInfluenceKind> {
2703 let capabilities = &plan.node_plans[node_id].controller_capabilities;
2704 let mut kinds = BTreeSet::new();
2705 if capabilities.contains(&ControllerCapability::PerformsInternalTuning)
2706 && base_influence_kind(plan, node_id) != Some(TrainingInfluenceKind::HpoSelection)
2707 {
2708 kinds.insert(TrainingInfluenceKind::HpoSelection);
2709 }
2710 if capabilities.contains(&ControllerCapability::UsesEarlyStopping) {
2711 kinds.insert(TrainingInfluenceKind::EarlyStopping);
2712 }
2713 if capabilities.contains(&ControllerCapability::UsesTrainingWeights) {
2714 kinds.insert(TrainingInfluenceKind::WeightingResampling);
2715 }
2716 kinds
2717}
2718
2719fn expected_influence_coordinates(
2720 request: &TrainingRequest,
2721 plan: &ExecutionPlan,
2722 closure: &BTreeSet<NodeId>,
2723) -> Result<ExpectedInfluenceCoordinates> {
2724 let fold_set = plan.fold_set.as_ref().ok_or_else(|| {
2725 DagMlError::CampaignValidation(
2726 "training influence scopes require an explicit fold_set".to_string(),
2727 )
2728 })?;
2729 let all_samples = fold_set.sample_ids.iter().cloned().collect::<BTreeSet<_>>();
2730 let mut expected = BTreeMap::new();
2731 for node_id in closure {
2732 let node_plan = &plan.node_plans[node_id];
2733 let Some(base_kind) = base_influence_kind(plan, node_id) else {
2734 continue;
2735 };
2736 let mut scopes = Vec::<(String, BTreeSet<SampleId>)>::new();
2737 if node_plan.supported_phases.contains(&Phase::FitCv) {
2738 match node_plan.fit_scope {
2739 ControllerFitScope::FoldTrain => {
2740 scopes.extend(fold_set.folds.iter().map(|fold| {
2741 (
2742 format!("fit_cv:{}", fold.fold_id),
2743 fold.train_sample_ids.iter().cloned().collect(),
2744 )
2745 }));
2746 }
2747 ControllerFitScope::FullTrain => {
2748 scopes.push(("fit_cv:full".to_string(), all_samples.clone()));
2752 }
2753 ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly => {}
2754 }
2755 }
2756 if request.options.refit && node_plan.supported_phases.contains(&Phase::Refit) {
2757 scopes.push(("refit:full".to_string(), all_samples.clone()));
2758 }
2759 for (scope_id, samples) in scopes {
2760 expected.insert((base_kind, scope_id, Some(node_id.clone())), samples);
2761 }
2762 }
2763 for requirement in &request.influence_requirements {
2764 let key = (
2765 requirement.kind,
2766 requirement.scope_id.clone(),
2767 Some(requirement.node_id.clone()),
2768 );
2769 if expected
2770 .insert(
2771 key,
2772 requirement.physical_sample_ids.iter().cloned().collect(),
2773 )
2774 .is_some()
2775 {
2776 return contract_error(format!(
2777 "controller influence scope `{}` collides with a derived influence coordinate",
2778 requirement.scope_id
2779 ));
2780 }
2781 }
2782 expected.insert(
2783 (
2784 TrainingInfluenceKind::HpoSelection,
2785 format!("select:{}", request.options.selection.id),
2786 None,
2787 ),
2788 all_samples,
2789 );
2790 Ok(expected)
2791}
2792
2793fn validate_influence_requirements(
2794 request: &TrainingRequest,
2795 plan: &ExecutionPlan,
2796 closure: &BTreeSet<NodeId>,
2797) -> Result<()> {
2798 let fold_set = plan.fold_set.as_ref().ok_or_else(|| {
2799 DagMlError::CampaignValidation(
2800 "controller influence requirements need an explicit fold_set".to_string(),
2801 )
2802 })?;
2803 let all_samples = fold_set.sample_ids.iter().cloned().collect::<BTreeSet<_>>();
2804 let mut expected_slots = BTreeMap::<InfluenceCapabilitySlot, BTreeSet<SampleId>>::new();
2805 for node_id in closure {
2806 let node_plan = &plan.node_plans[node_id];
2807 if base_influence_kind(plan, node_id).is_none() {
2808 continue;
2809 }
2810 let kinds = capability_influence_kinds(plan, node_id);
2811 if node_plan.supported_phases.contains(&Phase::FitCv) {
2812 match node_plan.fit_scope {
2813 ControllerFitScope::FoldTrain => {
2814 for fold in &fold_set.folds {
2815 for kind in &kinds {
2816 expected_slots.insert(
2817 (
2818 node_id.clone(),
2819 *kind,
2820 Phase::FitCv,
2821 Some(fold.fold_id.clone()),
2822 ),
2823 fold.train_sample_ids.iter().cloned().collect(),
2824 );
2825 }
2826 }
2827 }
2828 ControllerFitScope::FullTrain => {
2829 for kind in &kinds {
2833 expected_slots.insert(
2834 (node_id.clone(), *kind, Phase::FitCv, None),
2835 all_samples.clone(),
2836 );
2837 }
2838 }
2839 ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly => {}
2840 }
2841 }
2842 if request.options.refit && node_plan.supported_phases.contains(&Phase::Refit) {
2843 for kind in kinds {
2844 expected_slots.insert(
2845 (node_id.clone(), kind, Phase::Refit, None),
2846 all_samples.clone(),
2847 );
2848 }
2849 }
2850 }
2851 let mut actual_slots = BTreeSet::new();
2852 let mut previous: Option<(TrainingInfluenceKind, &str, &NodeId)> = None;
2853 for requirement in &request.influence_requirements {
2854 validate_identifier_text(
2855 "controller influence requirement scope_id",
2856 &requirement.scope_id,
2857 )?;
2858 validate_sorted_unique_ids(
2859 "controller influence requirement physical_sample_ids",
2860 &requirement.physical_sample_ids,
2861 true,
2862 )?;
2863 let key = (
2864 requirement.kind,
2865 requirement.scope_id.as_str(),
2866 &requirement.node_id,
2867 );
2868 if previous.as_ref().is_some_and(|previous| previous >= &key) {
2869 return contract_error(
2870 "controller influence requirements must be strictly sorted by (kind, scope_id, node_id)"
2871 .to_string(),
2872 );
2873 }
2874 previous = Some(key);
2875 if !closure.contains(&requirement.node_id) {
2876 return contract_error(format!(
2877 "controller influence requirement node `{}` is outside predictor closure",
2878 requirement.node_id
2879 ));
2880 }
2881 if !matches!(requirement.phase, Phase::FitCv | Phase::Refit) {
2882 return contract_error(format!(
2883 "controller influence scope `{}` uses non-training phase {:?}",
2884 requirement.scope_id, requirement.phase
2885 ));
2886 }
2887 let slot = (
2888 requirement.node_id.clone(),
2889 requirement.kind,
2890 requirement.phase,
2891 requirement.fold_id.clone(),
2892 );
2893 let eligible_samples = expected_slots.get(&slot).ok_or_else(|| {
2894 DagMlError::CampaignValidation(format!(
2895 "controller influence scope `{}` is not required by active controller capabilities",
2896 requirement.scope_id
2897 ))
2898 })?;
2899 let actual_samples = requirement
2900 .physical_sample_ids
2901 .iter()
2902 .cloned()
2903 .collect::<BTreeSet<_>>();
2904 if !actual_samples.is_subset(eligible_samples) {
2905 let outer_validation_overlap = requirement
2906 .fold_id
2907 .as_ref()
2908 .and_then(|fold_id| fold_set.folds.iter().find(|fold| &fold.fold_id == fold_id))
2909 .is_some_and(|fold| {
2910 fold.validation_sample_ids
2911 .iter()
2912 .any(|sample_id| actual_samples.contains(sample_id))
2913 });
2914 if outer_validation_overlap {
2915 return contract_error(format!(
2916 "controller influence scope `{}` leaks outer validation samples",
2917 requirement.scope_id
2918 ));
2919 }
2920 return contract_error(format!(
2921 "controller influence scope `{}` uses samples outside its training cohort",
2922 requirement.scope_id
2923 ));
2924 }
2925 match requirement.kind {
2926 TrainingInfluenceKind::WeightingResampling if actual_samples != *eligible_samples => {
2927 return contract_error(format!(
2928 "weighting influence scope `{}` must cover its complete fit cohort",
2929 requirement.scope_id
2930 ));
2931 }
2932 TrainingInfluenceKind::EarlyStopping
2933 if actual_samples.len() >= eligible_samples.len() =>
2934 {
2935 return contract_error(format!(
2936 "early-stopping influence scope `{}` must be a strict training-cohort subset",
2937 requirement.scope_id
2938 ));
2939 }
2940 _ => {}
2941 }
2942 if !actual_slots.insert(slot) {
2943 return contract_error(format!(
2944 "controller influence capability slot is declared more than once at `{}`",
2945 requirement.scope_id
2946 ));
2947 }
2948 }
2949 if actual_slots != expected_slots.into_keys().collect::<BTreeSet<_>>() {
2950 return contract_error(
2951 "controller influence requirements do not exactly cover active capability scopes"
2952 .to_string(),
2953 );
2954 }
2955 Ok(())
2956}
2957
2958fn validate_influence_identity_closure(
2959 entry: &TrainingInfluenceEntry,
2960 relations: &SampleRelationSet,
2961) -> Result<()> {
2962 let physical = entry.physical_sample_ids.iter().collect::<BTreeSet<_>>();
2963 let mut found = BTreeSet::new();
2964 let mut origins = BTreeSet::new();
2965 let mut groups = BTreeSet::new();
2966 for relation in &relations.records {
2967 if physical.contains(&relation.sample_id) {
2968 found.insert(&relation.sample_id);
2969 if let Some(origin) = &relation.origin_sample_id {
2970 origins.insert(origin.clone());
2971 }
2972 if let Some(group) = &relation.group_id {
2973 groups.insert(group.clone());
2974 }
2975 }
2976 }
2977 if found.len() != physical.len() {
2978 return contract_error(format!(
2979 "training influence `{}` contains physical samples absent from relation set",
2980 entry.scope_id
2981 ));
2982 }
2983 if entry
2984 .origin_sample_ids
2985 .iter()
2986 .cloned()
2987 .collect::<BTreeSet<_>>()
2988 != origins
2989 {
2990 return contract_error(format!(
2991 "training influence `{}` origin closure does not match relation set",
2992 entry.scope_id
2993 ));
2994 }
2995 if entry.group_ids.iter().cloned().collect::<BTreeSet<_>>() != groups {
2996 return contract_error(format!(
2997 "training influence `{}` group closure does not match relation set",
2998 entry.scope_id
2999 ));
3000 }
3001 Ok(())
3002}
3003
3004fn validate_output_controllers(
3005 plan: &ExecutionPlan,
3006 outputs: &[ResolvedTrainingOutput],
3007) -> Result<()> {
3008 for output in outputs {
3009 let node = &plan.node_plans[&output.node_id];
3010 if !node
3011 .controller_capabilities
3012 .contains(&ControllerCapability::EmitsPredictions)
3013 {
3014 return contract_error(format!(
3015 "training output node `{}` controller does not declare emits_predictions",
3016 output.node_id
3017 ));
3018 }
3019 }
3020 Ok(())
3021}
3022
3023fn validate_selection_output(
3024 plan: &ExecutionPlan,
3025 options: &TrainingOptions,
3026 outputs: &[ResolvedTrainingOutput],
3027) -> Result<()> {
3028 let output = outputs
3029 .iter()
3030 .find(|output| output.output_id == options.selection_output_id)
3031 .expect("TrainingOptions::validate resolved selection_output_id");
3032 let node_plan = &plan.node_plans[&output.node_id];
3033 if !node_plan.supported_phases.contains(&Phase::FitCv) {
3034 return contract_error(format!(
3035 "training selection output `{}` is not scorable in FIT_CV",
3036 output.output_id
3037 ));
3038 }
3039 let graph_node = plan
3040 .graph_plan
3041 .graph
3042 .nodes
3043 .iter()
3044 .find(|node| node.id == output.node_id)
3045 .expect("execution plan output node exists in graph");
3046 let binds_declared_prediction_port = graph_node
3047 .ports
3048 .outputs
3049 .iter()
3050 .any(|port| port.kind == PortKind::Prediction && port.name == output.port_name);
3051 if !binds_declared_prediction_port {
3052 return contract_error(format!(
3053 "training selection output `{}` port `{}.{}` is not a declared prediction port",
3054 output.output_id, output.node_id, output.port_name
3055 ));
3056 }
3057 let campaign_metric_level = plan.campaign.aggregation_policy.selection_metric_level;
3058 if output.prediction_level != campaign_metric_level {
3059 return contract_error(format!(
3060 "training selection output `{}` prediction level does not match campaign selection_metric_level",
3061 output.output_id
3062 ));
3063 }
3064 if options
3065 .selection
3066 .required_metric_level
3067 .is_some_and(|level| level != campaign_metric_level)
3068 {
3069 return contract_error(format!(
3070 "training selection output `{}` prediction level does not match selection.required_metric_level",
3071 output.output_id
3072 ));
3073 }
3074 let metric_name = options.selection.metric.name.as_str();
3075 let objective = options.selection.metric.objective;
3076 crate::metrics::RegressionMetricKind::resolve_for_prediction_kind(
3077 metric_name,
3078 objective,
3079 output.prediction_kind,
3080 )?;
3081 Ok(())
3082}
3083
3084fn validate_scheduler_capabilities(
3085 scheduler: &TrainingSchedulerOptions,
3086 plan: &ExecutionPlan,
3087 closure: &BTreeSet<NodeId>,
3088) -> Result<()> {
3089 let Some(backend) = scheduler.backend else {
3090 return Ok(());
3091 };
3092 for node_id in closure {
3093 let capabilities = &plan.node_plans[node_id].controller_capabilities;
3094 match backend {
3095 TrainingSchedulerBackend::Threads => {
3096 if !capabilities.contains(&ControllerCapability::ThreadSafe) {
3097 return contract_error(format!(
3098 "parallel thread scheduler requires thread_safe controller for `{node_id}`"
3099 ));
3100 }
3101 if capabilities.contains(&ControllerCapability::NeedsPythonGil) {
3102 return contract_error(format!(
3103 "parallel thread scheduler refuses needs_python_gil controller for `{node_id}`"
3104 ));
3105 }
3106 }
3107 TrainingSchedulerBackend::Processes => {
3108 if !capabilities.contains(&ControllerCapability::ProcessSafe) {
3109 return contract_error(format!(
3110 "parallel process scheduler requires process_safe controller for `{node_id}`"
3111 ));
3112 }
3113 }
3114 }
3115 }
3116 Ok(())
3117}
3118
3119fn validate_artifact_mode(
3120 artifacts: &TrainingArtifactOptions,
3121 plan: &ExecutionPlan,
3122 closure: &BTreeSet<NodeId>,
3123) -> Result<()> {
3124 if artifacts.fitted_artifacts != FittedArtifactMode::PortableRequired {
3125 return Ok(());
3126 }
3127 for node_id in closure {
3128 let node = &plan.node_plans[node_id];
3129 if node
3130 .controller_capabilities
3131 .contains(&ControllerCapability::EmitsArtifacts)
3132 && node.artifact_policy == ArtifactPolicy::HostOnly
3133 {
3134 return contract_error(format!(
3135 "portable_required training artifacts refuse host_only controller for `{node_id}`"
3136 ));
3137 }
3138 }
3139 Ok(())
3140}
3141
3142fn validate_training_data_identities(
3143 request: &TrainingRequest,
3144 plan: &ExecutionPlan,
3145) -> Result<()> {
3146 let mut expected = BTreeMap::new();
3147 for binding in plan.campaign.data_bindings.values().flatten() {
3148 let key = data_binding_requirement_key(&binding.node_id, &binding.input_name);
3149 if let Some(previous) = expected.insert(key.clone(), binding) {
3150 let previous_coordinates = (&previous.node_id, previous.input_name.as_str());
3151 let coordinates = (&binding.node_id, binding.input_name.as_str());
3152 let detail = if previous_coordinates == coordinates {
3153 "duplicate coordinates"
3154 } else {
3155 "distinct coordinates collide under the V1 node.input spelling"
3156 };
3157 return contract_error(format!(
3158 "training data bindings render duplicate requirement key `{key}`: {detail}"
3159 ));
3160 }
3161 }
3162 let mut actual = BTreeMap::new();
3163 let mut previous: Option<&str> = None;
3164 for identity in &request.data_identities {
3165 identity.validate()?;
3166 if previous.is_some_and(|previous| previous >= identity.requirement_key.as_str()) {
3167 return contract_error(
3168 "training data identities must be strictly sorted by requirement_key".to_string(),
3169 );
3170 }
3171 previous = Some(identity.requirement_key.as_str());
3172 actual.insert(identity.requirement_key.as_str(), identity);
3173 }
3174 if actual.keys().copied().collect::<BTreeSet<_>>()
3175 != expected.keys().map(String::as_str).collect::<BTreeSet<_>>()
3176 {
3177 return contract_error(
3178 "training data identities must exactly cover campaign data bindings".to_string(),
3179 );
3180 }
3181 for (key, binding) in expected {
3182 let identity = actual[&key.as_str()];
3183 if identity.schema_fingerprint != binding.schema_fingerprint
3184 || identity.plan_fingerprint != binding.plan_fingerprint
3185 || binding.relation_fingerprint.as_deref()
3186 != Some(identity.relation_fingerprint.as_str())
3187 {
3188 return contract_error(format!(
3189 "training data identity `{key}` does not match data binding fingerprints"
3190 ));
3191 }
3192 }
3193 Ok(())
3194}
3195
3196fn validate_package_data_identities(package: &PortablePredictorPackage) -> Result<()> {
3197 let expected = package
3198 .execution_bundle
3199 .data_requirements
3200 .iter()
3201 .map(|requirement| (requirement.key(), requirement))
3202 .collect::<BTreeMap<_, _>>();
3203 let mut actual = BTreeMap::new();
3204 let mut previous: Option<&str> = None;
3205 for identity in &package.data_identities {
3206 identity.validate()?;
3207 if previous.is_some_and(|previous| previous >= identity.requirement_key.as_str()) {
3208 return contract_error(
3209 "portable predictor data identities must be sorted by requirement_key".to_string(),
3210 );
3211 }
3212 previous = Some(identity.requirement_key.as_str());
3213 actual.insert(identity.requirement_key.clone(), identity);
3214 }
3215 if actual.keys().collect::<BTreeSet<_>>() != expected.keys().collect::<BTreeSet<_>>() {
3216 return contract_error(
3217 "portable predictor data identities do not exactly match bundle requirements"
3218 .to_string(),
3219 );
3220 }
3221 for (key, requirement) in expected {
3222 let identity = actual[&key];
3223 if identity.schema_fingerprint != requirement.schema_fingerprint
3224 || identity.plan_fingerprint != requirement.plan_fingerprint
3225 || requirement.relation_fingerprint.as_deref()
3226 != Some(identity.relation_fingerprint.as_str())
3227 {
3228 return contract_error(format!(
3229 "portable predictor data identity `{key}` does not match bundle fingerprints"
3230 ));
3231 }
3232 }
3233 Ok(())
3234}
3235
3236fn influence_identity_closure_for_samples(
3237 scope_id: &str,
3238 samples: &BTreeSet<SampleId>,
3239 relations: &SampleRelationSet,
3240) -> Result<(Vec<SampleId>, Vec<GroupId>)> {
3241 let mut found = BTreeSet::new();
3242 let mut origins = BTreeSet::new();
3243 let mut groups = BTreeSet::new();
3244 for relation in &relations.records {
3245 if samples.contains(&relation.sample_id) {
3246 found.insert(&relation.sample_id);
3247 if let Some(origin) = &relation.origin_sample_id {
3248 origins.insert(origin.clone());
3249 }
3250 if let Some(group) = &relation.group_id {
3251 groups.insert(group.clone());
3252 }
3253 }
3254 }
3255 if found.len() != samples.len() {
3256 return contract_error(format!(
3257 "training influence `{scope_id}` contains physical samples absent from relation set"
3258 ));
3259 }
3260 Ok((origins.into_iter().collect(), groups.into_iter().collect()))
3261}
3262
3263fn validate_package_base_influence(
3264 influence: &TrainingInfluenceManifest,
3265 plan: &ExecutionPlan,
3266 closure: &BTreeSet<NodeId>,
3267) -> Result<()> {
3268 let expected = closure
3269 .iter()
3270 .filter(|node_id| {
3271 plan.node_plans[*node_id]
3272 .supported_phases
3273 .contains(&Phase::FitCv)
3274 })
3275 .filter_map(|node_id| base_influence_kind(plan, node_id).map(|kind| (node_id, kind)))
3276 .collect::<BTreeMap<_, _>>();
3277 let base_kinds = [
3278 TrainingInfluenceKind::TransformFit,
3279 TrainingInfluenceKind::ModelFit,
3280 TrainingInfluenceKind::HpoSelection,
3281 TrainingInfluenceKind::TrainedMetaAggregation,
3282 ]
3283 .into_iter()
3284 .collect::<BTreeSet<_>>();
3285 let mut actual = BTreeMap::<&NodeId, Vec<TrainingInfluenceKind>>::new();
3286 for entry in &influence.entries {
3287 if let Some(node_id) = entry.node_id.as_ref() {
3288 if base_kinds.contains(&entry.kind) {
3289 actual.entry(node_id).or_default().push(entry.kind);
3290 }
3291 }
3292 }
3293 if actual.keys().copied().collect::<BTreeSet<_>>()
3294 != expected.keys().copied().collect::<BTreeSet<_>>()
3295 {
3296 return contract_error(
3297 "portable predictor base-influence nodes do not exactly match predictor closure"
3298 .to_string(),
3299 );
3300 }
3301 for (node_id, expected_kind) in expected {
3302 let kinds = &actual[node_id];
3303 if kinds.is_empty() || kinds.iter().any(|kind| *kind != expected_kind) {
3304 return contract_error(format!(
3305 "portable predictor influence node `{node_id}` entries do not all have expected kind `{:?}`",
3306 expected_kind
3307 ));
3308 }
3309 }
3310 Ok(())
3311}
3312
3313fn validate_package_artifact_bindings(package: &PortablePredictorPackage) -> Result<()> {
3314 let mut previous: Option<&ArtifactId> = None;
3315 for binding in &package.artifact_bindings {
3316 if previous.is_some_and(|previous| previous >= &binding.artifact_id) {
3317 return contract_error(
3318 "portable predictor artifact bindings must be strictly sorted by artifact_id"
3319 .to_string(),
3320 );
3321 }
3322 previous = Some(&binding.artifact_id);
3323 }
3324 let expected = package
3325 .execution_bundle
3326 .refit_artifacts
3327 .iter()
3328 .map(|record| record.artifact.id.clone())
3329 .collect::<BTreeSet<_>>();
3330 let actual = package
3331 .artifact_bindings
3332 .iter()
3333 .map(|binding| binding.artifact_id.clone())
3334 .collect::<BTreeSet<_>>();
3335 if actual != expected {
3336 return contract_error(
3337 "portable predictor artifact bindings do not exactly match bundle artifacts"
3338 .to_string(),
3339 );
3340 }
3341 for binding in &package.artifact_bindings {
3342 let record = package
3343 .execution_bundle
3344 .refit_artifacts
3345 .iter()
3346 .find(|record| record.artifact.id == binding.artifact_id)
3347 .expect("artifact id sets were checked above");
3348 let node_plan = &package.effective_plan.node_plans[&record.node_id];
3349 match binding.load_mode {
3350 ArtifactLoadMode::NativePortable => {
3351 record.artifact.validate_portable()?;
3352 if node_plan.artifact_policy == ArtifactPolicy::HostOnly {
3353 return contract_error(format!(
3354 "host_only artifact `{}` cannot be classified native_portable",
3355 binding.artifact_id
3356 ));
3357 }
3358 }
3359 ArtifactLoadMode::HostSidecar => {
3360 if package.fitted_artifact_mode != FittedArtifactMode::AllowHostSidecar {
3361 return contract_error(format!(
3362 "portable_required package forbids host sidecar artifact `{}`",
3363 binding.artifact_id
3364 ));
3365 }
3366 }
3367 }
3368 }
3369 Ok(())
3370}
3371
3372pub(crate) fn contains_runtime_handle(value: &serde_json::Value) -> bool {
3373 match value {
3374 serde_json::Value::Array(values) => values.iter().any(contains_runtime_handle),
3375 serde_json::Value::Object(values) => {
3376 values.keys().any(|key| {
3377 let key = key.to_ascii_lowercase();
3378 key == "handle" || key.ends_with("_handle") || key.ends_with("_handles")
3379 }) || values.values().any(contains_runtime_handle)
3380 }
3381 _ => false,
3382 }
3383}
3384
3385fn tcv1_fingerprint<T: Serialize + ?Sized>(value: &T, label: &str) -> Result<String> {
3386 let json = serde_json::to_string(value)?;
3387 parse_typed_json(&json)
3388 .and_then(|value| value.fingerprint())
3389 .map_err(|error| DagMlError::RuntimeValidation(format!("{label} is outside TCV1: {error}")))
3390}
3391
3392fn tcv1_fingerprint_without<T: Serialize>(value: &T, field: &str, label: &str) -> Result<String> {
3393 let json = serde_json::to_string(value)?;
3394 parse_typed_json(&json)
3395 .and_then(|value| value.fingerprint_without(field))
3396 .map_err(|error| DagMlError::RuntimeValidation(format!("{label} is outside TCV1: {error}")))
3397}
3398
3399fn strict_tcv1_fingerprint_without(json: &str, field: &str, label: &str) -> Result<String> {
3400 parse_typed_json(json)
3401 .and_then(|value| value.fingerprint_without(field))
3402 .map_err(|error| {
3403 DagMlError::RuntimeValidation(format!("{label} is outside strict TCV1: {error}"))
3404 })
3405}
3406
3407fn validate_sha256(label: &str, value: &str) -> Result<()> {
3408 if value.len() != 64
3409 || !value
3410 .bytes()
3411 .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
3412 {
3413 return contract_error(format!(
3414 "{label} fingerprint must be 64 lowercase hexadecimal characters"
3415 ));
3416 }
3417 Ok(())
3418}
3419
3420fn zero_fingerprint() -> String {
3421 "0".repeat(64)
3422}
3423
3424fn validate_identifier_text(label: &str, value: &str) -> Result<()> {
3425 if value.is_empty()
3426 || value.len() > 128
3427 || !value
3428 .bytes()
3429 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':'))
3430 {
3431 return contract_error(format!("{label} is not a valid DAG-ML identifier"));
3432 }
3433 Ok(())
3434}
3435
3436fn validate_non_empty(label: &str, value: &str) -> Result<()> {
3437 if value.trim().is_empty() {
3438 return contract_error(format!("{label} must be non-empty"));
3439 }
3440 Ok(())
3441}
3442
3443fn validate_sorted_unique_text(
3444 label: &str,
3445 values: &[String],
3446 require_non_empty: bool,
3447) -> Result<()> {
3448 if require_non_empty && values.is_empty() {
3449 return contract_error(format!("{label} must be non-empty"));
3450 }
3451 let mut previous: Option<&str> = None;
3452 for value in values {
3453 validate_non_empty(label, value)?;
3454 if previous.is_some_and(|previous| previous >= value.as_str()) {
3455 return contract_error(format!("{label} must be strictly sorted and unique"));
3456 }
3457 previous = Some(value.as_str());
3458 }
3459 Ok(())
3460}
3461
3462fn validate_unique_text(label: &str, values: &[String], require_non_empty: bool) -> Result<()> {
3463 if require_non_empty && values.is_empty() {
3464 return contract_error(format!("{label} must be non-empty"));
3465 }
3466 let mut seen = BTreeSet::new();
3467 for value in values {
3468 validate_non_empty(label, value)?;
3469 if !seen.insert(value.as_str()) {
3470 return contract_error(format!("{label} must be unique"));
3471 }
3472 }
3473 Ok(())
3474}
3475
3476fn validate_sorted_unique_ids<T: Ord + std::fmt::Display>(
3477 label: &str,
3478 values: &[T],
3479 require_non_empty: bool,
3480) -> Result<()> {
3481 if require_non_empty && values.is_empty() {
3482 return contract_error(format!("{label} must be non-empty"));
3483 }
3484 if values.windows(2).any(|pair| pair[0] >= pair[1]) {
3485 return contract_error(format!("{label} must be strictly sorted and unique"));
3486 }
3487 Ok(())
3488}
3489
3490fn unsupported_version<T>(label: &str, actual: u32, expected: u32) -> Result<T> {
3491 contract_error(format!(
3492 "{label} uses unsupported schema_version {actual}, expected {expected}"
3493 ))
3494}
3495
3496fn contract_error<T>(message: String) -> Result<T> {
3497 Err(DagMlError::CampaignValidation(message))
3498}
3499
3500#[cfg(test)]
3501mod tests {
3502 use serde_json::{json, Value};
3503
3504 use super::*;
3505 use crate::ids::{BundleId, ControllerId, ObservationId, VariantId};
3506 use crate::relation::SampleRelation;
3507 use crate::selection::{MetricObjective, SelectionMetric};
3508
3509 fn manifests() -> Vec<ControllerManifest> {
3510 let all: Vec<ControllerManifest> = serde_json::from_str(include_str!(
3511 "../tests/fixtures/package/controller_manifests.json"
3512 ))
3513 .unwrap();
3514 let mut selected = all
3515 .into_iter()
3516 .filter(|manifest| {
3517 matches!(
3518 manifest.operator_kind,
3519 NodeKind::Transform | NodeKind::Model
3520 )
3521 })
3522 .collect::<Vec<_>>();
3523 selected.sort_by(|left, right| left.controller_id.cmp(&right.controller_id));
3524 selected
3525 }
3526
3527 fn data_identity(campaign: &CampaignSpec) -> TrainingDataIdentity {
3528 let binding = &campaign.data_bindings[&NodeId::new("model:base").unwrap()][0];
3529 let mut identity = TrainingDataIdentity {
3530 requirement_key: "model:base.x".to_string(),
3531 schema_fingerprint: binding.schema_fingerprint.clone(),
3532 plan_fingerprint: binding.plan_fingerprint.clone(),
3533 relation_fingerprint: binding.relation_fingerprint.clone().unwrap(),
3534 data_content_fingerprint: "1".repeat(64),
3535 target_content_fingerprint: "2".repeat(64),
3536 identity_fingerprint: zero_fingerprint(),
3537 };
3538 identity.identity_fingerprint = identity.compute_fingerprint().unwrap();
3539 identity
3540 }
3541
3542 fn request() -> TrainingRequest {
3543 let graph: GraphSpec =
3544 serde_json::from_str(include_str!("../tests/fixtures/package/minimal_graph.json"))
3545 .unwrap();
3546 let campaign: CampaignSpec = serde_json::from_str(include_str!(
3547 "../tests/fixtures/package/campaign_oof_generation.json"
3548 ))
3549 .unwrap();
3550 let mut request = TrainingRequest {
3551 schema_version: TRAINING_REQUEST_SCHEMA_VERSION,
3552 request_id: "training:request.test".to_string(),
3553 plan_id: "plan:training.test".to_string(),
3554 graph,
3555 data_identities: vec![data_identity(&campaign)],
3556 campaign,
3557 controller_manifests: manifests(),
3558 parameter_patches: Vec::new(),
3559 patch_policies: Vec::new(),
3560 influence_requirements: Vec::new(),
3561 training_losses: Vec::new(),
3562 options: TrainingOptions {
3563 refit: true,
3564 refit_strategy: Some(RefitStrategy::RefitOne),
3565 seed: 12345,
3566 selection: SelectionPolicy {
3567 id: "selection:rmse".to_string(),
3568 metric: SelectionMetric {
3569 name: "rmse".to_string(),
3570 objective: MetricObjective::Minimize,
3571 },
3572 required_metric_level: None,
3573 require_finite: true,
3574 evaluation_scope: None,
3575 refit_slot_plan: None,
3576 stacking_fit_contract: None,
3577 reduction_id: None,
3578 },
3579 selection_output_id: "output:prediction".to_string(),
3580 outputs: vec![TrainingOutputRequest {
3581 output_id: "output:prediction".to_string(),
3582 node_id: NodeId::new("model:base").unwrap(),
3583 port_name: None,
3584 prediction_level: PredictionLevel::Sample,
3585 unit_level: Some(EntityUnitLevel::PhysicalSample),
3586 prediction_kind: PredictionKind::RegressionPoint,
3587 target_names: vec!["protein".to_string()],
3588 target_units: vec![Some("percent".to_string())],
3589 class_labels: vec![Vec::new()],
3590 output_order: OutputOrder::TargetOrder,
3591 target_space: "raw".to_string(),
3592 }],
3593 scheduler: TrainingSchedulerOptions {
3594 kind: TrainingSchedulerKind::Sequential,
3595 backend: None,
3596 workers: 1,
3597 },
3598 resources: TrainingResourceLimits {
3599 cpu_threads: 1,
3600 memory_bytes: Some(1024),
3601 gpu_devices: Vec::new(),
3602 wall_time_ms: Some(10_000),
3603 },
3604 artifacts: TrainingArtifactOptions {
3605 cv_artifacts: CvArtifactRetention::MetadataOnly,
3606 prediction_caches: PredictionCacheRetention::Retain,
3607 fitted_artifacts: FittedArtifactMode::AllowHostSidecar,
3608 },
3609 },
3610 request_fingerprint: zero_fingerprint(),
3611 };
3612 request.request_fingerprint = request.compute_fingerprint().unwrap();
3613 request
3614 }
3615
3616 fn resign_request(request: &mut TrainingRequest) {
3617 request.request_fingerprint = zero_fingerprint();
3618 request.request_fingerprint = request.compute_fingerprint().unwrap();
3619 }
3620
3621 #[test]
3622 fn training_request_projects_identically_for_refit_on_and_off() {
3623 let request = request();
3624 let refit = request.project().unwrap();
3625 assert_eq!(refit.outputs[0].port_name, "oof");
3626 assert!(!refit.parameters.requires_recompile);
3627 assert_eq!(
3628 refit.predictor_node_ids,
3629 BTreeSet::from([
3630 NodeId::new("model:base").unwrap(),
3631 NodeId::new("transform:snv").unwrap(),
3632 ])
3633 );
3634
3635 let mut no_refit = request;
3636 no_refit.options.refit = false;
3637 no_refit.options.refit_strategy = None;
3638 resign_request(&mut no_refit);
3639 let projection = no_refit.project().unwrap();
3640 assert_eq!(projection.outputs, refit.outputs);
3641 assert_eq!(projection.plan, refit.plan);
3642 }
3643
3644 #[test]
3645 fn training_request_rejects_duplicate_rendered_data_requirement_keys() {
3646 let mut request = request();
3647 let node_id = NodeId::new("model:base").unwrap();
3648 let duplicate = request.campaign.data_bindings[&node_id][0].clone();
3649 request
3650 .campaign
3651 .data_bindings
3652 .get_mut(&node_id)
3653 .unwrap()
3654 .push(duplicate);
3655 resign_request(&mut request);
3656
3657 let error = request.project().unwrap_err();
3658 assert!(error
3659 .to_string()
3660 .contains("render duplicate requirement key `model:base.x`"));
3661 }
3662
3663 #[test]
3664 fn training_options_reject_unknown_fields_and_binary64_integer_substitution() {
3665 let value = serde_json::to_value(request()).unwrap();
3666 let mut unknown = value.clone();
3667 unknown["options"]["mystery"] = json!(true);
3668 let error = serde_json::from_value::<TrainingRequest>(unknown).unwrap_err();
3669 assert!(error.to_string().contains("unknown field"));
3670
3671 let mut binary64_seed = value;
3672 binary64_seed["options"]["seed"] = json!(12345.0);
3673 assert!(serde_json::from_value::<TrainingRequest>(binary64_seed).is_err());
3674 }
3675
3676 #[test]
3677 fn training_request_from_json_requires_explicit_patch_collections() {
3678 let request_json = serde_json::to_value(request()).unwrap();
3679 for field in ["parameter_patches", "patch_policies"] {
3680 let mut missing = request_json.clone();
3681 missing.as_object_mut().unwrap().remove(field);
3682 let error = TrainingRequest::from_json(&serde_json::to_string(&missing).unwrap())
3683 .expect_err("required patch collection omission must fail closed");
3684 assert!(error.to_string().contains(field), "{field}: {error}");
3685 }
3686 }
3687
3688 #[test]
3695 fn required_nullable_fields_reject_omission_but_accept_explicit_null() {
3696 let base = request();
3697
3698 let output = serde_json::to_value(&base.options.outputs[0]).unwrap();
3700 assert!(output.get("unit_level").is_some());
3701 let mut missing = output.clone();
3702 missing.as_object_mut().unwrap().remove("unit_level");
3703 let error = serde_json::from_value::<TrainingOutputRequest>(missing).unwrap_err();
3704 assert!(
3705 error.to_string().contains("missing field") && error.to_string().contains("unit_level"),
3706 "unit_level omission: {error}"
3707 );
3708 let mut null_unit = output;
3709 null_unit["unit_level"] = json!(null);
3710 assert!(serde_json::from_value::<TrainingOutputRequest>(null_unit)
3711 .unwrap()
3712 .unit_level
3713 .is_none());
3714
3715 let resources = serde_json::to_value(&base.options.resources).unwrap();
3717 assert_eq!(resources["gpu_devices"], json!([]));
3718 let mut missing = resources.clone();
3719 missing.as_object_mut().unwrap().remove("gpu_devices");
3720 let error = serde_json::from_value::<TrainingResourceLimits>(missing).unwrap_err();
3721 assert!(
3722 error.to_string().contains("missing field")
3723 && error.to_string().contains("gpu_devices"),
3724 "gpu_devices omission: {error}"
3725 );
3726 assert!(serde_json::from_value::<TrainingResourceLimits>(resources)
3727 .unwrap()
3728 .gpu_devices
3729 .is_empty());
3730
3731 let options = serde_json::to_value(&base.options).unwrap();
3733 assert!(options.get("refit_strategy").is_some());
3734 let mut missing = options.clone();
3735 missing.as_object_mut().unwrap().remove("refit_strategy");
3736 let error = serde_json::from_value::<TrainingOptions>(missing).unwrap_err();
3737 assert!(
3738 error.to_string().contains("missing field")
3739 && error.to_string().contains("refit_strategy"),
3740 "refit_strategy omission: {error}"
3741 );
3742 let mut null_strategy = options;
3743 null_strategy["refit_strategy"] = json!(null);
3744 assert!(serde_json::from_value::<TrainingOptions>(null_strategy)
3745 .unwrap()
3746 .refit_strategy
3747 .is_none());
3748
3749 let requirement = serde_json::to_value(ControllerInfluenceRequirement {
3751 node_id: NodeId::new("model:base").unwrap(),
3752 kind: TrainingInfluenceKind::EarlyStopping,
3753 scope_id: "early:fold:0".to_string(),
3754 phase: Phase::FitCv,
3755 fold_id: Some(FoldId::new("fold:0").unwrap()),
3756 physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
3757 })
3758 .unwrap();
3759 assert!(requirement.get("fold_id").is_some());
3760 let mut missing = requirement.clone();
3761 missing.as_object_mut().unwrap().remove("fold_id");
3762 let error = serde_json::from_value::<ControllerInfluenceRequirement>(missing).unwrap_err();
3763 assert!(
3764 error.to_string().contains("missing field") && error.to_string().contains("fold_id"),
3765 "fold_id omission: {error}"
3766 );
3767 let mut null_fold = requirement;
3768 null_fold["fold_id"] = json!(null);
3769 assert!(
3770 serde_json::from_value::<ControllerInfluenceRequirement>(null_fold)
3771 .unwrap()
3772 .fold_id
3773 .is_none()
3774 );
3775
3776 let entry = serde_json::to_value(TrainingInfluenceEntry {
3778 kind: TrainingInfluenceKind::ModelFit,
3779 scope_id: "model:base:fold:0".to_string(),
3780 node_id: Some(NodeId::new("model:base").unwrap()),
3781 physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
3782 origin_sample_ids: Vec::new(),
3783 group_ids: Vec::new(),
3784 })
3785 .unwrap();
3786 assert!(entry.get("node_id").is_some());
3787 let mut missing = entry.clone();
3788 missing.as_object_mut().unwrap().remove("node_id");
3789 let error = serde_json::from_value::<TrainingInfluenceEntry>(missing).unwrap_err();
3790 assert!(
3791 error.to_string().contains("missing field") && error.to_string().contains("node_id"),
3792 "node_id omission: {error}"
3793 );
3794 let mut null_node = entry;
3795 null_node["node_id"] = json!(null);
3796 assert!(serde_json::from_value::<TrainingInfluenceEntry>(null_node)
3797 .unwrap()
3798 .node_id
3799 .is_none());
3800 }
3801
3802 #[test]
3803 fn output_resolution_rejects_no_output_ambiguity_and_non_prediction_ports() {
3804 let request = request();
3805 let output = &request.options.outputs[0];
3806 let mut no_output = request.graph.clone();
3807 no_output.nodes[1].ports.outputs.clear();
3808 assert!(output
3809 .resolve(&no_output)
3810 .unwrap_err()
3811 .to_string()
3812 .contains("no prediction output"));
3813
3814 let mut ambiguous = request.graph.clone();
3815 let mut second = ambiguous.nodes[1].ports.outputs[0].clone();
3816 second.name = "probability".to_string();
3817 ambiguous.nodes[1].ports.outputs.push(second);
3818 assert!(output
3819 .resolve(&ambiguous)
3820 .unwrap_err()
3821 .to_string()
3822 .contains("multiple prediction outputs"));
3823
3824 let mut explicit = output.clone();
3825 explicit.port_name = Some("x".to_string());
3826 assert!(explicit.resolve(&request.graph).is_err());
3827 }
3828
3829 #[test]
3830 fn output_target_and_class_orders_are_semantic_not_lexically_sorted() {
3831 let graph = request().graph;
3832 let mut binding = OutputBinding {
3833 schema_version: OUTPUT_BINDING_SCHEMA_VERSION,
3834 binding_id: "output:ordered".to_string(),
3835 node_id: NodeId::new("model:base").unwrap(),
3836 port_name: "oof".to_string(),
3837 prediction_level: PredictionLevel::Sample,
3838 unit_level: Some(EntityUnitLevel::PhysicalSample),
3839 prediction_kind: PredictionKind::RegressionPoint,
3840 prediction_source: PredictionSource::FinalRefit,
3841 refit_strategy: Some(RefitStrategy::RefitOne),
3842 aggregation_fingerprint: "6".repeat(64),
3843 target_names: vec!["z_target".to_string(), "a_target".to_string()],
3844 target_units: vec![Some("z_unit".to_string()), Some("a_unit".to_string())],
3845 class_labels: vec![Vec::new(), Vec::new()],
3846 output_order: OutputOrder::TargetOrder,
3847 target_space: "raw".to_string(),
3848 binding_fingerprint: zero_fingerprint(),
3849 };
3850 binding.binding_fingerprint = binding.compute_fingerprint().unwrap();
3851 binding.validate(&graph).unwrap();
3852 let original = binding.binding_fingerprint.clone();
3853 binding.target_names.swap(0, 1);
3854 binding.binding_fingerprint = zero_fingerprint();
3855 binding.binding_fingerprint = binding.compute_fingerprint().unwrap();
3856 binding.validate(&graph).unwrap();
3857 assert_ne!(original, binding.binding_fingerprint);
3858 }
3859
3860 #[test]
3861 fn output_unit_levels_and_class_vocabularies_match_w0_contract() {
3862 let request = request();
3863 let graph = &request.graph;
3864 let mut output = request.options.outputs[0].clone();
3865
3866 output.unit_level = None;
3867 assert!(output
3868 .resolve(graph)
3869 .unwrap_err()
3870 .to_string()
3871 .contains("physical_sample"));
3872
3873 output.prediction_level = PredictionLevel::Target;
3874 output.unit_level = Some(EntityUnitLevel::PhysicalSample);
3875 assert!(output
3876 .resolve(graph)
3877 .unwrap_err()
3878 .to_string()
3879 .contains("unit_level=null"));
3880 output.unit_level = None;
3881 let target_wire = serde_json::to_value(&output).unwrap();
3882 assert_eq!(target_wire.get("unit_level"), Some(&Value::Null));
3883 output.resolve(graph).unwrap();
3884
3885 output.prediction_level = PredictionLevel::Sample;
3886 output.unit_level = Some(EntityUnitLevel::PhysicalSample);
3887 output.prediction_kind = PredictionKind::ClassLabel;
3888 output.class_labels = vec![Vec::new()];
3889 output.output_order = OutputOrder::TargetOrder;
3890 output.resolve(graph).unwrap();
3891 output.class_labels = vec![vec!["low".to_string(), "high".to_string()]];
3892 output.resolve(graph).unwrap();
3893
3894 output.prediction_kind = PredictionKind::DecisionScore;
3895 output.resolve(graph).unwrap();
3896 output.class_labels = vec![Vec::new()];
3897 output.resolve(graph).unwrap();
3898
3899 output.prediction_kind = PredictionKind::RegressionPoint;
3900 output.class_labels = vec![vec!["not-a-regression-class".to_string()]];
3901 assert!(output.resolve(graph).is_err());
3902
3903 output.prediction_kind = PredictionKind::ClassProbability;
3904 output.output_order = OutputOrder::TargetMajorClassMinor;
3905 output.class_labels = vec![Vec::new()];
3906 assert!(output.resolve(graph).is_err());
3907 output.class_labels = vec![vec!["low".to_string(), "high".to_string()]];
3908 output.resolve(graph).unwrap();
3909 }
3910
3911 #[test]
3912 fn selection_output_metric_matrix_is_explicit() {
3913 let mut level_mismatch = request();
3914 level_mismatch.options.outputs[0].prediction_level = PredictionLevel::Target;
3915 level_mismatch.options.outputs[0].unit_level = None;
3916 assert!(level_mismatch
3917 .options
3918 .selection
3919 .required_metric_level
3920 .is_none());
3921 resign_request(&mut level_mismatch);
3922 assert!(level_mismatch
3923 .validate()
3924 .unwrap_err()
3925 .to_string()
3926 .contains("campaign selection_metric_level"));
3927
3928 let mut regression = request();
3929 regression.options.selection.metric.objective = MetricObjective::Maximize;
3930 resign_request(&mut regression);
3931 assert!(regression
3932 .validate()
3933 .unwrap_err()
3934 .to_string()
3935 .contains("not supported for RegressionPoint"));
3936
3937 let mut class_label = request();
3938 class_label.options.outputs[0].prediction_kind = PredictionKind::ClassLabel;
3939 class_label.options.outputs[0].class_labels =
3940 vec![vec!["low".to_string(), "high".to_string()]];
3941 class_label.options.selection.metric.name = "accuracy".to_string();
3942 class_label.options.selection.metric.objective = MetricObjective::Maximize;
3943 resign_request(&mut class_label);
3944 class_label.validate().unwrap();
3945
3946 let mut probability = class_label.clone();
3947 probability.options.outputs[0].prediction_kind = PredictionKind::ClassProbability;
3948 probability.options.outputs[0].output_order = OutputOrder::TargetMajorClassMinor;
3949 resign_request(&mut probability);
3950 assert!(probability
3951 .validate()
3952 .unwrap_err()
3953 .to_string()
3954 .contains("not supported for ClassProbability"));
3955
3956 let mut decision = class_label;
3957 decision.options.outputs[0].prediction_kind = PredictionKind::DecisionScore;
3958 resign_request(&mut decision);
3959 assert!(decision
3960 .validate()
3961 .unwrap_err()
3962 .to_string()
3963 .contains("not supported for DecisionScore"));
3964 }
3965
3966 fn request_with_nested_params() -> TrainingRequest {
3967 let mut request = request();
3968 let model = request
3969 .graph
3970 .nodes
3971 .iter_mut()
3972 .find(|node| node.id.as_str() == "model:base")
3973 .unwrap();
3974 model.params.insert(
3975 "nested".to_string(),
3976 json!({"depth": {"alpha": 1}, "array": [1, 2]}),
3977 );
3978 resign_request(&mut request);
3979 request
3980 }
3981
3982 fn patch(namespace: ParameterNamespace, path: &[&str], value: Value) -> ParameterPatch {
3983 ParameterPatch {
3984 schema_version: PARAMETER_PATCH_SCHEMA_VERSION,
3985 node_id: NodeId::new("model:base").unwrap(),
3986 namespace,
3987 path: path.iter().map(|part| (*part).to_string()).collect(),
3988 value,
3989 }
3990 }
3991
3992 fn patch_policy(namespaces: &[ParameterNamespace]) -> NodePatchPolicy {
3993 NodePatchPolicy {
3994 node_id: NodeId::new("model:base").unwrap(),
3995 allowed_namespaces: namespaces.iter().copied().collect(),
3996 }
3997 }
3998
3999 #[test]
4000 fn namespaced_deep_patch_is_isolated_bijective_and_structural() {
4001 let plan = request_with_nested_params().project().unwrap().plan;
4002 let original = plan.node_plans[&NodeId::new("model:base").unwrap()]
4003 .params
4004 .clone();
4005 let patches = vec![
4006 patch(
4007 ParameterNamespace::Operator,
4008 &["nested", "depth", "alpha"],
4009 json!(2),
4010 ),
4011 patch(ParameterNamespace::Fit, &["epochs"], json!(12)),
4012 patch(ParameterNamespace::Structural, &["topology"], json!("wide")),
4013 ];
4014 let projection = project_parameter_patches(
4015 &plan,
4016 &patches,
4017 &[patch_policy(&[
4018 ParameterNamespace::Operator,
4019 ParameterNamespace::Fit,
4020 ParameterNamespace::Structural,
4021 ])],
4022 )
4023 .unwrap();
4024 let node = &projection.nodes[&NodeId::new("model:base").unwrap()];
4025 assert_eq!(node.params["nested"]["depth"]["alpha"], json!(2));
4026 assert_eq!(node.fit_params["epochs"], json!(12));
4027 assert_eq!(node.structural_params["topology"], json!("wide"));
4028 assert!(projection.requires_recompile);
4029 assert_eq!(
4030 plan.node_plans[&NodeId::new("model:base").unwrap()].params,
4031 original
4032 );
4033 assert_eq!(ParameterNamespace::Operator.plan_root(), "params");
4034 assert_eq!(ParameterNamespace::Fit.plan_root(), "fit_params");
4035 assert_eq!(ParameterNamespace::Control.plan_root(), "control_params");
4036 assert_eq!(
4037 ParameterNamespace::Structural.plan_root(),
4038 "structural_params"
4039 );
4040
4041 let mut dishonest = projection.clone();
4042 dishonest.requires_recompile = false;
4043 dishonest.projection_fingerprint = zero_fingerprint();
4044 dishonest.projection_fingerprint = dishonest.compute_fingerprint().unwrap();
4045 assert!(dishonest.validate().is_err());
4046 }
4047
4048 #[test]
4049 fn patch_projection_rejects_namespace_order_duplicates_parent_child_and_arrays() {
4050 let plan = request_with_nested_params().project().unwrap().plan;
4051 let operator = patch_policy(&[ParameterNamespace::Operator]);
4052
4053 let forbidden = patch(ParameterNamespace::Fit, &["epochs"], json!(3));
4054 assert!(
4055 project_parameter_patches(&plan, &[forbidden], std::slice::from_ref(&operator))
4056 .unwrap_err()
4057 .to_string()
4058 .contains("forbidden")
4059 );
4060
4061 let duplicate = patch(ParameterNamespace::Operator, &["x"], json!(1));
4062 assert!(project_parameter_patches(
4063 &plan,
4064 &[duplicate.clone(), duplicate],
4065 std::slice::from_ref(&operator)
4066 )
4067 .is_err());
4068
4069 let parent = patch(
4070 ParameterNamespace::Operator,
4071 &["nested", "depth"],
4072 json!({"alpha": 2}),
4073 );
4074 let child = patch(
4075 ParameterNamespace::Operator,
4076 &["nested", "depth", "alpha"],
4077 json!(3),
4078 );
4079 assert!(project_parameter_patches(
4080 &plan,
4081 &[parent, child],
4082 std::slice::from_ref(&operator),
4083 )
4084 .unwrap_err()
4085 .to_string()
4086 .contains("parent/child"));
4087
4088 let array = patch(
4089 ParameterNamespace::Operator,
4090 &["nested", "array", "0"],
4091 json!(9),
4092 );
4093 assert!(
4094 project_parameter_patches(&plan, &[array], std::slice::from_ref(&operator)).is_err()
4095 );
4096
4097 let out_of_order = vec![
4098 patch(ParameterNamespace::Operator, &["z"], json!(1)),
4099 patch(ParameterNamespace::Operator, &["a"], json!(1)),
4100 ];
4101 assert!(project_parameter_patches(&plan, &out_of_order, &[operator]).is_err());
4102
4103 assert!(project_parameter_patches(
4104 &plan,
4105 &[],
4106 &[patch_policy(&[ParameterNamespace::Operator])]
4107 )
4108 .is_err());
4109 }
4110
4111 #[test]
4112 fn patch_tcv1_distinguishes_integer_and_binary64() {
4113 let integer = vec![patch(ParameterNamespace::Operator, &["x"], json!(2))];
4114 let binary64 = vec![patch(ParameterNamespace::Operator, &["x"], json!(2.0))];
4115 assert_ne!(
4116 tcv1_fingerprint(&integer, "integer patch").unwrap(),
4117 tcv1_fingerprint(&binary64, "binary64 patch").unwrap()
4118 );
4119 }
4120
4121 fn cache_namespace() -> CacheNamespace {
4122 let mut namespace = CacheNamespace {
4123 schema_version: CACHE_NAMESPACE_SCHEMA_VERSION,
4124 prediction_requirement_key: bundle_prediction_requirement_key(
4125 &NodeId::new("model:base").unwrap(),
4126 "oof",
4127 &NodeId::new("model:meta").unwrap(),
4128 "stacked",
4129 ),
4130 data_requirement_key: "model:base.x".to_string(),
4131 producer_node_id: NodeId::new("model:base").unwrap(),
4132 source_port_name: "oof".to_string(),
4133 consumer_node_id: NodeId::new("model:meta").unwrap(),
4134 target_port_name: "stacked".to_string(),
4135 phase: Phase::FitCv,
4136 params_fingerprint: "a".repeat(64),
4137 data_identity_fingerprint: "b".repeat(64),
4138 fold_id: FoldId::new("fold:0").unwrap(),
4139 trial_id: "trial:0".to_string(),
4140 seed: 7,
4141 namespace_fingerprint: zero_fingerprint(),
4142 };
4143 namespace.namespace_fingerprint = namespace.compute_fingerprint().unwrap();
4144 namespace
4145 }
4146
4147 #[test]
4148 fn cache_namespace_is_candidate_dataset_fold_trial_and_seed_specific() {
4149 let identity = request().data_identities.remove(0);
4150 let mut base = cache_namespace();
4151 base.data_identity_fingerprint = identity.identity_fingerprint.clone();
4152 base.namespace_fingerprint = zero_fingerprint();
4153 base.namespace_fingerprint = base.compute_fingerprint().unwrap();
4154 base.validate_for_identity(&identity).unwrap();
4155 for mutation in 0..5 {
4156 let mut changed = base.clone();
4157 match mutation {
4158 0 => changed.params_fingerprint = "d".repeat(64),
4159 1 => changed.data_identity_fingerprint = "e".repeat(64),
4160 2 => changed.fold_id = FoldId::new("fold:1").unwrap(),
4161 3 => changed.trial_id = "trial:1".to_string(),
4162 _ => changed.seed += 1,
4163 }
4164 changed.namespace_fingerprint = zero_fingerprint();
4165 changed.namespace_fingerprint = changed.compute_fingerprint().unwrap();
4166 changed.validate().unwrap();
4167 assert_ne!(base.namespace_fingerprint, changed.namespace_fingerprint);
4168 }
4169
4170 let mut value = serde_json::to_value(&base).unwrap();
4171 value["seed"] = json!(7.0);
4172 assert!(serde_json::from_value::<CacheNamespace>(value).is_err());
4173
4174 let mut relation_changed = identity.clone();
4175 relation_changed.relation_fingerprint = "d".repeat(64);
4176 relation_changed.identity_fingerprint = zero_fingerprint();
4177 relation_changed.identity_fingerprint = relation_changed.compute_fingerprint().unwrap();
4178 assert!(base.validate_for_identity(&relation_changed).is_err());
4179 let mut other_dataset_namespace = base.clone();
4180 other_dataset_namespace.data_identity_fingerprint =
4181 relation_changed.identity_fingerprint.clone();
4182 other_dataset_namespace.namespace_fingerprint = zero_fingerprint();
4183 other_dataset_namespace.namespace_fingerprint =
4184 other_dataset_namespace.compute_fingerprint().unwrap();
4185 assert_ne!(
4186 base.namespace_fingerprint,
4187 other_dataset_namespace.namespace_fingerprint
4188 );
4189
4190 let mut other_output = base.clone();
4191 other_output.source_port_name = "probability".to_string();
4192 other_output.prediction_requirement_key = bundle_prediction_requirement_key(
4193 &other_output.producer_node_id,
4194 &other_output.source_port_name,
4195 &other_output.consumer_node_id,
4196 &other_output.target_port_name,
4197 );
4198 other_output.namespace_fingerprint = zero_fingerprint();
4199 other_output.namespace_fingerprint = other_output.compute_fingerprint().unwrap();
4200 other_output.validate().unwrap();
4201 assert_ne!(
4202 base.namespace_fingerprint,
4203 other_output.namespace_fingerprint
4204 );
4205
4206 let mut wrong_phase = base.clone();
4207 wrong_phase.phase = Phase::Refit;
4208 wrong_phase.namespace_fingerprint = zero_fingerprint();
4209 wrong_phase.namespace_fingerprint = wrong_phase.compute_fingerprint().unwrap();
4210 assert!(wrong_phase.validate().is_err());
4211 }
4212
4213 fn relations() -> SampleRelationSet {
4214 let records = (1..=4)
4215 .map(|index| {
4216 let mut relation = SampleRelation::new(
4217 ObservationId::new(format!("observation:{index}")).unwrap(),
4218 SampleId::new(format!("sample:{index}")).unwrap(),
4219 );
4220 relation.group_id =
4221 Some(GroupId::new(if index <= 2 { "group:0" } else { "group:1" }).unwrap());
4222 relation
4223 })
4224 .collect();
4225 SampleRelationSet { records }
4226 }
4227
4228 fn request_for_relations(relations: &SampleRelationSet) -> TrainingRequest {
4229 let mut request = request();
4230 let fingerprint = relations.fingerprint().unwrap();
4231 request
4232 .campaign
4233 .data_bindings
4234 .get_mut(&NodeId::new("model:base").unwrap())
4235 .unwrap()[0]
4236 .relation_fingerprint = Some(fingerprint.clone());
4237 request.data_identities[0].relation_fingerprint = fingerprint;
4238 request.data_identities[0].identity_fingerprint = zero_fingerprint();
4239 request.data_identities[0].identity_fingerprint =
4240 request.data_identities[0].compute_fingerprint().unwrap();
4241 resign_request(&mut request);
4242 request
4243 }
4244
4245 fn influence_manifest(
4246 request: &TrainingRequest,
4247 projection: &TrainingContractProjection,
4248 relations: &SampleRelationSet,
4249 ) -> TrainingInfluenceManifest {
4250 let expected = expected_influence_coordinates(
4251 request,
4252 &projection.plan,
4253 &projection.predictor_node_ids,
4254 )
4255 .unwrap();
4256 let entries = expected
4257 .into_iter()
4258 .map(|((kind, scope_id, node_id), samples)| {
4259 let groups = relations
4260 .records
4261 .iter()
4262 .filter(|relation| samples.contains(&relation.sample_id))
4263 .filter_map(|relation| relation.group_id.clone())
4264 .collect::<BTreeSet<_>>()
4265 .into_iter()
4266 .collect();
4267 TrainingInfluenceEntry {
4268 kind,
4269 scope_id,
4270 node_id,
4271 physical_sample_ids: samples.into_iter().collect(),
4272 origin_sample_ids: Vec::new(),
4273 group_ids: groups,
4274 }
4275 })
4276 .collect();
4277 let mut manifest = TrainingInfluenceManifest {
4278 schema_version: TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
4279 relation_fingerprint: relations.fingerprint().unwrap(),
4280 entries,
4281 manifest_fingerprint: zero_fingerprint(),
4282 };
4283 manifest.manifest_fingerprint = manifest.compute_fingerprint().unwrap();
4284 manifest
4285 }
4286
4287 fn resign_manifest(manifest: &mut TrainingInfluenceManifest) {
4288 manifest.manifest_fingerprint = zero_fingerprint();
4289 manifest.manifest_fingerprint = manifest.compute_fingerprint().unwrap();
4290 }
4291
4292 fn early_stopping_requirements() -> Vec<ControllerInfluenceRequirement> {
4293 vec![
4294 ControllerInfluenceRequirement {
4295 node_id: NodeId::new("model:base").unwrap(),
4296 kind: TrainingInfluenceKind::EarlyStopping,
4297 scope_id: "early:fold:0".to_string(),
4298 phase: Phase::FitCv,
4299 fold_id: Some(FoldId::new("fold:0").unwrap()),
4300 physical_sample_ids: vec![SampleId::new("sample:3").unwrap()],
4301 },
4302 ControllerInfluenceRequirement {
4303 node_id: NodeId::new("model:base").unwrap(),
4304 kind: TrainingInfluenceKind::EarlyStopping,
4305 scope_id: "early:fold:1".to_string(),
4306 phase: Phase::FitCv,
4307 fold_id: Some(FoldId::new("fold:1").unwrap()),
4308 physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
4309 },
4310 ControllerInfluenceRequirement {
4311 node_id: NodeId::new("model:base").unwrap(),
4312 kind: TrainingInfluenceKind::EarlyStopping,
4313 scope_id: "early:refit".to_string(),
4314 phase: Phase::Refit,
4315 fold_id: None,
4316 physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
4317 },
4318 ]
4319 }
4320
4321 fn full_scope_requirements(
4322 kind: TrainingInfluenceKind,
4323 prefix: &str,
4324 ) -> Vec<ControllerInfluenceRequirement> {
4325 vec![
4326 ControllerInfluenceRequirement {
4327 node_id: NodeId::new("model:base").unwrap(),
4328 kind,
4329 scope_id: format!("{prefix}:fold:0"),
4330 phase: Phase::FitCv,
4331 fold_id: Some(FoldId::new("fold:0").unwrap()),
4332 physical_sample_ids: vec![
4333 SampleId::new("sample:3").unwrap(),
4334 SampleId::new("sample:4").unwrap(),
4335 ],
4336 },
4337 ControllerInfluenceRequirement {
4338 node_id: NodeId::new("model:base").unwrap(),
4339 kind,
4340 scope_id: format!("{prefix}:fold:1"),
4341 phase: Phase::FitCv,
4342 fold_id: Some(FoldId::new("fold:1").unwrap()),
4343 physical_sample_ids: vec![
4344 SampleId::new("sample:1").unwrap(),
4345 SampleId::new("sample:2").unwrap(),
4346 ],
4347 },
4348 ControllerInfluenceRequirement {
4349 node_id: NodeId::new("model:base").unwrap(),
4350 kind,
4351 scope_id: format!("{prefix}:refit"),
4352 phase: Phase::Refit,
4353 fold_id: None,
4354 physical_sample_ids: (1..=4)
4355 .map(|index| SampleId::new(format!("sample:{index}")).unwrap())
4356 .collect(),
4357 },
4358 ]
4359 }
4360
4361 #[test]
4362 fn influence_evidence_is_capability_complete_and_relation_closed() {
4363 let relations = relations();
4364 let request = request_for_relations(&relations);
4365 let projection = request.project().unwrap();
4366 let manifest = influence_manifest(&request, &projection, &relations);
4367 assert_eq!(manifest.entries.len(), 7);
4368 manifest
4369 .validate_for_projection(&projection, &request, &relations)
4370 .unwrap();
4371
4372 let mut missing = manifest.clone();
4373 missing.entries.remove(1);
4374 resign_manifest(&mut missing);
4375 assert!(missing
4376 .validate_for_projection(&projection, &request, &relations)
4377 .unwrap_err()
4378 .to_string()
4379 .contains("phase scopes"));
4380
4381 let mut wrong_group = manifest;
4382 wrong_group.entries[0].group_ids.pop();
4383 resign_manifest(&mut wrong_group);
4384 assert!(wrong_group
4385 .validate_for_projection(&projection, &request, &relations)
4386 .unwrap_err()
4387 .to_string()
4388 .contains("group closure"));
4389 }
4390
4391 #[test]
4392 fn influence_capabilities_require_every_fold_and_refit_scope_and_refuse_extra() {
4393 let relations = relations();
4394 let mut request = request_for_relations(&relations);
4395 let model_manifest = request
4396 .controller_manifests
4397 .iter_mut()
4398 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4399 .unwrap();
4400 model_manifest
4401 .capabilities
4402 .insert(ControllerCapability::UsesEarlyStopping);
4403 request.influence_requirements = early_stopping_requirements();
4404 resign_request(&mut request);
4405 let projection = request.project().unwrap();
4406 let mut manifest = influence_manifest(&request, &projection, &relations);
4407 assert_eq!(
4408 manifest
4409 .entries
4410 .iter()
4411 .filter(|entry| entry.kind == TrainingInfluenceKind::EarlyStopping)
4412 .count(),
4413 3
4414 );
4415 let removed = manifest
4416 .entries
4417 .iter()
4418 .position(|entry| entry.kind == TrainingInfluenceKind::EarlyStopping)
4419 .unwrap();
4420 manifest.entries.remove(removed);
4421 resign_manifest(&mut manifest);
4422 assert!(manifest
4423 .validate_for_projection(&projection, &request, &relations)
4424 .unwrap_err()
4425 .to_string()
4426 .contains("phase scopes"));
4427
4428 let mut leaked_request = request.clone();
4429 leaked_request.influence_requirements[0].physical_sample_ids =
4430 vec![SampleId::new("sample:1").unwrap()];
4431 resign_request(&mut leaked_request);
4432 assert!(leaked_request
4433 .project()
4434 .unwrap_err()
4435 .to_string()
4436 .contains("outer validation"));
4437
4438 let base_request = request_for_relations(&relations);
4439 let base_projection = base_request.project().unwrap();
4440 let mut extra = influence_manifest(&base_request, &base_projection, &relations);
4441 let model_entry = extra
4442 .entries
4443 .iter()
4444 .find(|entry| {
4445 entry.kind == TrainingInfluenceKind::ModelFit
4446 && entry.scope_id.starts_with("fit_cv:")
4447 })
4448 .unwrap()
4449 .clone();
4450 extra.entries.push(TrainingInfluenceEntry {
4451 kind: TrainingInfluenceKind::EarlyStopping,
4452 ..model_entry
4453 });
4454 extra.entries.sort_by(|left, right| {
4455 (left.kind, left.scope_id.as_str(), left.node_id.as_ref()).cmp(&(
4456 right.kind,
4457 right.scope_id.as_str(),
4458 right.node_id.as_ref(),
4459 ))
4460 });
4461 resign_manifest(&mut extra);
4462 assert!(extra
4463 .validate_for_projection(&base_projection, &base_request, &relations)
4464 .unwrap_err()
4465 .to_string()
4466 .contains("undeclared coordinate"));
4467 }
4468
4469 #[test]
4470 fn influence_requirement_cannot_claim_a_capability_the_controller_lacks() {
4471 let mut request = request();
4472 request.influence_requirements = early_stopping_requirements();
4473 resign_request(&mut request);
4474 assert!(request
4475 .project()
4476 .unwrap_err()
4477 .to_string()
4478 .contains("not required by active controller capabilities"));
4479 }
4480
4481 #[test]
4482 fn influence_capability_matrix_covers_weights_internal_tuning_and_trained_aggregation() {
4483 let relations = relations();
4484 for (capability, kind, prefix) in [
4485 (
4486 ControllerCapability::UsesTrainingWeights,
4487 TrainingInfluenceKind::WeightingResampling,
4488 "weighting",
4489 ),
4490 (
4491 ControllerCapability::PerformsInternalTuning,
4492 TrainingInfluenceKind::HpoSelection,
4493 "internal_hpo",
4494 ),
4495 ] {
4496 let mut request = request_for_relations(&relations);
4497 let model = request
4498 .controller_manifests
4499 .iter_mut()
4500 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4501 .unwrap();
4502 model.capabilities.insert(capability);
4503 if capability == ControllerCapability::UsesTrainingWeights {
4504 model
4505 .capabilities
4506 .insert(ControllerCapability::SupportsSampleWeights);
4507 }
4508 request.influence_requirements = full_scope_requirements(kind, prefix);
4509 resign_request(&mut request);
4510 let projection = request.project().unwrap();
4511 let mut manifest = influence_manifest(&request, &projection, &relations);
4512 assert_eq!(
4513 manifest
4514 .entries
4515 .iter()
4516 .filter(|entry| entry.kind == kind && entry.node_id.is_some())
4517 .count(),
4518 3
4519 );
4520 manifest
4521 .validate_for_projection(&projection, &request, &relations)
4522 .unwrap();
4523 let removed = manifest
4524 .entries
4525 .iter()
4526 .position(|entry| entry.kind == kind && entry.node_id.is_some())
4527 .unwrap();
4528 manifest.entries.remove(removed);
4529 resign_manifest(&mut manifest);
4530 assert!(manifest
4531 .validate_for_projection(&projection, &request, &relations)
4532 .unwrap_err()
4533 .to_string()
4534 .contains("phase scopes"));
4535 }
4536
4537 let mut aggregation = request_for_relations(&relations);
4538 let model = aggregation
4539 .controller_manifests
4540 .iter_mut()
4541 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4542 .unwrap();
4543 model
4544 .capabilities
4545 .insert(ControllerCapability::TrainsAggregation);
4546 resign_request(&mut aggregation);
4547 assert!(aggregation.validate().is_err());
4548
4549 let model = aggregation
4550 .controller_manifests
4551 .iter_mut()
4552 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4553 .unwrap();
4554 model
4555 .capabilities
4556 .insert(ControllerCapability::AggregatesPredictions);
4557 resign_request(&mut aggregation);
4558 let projection = aggregation.project().unwrap();
4559 let mut manifest = influence_manifest(&aggregation, &projection, &relations);
4560 assert!(manifest.entries.iter().any(|entry| {
4561 entry
4562 .node_id
4563 .as_ref()
4564 .is_some_and(|node_id| node_id.as_str() == "model:base")
4565 && entry.kind == TrainingInfluenceKind::TrainedMetaAggregation
4566 }));
4567 assert!(!manifest.entries.iter().any(|entry| {
4568 entry
4569 .node_id
4570 .as_ref()
4571 .is_some_and(|node_id| node_id.as_str() == "model:base")
4572 && entry.kind == TrainingInfluenceKind::ModelFit
4573 }));
4574 manifest
4575 .validate_for_projection(&projection, &aggregation, &relations)
4576 .unwrap();
4577 let removed = manifest
4578 .entries
4579 .iter()
4580 .position(|entry| {
4581 entry
4582 .node_id
4583 .as_ref()
4584 .is_some_and(|node_id| node_id.as_str() == "model:base")
4585 && entry.kind == TrainingInfluenceKind::TrainedMetaAggregation
4586 })
4587 .unwrap();
4588 manifest.entries.remove(removed);
4589 resign_manifest(&mut manifest);
4590 assert!(manifest
4591 .validate_for_projection(&projection, &aggregation, &relations)
4592 .is_err());
4593 }
4594
4595 #[test]
4596 fn parallel_scheduler_is_bound_to_thread_or_process_capabilities() {
4597 let mut threaded = request();
4598 threaded.options.scheduler = TrainingSchedulerOptions {
4599 kind: TrainingSchedulerKind::Parallel,
4600 backend: Some(TrainingSchedulerBackend::Threads),
4601 workers: 2,
4602 };
4603 threaded.options.resources.cpu_threads = 2;
4604 resign_request(&mut threaded);
4605 threaded.validate().unwrap();
4606
4607 let mut unsafe_threads = threaded.clone();
4608 unsafe_threads
4609 .controller_manifests
4610 .iter_mut()
4611 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4612 .unwrap()
4613 .capabilities
4614 .remove(&ControllerCapability::ThreadSafe);
4615 resign_request(&mut unsafe_threads);
4616 assert!(unsafe_threads
4617 .validate()
4618 .unwrap_err()
4619 .to_string()
4620 .contains("thread_safe"));
4621
4622 let mut gil_threads = threaded.clone();
4623 gil_threads
4624 .controller_manifests
4625 .iter_mut()
4626 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4627 .unwrap()
4628 .capabilities
4629 .insert(ControllerCapability::NeedsPythonGil);
4630 resign_request(&mut gil_threads);
4631 assert!(gil_threads
4632 .validate()
4633 .unwrap_err()
4634 .to_string()
4635 .contains("needs_python_gil"));
4636
4637 gil_threads.options.scheduler.backend = Some(TrainingSchedulerBackend::Processes);
4638 resign_request(&mut gil_threads);
4639 gil_threads.validate().unwrap();
4640 }
4641
4642 #[test]
4643 fn portable_required_artifact_mode_rejects_host_only_controller() {
4644 let mut request = request();
4645 request.options.artifacts.fitted_artifacts = FittedArtifactMode::PortableRequired;
4646 request
4647 .controller_manifests
4648 .iter_mut()
4649 .find(|manifest| manifest.operator_kind == NodeKind::Model)
4650 .unwrap()
4651 .artifact_policy = ArtifactPolicy::HostOnly;
4652 resign_request(&mut request);
4653 assert!(request
4654 .validate()
4655 .unwrap_err()
4656 .to_string()
4657 .contains("host_only"));
4658 request.options.artifacts.fitted_artifacts = FittedArtifactMode::AllowHostSidecar;
4659 resign_request(&mut request);
4660 request.validate().unwrap();
4661 }
4662
4663 fn portable_refit_recipe() -> PortableRefitRecipe {
4664 let mut recipe = PortableRefitRecipe {
4665 schema_version: PORTABLE_REFIT_RECIPE_SCHEMA_VERSION,
4666 recipe_id: "recipe:full.test".to_string(),
4667 mode: PortableRefitMode::Full,
4668 parent_package_fingerprint: "1".repeat(64),
4669 parent_outcome: TrainingOutcomeRef {
4670 outcome_id: "outcome:parent".to_string(),
4671 outcome_fingerprint: "2".repeat(64),
4672 pre_conformal_outcome_fingerprint: None,
4673 training_request_fingerprint: "3".repeat(64),
4674 effective_plan_fingerprint: "4".repeat(64),
4675 execution_bundle_id: BundleId::new("bundle:parent").unwrap(),
4676 execution_bundle_fingerprint: "5".repeat(64),
4677 output_binding_fingerprints: vec!["6".repeat(64)],
4678 training_influence_fingerprint: "7".repeat(64),
4679 data_identities_fingerprint: "8".repeat(64),
4680 },
4681 effective_plan_fingerprint: "4".repeat(64),
4682 selected_variant_id: VariantId::new("variant:selected").unwrap(),
4683 selected_variant_fingerprint: "9".repeat(64),
4684 selected_parameter_projection_fingerprint: "a".repeat(64),
4685 target_binding_fingerprints: vec!["b".repeat(64)],
4686 target_schema_fingerprint: "c".repeat(64),
4687 controllers: vec![PortableRefitController {
4688 node_id: NodeId::new("model:selected").unwrap(),
4689 controller_id: ControllerId::new("controller:methods").unwrap(),
4690 controller_version: "1.0.0".to_string(),
4691 manifest_fingerprint: "d".repeat(64),
4692 capabilities: BTreeSet::from([
4693 ControllerCapability::Deterministic,
4694 ControllerCapability::SupportsPortableFullRefit,
4695 ]),
4696 }],
4697 recipe_fingerprint: zero_fingerprint(),
4698 };
4699 recipe.recipe_fingerprint = recipe.compute_fingerprint().unwrap();
4700 recipe
4701 }
4702
4703 #[test]
4704 fn portable_refit_recipe_is_closed_full_only_and_tcv1_attested() {
4705 let recipe = portable_refit_recipe();
4706 recipe.validate().unwrap();
4707 PortableRefitRecipe::from_json(&serde_json::to_string(&recipe).unwrap()).unwrap();
4708
4709 let mut transfer = recipe.clone();
4710 transfer.mode = PortableRefitMode::Transfer;
4711 transfer.recipe_fingerprint = transfer.compute_fingerprint().unwrap();
4712 assert!(transfer
4713 .validate()
4714 .unwrap_err()
4715 .to_string()
4716 .contains("only full mode"));
4717
4718 let mut unqualified = recipe.clone();
4719 unqualified.controllers[0]
4720 .capabilities
4721 .remove(&ControllerCapability::SupportsPortableFullRefit);
4722 unqualified.recipe_fingerprint = unqualified.compute_fingerprint().unwrap();
4723 assert!(unqualified
4724 .validate()
4725 .unwrap_err()
4726 .to_string()
4727 .contains("supports_portable_full_refit"));
4728 }
4729
4730 #[test]
4731 fn portable_refit_provenance_requires_a_fresh_target_cohort() {
4732 let recipe = portable_refit_recipe();
4733 let request = request();
4734 let projection = request.project().unwrap();
4735 let relations = relations();
4736 let influence = influence_manifest(&request, &projection, &relations);
4737 let mut identity = data_identity(&request.campaign);
4738 identity.relation_fingerprint = influence.relation_fingerprint.clone();
4739 identity.identity_fingerprint = zero_fingerprint();
4740 identity.identity_fingerprint = identity.compute_fingerprint().unwrap();
4741 let provenance = PortableRefitProvenance::from_target_cohort(
4742 &recipe,
4743 "e".repeat(64),
4744 &[identity.clone()],
4745 &influence,
4746 )
4747 .unwrap();
4748 provenance.validate_against_recipe(&recipe).unwrap();
4749 PortableRefitProvenance::from_json_for_recipe(
4750 &serde_json::to_string(&provenance).unwrap(),
4751 &recipe,
4752 )
4753 .unwrap();
4754
4755 let error = PortableRefitProvenance::from_target_cohort(
4756 &recipe,
4757 recipe.parent_outcome.training_request_fingerprint.clone(),
4758 &[identity],
4759 &influence,
4760 )
4761 .unwrap_err()
4762 .to_string();
4763 assert!(error.contains("must not reuse the parent"));
4764 }
4765
4766 #[cfg(dag_ml_workspace_contract_fixtures)]
4767 fn package() -> PortablePredictorPackage {
4768 let outcome: Value = serde_json::from_str(include_str!(
4769 "../../../examples/fixtures/estimator/training_outcome_refit.v1.json"
4770 ))
4771 .unwrap();
4772 let effective_plan: ExecutionPlan =
4773 serde_json::from_value(outcome["effective_plan"].clone()).unwrap();
4774 let mut execution_bundle: ExecutionBundle =
4775 serde_json::from_value(outcome["execution_bundle"].clone()).unwrap();
4776 execution_bundle.schema_version = crate::bundle::EXECUTION_BUNDLE_SCHEMA_VERSION;
4780 if let Some(scores) = &mut execution_bundle.scores {
4781 scores.schema_version = crate::metrics::SCORE_SET_SCHEMA_VERSION;
4782 for report in &mut scores.reports {
4783 report.producer_port = Some("oof".to_string());
4784 }
4785 }
4786 for cache in &mut execution_bundle.prediction_caches {
4787 cache.format = crate::bundle::BUNDLE_PREDICTION_CACHE_FORMAT.to_string();
4788 }
4789 let output_bindings = outcome["outputs"]
4790 .as_array()
4791 .unwrap()
4792 .iter()
4793 .map(|output| serde_json::from_value(output["binding"].clone()).unwrap())
4794 .collect::<Vec<OutputBinding>>();
4795 let training_influence: TrainingInfluenceManifest =
4796 serde_json::from_value(outcome["training_influence"].clone()).unwrap();
4797 let mut template = PredictorTemplate {
4798 graph: effective_plan.graph_plan.graph.clone(),
4799 campaign: effective_plan.campaign.clone(),
4800 controller_manifests: effective_plan.controller_manifests.clone(),
4801 template_fingerprint: zero_fingerprint(),
4802 };
4803 template.template_fingerprint = template.compute_fingerprint().unwrap();
4804 let mut data_identities = execution_bundle
4805 .data_requirements
4806 .iter()
4807 .map(|requirement| {
4808 let mut identity = TrainingDataIdentity {
4809 requirement_key: requirement.key(),
4810 schema_fingerprint: requirement.schema_fingerprint.clone(),
4811 plan_fingerprint: requirement.plan_fingerprint.clone(),
4812 relation_fingerprint: requirement.relation_fingerprint.clone().unwrap(),
4813 data_content_fingerprint: "3".repeat(64),
4814 target_content_fingerprint: "4".repeat(64),
4815 identity_fingerprint: zero_fingerprint(),
4816 };
4817 identity.identity_fingerprint = identity.compute_fingerprint().unwrap();
4818 identity
4819 })
4820 .collect::<Vec<_>>();
4821 data_identities.sort_by(|left, right| left.requirement_key.cmp(&right.requirement_key));
4822 let closure = predictor_closure(
4823 &effective_plan,
4824 output_bindings.iter().map(|binding| &binding.node_id),
4825 )
4826 .unwrap();
4827 let mut artifact_bindings = execution_bundle
4828 .refit_artifacts
4829 .iter()
4830 .map(|record| PackageArtifactBinding {
4831 artifact_id: record.artifact.id.clone(),
4832 load_mode: ArtifactLoadMode::HostSidecar,
4833 })
4834 .collect::<Vec<_>>();
4835 artifact_bindings.sort_by(|left, right| left.artifact_id.cmp(&right.artifact_id));
4836 let output_binding_fingerprints = output_bindings
4837 .iter()
4838 .map(|binding| binding.binding_fingerprint.clone())
4839 .collect::<Vec<_>>();
4840 let execution_bundle_fingerprint =
4841 tcv1_fingerprint(&execution_bundle, "test execution bundle").unwrap();
4842 let data_identities_fingerprint =
4843 tcv1_fingerprint(&data_identities, "test data identities").unwrap();
4844 let mut package = PortablePredictorPackage {
4845 schema_version: PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION,
4846 package_id: "predictor:package.test".to_string(),
4847 template,
4848 training_request_fingerprint: "5".repeat(64),
4849 training_outcome: TrainingOutcomeRef {
4850 outcome_id: outcome["outcome_id"].as_str().unwrap().to_string(),
4851 outcome_fingerprint: outcome["outcome_fingerprint"].as_str().unwrap().to_string(),
4852 pre_conformal_outcome_fingerprint: None,
4853 training_request_fingerprint: "5".repeat(64),
4854 effective_plan_fingerprint: outcome["effective_plan_fingerprint"]
4855 .as_str()
4856 .unwrap()
4857 .to_string(),
4858 execution_bundle_id: execution_bundle.bundle_id.clone(),
4859 execution_bundle_fingerprint,
4860 output_binding_fingerprints,
4861 training_influence_fingerprint: training_influence.manifest_fingerprint.clone(),
4862 data_identities_fingerprint,
4863 },
4864 effective_plan,
4865 execution_bundle,
4866 conformal_calibration: None,
4867 conformal_calibration_replay: None,
4868 output_bindings,
4869 predictor_node_ids: closure.into_iter().collect(),
4870 training_influence,
4871 data_identities,
4872 fitted_artifact_mode: FittedArtifactMode::AllowHostSidecar,
4873 artifact_bindings,
4874 package_fingerprint: zero_fingerprint(),
4875 };
4876 package.package_fingerprint = package.compute_fingerprint().unwrap();
4877 package
4878 }
4879
4880 #[cfg(dag_ml_workspace_contract_fixtures)]
4881 #[test]
4882 fn portable_package_round_trips_loads_sidecar_and_rejects_tamper_and_future() {
4883 let package = package();
4884 package.validate().unwrap();
4885 let json = serde_json::to_string(&package).unwrap();
4886 PortablePredictorPackage::from_json(&json).unwrap();
4887
4888 let loaded = package
4889 .clone()
4890 .load_with(|record| Ok(format!("handle:{}", record.artifact.id)))
4891 .unwrap();
4892 let first = &package.artifact_bindings[0].artifact_id;
4893 assert_eq!(loaded.artifact(first).unwrap(), &format!("handle:{first}"));
4894
4895 let mut missing_handles = BTreeMap::new();
4896 missing_handles.insert(first.clone(), "only-one".to_string());
4897 assert!(LoadedPredictor::new(package.clone(), missing_handles).is_err());
4898
4899 let mut tampered = package.clone();
4900 tampered.output_bindings[0].target_space = "tampered".to_string();
4901 assert!(tampered.validate().is_err());
4902
4903 let mut future = package.clone();
4904 future.schema_version += 1;
4905 future.package_fingerprint = zero_fingerprint();
4906 future.package_fingerprint = future.compute_fingerprint().unwrap();
4907 assert!(future.validate().is_err());
4908
4909 let mut binary64 = serde_json::to_value(package).unwrap();
4910 binary64["effective_plan"]["campaign"]["root_seed"] = json!(12345.0);
4911 assert!(serde_json::from_value::<PortablePredictorPackage>(binary64).is_err());
4912 }
4913
4914 #[cfg(dag_ml_workspace_contract_fixtures)]
4915 #[test]
4916 fn portable_refit_recipe_refuses_host_sidecars_before_controller_inspection() {
4917 let package = package();
4918 let error = PortableRefitRecipe::derive_from_package(&package, "recipe:host-refusal")
4919 .unwrap_err()
4920 .to_string();
4921 assert!(error.contains("only native_portable artifacts"));
4922 }
4923
4924 #[cfg(dag_ml_workspace_contract_fixtures)]
4925 #[test]
4926 fn portable_package_strict_parser_rejects_duplicate_and_nfc_colliding_keys() {
4927 let json = serde_json::to_string(&package()).unwrap();
4928 let duplicate = json.replacen(
4929 "\"schema_version\":1",
4930 "\"schema_version\":1,\"schema_version\":1",
4931 1,
4932 );
4933 assert!(PortablePredictorPackage::from_json(&duplicate)
4934 .unwrap_err()
4935 .to_string()
4936 .contains("duplicate JSON object key"));
4937
4938 let collision = json.replacen(
4939 "\"metadata\":{}",
4940 "\"metadata\":{\"é\":1,\"e\\u0301\":2}",
4941 1,
4942 );
4943 assert!(PortablePredictorPackage::from_json(&collision)
4944 .unwrap_err()
4945 .to_string()
4946 .contains("NFC-colliding"));
4947 }
4948
4949 #[cfg(dag_ml_workspace_contract_fixtures)]
4950 #[test]
4951 fn portable_package_rejects_refingerprinted_crosslink_and_relation_drift() {
4952 let mut plan_drift = package();
4953 plan_drift.training_outcome.effective_plan_fingerprint = "f".repeat(64);
4954 plan_drift.package_fingerprint = zero_fingerprint();
4955 plan_drift.package_fingerprint = plan_drift.compute_fingerprint().unwrap();
4956 assert!(plan_drift
4957 .validate()
4958 .unwrap_err()
4959 .to_string()
4960 .contains("effective plan fingerprint"));
4961
4962 let mut binding_drift = package();
4963 binding_drift.output_bindings[0].target_space = "other".to_string();
4964 binding_drift.output_bindings[0].binding_fingerprint = zero_fingerprint();
4965 binding_drift.output_bindings[0].binding_fingerprint = binding_drift.output_bindings[0]
4966 .compute_fingerprint()
4967 .unwrap();
4968 binding_drift.package_fingerprint = zero_fingerprint();
4969 binding_drift.package_fingerprint = binding_drift.compute_fingerprint().unwrap();
4970 assert!(binding_drift
4971 .validate()
4972 .unwrap_err()
4973 .to_string()
4974 .contains("output bindings are not cross-linked"));
4975
4976 let mut relation_drift = package();
4977 relation_drift.data_identities[0].relation_fingerprint = "e".repeat(64);
4978 relation_drift.data_identities[0].identity_fingerprint = zero_fingerprint();
4979 relation_drift.data_identities[0].identity_fingerprint = relation_drift.data_identities[0]
4980 .compute_fingerprint()
4981 .unwrap();
4982 relation_drift.package_fingerprint = zero_fingerprint();
4983 relation_drift.package_fingerprint = relation_drift.compute_fingerprint().unwrap();
4984 assert!(relation_drift
4985 .validate()
4986 .unwrap_err()
4987 .to_string()
4988 .contains("bundle fingerprints"));
4989
4990 let mut content_drift = package();
4991 content_drift.data_identities[0].data_content_fingerprint = "d".repeat(64);
4992 content_drift.data_identities[0].identity_fingerprint = zero_fingerprint();
4993 content_drift.data_identities[0].identity_fingerprint = content_drift.data_identities[0]
4994 .compute_fingerprint()
4995 .unwrap();
4996 content_drift.package_fingerprint = zero_fingerprint();
4997 content_drift.package_fingerprint = content_drift.compute_fingerprint().unwrap();
4998 assert!(content_drift
4999 .validate()
5000 .unwrap_err()
5001 .to_string()
5002 .contains("data identity content"));
5003
5004 let mut bundle_drift = package();
5005 bundle_drift
5006 .execution_bundle
5007 .metadata
5008 .insert("same_id_drift".to_string(), json!(true));
5009 bundle_drift.package_fingerprint = zero_fingerprint();
5010 bundle_drift.package_fingerprint = bundle_drift.compute_fingerprint().unwrap();
5011 assert!(bundle_drift
5012 .validate()
5013 .unwrap_err()
5014 .to_string()
5015 .contains("execution bundle content"));
5016 }
5017
5018 #[cfg(dag_ml_workspace_contract_fixtures)]
5019 #[test]
5020 fn portable_required_package_has_no_host_sidecar_subset() {
5021 let mut package = package();
5022 package.fitted_artifact_mode = FittedArtifactMode::PortableRequired;
5023 for binding in &mut package.artifact_bindings {
5024 binding.load_mode = ArtifactLoadMode::NativePortable;
5025 }
5026 package.package_fingerprint = zero_fingerprint();
5027 package.package_fingerprint = package.compute_fingerprint().unwrap();
5028 package.validate().unwrap();
5029 let loaded = LoadedPredictor::<String>::new(package, BTreeMap::new()).unwrap();
5030 assert!(loaded.artifacts.is_empty());
5031 }
5032
5033 #[cfg(dag_ml_workspace_contract_fixtures)]
5034 #[test]
5035 fn package_refuses_runtime_handle_shape_even_when_nested_in_metadata() {
5036 for payload in [
5037 json!({"handle": 9, "owner_controller": "controller:model.mock"}),
5038 json!({"nested": {"model_handle": 9}}),
5039 json!({"nested": [{"runtime_handles": [9]}]}),
5040 ] {
5041 let mut package = package();
5042 package
5043 .execution_bundle
5044 .metadata
5045 .insert("forbidden".to_string(), payload);
5046 package.training_outcome.execution_bundle_fingerprint =
5047 tcv1_fingerprint(&package.execution_bundle, "runtime-handle test bundle").unwrap();
5048 package.package_fingerprint = zero_fingerprint();
5049 package.package_fingerprint = package.compute_fingerprint().unwrap();
5050 assert!(package
5051 .validate()
5052 .unwrap_err()
5053 .to_string()
5054 .contains("runtime handles"));
5055 }
5056 }
5057
5058 #[cfg(dag_ml_workspace_contract_fixtures)]
5059 #[test]
5060 fn portable_w0_output_binding_and_influence_fingerprints_match_production_tcv1() {
5061 let package = package();
5062 for binding in &package.output_bindings {
5063 assert_eq!(
5064 binding.binding_fingerprint,
5065 binding.compute_fingerprint().unwrap()
5066 );
5067 }
5068 assert_eq!(
5069 package.training_influence.manifest_fingerprint,
5070 package.training_influence.compute_fingerprint().unwrap()
5071 );
5072 }
5073
5074 #[cfg(dag_ml_workspace_contract_fixtures)]
5075 #[test]
5076 fn portable_package_accepts_multi_scope_base_influence_per_node() {
5077 let mut package = package();
5078 let base_kinds = [
5079 TrainingInfluenceKind::TransformFit,
5080 TrainingInfluenceKind::ModelFit,
5081 TrainingInfluenceKind::HpoSelection,
5082 TrainingInfluenceKind::TrainedMetaAggregation,
5083 ]
5084 .into_iter()
5085 .collect::<BTreeSet<_>>();
5086 let mut entries = Vec::new();
5087 for entry in package.training_influence.entries.clone() {
5088 if entry.node_id.is_some() && base_kinds.contains(&entry.kind) {
5089 for suffix in ["fit_cv:fold:0", "fit_cv:fold:1", "refit:full"] {
5090 let mut scoped = entry.clone();
5091 scoped.scope_id = format!("{suffix}:{}", entry.scope_id);
5092 entries.push(scoped);
5093 }
5094 } else {
5095 entries.push(entry);
5096 }
5097 }
5098 entries.sort_by(|left, right| {
5099 (left.kind, &left.scope_id, &left.node_id).cmp(&(
5100 right.kind,
5101 &right.scope_id,
5102 &right.node_id,
5103 ))
5104 });
5105 package.training_influence.entries = entries;
5106 package.training_influence.manifest_fingerprint = zero_fingerprint();
5107 package.training_influence.manifest_fingerprint =
5108 package.training_influence.compute_fingerprint().unwrap();
5109 package.training_outcome.training_influence_fingerprint =
5110 package.training_influence.manifest_fingerprint.clone();
5111 package.package_fingerprint = zero_fingerprint();
5112 package.package_fingerprint = package.compute_fingerprint().unwrap();
5113 package.validate().unwrap();
5114 }
5115
5116 #[cfg(dag_ml_workspace_contract_fixtures)]
5117 #[test]
5118 fn controller_id_import_remains_the_same_public_type() {
5119 let id = ControllerId::new("controller:model.mock").unwrap();
5122 assert!(package().template.controller_manifests.contains_key(&id));
5123 }
5124
5125 #[cfg(dag_ml_workspace_contract_fixtures)]
5126 #[test]
5127 fn committed_w1_fixtures_match_rust_and_independent_tcv1_oracle() {
5128 let refit_json =
5129 include_str!("../../../examples/fixtures/training/training_request_refit.v1.json");
5130 let refit = TrainingRequest::from_json(refit_json).unwrap();
5131 let no_refit = TrainingRequest::from_json(include_str!(
5132 "../../../examples/fixtures/training/training_request_no_refit.v1.json"
5133 ))
5134 .unwrap();
5135 let active_influence = TrainingRequest::from_json(include_str!(
5136 "../../../examples/fixtures/training/training_request_active_influence.v1.json"
5137 ))
5138 .unwrap();
5139 let package_request = TrainingRequest::from_json(include_str!(
5140 "../../../examples/fixtures/training/training_request_package_refit.v1.json"
5141 ))
5142 .unwrap();
5143 assert!(refit.options.refit);
5144 assert!(!no_refit.options.refit);
5145 assert_eq!(active_influence.influence_requirements.len(), 6);
5146
5147 let package_json =
5148 include_str!("../../../examples/fixtures/training/portable_predictor_package.v1.json");
5149 let package = PortablePredictorPackage::from_json(package_json).unwrap();
5150 assert_eq!(
5151 package.training_request_fingerprint,
5152 package_request.request_fingerprint
5153 );
5154 assert_eq!(package.data_identities, package_request.data_identities);
5155 let namespace = CacheNamespace::from_json(include_str!(
5156 "../../../examples/fixtures/training/cache_namespace_fit_cv.v1.json"
5157 ))
5158 .unwrap();
5159 let identity = package
5160 .data_identities
5161 .iter()
5162 .find(|identity| identity.requirement_key == namespace.data_requirement_key)
5163 .unwrap();
5164 namespace.validate_for_identity(identity).unwrap();
5165
5166 let projection: ParameterProjection = serde_json::from_str(include_str!(
5167 "../../../examples/fixtures/training/parameter_projection_empty.v1.json"
5168 ))
5169 .unwrap();
5170 projection.validate().unwrap();
5171
5172 let negatives: serde_json::Value = serde_json::from_str(include_str!(
5173 "../../../examples/fixtures/training/negative_cases.v1.json"
5174 ))
5175 .unwrap();
5176 for case in negatives["cases"].as_array().unwrap() {
5177 let document = serde_json::to_string(&case["document"]).unwrap();
5178 let error = match case["contract"].as_str().unwrap() {
5179 "cache_namespace" => CacheNamespace::from_json(&document).unwrap_err(),
5180 "portable_predictor_package" => {
5181 PortablePredictorPackage::from_json(&document).unwrap_err()
5182 }
5183 "training_outcome" => {
5184 crate::training_runtime::TrainingOutcome::from_json(&document).unwrap_err()
5185 }
5186 "training_request" => TrainingRequest::from_json(&document).unwrap_err(),
5187 other => panic!("unknown negative contract {other}"),
5188 };
5189 assert!(
5190 error
5191 .to_string()
5192 .contains(case["expected_error"].as_str().unwrap()),
5193 "{}: {error}",
5194 case["id"]
5195 );
5196 }
5197 }
5198
5199 #[test]
5200 fn projection_strict_parsers_reject_duplicate_and_nfc_colliding_keys() {
5201 let mut projection = ParameterProjection {
5202 schema_version: PARAMETER_PROJECTION_SCHEMA_VERSION,
5203 nodes: BTreeMap::new(),
5204 requires_recompile: false,
5205 structural_patch_count: 0,
5206 patches_fingerprint: tcv1_fingerprint(&Vec::<ParameterPatch>::new(), "test patches")
5207 .unwrap(),
5208 projection_fingerprint: zero_fingerprint(),
5209 };
5210 projection.projection_fingerprint = projection.compute_fingerprint().unwrap();
5211 let parameter_json = serde_json::to_string(&projection).unwrap();
5212 ParameterProjection::from_json(¶meter_json).unwrap();
5213 let duplicate = parameter_json.replacen(
5214 "\"schema_version\":1",
5215 "\"schema_version\":1,\"schema_version\":1",
5216 1,
5217 );
5218 assert!(ParameterProjection::from_json(&duplicate)
5219 .unwrap_err()
5220 .to_string()
5221 .contains("duplicate JSON object key"));
5222 let collision = parameter_json.replacen('{', "{\"é\":1,\"e\\u0301\":2,", 1);
5223 assert!(ParameterProjection::from_json(&collision)
5224 .unwrap_err()
5225 .to_string()
5226 .contains("NFC-colliding"));
5227
5228 let request = request();
5229 let projection = request.project().unwrap();
5230 let projection_json = serde_json::to_string(&projection).unwrap();
5231 TrainingContractProjection::from_json(&projection_json).unwrap();
5232 let duplicate = projection_json.replacen(
5233 "\"request_id\":",
5234 "\"request_id\":\"duplicate\",\"request_id\":",
5235 1,
5236 );
5237 assert!(TrainingContractProjection::from_json(&duplicate)
5238 .unwrap_err()
5239 .to_string()
5240 .contains("duplicate JSON object key"));
5241 let collision = projection_json.replacen('{', "{\"é\":1,\"e\\u0301\":2,", 1);
5242 assert!(TrainingContractProjection::from_json(&collision)
5243 .unwrap_err()
5244 .to_string()
5245 .contains("NFC-colliding"));
5246
5247 for path in [
5248 &["plan", "graph_plan", "graph"][..],
5249 &["plan", "campaign"][..],
5250 ] {
5251 let mut unknown: serde_json::Value = serde_json::from_str(&projection_json).unwrap();
5252 let mut parent = &mut unknown;
5253 for segment in path {
5254 parent = &mut parent[*segment];
5255 }
5256 parent["unknown_projection_field"] = json!(true);
5257 let error =
5258 TrainingContractProjection::from_json(&serde_json::to_string(&unknown).unwrap())
5259 .unwrap_err();
5260 let expected_path = format!("{}.unknown_projection_field", path.join("."));
5261 assert!(error.to_string().contains("unknown field"), "{error}");
5262 assert!(error.to_string().contains(&expected_path), "{error}");
5263 }
5264 }
5265}