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