Skip to main content

dag_ml_core/
training.rs

1//! Frozen W1 training and portable-predictor contracts.
2//!
3//! This module is intentionally contract-only. It validates and projects the
4//! information needed by the future native training operation without running
5//! controllers or duplicating scheduler logic. Historical graph, campaign,
6//! controller and plan fingerprints keep their existing serde-JSON profile;
7//! only the new contracts in this module use TCV1.
8
9use 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
48/// Deserialize a **required but nullable** field.
49///
50/// Wire semantics: the key MUST be present, yet its value may be an explicit
51/// JSON `null`. Paired with `#[serde(deserialize_with = ...)]` and **no**
52/// `#[serde(default)]`, this keeps "absent" and "present-and-null" distinct on
53/// the wire: an omitted key is a hard `missing field` error, while an explicit
54/// `null` maps to `None`. This differs from serde's default treatment of an
55/// `Option<T>` field, where omission silently becomes `None`. It matches the
56/// W1 JSON schemas that list these fields as `required` with an
57/// `anyOf [ T, null ]` value.
58fn 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/// Requested output metadata before the producing port has been resolved.
93#[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    /// Resolve the only prediction port when omitted, or validate an explicit
154    /// port. A producer with zero or multiple prediction ports is never guessed.
155    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    // Required by schema: an empty list is valid, but omitting the key is not,
277    // so this field intentionally carries no `#[serde(default)]`.
278    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/// Content identity paired with one external data requirement.
408#[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    /// Build the complete, signed content identity for one exact data binding.
422    ///
423    /// Prediction-only or legacy envelopes may omit content fingerprints, but
424    /// such envelopes cannot attest a native training operation and are
425    /// rejected here.
426    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    /// Return the frozen internal `ExecutionPlan` root for this public wire
500    /// namespace. The mapping is bijective and must not be renamed silently.
501    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
651/// Clone and deeply apply typed patches. Intermediate path segments must
652/// already exist and be objects; only the final object key may be new. Arrays
653/// are never addressable in V1.
654pub 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            &registry,
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/// Candidate-cache identity. Every field that can change predictions is part
1200/// of the TCV1 namespace; a requirement key alone is deliberately insufficient.
1201#[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/// A resolved W0 OutputBinding with native port validation.
1346#[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/// Portable deployment package. It contains only JSON-safe contracts and
1536/// artifact descriptors; process-local handles live exclusively in
1537/// [`LoadedPredictor`].
1538#[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        // A portable package is a deployable predictor, so it must independently
1738        // prove PREDICT replayability from its own plan/closure/retained artifacts
1739        // — never infer portability from a merely non-empty claimed phase set. An
1740        // outcome that only reaches REFIT (skipped refit) or has no honest replay
1741        // mode ([]) carries no full-training predictor and is refused here.
1742        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
1790/// Process-local sidecar. It deliberately implements neither `Serialize` nor
1791/// `Deserialize`, so opaque host objects cannot leak into portable packages.
1792pub 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                    // ControllerManifest::validate rejects FullTrain + FIT_CV.
2077                    // Keep exhaustive all-sample accounting as defense in depth
2078                    // if a hand-built plan ever bypasses that invariant.
2079                    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                    // Unreachable for a validated manifest (FullTrain cannot
2158                    // support FIT_CV), but never under-report influence if an
2159                    // invalid hand-built plan reaches this helper.
2160                    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    /// Presence-strictness (W1-0): five schema-`required` fields — four of them
3069    /// nullable — must reject key omission at serde deserialization with a
3070    /// field-specific `missing field` error, while still accepting the valid
3071    /// explicit values `null` / `[]`. These deserialize the individual
3072    /// sub-structs directly, so no outer fingerprint can mask or trigger the
3073    /// failure: presence is the only thing under test.
3074    #[test]
3075    fn required_nullable_fields_reject_omission_but_accept_explicit_null() {
3076        let base = request();
3077
3078        // 1. TrainingOutputRequest.unit_level (required, nullable).
3079        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        // 2. TrainingResourceLimits.gpu_devices (required, non-nullable array).
3096        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        // 3. TrainingOptions.refit_strategy (required, nullable).
3112        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        // 4. ControllerInfluenceRequirement.fold_id (required, nullable).
3130        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        // 5. TrainingInfluenceEntry.node_id (required, nullable).
3157        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        // Guards the package/template key type against accidental string-only
4364        // drift while keeping this test module's import exercised.
4365        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}