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