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