1use std::cell::RefCell;
2use std::collections::{BTreeMap, BTreeSet};
3
4use serde::{Deserialize, Serialize};
5
6use crate::campaign::stable_json_fingerprint;
7use crate::error::{DagMlError, Result};
8use crate::fold::FoldSet;
9use crate::ids::{ControllerId, FoldId, NodeId, RunId, SampleId, VariantId};
10use crate::phase::Phase;
11use crate::policy::FitInfluencePolicy;
12use crate::relation::{EntityUnitLevel, SampleRelationSet};
13use crate::runtime::{
14 DataMaterializationRequest, DataProviderViewSpec, DataViewRequest, HandleKind, HandleRef,
15 RuntimeDataProvider,
16};
17
18pub const EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1: u32 = 1;
20pub const EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V2: u32 = 2;
23pub const EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION: u32 =
26 EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1;
27pub const MODEL_INPUT_SPEC_SCHEMA_VERSION: u32 = 1;
28pub const MODEL_INPUT_SPEC_SCHEMA_ID: &str =
29 "https://github.com/GBeurier/dag-ml/schemas/model_input_spec.v1.schema.json";
30pub const DATA_PLAN_SCHEMA_VERSION: u32 = 1;
31pub const DATA_PLAN_SCHEMA_ID: &str =
32 "https://github.com/GBeurier/dag-ml/schemas/data_plan.v1.schema.json";
33pub const SOURCE_INDEX_METADATA_KEY: &str = "source_index";
34
35fn default_external_data_plan_envelope_schema_version() -> u32 {
36 EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1
37}
38
39fn default_model_input_spec_schema_version() -> u32 {
40 MODEL_INPUT_SPEC_SCHEMA_VERSION
41}
42
43fn default_data_plan_schema_version() -> u32 {
44 DATA_PLAN_SCHEMA_VERSION
45}
46
47#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
48#[serde(rename_all = "snake_case")]
49pub enum DataRequestPartition {
50 FoldTrain,
51 FoldValidation,
52 FullTrain,
53 Predict,
54}
55
56#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
57#[serde(rename_all = "snake_case")]
58pub enum ModelInputFusionMode {
59 SingleSource,
60 ConcatenateFeatures,
61 StackSamples,
62 DictBySource,
63 Custom,
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
67#[serde(rename_all = "snake_case")]
68pub enum BranchViewMode {
69 Separation,
70 BySource,
71 ByMetadata,
72 ByTag,
73 ByFilter,
74}
75
76#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
77pub struct DataViewSelector {
78 #[serde(default, skip_serializing_if = "Vec::is_empty")]
79 pub source_ids: Vec<String>,
80 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
81 pub metadata: BTreeMap<String, serde_json::Value>,
82 #[serde(default, skip_serializing_if = "Vec::is_empty")]
83 pub tags: Vec<String>,
84 #[serde(default, skip_serializing_if = "Option::is_none")]
85 pub filter: Option<serde_json::Value>,
86}
87
88impl DataViewSelector {
89 pub fn validate(&self, label: &str) -> Result<()> {
90 if self.source_ids.is_empty()
91 && self.metadata.is_empty()
92 && self.tags.is_empty()
93 && self.filter.is_none()
94 {
95 return Err(DagMlError::CampaignValidation(format!(
96 "{label} selector must constrain source_ids, metadata, tags or filter"
97 )));
98 }
99 validate_string_list_entries(&format!("{label} selector source_ids"), &self.source_ids)?;
100 validate_unique_strings(&format!("{label} selector source_ids"), &self.source_ids)?;
101 validate_string_list_entries(&format!("{label} selector tags"), &self.tags)?;
102 validate_unique_strings(&format!("{label} selector tags"), &self.tags)?;
103 for key in self.metadata.keys() {
104 if key.trim().is_empty() {
105 return Err(DagMlError::CampaignValidation(format!(
106 "{label} selector contains an empty metadata key"
107 )));
108 }
109 }
110 if matches!(self.filter, Some(serde_json::Value::Null)) {
111 return Err(DagMlError::CampaignValidation(format!(
112 "{label} selector filter must not be null"
113 )));
114 }
115 Ok(())
116 }
117}
118
119#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
120pub struct BranchViewPlan {
121 pub view_id: String,
122 pub branch_id: String,
123 pub mode: BranchViewMode,
124 pub selector: DataViewSelector,
125 #[serde(default)]
126 pub allow_overlap: bool,
127 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
128 pub metadata: BTreeMap<String, serde_json::Value>,
129}
130
131impl BranchViewPlan {
132 pub fn validate(&self) -> Result<()> {
133 validate_non_empty("branch view plan view_id", &self.view_id)?;
134 validate_non_empty("branch view plan branch_id", &self.branch_id)?;
135 self.selector
136 .validate(&format!("branch view `{}`", self.view_id))?;
137 match self.mode {
138 BranchViewMode::BySource if self.selector.source_ids.is_empty() => {
139 return Err(DagMlError::CampaignValidation(format!(
140 "branch view `{}` mode=by_source requires source_ids",
141 self.view_id
142 )));
143 }
144 BranchViewMode::ByMetadata if self.selector.metadata.is_empty() => {
145 return Err(DagMlError::CampaignValidation(format!(
146 "branch view `{}` mode=by_metadata requires metadata",
147 self.view_id
148 )));
149 }
150 BranchViewMode::ByTag if self.selector.tags.is_empty() => {
151 return Err(DagMlError::CampaignValidation(format!(
152 "branch view `{}` mode=by_tag requires tags",
153 self.view_id
154 )));
155 }
156 BranchViewMode::ByFilter if self.selector.filter.is_none() => {
157 return Err(DagMlError::CampaignValidation(format!(
158 "branch view `{}` mode=by_filter requires filter",
159 self.view_id
160 )));
161 }
162 _ => {}
163 }
164 for key in self.metadata.keys() {
165 if key.trim().is_empty() {
166 return Err(DagMlError::CampaignValidation(format!(
167 "branch view `{}` metadata contains an empty key",
168 self.view_id
169 )));
170 }
171 }
172 Ok(())
173 }
174}
175
176#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
177#[serde(rename_all = "snake_case")]
178pub enum CombinationMode {
179 #[default]
180 Cartesian,
181 Zip,
182 MatchBy,
183 SampleK,
184 ReferenceBroadcast,
185}
186
187#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
188#[serde(rename_all = "snake_case")]
189pub enum RepresentationMissingSourcePolicy {
190 Strict,
191 Warn,
192 DropIncomplete,
193 ImputeDeclared,
194 Mask,
195 PartialModel,
196 Pad,
197}
198
199#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
200#[serde(rename_all = "snake_case")]
201pub enum RepresentationCardinality {
202 OneToOne,
203 OneToMany,
204 ManyToOne,
205 ManyToMany,
206 BoundedMany,
207}
208
209#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
210#[serde(deny_unknown_fields)]
211pub struct CombinationPlan {
212 pub mode: CombinationMode,
213 #[serde(default, skip_serializing_if = "Vec::is_empty")]
214 pub component_source_ids: Vec<String>,
215 #[serde(default, skip_serializing_if = "Vec::is_empty")]
216 pub component_unit_ids: Vec<String>,
217 #[serde(default, skip_serializing_if = "Option::is_none")]
218 pub match_key: Option<String>,
219 #[serde(default, skip_serializing_if = "Option::is_none")]
220 pub reference_source_id: Option<String>,
221 #[serde(default, skip_serializing_if = "Option::is_none")]
222 pub seed: Option<u64>,
223 #[serde(default, skip_serializing_if = "Option::is_none")]
224 pub cap: Option<usize>,
225 #[serde(default, skip_serializing_if = "Option::is_none")]
226 pub budget: Option<usize>,
227 #[serde(default, skip_serializing_if = "Option::is_none")]
228 pub missing_source_policy: Option<RepresentationMissingSourcePolicy>,
229 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
230 pub metadata: BTreeMap<String, serde_json::Value>,
231}
232
233impl CombinationPlan {
234 pub fn validate(&self) -> Result<()> {
235 validate_string_list_entries(
236 "combination plan component_source_ids",
237 &self.component_source_ids,
238 )?;
239 validate_unique_strings(
240 "combination plan component_source_ids",
241 &self.component_source_ids,
242 )?;
243 validate_string_list_entries(
244 "combination plan component_unit_ids",
245 &self.component_unit_ids,
246 )?;
247 validate_unique_strings(
248 "combination plan component_unit_ids",
249 &self.component_unit_ids,
250 )?;
251 validate_optional_non_empty("combination plan match_key", &self.match_key)?;
252 validate_optional_non_empty(
253 "combination plan reference_source_id",
254 &self.reference_source_id,
255 )?;
256 if self.cap == Some(0) {
257 return Err(DagMlError::CampaignValidation(
258 "combination plan cap must be positive when present".to_string(),
259 ));
260 }
261 if self.budget == Some(0) {
262 return Err(DagMlError::CampaignValidation(
263 "combination plan budget must be positive when present".to_string(),
264 ));
265 }
266 match self.mode {
267 CombinationMode::Cartesian => {
268 if self.component_source_ids.len() < 2 {
269 return Err(DagMlError::CampaignValidation(
270 "cartesian combination requires at least two component_source_ids"
271 .to_string(),
272 ));
273 }
274 }
275 CombinationMode::Zip => {
276 if self.component_source_ids.len() < 2 {
277 return Err(DagMlError::CampaignValidation(
278 "zip combination requires at least two component_source_ids".to_string(),
279 ));
280 }
281 }
282 CombinationMode::MatchBy => {
283 if self.match_key.is_none() {
284 return Err(DagMlError::CampaignValidation(
285 "match_by combination requires match_key".to_string(),
286 ));
287 }
288 }
289 CombinationMode::SampleK => {
290 if self.seed.is_none() {
291 return Err(DagMlError::CampaignValidation(
292 "sample_k combination requires seed".to_string(),
293 ));
294 }
295 if self.cap.is_none() {
296 return Err(DagMlError::CampaignValidation(
297 "sample_k combination requires cap".to_string(),
298 ));
299 }
300 }
301 CombinationMode::ReferenceBroadcast => {
302 let Some(reference) = &self.reference_source_id else {
303 return Err(DagMlError::CampaignValidation(
304 "reference_broadcast combination requires reference_source_id".to_string(),
305 ));
306 };
307 if !self.component_source_ids.is_empty()
308 && !self
309 .component_source_ids
310 .iter()
311 .any(|source| source == reference)
312 {
313 return Err(DagMlError::CampaignValidation(format!(
314 "reference_broadcast reference_source_id `{reference}` is not in component_source_ids"
315 )));
316 }
317 }
318 }
319 for key in self.metadata.keys() {
320 if key.trim().is_empty() {
321 return Err(DagMlError::CampaignValidation(
322 "combination plan metadata contains an empty key".to_string(),
323 ));
324 }
325 }
326 Ok(())
327 }
328}
329
330#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
331#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
332pub enum RepresentationPlan {
333 Aggregate(AggregateRepresentation),
334 CartesianProduct(CartesianProductRepresentation),
335 MonteCarloCartesian(MonteCarloCartesianRepresentation),
336 StackFixed(StackFixedRepresentation),
337 StackPaddedMasked(StackPaddedMaskedRepresentation),
338}
339
340impl RepresentationPlan {
341 pub fn validate(&self) -> Result<()> {
342 match self {
343 Self::Aggregate(plan) => plan.validate(),
344 Self::CartesianProduct(plan) => plan.validate(),
345 Self::MonteCarloCartesian(plan) => plan.validate(),
346 Self::StackFixed(plan) => plan.validate(),
347 Self::StackPaddedMasked(plan) => plan.validate(),
348 }
349 }
350
351 pub fn output_unit_level(&self) -> EntityUnitLevel {
352 match self {
353 Self::Aggregate(plan) => plan.output_unit_level,
354 Self::CartesianProduct(plan) => plan.output_unit_level,
355 Self::MonteCarloCartesian(plan) => plan.output_unit_level,
356 Self::StackFixed(plan) => plan.output_unit_level,
357 Self::StackPaddedMasked(plan) => plan.output_unit_level,
358 }
359 }
360}
361
362#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
363#[serde(deny_unknown_fields)]
364pub struct AggregateRepresentation {
365 pub input_unit_level: EntityUnitLevel,
366 pub output_unit_level: EntityUnitLevel,
367 #[serde(default, skip_serializing_if = "Option::is_none")]
368 pub reducer_id: Option<String>,
369 #[serde(default, skip_serializing_if = "Option::is_none")]
370 pub method: Option<String>,
371 pub cardinality: RepresentationCardinality,
372}
373
374impl AggregateRepresentation {
375 pub fn validate(&self) -> Result<()> {
376 validate_optional_non_empty("aggregate representation reducer_id", &self.reducer_id)?;
377 validate_optional_non_empty("aggregate representation method", &self.method)?;
378 if self.cardinality != RepresentationCardinality::ManyToOne {
379 return Err(DagMlError::CampaignValidation(
380 "aggregate representation cardinality must be many_to_one".to_string(),
381 ));
382 }
383 Ok(())
384 }
385}
386
387#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
388#[serde(deny_unknown_fields)]
389pub struct CartesianProductRepresentation {
390 pub combination_plan: CombinationPlan,
391 pub output_unit_level: EntityUnitLevel,
392 pub cardinality: RepresentationCardinality,
393 #[serde(default = "default_true")]
394 pub preserve_provenance: bool,
395}
396
397impl CartesianProductRepresentation {
398 pub fn validate(&self) -> Result<()> {
399 self.combination_plan.validate()?;
400 if self.combination_plan.mode != CombinationMode::Cartesian {
401 return Err(DagMlError::CampaignValidation(
402 "cartesian_product representation requires combination_plan.mode=cartesian"
403 .to_string(),
404 ));
405 }
406 validate_combo_like_output("cartesian_product", self.output_unit_level)?;
407 if self.cardinality != RepresentationCardinality::ManyToMany {
408 return Err(DagMlError::CampaignValidation(
409 "cartesian_product representation cardinality must be many_to_many".to_string(),
410 ));
411 }
412 Ok(())
413 }
414}
415
416#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
417#[serde(deny_unknown_fields)]
418pub struct MonteCarloCartesianRepresentation {
419 pub combination_plan: CombinationPlan,
420 pub output_unit_level: EntityUnitLevel,
421 pub cardinality: RepresentationCardinality,
422 #[serde(default = "default_true")]
423 pub preserve_provenance: bool,
424}
425
426impl MonteCarloCartesianRepresentation {
427 pub fn validate(&self) -> Result<()> {
428 self.combination_plan.validate()?;
429 if self.combination_plan.mode != CombinationMode::SampleK {
430 return Err(DagMlError::CampaignValidation(
431 "monte_carlo_cartesian representation requires combination_plan.mode=sample_k"
432 .to_string(),
433 ));
434 }
435 validate_combo_like_output("monte_carlo_cartesian", self.output_unit_level)?;
436 if self.cardinality != RepresentationCardinality::BoundedMany {
437 return Err(DagMlError::CampaignValidation(
438 "monte_carlo_cartesian representation cardinality must be bounded_many".to_string(),
439 ));
440 }
441 Ok(())
442 }
443}
444
445#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
446#[serde(deny_unknown_fields)]
447pub struct StackFixedRepresentation {
448 pub output_unit_level: EntityUnitLevel,
449 pub cardinality: RepresentationCardinality,
450 pub expected_cardinality: usize,
451 #[serde(default, skip_serializing_if = "Vec::is_empty")]
452 pub component_source_ids: Vec<String>,
453}
454
455impl StackFixedRepresentation {
456 pub fn validate(&self) -> Result<()> {
457 if self.expected_cardinality == 0 {
458 return Err(DagMlError::CampaignValidation(
459 "stack_fixed representation expected_cardinality must be positive".to_string(),
460 ));
461 }
462 if self.cardinality != RepresentationCardinality::OneToMany {
463 return Err(DagMlError::CampaignValidation(
464 "stack_fixed representation cardinality must be one_to_many".to_string(),
465 ));
466 }
467 validate_string_list_entries(
468 "stack_fixed representation component_source_ids",
469 &self.component_source_ids,
470 )?;
471 validate_unique_strings(
472 "stack_fixed representation component_source_ids",
473 &self.component_source_ids,
474 )
475 }
476}
477
478#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
479#[serde(deny_unknown_fields)]
480pub struct StackPaddedMaskedRepresentation {
481 pub output_unit_level: EntityUnitLevel,
482 pub cardinality: RepresentationCardinality,
483 pub expected_cardinality: usize,
484 pub missing_source_policy: RepresentationMissingSourcePolicy,
485 #[serde(default = "default_true")]
486 pub requires_missing_masks: bool,
487 #[serde(default, skip_serializing_if = "Vec::is_empty")]
488 pub component_source_ids: Vec<String>,
489}
490
491impl StackPaddedMaskedRepresentation {
492 pub fn validate(&self) -> Result<()> {
493 if self.expected_cardinality == 0 {
494 return Err(DagMlError::CampaignValidation(
495 "stack_padded_masked representation expected_cardinality must be positive"
496 .to_string(),
497 ));
498 }
499 if self.cardinality != RepresentationCardinality::BoundedMany {
500 return Err(DagMlError::CampaignValidation(
501 "stack_padded_masked representation cardinality must be bounded_many".to_string(),
502 ));
503 }
504 if !matches!(
505 self.missing_source_policy,
506 RepresentationMissingSourcePolicy::Mask | RepresentationMissingSourcePolicy::Pad
507 ) {
508 return Err(DagMlError::CampaignValidation(
509 "stack_padded_masked representation requires missing_source_policy=mask or pad"
510 .to_string(),
511 ));
512 }
513 if !self.requires_missing_masks {
514 return Err(DagMlError::CampaignValidation(
515 "stack_padded_masked representation requires missing-mask controller support"
516 .to_string(),
517 ));
518 }
519 validate_string_list_entries(
520 "stack_padded_masked representation component_source_ids",
521 &self.component_source_ids,
522 )?;
523 validate_unique_strings(
524 "stack_padded_masked representation component_source_ids",
525 &self.component_source_ids,
526 )
527 }
528}
529
530#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
531#[serde(deny_unknown_fields)]
532pub struct RepresentationSampleObservationMapping {
533 pub physical_sample_id: String,
534 pub source_id: String,
535 pub observation_ids: Vec<String>,
536}
537
538impl RepresentationSampleObservationMapping {
539 pub fn validate(&self) -> Result<()> {
540 validate_non_empty(
541 "representation sample observation mapping physical_sample_id",
542 &self.physical_sample_id,
543 )?;
544 validate_non_empty(
545 "representation sample observation mapping source_id",
546 &self.source_id,
547 )?;
548 validate_non_empty_list(
549 "representation sample observation mapping observation_ids",
550 &self.observation_ids,
551 )?;
552 validate_unique_strings(
553 "representation sample observation mapping observation_ids",
554 &self.observation_ids,
555 )
556 }
557}
558
559#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
560#[serde(deny_unknown_fields)]
561pub struct RepresentationComboSelectionRecord {
562 pub combo_unit_id: String,
563 pub physical_sample_id: String,
564 pub component_observation_ids: Vec<String>,
565 #[serde(default, skip_serializing_if = "Option::is_none")]
566 pub seed: Option<u64>,
567}
568
569impl RepresentationComboSelectionRecord {
570 pub fn validate(&self) -> Result<()> {
571 validate_non_empty(
572 "representation combo selection combo_unit_id",
573 &self.combo_unit_id,
574 )?;
575 validate_non_empty(
576 "representation combo selection physical_sample_id",
577 &self.physical_sample_id,
578 )?;
579 validate_non_empty_list(
580 "representation combo selection component_observation_ids",
581 &self.component_observation_ids,
582 )?;
583 validate_unique_strings(
584 "representation combo selection component_observation_ids",
585 &self.component_observation_ids,
586 )
587 }
588}
589
590#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
591#[serde(rename_all = "snake_case")]
592pub enum RepresentationCompatibilitySeverity {
593 Info,
594 Warning,
595 Error,
596}
597
598#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
599#[serde(rename_all = "snake_case")]
600pub enum RepresentationCompatibilityOutcome {
601 Compatible,
602 CompatibleWithFallback,
603 Incompatible,
604}
605
606#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
607#[serde(deny_unknown_fields)]
608pub struct RepresentationCompatibilityReport {
609 pub policy: RepresentationMissingSourcePolicy,
610 pub outcome: RepresentationCompatibilityOutcome,
611 #[serde(default, skip_serializing_if = "Option::is_none")]
612 pub fallback_used: Option<String>,
613 #[serde(default, skip_serializing_if = "Option::is_none")]
614 pub warning_severity: Option<RepresentationCompatibilitySeverity>,
615 #[serde(default)]
616 pub affected_source_count: u64,
617 #[serde(default)]
618 pub affected_repetition_count: u64,
619 #[serde(default)]
620 pub affected_sample_count: u64,
621 #[serde(default, skip_serializing_if = "Option::is_none")]
622 pub train_relation_fingerprint: Option<String>,
623 #[serde(default, skip_serializing_if = "Option::is_none")]
624 pub predict_relation_fingerprint: Option<String>,
625 #[serde(default, skip_serializing_if = "Option::is_none")]
626 pub train_unit_count: Option<u64>,
627 #[serde(default, skip_serializing_if = "Option::is_none")]
628 pub predict_unit_count: Option<u64>,
629 #[serde(default)]
630 pub fixed_width_required: bool,
631 #[serde(default)]
632 pub final_reducer_stabilizes_output: bool,
633 #[serde(default)]
634 pub cartesian_combo_count_changed: bool,
635 #[serde(default)]
636 pub late_fusion_branch_delta: bool,
637 #[serde(default, skip_serializing_if = "Vec::is_empty")]
638 pub messages: Vec<String>,
639 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
640 pub metadata: BTreeMap<String, serde_json::Value>,
641}
642
643impl RepresentationCompatibilityReport {
644 pub fn validate(&self) -> Result<()> {
645 validate_optional_non_empty(
646 "representation compatibility fallback_used",
647 &self.fallback_used,
648 )?;
649 if let Some(fingerprint) = &self.train_relation_fingerprint {
650 validate_fingerprint("representation compatibility train relation", fingerprint)?;
651 }
652 if let Some(fingerprint) = &self.predict_relation_fingerprint {
653 validate_fingerprint("representation compatibility predict relation", fingerprint)?;
654 }
655 validate_string_list_entries("representation compatibility messages", &self.messages)?;
656 for key in self.metadata.keys() {
657 if key.trim().is_empty() {
658 return Err(DagMlError::CampaignValidation(
659 "representation compatibility metadata contains an empty key".to_string(),
660 ));
661 }
662 }
663
664 let affected_total = self
665 .affected_source_count
666 .saturating_add(self.affected_repetition_count)
667 .saturating_add(self.affected_sample_count);
668 let relation_fingerprint_changed = matches!(
669 (
670 self.train_relation_fingerprint.as_deref(),
671 self.predict_relation_fingerprint.as_deref()
672 ),
673 (Some(train), Some(predict)) if train != predict
674 );
675 let unit_count_changed = matches!(
676 (self.train_unit_count, self.predict_unit_count),
677 (Some(train), Some(predict)) if train != predict
678 );
679 if affected_total == 0 {
680 if relation_fingerprint_changed {
681 return Err(DagMlError::CampaignValidation(
682 "representation compatibility relation fingerprint mismatch requires affected units"
683 .to_string(),
684 ));
685 }
686 if unit_count_changed {
687 return Err(DagMlError::CampaignValidation(
688 "representation compatibility unit count mismatch requires affected units"
689 .to_string(),
690 ));
691 }
692 if self.outcome == RepresentationCompatibilityOutcome::CompatibleWithFallback {
693 return Err(DagMlError::CampaignValidation(
694 "representation compatibility cannot use fallback when no units are affected"
695 .to_string(),
696 ));
697 }
698 if self.warning_severity.is_some() {
699 return Err(DagMlError::CampaignValidation(
700 "representation compatibility warning_severity requires affected units"
701 .to_string(),
702 ));
703 }
704 } else if self.policy == RepresentationMissingSourcePolicy::Strict {
705 if self.outcome != RepresentationCompatibilityOutcome::Incompatible {
706 return Err(DagMlError::CampaignValidation(
707 "strict representation compatibility with affected units must be incompatible"
708 .to_string(),
709 ));
710 }
711 if self.fallback_used.is_some() {
712 return Err(DagMlError::CampaignValidation(
713 "strict representation compatibility cannot declare fallback_used".to_string(),
714 ));
715 }
716 } else {
717 if self.warning_severity.is_none() {
718 return Err(DagMlError::CampaignValidation(
719 "non-strict representation compatibility with affected units requires warning_severity"
720 .to_string(),
721 ));
722 }
723 if self.outcome == RepresentationCompatibilityOutcome::Compatible {
724 return Err(DagMlError::CampaignValidation(
725 "representation compatibility with affected units cannot be compatible"
726 .to_string(),
727 ));
728 }
729 if self.outcome == RepresentationCompatibilityOutcome::CompatibleWithFallback
730 && self.fallback_used.is_none()
731 {
732 return Err(DagMlError::CampaignValidation(
733 "compatible_with_fallback representation compatibility requires fallback_used"
734 .to_string(),
735 ));
736 }
737 }
738
739 if self.outcome == RepresentationCompatibilityOutcome::Incompatible
740 && self.fallback_used.is_some()
741 {
742 return Err(DagMlError::CampaignValidation(
743 "incompatible representation compatibility cannot declare fallback_used"
744 .to_string(),
745 ));
746 }
747
748 if self.fixed_width_required && unit_count_changed && !self.allows_fixed_width_fallback() {
749 if self.outcome == RepresentationCompatibilityOutcome::Incompatible {
750 return Ok(());
751 }
752 return Err(DagMlError::CampaignValidation(
753 "fixed-width representation compatibility mismatch requires mask or pad policy/fallback"
754 .to_string(),
755 ));
756 }
757 if self.cartesian_combo_count_changed && !self.final_reducer_stabilizes_output {
758 if self.outcome == RepresentationCompatibilityOutcome::Incompatible {
759 return Ok(());
760 }
761 return Err(DagMlError::CampaignValidation(
762 "cartesian representation combo count may vary only when final reducer stabilizes output"
763 .to_string(),
764 ));
765 }
766 if self.late_fusion_branch_delta && !self.allows_late_fusion_delta() {
767 if self.outcome == RepresentationCompatibilityOutcome::Incompatible {
768 return Ok(());
769 }
770 return Err(DagMlError::CampaignValidation(
771 "late-fusion source deltas require an explicit drop/impute/mask/partial-model/pad policy or fallback"
772 .to_string(),
773 ));
774 }
775 Ok(())
776 }
777
778 fn allows_fixed_width_fallback(&self) -> bool {
779 matches!(
780 self.policy,
781 RepresentationMissingSourcePolicy::Mask | RepresentationMissingSourcePolicy::Pad
782 ) || self
783 .fallback_used
784 .as_deref()
785 .is_some_and(|fallback| matches!(fallback, "mask" | "pad"))
786 }
787
788 fn allows_late_fusion_delta(&self) -> bool {
789 matches!(
790 self.policy,
791 RepresentationMissingSourcePolicy::DropIncomplete
792 | RepresentationMissingSourcePolicy::ImputeDeclared
793 | RepresentationMissingSourcePolicy::Mask
794 | RepresentationMissingSourcePolicy::PartialModel
795 | RepresentationMissingSourcePolicy::Pad
796 ) || self.fallback_used.as_deref().is_some_and(|fallback| {
797 matches!(
798 fallback,
799 "drop_incomplete" | "impute_declared" | "mask" | "partial_model" | "pad"
800 )
801 })
802 }
803}
804
805#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
806#[serde(deny_unknown_fields)]
807pub struct RepresentationReplayManifest {
808 pub manifest_id: String,
809 pub representation_plan: RepresentationPlan,
810 #[serde(default, skip_serializing_if = "Option::is_none")]
811 pub combination_plan: Option<CombinationPlan>,
812 pub output_unit_level: EntityUnitLevel,
813 #[serde(default, skip_serializing_if = "Option::is_none")]
814 pub output_representation: Option<String>,
815 #[serde(default, skip_serializing_if = "Option::is_none")]
816 pub relation_fingerprint: Option<String>,
817 #[serde(default, skip_serializing_if = "Option::is_none")]
818 pub feature_schema_fingerprint: Option<String>,
819 #[serde(default, skip_serializing_if = "Option::is_none")]
820 pub final_reduction_id: Option<String>,
821 #[serde(default, skip_serializing_if = "Vec::is_empty")]
822 pub sample_observation_mapping: Vec<RepresentationSampleObservationMapping>,
823 #[serde(default, skip_serializing_if = "Vec::is_empty")]
824 pub combo_selection: Vec<RepresentationComboSelectionRecord>,
825 #[serde(default, skip_serializing_if = "Vec::is_empty")]
826 pub qc_policy_refs: Vec<String>,
827 #[serde(default, skip_serializing_if = "Vec::is_empty")]
828 pub outlier_policy_refs: Vec<String>,
829 #[serde(default, skip_serializing_if = "Option::is_none")]
830 pub missing_source_policy: Option<RepresentationMissingSourcePolicy>,
831 #[serde(default, skip_serializing_if = "Option::is_none")]
832 pub missing_repetition_policy: Option<RepresentationMissingSourcePolicy>,
833 #[serde(default, skip_serializing_if = "Option::is_none")]
834 pub prediction_representation: Option<String>,
835 #[serde(default, skip_serializing_if = "Option::is_none")]
836 pub final_output_unit_level: Option<EntityUnitLevel>,
837 #[serde(default, skip_serializing_if = "Option::is_none")]
838 pub train_compatibility: Option<RepresentationCompatibilityReport>,
839 #[serde(default, skip_serializing_if = "Option::is_none")]
840 pub predict_compatibility: Option<RepresentationCompatibilityReport>,
841 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
842 pub metadata: BTreeMap<String, serde_json::Value>,
843}
844
845impl RepresentationReplayManifest {
846 pub fn validate(&self) -> Result<()> {
847 validate_non_empty("representation replay manifest_id", &self.manifest_id)?;
848 self.representation_plan.validate()?;
849 if let Some(combination_plan) = &self.combination_plan {
850 combination_plan.validate()?;
851 }
852 if self.output_unit_level != self.representation_plan.output_unit_level() {
853 return Err(DagMlError::CampaignValidation(
854 "representation replay output_unit_level must match representation_plan"
855 .to_string(),
856 ));
857 }
858 validate_optional_non_empty(
859 "representation replay output_representation",
860 &self.output_representation,
861 )?;
862 validate_optional_non_empty(
863 "representation replay final_reduction_id",
864 &self.final_reduction_id,
865 )?;
866 validate_string_list_entries("representation replay qc_policy_refs", &self.qc_policy_refs)?;
867 validate_unique_strings("representation replay qc_policy_refs", &self.qc_policy_refs)?;
868 validate_string_list_entries(
869 "representation replay outlier_policy_refs",
870 &self.outlier_policy_refs,
871 )?;
872 validate_unique_strings(
873 "representation replay outlier_policy_refs",
874 &self.outlier_policy_refs,
875 )?;
876 validate_optional_non_empty(
877 "representation replay prediction_representation",
878 &self.prediction_representation,
879 )?;
880 let mut sample_source_pairs = BTreeSet::new();
881 for mapping in &self.sample_observation_mapping {
882 mapping.validate()?;
883 if !sample_source_pairs.insert((
884 mapping.physical_sample_id.as_str(),
885 mapping.source_id.as_str(),
886 )) {
887 return Err(DagMlError::CampaignValidation(format!(
888 "representation replay sample_observation_mapping contains duplicate physical_sample_id/source_id `{}`/`{}`",
889 mapping.physical_sample_id, mapping.source_id
890 )));
891 }
892 }
893 let mut combo_unit_ids = BTreeSet::new();
894 for record in &self.combo_selection {
895 record.validate()?;
896 if !combo_unit_ids.insert(record.combo_unit_id.as_str()) {
897 return Err(DagMlError::CampaignValidation(format!(
898 "representation replay combo_selection contains duplicate combo_unit_id `{}`",
899 record.combo_unit_id
900 )));
901 }
902 }
903 if let Some(report) = &self.train_compatibility {
904 report.validate()?;
905 }
906 if let Some(report) = &self.predict_compatibility {
907 report.validate()?;
908 }
909 if let Some(fingerprint) = &self.relation_fingerprint {
910 validate_fingerprint("representation replay relation", fingerprint)?;
911 }
912 if let Some(fingerprint) = &self.feature_schema_fingerprint {
913 validate_fingerprint("representation replay feature schema", fingerprint)?;
914 }
915 for key in self.metadata.keys() {
916 if key.trim().is_empty() {
917 return Err(DagMlError::CampaignValidation(
918 "representation replay metadata contains an empty key".to_string(),
919 ));
920 }
921 }
922 Ok(())
923 }
924}
925
926#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
927#[serde(deny_unknown_fields)]
928pub struct ModelInputFusionPolicy {
929 pub mode: ModelInputFusionMode,
930 #[serde(default)]
931 pub alignment: Option<String>,
932 #[serde(default)]
933 pub adapter_id: Option<String>,
934 #[serde(default, skip_serializing_if = "Option::is_none")]
935 pub representation_plan: Option<RepresentationPlan>,
936 #[serde(default)]
937 pub params: BTreeMap<String, serde_json::Value>,
938}
939
940impl ModelInputFusionPolicy {
941 pub fn validate(&self) -> Result<()> {
942 if self
943 .alignment
944 .as_ref()
945 .is_some_and(|alignment| alignment.trim().is_empty())
946 {
947 return Err(DagMlError::CampaignValidation(
948 "model input fusion policy has empty alignment".to_string(),
949 ));
950 }
951 if self
952 .adapter_id
953 .as_ref()
954 .is_some_and(|adapter_id| adapter_id.trim().is_empty())
955 {
956 return Err(DagMlError::CampaignValidation(
957 "model input fusion policy has empty adapter_id".to_string(),
958 ));
959 }
960 if self.mode == ModelInputFusionMode::Custom && self.adapter_id.is_none() {
961 return Err(DagMlError::CampaignValidation(
962 "custom model input fusion policy requires adapter_id".to_string(),
963 ));
964 }
965 if let Some(representation_plan) = &self.representation_plan {
966 representation_plan.validate()?;
967 }
968 for key in self.params.keys() {
969 if key.trim().is_empty() {
970 return Err(DagMlError::CampaignValidation(
971 "model input fusion policy contains an empty param key".to_string(),
972 ));
973 }
974 }
975 Ok(())
976 }
977}
978
979#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
980#[serde(deny_unknown_fields)]
981pub struct ModelInputPortSpec {
982 pub name: String,
983 pub accepted_representations: Vec<String>,
984 pub accepted_types: Vec<String>,
985 #[serde(default)]
986 pub rank: Option<u32>,
987 #[serde(default)]
988 pub multi_source: bool,
989 #[serde(default)]
990 pub optional: bool,
991 #[serde(default)]
992 pub metadata: BTreeMap<String, serde_json::Value>,
993}
994
995impl ModelInputPortSpec {
996 pub fn validate(&self) -> Result<()> {
997 validate_non_empty("model input port name", &self.name)?;
998 validate_non_empty_list(
999 "model input port accepted_representations",
1000 &self.accepted_representations,
1001 )?;
1002 validate_non_empty_list("model input port accepted_types", &self.accepted_types)?;
1003 validate_unique_strings(
1004 "model input port accepted_representations",
1005 &self.accepted_representations,
1006 )?;
1007 validate_unique_strings("model input port accepted_types", &self.accepted_types)?;
1008 if self.rank.is_some_and(|rank| rank > 16) {
1009 return Err(DagMlError::CampaignValidation(format!(
1010 "model input port `{}` rank must be <= 16",
1011 self.name
1012 )));
1013 }
1014 for key in self.metadata.keys() {
1015 if key.trim().is_empty() {
1016 return Err(DagMlError::CampaignValidation(format!(
1017 "model input port `{}` contains an empty metadata key",
1018 self.name
1019 )));
1020 }
1021 }
1022 Ok(())
1023 }
1024}
1025
1026#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1027#[serde(deny_unknown_fields)]
1028pub struct ModelInputSpec {
1029 #[serde(default = "default_model_input_spec_schema_version")]
1030 pub schema_version: u32,
1031 pub ports: Vec<ModelInputPortSpec>,
1032 #[serde(default)]
1033 pub default_fusion: Option<ModelInputFusionPolicy>,
1034 #[serde(default, skip_serializing_if = "Option::is_none")]
1035 pub fit_influence_policy: Option<FitInfluencePolicy>,
1036 #[serde(default)]
1037 pub metadata: BTreeMap<String, serde_json::Value>,
1038}
1039
1040impl ModelInputSpec {
1041 pub fn validate(&self) -> Result<()> {
1042 if self.schema_version != MODEL_INPUT_SPEC_SCHEMA_VERSION {
1043 return Err(DagMlError::CampaignValidation(format!(
1044 "model input spec uses unsupported schema_version {}, expected {}",
1045 self.schema_version, MODEL_INPUT_SPEC_SCHEMA_VERSION
1046 )));
1047 }
1048 if self.ports.is_empty() {
1049 return Err(DagMlError::CampaignValidation(
1050 "model input spec must declare at least one port".to_string(),
1051 ));
1052 }
1053 let mut names = BTreeSet::new();
1054 for port in &self.ports {
1055 port.validate()?;
1056 if !names.insert(port.name.as_str()) {
1057 return Err(DagMlError::CampaignValidation(format!(
1058 "model input spec contains duplicate port `{}`",
1059 port.name
1060 )));
1061 }
1062 }
1063 if let Some(default_fusion) = &self.default_fusion {
1064 default_fusion.validate()?;
1065 }
1066 for key in self.metadata.keys() {
1067 if key.trim().is_empty() {
1068 return Err(DagMlError::CampaignValidation(
1069 "model input spec contains an empty metadata key".to_string(),
1070 ));
1071 }
1072 }
1073 Ok(())
1074 }
1075}
1076
1077#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1078#[serde(rename_all = "snake_case")]
1079pub enum DataPlanStepKind {
1080 Materialize,
1081 Adapt,
1082 Align,
1083 Join,
1084 Collate,
1085}
1086
1087#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1088#[serde(deny_unknown_fields)]
1089pub struct DataPlanStep {
1090 pub kind: DataPlanStepKind,
1091 #[serde(default)]
1092 pub inputs: Vec<String>,
1093 pub output: String,
1094 #[serde(default)]
1095 pub adapter_id: Option<String>,
1096 #[serde(default)]
1097 pub params: BTreeMap<String, serde_json::Value>,
1098}
1099
1100impl DataPlanStep {
1101 pub fn validate(&self, previous_outputs: &BTreeSet<String>) -> Result<()> {
1102 validate_non_empty("data plan step output", &self.output)?;
1103 if self.kind != DataPlanStepKind::Materialize && self.inputs.is_empty() {
1104 return Err(DagMlError::CampaignValidation(format!(
1105 "data plan step `{}` requires at least one input",
1106 self.output
1107 )));
1108 }
1109 for (index, input) in self.inputs.iter().enumerate() {
1110 validate_non_empty("data plan step input", input)?;
1111 if self.kind != DataPlanStepKind::Materialize && !previous_outputs.contains(input) {
1112 return Err(DagMlError::CampaignValidation(format!(
1113 "data plan step `{}` input #{index} references `{input}` before it is produced",
1114 self.output
1115 )));
1116 }
1117 }
1118 if self
1119 .adapter_id
1120 .as_ref()
1121 .is_some_and(|adapter_id| adapter_id.trim().is_empty())
1122 {
1123 return Err(DagMlError::CampaignValidation(format!(
1124 "data plan step `{}` has empty adapter_id",
1125 self.output
1126 )));
1127 }
1128 for key in self.params.keys() {
1129 if key.trim().is_empty() {
1130 return Err(DagMlError::CampaignValidation(format!(
1131 "data plan step `{}` contains an empty param key",
1132 self.output
1133 )));
1134 }
1135 }
1136 Ok(())
1137 }
1138}
1139
1140#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1141#[serde(deny_unknown_fields)]
1142pub struct DataPlan {
1143 #[serde(default = "default_data_plan_schema_version")]
1144 pub schema_version: u32,
1145 pub id: String,
1146 pub steps: Vec<DataPlanStep>,
1147 pub output_ports: BTreeMap<String, String>,
1148 #[serde(default)]
1149 pub warnings: Vec<String>,
1150 #[serde(default)]
1151 pub requires_user_choice: Vec<String>,
1152 #[serde(default)]
1153 pub metadata: BTreeMap<String, serde_json::Value>,
1154}
1155
1156impl DataPlan {
1157 pub fn validate(&self) -> Result<()> {
1158 if self.schema_version != DATA_PLAN_SCHEMA_VERSION {
1159 return Err(DagMlError::CampaignValidation(format!(
1160 "data plan uses unsupported schema_version {}, expected {}",
1161 self.schema_version, DATA_PLAN_SCHEMA_VERSION
1162 )));
1163 }
1164 validate_non_empty("data plan id", &self.id)?;
1165 if self.steps.is_empty() {
1166 return Err(DagMlError::CampaignValidation(format!(
1167 "data plan `{}` must contain at least one step",
1168 self.id
1169 )));
1170 }
1171 let mut outputs = BTreeSet::new();
1172 for step in &self.steps {
1173 step.validate(&outputs)?;
1174 if !outputs.insert(step.output.clone()) {
1175 return Err(DagMlError::CampaignValidation(format!(
1176 "data plan `{}` contains duplicate step output `{}`",
1177 self.id, step.output
1178 )));
1179 }
1180 }
1181 if self.output_ports.is_empty() {
1182 return Err(DagMlError::CampaignValidation(format!(
1183 "data plan `{}` must declare at least one output port",
1184 self.id
1185 )));
1186 }
1187 for (port_name, output) in &self.output_ports {
1188 validate_non_empty("data plan output port", port_name)?;
1189 validate_non_empty("data plan output reference", output)?;
1190 if !outputs.contains(output) {
1191 return Err(DagMlError::CampaignValidation(format!(
1192 "data plan `{}` output port `{port_name}` references unknown output `{output}`",
1193 self.id
1194 )));
1195 }
1196 }
1197 validate_string_list_entries("data plan warnings", &self.warnings)?;
1198 validate_string_list_entries("data plan requires_user_choice", &self.requires_user_choice)?;
1199 for key in self.metadata.keys() {
1200 if key.trim().is_empty() {
1201 return Err(DagMlError::CampaignValidation(format!(
1202 "data plan `{}` contains an empty metadata key",
1203 self.id
1204 )));
1205 }
1206 }
1207 Ok(())
1208 }
1209}
1210
1211#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1212pub struct DataViewPolicy {
1213 #[serde(default = "default_fit_partition")]
1214 pub fit_partition: DataRequestPartition,
1215 #[serde(default = "default_predict_partition")]
1216 pub predict_partition: DataRequestPartition,
1217 #[serde(default)]
1218 pub include_augmented_train: bool,
1219 #[serde(default)]
1220 pub include_augmented_validation: bool,
1221 #[serde(default)]
1222 pub include_excluded: bool,
1223 #[serde(default = "default_true")]
1224 pub require_sample_ids: bool,
1225 #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
1226 pub unsafe_flags: BTreeSet<String>,
1227}
1228
1229impl Default for DataViewPolicy {
1230 fn default() -> Self {
1231 Self {
1232 fit_partition: DataRequestPartition::FoldTrain,
1233 predict_partition: DataRequestPartition::FoldValidation,
1234 include_augmented_train: true,
1235 include_augmented_validation: false,
1236 include_excluded: false,
1237 require_sample_ids: true,
1238 unsafe_flags: BTreeSet::new(),
1239 }
1240 }
1241}
1242
1243impl DataViewPolicy {
1244 pub const ALLOW_FIT_CV_FULL_TRAIN_VIEW: &'static str = "allow_fit_cv_full_train_view";
1245 pub const ALLOW_FIT_CV_VALIDATION_VIEW: &'static str = "allow_fit_cv_validation_view";
1246 pub const ALLOW_AUGMENTED_VALIDATION_VIEW: &'static str = "allow_augmented_validation_view";
1247 pub const ALLOW_EXCLUDED_ROWS: &'static str = "allow_excluded_rows";
1248
1249 pub fn validate(&self) -> Result<()> {
1250 for unsafe_flag in &self.unsafe_flags {
1251 if unsafe_flag.trim().is_empty() {
1252 return Err(DagMlError::CampaignValidation(
1253 "data view policy contains an empty unsafe flag".to_string(),
1254 ));
1255 }
1256 }
1257 match self.fit_partition {
1258 DataRequestPartition::FoldTrain => {}
1259 DataRequestPartition::FullTrain
1260 if self
1261 .unsafe_flags
1262 .contains(Self::ALLOW_FIT_CV_FULL_TRAIN_VIEW) => {}
1263 DataRequestPartition::FoldValidation
1264 if self
1265 .unsafe_flags
1266 .contains(Self::ALLOW_FIT_CV_VALIDATION_VIEW) => {}
1267 DataRequestPartition::FullTrain => {
1268 return Err(DagMlError::CampaignValidation(
1269 "data view policy fit_partition=full_train would leak validation rows during FIT_CV; add explicit unsafe flag allow_fit_cv_full_train_view".to_string(),
1270 ));
1271 }
1272 DataRequestPartition::FoldValidation => {
1273 return Err(DagMlError::CampaignValidation(
1274 "data view policy fit_partition=fold_validation would train on validation rows during FIT_CV; add explicit unsafe flag allow_fit_cv_validation_view".to_string(),
1275 ));
1276 }
1277 DataRequestPartition::Predict => {
1278 return Err(DagMlError::CampaignValidation(
1279 "data view policy fit_partition=predict is not valid for FIT_CV".to_string(),
1280 ));
1281 }
1282 }
1283 match self.predict_partition {
1284 DataRequestPartition::FoldValidation | DataRequestPartition::Predict => {}
1285 DataRequestPartition::FoldTrain | DataRequestPartition::FullTrain => {
1286 return Err(DagMlError::CampaignValidation(format!(
1287 "data view policy predict_partition={:?} is not valid for validation/predict views",
1288 self.predict_partition
1289 )));
1290 }
1291 }
1292 if self.include_augmented_validation
1293 && !self
1294 .unsafe_flags
1295 .contains(Self::ALLOW_AUGMENTED_VALIDATION_VIEW)
1296 {
1297 return Err(DagMlError::CampaignValidation(
1298 "data view policy include_augmented_validation=true can leak augmented validation/test rows; add explicit unsafe flag allow_augmented_validation_view".to_string(),
1299 ));
1300 }
1301 if self.include_excluded && !self.unsafe_flags.contains(Self::ALLOW_EXCLUDED_ROWS) {
1302 return Err(DagMlError::CampaignValidation(
1303 "data view policy include_excluded=true requires explicit unsafe flag allow_excluded_rows".to_string(),
1304 ));
1305 }
1306 Ok(())
1307 }
1308}
1309
1310fn default_fit_partition() -> DataRequestPartition {
1311 DataRequestPartition::FoldTrain
1312}
1313
1314fn default_predict_partition() -> DataRequestPartition {
1315 DataRequestPartition::FoldValidation
1316}
1317
1318fn default_true() -> bool {
1319 true
1320}
1321
1322#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1323pub struct DataBinding {
1324 pub node_id: NodeId,
1325 pub input_name: String,
1326 pub request_id: String,
1327 pub schema_fingerprint: String,
1328 pub plan_fingerprint: String,
1329 #[serde(default)]
1330 pub relation_fingerprint: Option<String>,
1331 pub output_representation: String,
1332 #[serde(default)]
1333 pub feature_set_id: Option<String>,
1334 #[serde(default)]
1335 pub source_ids: Vec<String>,
1336 #[serde(default)]
1337 pub require_relations: bool,
1338 #[serde(default)]
1339 pub view_policy: DataViewPolicy,
1340 #[serde(default)]
1341 pub metadata: BTreeMap<String, serde_json::Value>,
1342}
1343
1344pub fn data_binding_requirement_key(node_id: &NodeId, input_name: &str) -> String {
1352 format!("{node_id}.{input_name}")
1353}
1354
1355impl DataBinding {
1356 pub fn validate(&self) -> Result<()> {
1357 self.view_policy.validate()?;
1358 if self.input_name.trim().is_empty() {
1359 return Err(DagMlError::CampaignValidation(format!(
1360 "data binding for `{}` has empty input_name",
1361 self.node_id
1362 )));
1363 }
1364 if self.request_id.trim().is_empty() {
1365 return Err(DagMlError::CampaignValidation(format!(
1366 "data binding `{}` on `{}` has empty request_id",
1367 self.input_name, self.node_id
1368 )));
1369 }
1370 validate_fingerprint("schema", &self.schema_fingerprint)?;
1371 validate_fingerprint("plan", &self.plan_fingerprint)?;
1372 if let Some(relation_fingerprint) = &self.relation_fingerprint {
1373 validate_fingerprint("relation", relation_fingerprint)?;
1374 } else if self.require_relations {
1375 return Err(DagMlError::CampaignValidation(format!(
1376 "data binding `{}` on `{}` requires relations but has no relation_fingerprint",
1377 self.input_name, self.node_id
1378 )));
1379 }
1380 if self.output_representation.trim().is_empty() {
1381 return Err(DagMlError::CampaignValidation(format!(
1382 "data binding `{}` on `{}` has empty output_representation",
1383 self.input_name, self.node_id
1384 )));
1385 }
1386 if let Some(feature_set_id) = &self.feature_set_id {
1387 if feature_set_id.trim().is_empty() {
1388 return Err(DagMlError::CampaignValidation(format!(
1389 "data binding `{}` on `{}` has empty feature_set_id",
1390 self.input_name, self.node_id
1391 )));
1392 }
1393 }
1394 for source_id in &self.source_ids {
1395 if source_id.trim().is_empty() {
1396 return Err(DagMlError::CampaignValidation(format!(
1397 "data binding `{}` on `{}` has empty source id",
1398 self.input_name, self.node_id
1399 )));
1400 }
1401 }
1402 validate_unique_strings(
1403 &format!(
1404 "data binding `{}` on `{}` source_ids",
1405 self.input_name, self.node_id
1406 ),
1407 &self.source_ids,
1408 )?;
1409 validate_source_index_metadata(
1410 &format!(
1411 "data binding `{}` on `{}` metadata.source_index",
1412 self.input_name, self.node_id
1413 ),
1414 self.metadata.get(SOURCE_INDEX_METADATA_KEY),
1415 &self.source_ids,
1416 )?;
1417 Ok(())
1418 }
1419
1420 pub fn feature_set_id(&self) -> &str {
1421 self.feature_set_id.as_deref().unwrap_or(&self.input_name)
1422 }
1423
1424 pub fn validate_envelope(&self, envelope: &ExternalDataPlanEnvelope) -> Result<()> {
1425 self.validate()?;
1426 envelope.validate()?;
1427 if self.schema_fingerprint != envelope.schema_fingerprint {
1428 return Err(DagMlError::CampaignValidation(format!(
1429 "data binding `{}` on `{}` schema fingerprint mismatch",
1430 self.input_name, self.node_id
1431 )));
1432 }
1433 if self.plan_fingerprint != envelope.plan_fingerprint {
1434 return Err(DagMlError::CampaignValidation(format!(
1435 "data binding `{}` on `{}` plan fingerprint mismatch",
1436 self.input_name, self.node_id
1437 )));
1438 }
1439 if self.relation_fingerprint != envelope.relation_fingerprint {
1440 return Err(DagMlError::CampaignValidation(format!(
1441 "data binding `{}` on `{}` relation fingerprint mismatch",
1442 self.input_name, self.node_id
1443 )));
1444 }
1445 if self.require_relations && envelope.coordinator_relations.is_none() {
1446 return Err(DagMlError::CampaignValidation(format!(
1447 "data binding `{}` on `{}` requires coordinator relations",
1448 self.input_name, self.node_id
1449 )));
1450 }
1451 Ok(())
1452 }
1453}
1454
1455pub(crate) fn validate_source_index_metadata(
1456 label: &str,
1457 value: Option<&serde_json::Value>,
1458 expected_sources: &[String],
1459) -> Result<()> {
1460 let Some(value) = value else {
1461 return Ok(());
1462 };
1463 let Some(source_index) = value.as_object() else {
1464 return Err(DagMlError::CampaignValidation(format!(
1465 "{label} must be an object mapping source id to feature-axis block index"
1466 )));
1467 };
1468 if source_index.is_empty() {
1469 return Err(DagMlError::CampaignValidation(format!(
1470 "{label} must not be empty"
1471 )));
1472 }
1473 let mut seen_indices = BTreeSet::new();
1474 for (source_id, index_value) in source_index {
1475 if source_id.trim().is_empty() {
1476 return Err(DagMlError::CampaignValidation(format!(
1477 "{label} contains an empty source id"
1478 )));
1479 }
1480 let Some(index) = index_value.as_u64() else {
1481 return Err(DagMlError::CampaignValidation(format!(
1482 "{label} entry `{source_id}` must be a non-negative integer"
1483 )));
1484 };
1485 if !seen_indices.insert(index) {
1486 return Err(DagMlError::CampaignValidation(format!(
1487 "{label} contains duplicate feature-axis block index `{index}`"
1488 )));
1489 }
1490 }
1491 if !expected_sources.is_empty() {
1492 let actual = source_index.keys().cloned().collect::<BTreeSet<_>>();
1493 let expected = expected_sources.iter().cloned().collect::<BTreeSet<_>>();
1494 if actual != expected {
1495 return Err(DagMlError::CampaignValidation(format!(
1496 "{label} keys must match data binding source_ids"
1497 )));
1498 }
1499 }
1500 Ok(())
1501}
1502
1503#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1509#[serde(rename_all = "snake_case")]
1510pub enum PredictCohortRole {
1511 ExternalTest,
1512 Inference,
1513}
1514
1515#[derive(Serialize)]
1516struct PredictCohortFingerprintInput<'a> {
1517 role: PredictCohortRole,
1518 physical_sample_ids: &'a [SampleId],
1519 origin_sample_ids: &'a [SampleId],
1520 target_names: &'a [String],
1521 relation_fingerprint: &'a str,
1522 relations: &'a SampleRelationSet,
1523 data_content_fingerprint: &'a str,
1524 target_content_fingerprint: Option<&'a str>,
1525}
1526
1527#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1534#[serde(deny_unknown_fields)]
1535pub struct PredictCohort {
1536 pub role: PredictCohortRole,
1537 pub physical_sample_ids: Vec<SampleId>,
1538 pub origin_sample_ids: Vec<SampleId>,
1539 pub target_names: Vec<String>,
1540 pub relation_fingerprint: String,
1541 pub relations: SampleRelationSet,
1542 pub data_content_fingerprint: String,
1543 #[serde(default, skip_serializing_if = "Option::is_none")]
1544 pub target_content_fingerprint: Option<String>,
1545 pub cohort_fingerprint: String,
1546}
1547
1548impl PredictCohort {
1549 pub fn from_relations(
1558 role: PredictCohortRole,
1559 relations: SampleRelationSet,
1560 target_names: Vec<String>,
1561 data_content_fingerprint: String,
1562 target_content_fingerprint: Option<String>,
1563 ) -> Result<Self> {
1564 relations.validate()?;
1565 let physical_sample_ids = relations
1566 .records
1567 .iter()
1568 .map(|record| record.sample_id.clone())
1569 .collect::<BTreeSet<_>>()
1570 .into_iter()
1571 .collect();
1572 let origin_sample_ids = relations
1573 .records
1574 .iter()
1575 .map(|record| {
1576 record
1577 .origin_sample_id
1578 .clone()
1579 .unwrap_or_else(|| record.sample_id.clone())
1580 })
1581 .collect::<BTreeSet<_>>()
1582 .into_iter()
1583 .collect();
1584 let relation_fingerprint = relations.fingerprint()?;
1585 let mut cohort = Self {
1586 role,
1587 physical_sample_ids,
1588 origin_sample_ids,
1589 target_names,
1590 relation_fingerprint,
1591 relations,
1592 data_content_fingerprint,
1593 target_content_fingerprint,
1594 cohort_fingerprint: String::new(),
1595 };
1596 cohort.validate_members()?;
1597 cohort.cohort_fingerprint = cohort.fingerprint()?;
1598 Ok(cohort)
1599 }
1600
1601 fn validate_members(&self) -> Result<()> {
1602 validate_sorted_unique_sample_ids(
1603 "predict cohort physical_sample_ids",
1604 &self.physical_sample_ids,
1605 )?;
1606 validate_sorted_unique_sample_ids(
1607 "predict cohort origin_sample_ids",
1608 &self.origin_sample_ids,
1609 )?;
1610 validate_string_list_entries("predict cohort target_names", &self.target_names)?;
1611 validate_unique_strings("predict cohort target_names", &self.target_names)?;
1612 validate_fingerprint("predict cohort relation", &self.relation_fingerprint)?;
1613 validate_fingerprint(
1614 "predict cohort data content",
1615 &self.data_content_fingerprint,
1616 )?;
1617 if let Some(target_content_fingerprint) = &self.target_content_fingerprint {
1618 validate_fingerprint("predict cohort target content", target_content_fingerprint)?;
1619 }
1620 match self.role {
1621 PredictCohortRole::ExternalTest if self.target_content_fingerprint.is_none() => {
1622 return Err(DagMlError::CampaignValidation(
1623 "external_test predict cohort requires target_content_fingerprint".to_string(),
1624 ));
1625 }
1626 PredictCohortRole::Inference if self.target_content_fingerprint.is_some() => {
1627 return Err(DagMlError::CampaignValidation(
1628 "inference predict cohort must not carry target_content_fingerprint"
1629 .to_string(),
1630 ));
1631 }
1632 _ => {}
1633 }
1634 self.relations.validate()?;
1635 let relation_fingerprint = self.relations.fingerprint()?;
1636 if self.relation_fingerprint != relation_fingerprint {
1637 return Err(DagMlError::CampaignValidation(
1638 "predict cohort relation_fingerprint does not match relations".to_string(),
1639 ));
1640 }
1641 let relation_samples = self
1642 .relations
1643 .records
1644 .iter()
1645 .map(|record| record.sample_id.clone())
1646 .collect::<BTreeSet<_>>();
1647 let relation_origins = self
1648 .relations
1649 .records
1650 .iter()
1651 .map(|record| {
1652 record
1653 .origin_sample_id
1654 .clone()
1655 .unwrap_or_else(|| record.sample_id.clone())
1656 })
1657 .collect::<BTreeSet<_>>();
1658 if relation_samples != self.physical_sample_ids.iter().cloned().collect() {
1659 return Err(DagMlError::CampaignValidation(
1660 "predict cohort physical_sample_ids do not exactly cover relation samples"
1661 .to_string(),
1662 ));
1663 }
1664 if relation_origins != self.origin_sample_ids.iter().cloned().collect() {
1665 return Err(DagMlError::CampaignValidation(
1666 "predict cohort origin_sample_ids do not exactly cover relation origins"
1667 .to_string(),
1668 ));
1669 }
1670 Ok(())
1671 }
1672
1673 pub fn fingerprint(&self) -> Result<String> {
1674 self.validate_members()?;
1675 stable_json_fingerprint(&PredictCohortFingerprintInput {
1676 role: self.role,
1677 physical_sample_ids: &self.physical_sample_ids,
1678 origin_sample_ids: &self.origin_sample_ids,
1679 target_names: &self.target_names,
1680 relation_fingerprint: &self.relation_fingerprint,
1681 relations: &self.relations,
1682 data_content_fingerprint: &self.data_content_fingerprint,
1683 target_content_fingerprint: self.target_content_fingerprint.as_deref(),
1684 })
1685 }
1686
1687 pub fn validate(&self) -> Result<()> {
1688 self.validate_members()?;
1689 validate_fingerprint("predict cohort", &self.cohort_fingerprint)?;
1690 if self.cohort_fingerprint != self.fingerprint()? {
1691 return Err(DagMlError::CampaignValidation(
1692 "predict cohort fingerprint does not match canonical content".to_string(),
1693 ));
1694 }
1695 Ok(())
1696 }
1697
1698 pub fn validate_against_cv_fold_set(&self, fold_set: &FoldSet) -> Result<()> {
1702 self.validate()?;
1703 fold_set.validate()?;
1704 if self.role == PredictCohortRole::Inference {
1705 return Ok(());
1706 }
1707 let cv_ids = fold_set.sample_ids.iter().collect::<BTreeSet<_>>();
1708 for sample_id in self
1709 .physical_sample_ids
1710 .iter()
1711 .chain(self.origin_sample_ids.iter())
1712 {
1713 if cv_ids.contains(sample_id) {
1714 return Err(DagMlError::CampaignValidation(format!(
1715 "external_test predict cohort overlaps CV fold sample or origin `{sample_id}`"
1716 )));
1717 }
1718 }
1719 Ok(())
1720 }
1721
1722 pub fn validate_against_cv_relations(&self, cv_relations: &SampleRelationSet) -> Result<()> {
1729 self.validate()?;
1730 cv_relations.validate()?;
1731 if self.role == PredictCohortRole::Inference {
1732 return Ok(());
1733 }
1734 let cv_identity_closure = cv_relations
1735 .records
1736 .iter()
1737 .flat_map(|record| {
1738 std::iter::once(record.sample_id.clone()).chain(record.origin_sample_id.clone())
1739 })
1740 .collect::<BTreeSet<_>>();
1741 for sample_id in self
1742 .physical_sample_ids
1743 .iter()
1744 .chain(self.origin_sample_ids.iter())
1745 {
1746 if cv_identity_closure.contains(sample_id) {
1747 return Err(DagMlError::CampaignValidation(format!(
1748 "external_test predict cohort overlaps CV relation identity closure `{sample_id}`"
1749 )));
1750 }
1751 }
1752 Ok(())
1753 }
1754}
1755
1756fn validate_sorted_unique_sample_ids(label: &str, values: &[SampleId]) -> Result<()> {
1757 if values.is_empty() {
1758 return Err(DagMlError::CampaignValidation(format!(
1759 "{label} must not be empty"
1760 )));
1761 }
1762 for pair in values.windows(2) {
1763 if pair[0] >= pair[1] {
1764 return Err(DagMlError::CampaignValidation(format!(
1765 "{label} must be strictly sorted and unique"
1766 )));
1767 }
1768 }
1769 Ok(())
1770}
1771
1772#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1773pub struct ExternalDataPlanEnvelope {
1774 #[serde(default = "default_external_data_plan_envelope_schema_version")]
1775 pub schema_version: u32,
1776 pub schema_fingerprint: String,
1777 pub plan_fingerprint: String,
1778 #[serde(default)]
1779 pub relation_fingerprint: Option<String>,
1780 #[serde(default, skip_serializing_if = "Option::is_none")]
1784 pub data_content_fingerprint: Option<String>,
1785 #[serde(default, skip_serializing_if = "Option::is_none")]
1789 pub target_content_fingerprint: Option<String>,
1790 #[serde(default)]
1791 pub coordinator_relations: Option<SampleRelationSet>,
1792 #[serde(default, skip_serializing_if = "Option::is_none")]
1795 pub predict_cohort: Option<PredictCohort>,
1796}
1797
1798impl ExternalDataPlanEnvelope {
1799 pub fn validate(&self) -> Result<()> {
1800 if !matches!(
1801 self.schema_version,
1802 EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1
1803 | EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V2
1804 ) {
1805 return Err(DagMlError::CampaignValidation(format!(
1806 "external data-plan envelope uses unsupported schema_version {}",
1807 self.schema_version
1808 )));
1809 }
1810 match (self.schema_version, &self.predict_cohort) {
1811 (EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1, Some(_)) => {
1812 return Err(DagMlError::CampaignValidation(
1813 "external data-plan envelope V1 cannot carry predict_cohort".to_string(),
1814 ));
1815 }
1816 (EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V2, None) => {
1817 return Err(DagMlError::CampaignValidation(
1818 "external data-plan envelope V2 requires predict_cohort".to_string(),
1819 ));
1820 }
1821 _ => {}
1822 }
1823 validate_fingerprint("schema", &self.schema_fingerprint)?;
1824 validate_fingerprint("plan", &self.plan_fingerprint)?;
1825 if let Some(relation_fingerprint) = &self.relation_fingerprint {
1826 validate_fingerprint("relation", relation_fingerprint)?;
1827 if self.coordinator_relations.is_none() {
1828 return Err(DagMlError::CampaignValidation(
1829 "relation_fingerprint requires coordinator_relations".to_string(),
1830 ));
1831 }
1832 }
1833 if let Some(data_content_fingerprint) = &self.data_content_fingerprint {
1834 validate_fingerprint("data content", data_content_fingerprint)?;
1835 }
1836 if let Some(target_content_fingerprint) = &self.target_content_fingerprint {
1837 validate_fingerprint("target content", target_content_fingerprint)?;
1838 }
1839 if let Some(relations) = &self.coordinator_relations {
1840 relations.validate()?;
1841 }
1842 if let Some(predict_cohort) = &self.predict_cohort {
1843 predict_cohort.validate()?;
1844 if predict_cohort.role == PredictCohortRole::ExternalTest {
1845 let cv_relations = self.coordinator_relations.as_ref().ok_or_else(|| {
1846 DagMlError::CampaignValidation(
1847 "external_test predict cohort requires coordinator_relations to prove CV disjointness"
1848 .to_string(),
1849 )
1850 })?;
1851 predict_cohort.validate_against_cv_relations(cv_relations)?;
1852 }
1853 }
1854 Ok(())
1855 }
1856}
1857
1858pub fn validate_data_binding_envelope(
1859 binding: &DataBinding,
1860 envelope: &ExternalDataPlanEnvelope,
1861) -> Result<()> {
1862 binding.validate_envelope(envelope)
1863}
1864
1865#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
1866struct DataEnvelopeKey {
1867 schema_fingerprint: String,
1868 plan_fingerprint: String,
1869 relation_fingerprint: Option<String>,
1870}
1871
1872impl DataEnvelopeKey {
1873 fn from_binding(binding: &DataBinding) -> Self {
1874 Self {
1875 schema_fingerprint: binding.schema_fingerprint.clone(),
1876 plan_fingerprint: binding.plan_fingerprint.clone(),
1877 relation_fingerprint: binding.relation_fingerprint.clone(),
1878 }
1879 }
1880
1881 fn from_envelope(envelope: &ExternalDataPlanEnvelope) -> Self {
1882 Self {
1883 schema_fingerprint: envelope.schema_fingerprint.clone(),
1884 plan_fingerprint: envelope.plan_fingerprint.clone(),
1885 relation_fingerprint: envelope.relation_fingerprint.clone(),
1886 }
1887 }
1888}
1889
1890#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1891pub struct DataHandleRecord {
1892 pub handle: HandleRef,
1893 pub run_id: RunId,
1894 pub node_id: NodeId,
1895 pub input_name: String,
1896 pub phase: Phase,
1897 pub variant_id: Option<VariantId>,
1898 pub fold_id: Option<FoldId>,
1899 pub request_id: String,
1900 pub schema_fingerprint: String,
1901 pub plan_fingerprint: String,
1902 pub relation_fingerprint: Option<String>,
1903 pub output_representation: String,
1904 #[serde(default)]
1905 pub feature_set_id: Option<String>,
1906 #[serde(default)]
1907 pub source_ids: Vec<String>,
1908 pub relation_record_count: Option<usize>,
1909}
1910
1911#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1912pub struct DataViewHandleRecord {
1913 pub handle: HandleRef,
1914 pub parent_handle: HandleRef,
1915 pub run_id: RunId,
1916 pub node_id: NodeId,
1917 pub input_name: String,
1918 pub phase: Phase,
1919 pub variant_id: Option<VariantId>,
1920 pub fold_id: Option<FoldId>,
1921 pub request_id: String,
1922 pub feature_set_id: String,
1923 pub view: DataProviderViewSpec,
1924}
1925
1926#[derive(Debug)]
1927pub struct InMemoryDataProvider {
1928 owner_controller: ControllerId,
1929 envelopes: BTreeMap<DataEnvelopeKey, ExternalDataPlanEnvelope>,
1930 next_handle: RefCell<u64>,
1931 records: RefCell<BTreeMap<u64, DataHandleRecord>>,
1932 view_records: RefCell<BTreeMap<u64, DataViewHandleRecord>>,
1933}
1934
1935impl InMemoryDataProvider {
1936 pub fn new(owner_controller: ControllerId) -> Self {
1937 Self {
1938 owner_controller,
1939 envelopes: BTreeMap::new(),
1940 next_handle: RefCell::new(1),
1941 records: RefCell::new(BTreeMap::new()),
1942 view_records: RefCell::new(BTreeMap::new()),
1943 }
1944 }
1945
1946 pub fn with_envelope(
1947 owner_controller: ControllerId,
1948 envelope: ExternalDataPlanEnvelope,
1949 ) -> Result<Self> {
1950 let mut provider = Self::new(owner_controller);
1951 provider.register_envelope(envelope)?;
1952 Ok(provider)
1953 }
1954
1955 pub fn register_envelope(&mut self, envelope: ExternalDataPlanEnvelope) -> Result<()> {
1956 envelope.validate()?;
1957 let key = DataEnvelopeKey::from_envelope(&envelope);
1958 if let Some(existing) = self.envelopes.get(&key) {
1959 if existing == &envelope {
1960 return Ok(());
1961 }
1962 return Err(DagMlError::RuntimeValidation(
1963 "duplicate external data-plan envelope with different payload".to_string(),
1964 ));
1965 }
1966 self.envelopes.insert(key, envelope);
1967 Ok(())
1968 }
1969
1970 pub fn handle_record(&self, handle: u64) -> Option<DataHandleRecord> {
1971 self.records.borrow().get(&handle).cloned()
1972 }
1973
1974 pub fn handle_records(&self) -> Vec<DataHandleRecord> {
1975 self.records.borrow().values().cloned().collect()
1976 }
1977
1978 pub fn view_record(&self, handle: u64) -> Option<DataViewHandleRecord> {
1979 self.view_records.borrow().get(&handle).cloned()
1980 }
1981
1982 pub fn view_records(&self) -> Vec<DataViewHandleRecord> {
1983 self.view_records.borrow().values().cloned().collect()
1984 }
1985
1986 fn next_handle(&self) -> u64 {
1987 let mut next = self.next_handle.borrow_mut();
1988 let handle = *next;
1989 *next += 1;
1990 handle
1991 }
1992}
1993
1994impl RuntimeDataProvider for InMemoryDataProvider {
1995 fn materialize(&self, request: &DataMaterializationRequest) -> Result<HandleRef> {
1996 if request.node_id != request.binding.node_id {
1997 return Err(DagMlError::RuntimeValidation(format!(
1998 "data materialization request node `{}` does not match binding node `{}`",
1999 request.node_id, request.binding.node_id
2000 )));
2001 }
2002 if request.input_name != request.binding.input_name {
2003 return Err(DagMlError::RuntimeValidation(format!(
2004 "data materialization request input `{}` does not match binding input `{}`",
2005 request.input_name, request.binding.input_name
2006 )));
2007 }
2008 let envelope = self
2009 .envelopes
2010 .get(&DataEnvelopeKey::from_binding(&request.binding))
2011 .ok_or_else(|| {
2012 DagMlError::RuntimeValidation(format!(
2013 "no external data-plan envelope registered for binding `{}` on `{}`",
2014 request.binding.input_name, request.binding.node_id
2015 ))
2016 })?;
2017 request.binding.validate_envelope(envelope)?;
2018
2019 let handle = HandleRef {
2020 handle: self.next_handle(),
2021 kind: HandleKind::Data,
2022 owner_controller: self.owner_controller.clone(),
2023 };
2024 let record = DataHandleRecord {
2025 handle: handle.clone(),
2026 run_id: request.run_id.clone(),
2027 node_id: request.node_id.clone(),
2028 input_name: request.input_name.clone(),
2029 phase: request.phase,
2030 variant_id: request.variant_id.clone(),
2031 fold_id: request.fold_id.clone(),
2032 request_id: request.binding.request_id.clone(),
2033 schema_fingerprint: request.binding.schema_fingerprint.clone(),
2034 plan_fingerprint: request.binding.plan_fingerprint.clone(),
2035 relation_fingerprint: request.binding.relation_fingerprint.clone(),
2036 output_representation: request.binding.output_representation.clone(),
2037 feature_set_id: request.binding.feature_set_id.clone(),
2038 source_ids: request.binding.source_ids.clone(),
2039 relation_record_count: envelope
2040 .coordinator_relations
2041 .as_ref()
2042 .map(|relations| relations.records.len()),
2043 };
2044 self.records.borrow_mut().insert(handle.handle, record);
2045 Ok(handle)
2046 }
2047
2048 fn make_view(&self, request: &DataViewRequest) -> Result<HandleRef> {
2049 request.view.validate()?;
2050 if request.node_id != request.binding.node_id {
2051 return Err(DagMlError::RuntimeValidation(format!(
2052 "data view request node `{}` does not match binding node `{}`",
2053 request.node_id, request.binding.node_id
2054 )));
2055 }
2056 if request.input_name != request.binding.input_name {
2057 return Err(DagMlError::RuntimeValidation(format!(
2058 "data view request input `{}` does not match binding input `{}`",
2059 request.input_name, request.binding.input_name
2060 )));
2061 }
2062 if request.data_handle.kind != HandleKind::Data {
2063 return Err(DagMlError::RuntimeValidation(format!(
2064 "data view request for `{}` on `{}` received non-data parent handle",
2065 request.input_name, request.node_id
2066 )));
2067 }
2068 let parent = self
2069 .records
2070 .borrow()
2071 .get(&request.data_handle.handle)
2072 .cloned()
2073 .ok_or_else(|| {
2074 DagMlError::RuntimeValidation(format!(
2075 "unknown data handle `{}` for view request `{}` on `{}`",
2076 request.data_handle.handle, request.input_name, request.node_id
2077 ))
2078 })?;
2079 if parent.handle != request.data_handle {
2080 return Err(DagMlError::RuntimeValidation(format!(
2081 "data view request parent handle `{}` does not match provider record",
2082 request.data_handle.handle
2083 )));
2084 }
2085 request.binding.validate()?;
2086 let handle = HandleRef {
2087 handle: self.next_handle(),
2088 kind: HandleKind::DataView,
2089 owner_controller: self.owner_controller.clone(),
2090 };
2091 let record = DataViewHandleRecord {
2092 handle: handle.clone(),
2093 parent_handle: request.data_handle.clone(),
2094 run_id: request.run_id.clone(),
2095 node_id: request.node_id.clone(),
2096 input_name: request.input_name.clone(),
2097 phase: request.phase,
2098 variant_id: request.variant_id.clone(),
2099 fold_id: request.fold_id.clone(),
2100 request_id: request.binding.request_id.clone(),
2101 feature_set_id: request.binding.feature_set_id().to_string(),
2102 view: request.view.clone(),
2103 };
2104 self.view_records.borrow_mut().insert(handle.handle, record);
2105 Ok(handle)
2106 }
2107
2108 fn training_data_identity(
2109 &self,
2110 binding: &DataBinding,
2111 ) -> Result<Option<crate::training::TrainingDataIdentity>> {
2112 let envelope = self
2113 .envelopes
2114 .get(&DataEnvelopeKey::from_binding(binding))
2115 .ok_or_else(|| {
2116 DagMlError::RuntimeValidation(format!(
2117 "no external data-plan envelope registered for binding `{}` on `{}`",
2118 binding.input_name, binding.node_id
2119 ))
2120 })?;
2121 binding.validate_envelope(envelope)?;
2122 if envelope.relation_fingerprint.is_none()
2123 || envelope.data_content_fingerprint.is_none()
2124 || envelope.target_content_fingerprint.is_none()
2125 {
2126 return Ok(None);
2127 }
2128 Ok(Some(
2129 crate::training::TrainingDataIdentity::from_binding_envelope(binding, envelope)?,
2130 ))
2131 }
2132
2133 fn coordinator_relations(&self, binding: &DataBinding) -> Result<Option<SampleRelationSet>> {
2134 let envelope = self
2135 .envelopes
2136 .get(&DataEnvelopeKey::from_binding(binding))
2137 .ok_or_else(|| {
2138 DagMlError::RuntimeValidation(format!(
2139 "no external data-plan envelope registered for binding `{}` on `{}`",
2140 binding.input_name, binding.node_id
2141 ))
2142 })?;
2143 binding.validate_envelope(envelope)?;
2144 Ok(envelope.coordinator_relations.clone())
2145 }
2146
2147 fn predict_cohort(&self, binding: &DataBinding, phase: Phase) -> Result<Option<PredictCohort>> {
2148 if phase != Phase::Predict {
2149 return Err(DagMlError::RuntimeValidation(format!(
2150 "predict cohort may be requested only during PREDICT, got {phase:?}"
2151 )));
2152 }
2153 let envelope = self
2154 .envelopes
2155 .get(&DataEnvelopeKey::from_binding(binding))
2156 .ok_or_else(|| {
2157 DagMlError::RuntimeValidation(format!(
2158 "no external data-plan envelope registered for binding `{}` on `{}`",
2159 binding.input_name, binding.node_id
2160 ))
2161 })?;
2162 binding.validate_envelope(envelope)?;
2163 Ok(envelope.predict_cohort.clone())
2164 }
2165}
2166
2167fn validate_fingerprint(label: &str, value: &str) -> Result<()> {
2168 if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
2169 return Err(DagMlError::CampaignValidation(format!(
2170 "{label} fingerprint must be a 64-character hex digest"
2171 )));
2172 }
2173 Ok(())
2174}
2175
2176fn validate_non_empty(label: &str, value: &str) -> Result<()> {
2177 if value.trim().is_empty() {
2178 return Err(DagMlError::CampaignValidation(format!(
2179 "{label} must be a non-empty string"
2180 )));
2181 }
2182 Ok(())
2183}
2184
2185fn validate_optional_non_empty(label: &str, value: &Option<String>) -> Result<()> {
2186 if let Some(value) = value {
2187 validate_non_empty(label, value)?;
2188 }
2189 Ok(())
2190}
2191
2192fn validate_combo_like_output(label: &str, unit_level: EntityUnitLevel) -> Result<()> {
2193 if matches!(
2194 unit_level,
2195 EntityUnitLevel::Combo | EntityUnitLevel::Observation
2196 ) {
2197 return Ok(());
2198 }
2199 Err(DagMlError::CampaignValidation(format!(
2200 "{label} representation output_unit_level must be combo or observation"
2201 )))
2202}
2203
2204fn validate_non_empty_list(label: &str, values: &[String]) -> Result<()> {
2205 if values.is_empty() {
2206 return Err(DagMlError::CampaignValidation(format!(
2207 "{label} must be a non-empty list"
2208 )));
2209 }
2210 validate_string_list_entries(label, values)
2211}
2212
2213fn validate_string_list_entries(label: &str, values: &[String]) -> Result<()> {
2214 for (index, value) in values.iter().enumerate() {
2215 if value.trim().is_empty() {
2216 return Err(DagMlError::CampaignValidation(format!(
2217 "{label}[{index}] must be a non-empty string"
2218 )));
2219 }
2220 }
2221 Ok(())
2222}
2223
2224fn validate_unique_strings(label: &str, values: &[String]) -> Result<()> {
2225 let mut seen = BTreeSet::new();
2226 for value in values {
2227 if !seen.insert(value.as_str()) {
2228 return Err(DagMlError::CampaignValidation(format!(
2229 "{label} contains duplicate value `{value}`"
2230 )));
2231 }
2232 }
2233 Ok(())
2234}
2235
2236#[cfg(test)]
2237mod tests {
2238 use super::*;
2239 use crate::fold::{FoldAssignment, FoldPartitionMode};
2240 use crate::ids::{ControllerId, FoldId, NodeId};
2241 use crate::runtime::{DataMaterializationRequest, RuntimeDataProvider};
2242
2243 fn binding() -> DataBinding {
2244 DataBinding {
2245 node_id: NodeId::new("model:base").unwrap(),
2246 input_name: "x".to_string(),
2247 request_id: "nir-to-tabular".to_string(),
2248 schema_fingerprint: "f97b37872fa22134b508f98fd8e207e5b776b52594fb8f6f5c3e15bee212246b"
2249 .to_string(),
2250 plan_fingerprint: "7c5431d85574b3f337022fa5d25971d5b5cf445b90331b49938f573ff6901e4d"
2251 .to_string(),
2252 relation_fingerprint: Some(
2253 "a3a7e329df35db9f2883a17b8611b7fae6dcaa031875e3ec2c9be1b9e29cbe10".to_string(),
2254 ),
2255 output_representation: "tabular_numeric".to_string(),
2256 feature_set_id: Some("x".to_string()),
2257 source_ids: vec!["nir".to_string()],
2258 require_relations: true,
2259 view_policy: DataViewPolicy::default(),
2260 metadata: BTreeMap::new(),
2261 }
2262 }
2263
2264 #[test]
2265 fn validates_data_binding_contract() {
2266 let binding = binding();
2267 binding.validate().unwrap();
2268 assert_eq!(binding.feature_set_id(), "x");
2269 }
2270
2271 #[cfg(dag_ml_workspace_contract_fixtures)]
2272 #[test]
2273 fn published_model_input_and_data_plan_schemas_declare_current_contract() {
2274 let model_input_schema: serde_json::Value = serde_json::from_str(include_str!(
2275 "../../../docs/contracts/model_input_spec.schema.json"
2276 ))
2277 .unwrap();
2278 assert_eq!(model_input_schema["$id"], MODEL_INPUT_SPEC_SCHEMA_ID);
2279 assert_eq!(
2280 model_input_schema["properties"]["schema_version"]["const"].as_u64(),
2281 Some(MODEL_INPUT_SPEC_SCHEMA_VERSION as u64)
2282 );
2283 assert!(model_input_schema["$defs"]["input_port"]["required"]
2284 .as_array()
2285 .unwrap()
2286 .iter()
2287 .any(|field| field.as_str() == Some("accepted_representations")));
2288 assert!(model_input_schema["$defs"]["fusion_policy"]["properties"]
2289 .as_object()
2290 .unwrap()
2291 .contains_key("representation_plan"));
2292 assert!(model_input_schema["$defs"]
2293 .as_object()
2294 .unwrap()
2295 .contains_key("combination_plan"));
2296 assert!(model_input_schema["$defs"]
2297 .as_object()
2298 .unwrap()
2299 .contains_key("representation_plan"));
2300
2301 let data_plan_schema: serde_json::Value = serde_json::from_str(include_str!(
2302 "../../../docs/contracts/data_plan.schema.json"
2303 ))
2304 .unwrap();
2305 assert_eq!(data_plan_schema["$id"], DATA_PLAN_SCHEMA_ID);
2306 assert_eq!(
2307 data_plan_schema["properties"]["schema_version"]["const"].as_u64(),
2308 Some(DATA_PLAN_SCHEMA_VERSION as u64)
2309 );
2310 assert!(data_plan_schema["$defs"]["data_plan_step_kind"]["enum"]
2311 .as_array()
2312 .unwrap()
2313 .iter()
2314 .any(|kind| kind.as_str() == Some("collate")));
2315 }
2316
2317 #[test]
2318 fn validates_model_input_and_data_plan_fixtures() {
2319 let model_input: ModelInputSpec = serde_json::from_str(include_str!(
2320 "../tests/fixtures/package/data/model_input_spec_tabular_regressor.json"
2321 ))
2322 .unwrap();
2323 model_input.validate().unwrap();
2324 assert_eq!(model_input.ports[0].rank, Some(2));
2325 assert!(model_input.ports[0].multi_source);
2326
2327 let data_plan: DataPlan = serde_json::from_str(include_str!(
2328 "../tests/fixtures/package/data/data_plan_tabular_fusion.json"
2329 ))
2330 .unwrap();
2331 data_plan.validate().unwrap();
2332 assert_eq!(data_plan.output_ports.get("x").unwrap(), "x_collated");
2333 }
2334
2335 #[test]
2336 fn data_plan_rejects_forward_step_references() {
2337 let data_plan = DataPlan {
2338 schema_version: DATA_PLAN_SCHEMA_VERSION,
2339 id: "data-plan:bad".to_string(),
2340 steps: vec![DataPlanStep {
2341 kind: DataPlanStepKind::Adapt,
2342 inputs: vec!["missing".to_string()],
2343 output: "adapted".to_string(),
2344 adapter_id: Some("adapter:adapt".to_string()),
2345 params: BTreeMap::new(),
2346 }],
2347 output_ports: BTreeMap::from([("x".to_string(), "adapted".to_string())]),
2348 warnings: Vec::new(),
2349 requires_user_choice: Vec::new(),
2350 metadata: BTreeMap::new(),
2351 };
2352
2353 let error = data_plan.validate().unwrap_err().to_string();
2354 assert!(error.contains("before it is produced"));
2355 }
2356
2357 #[test]
2358 fn data_view_policy_rejects_unsafe_fit_and_validation_augmentation_by_default() {
2359 let mut full_train_binding = binding();
2360 full_train_binding.view_policy.fit_partition = DataRequestPartition::FullTrain;
2361 let full_train_error = full_train_binding.validate().unwrap_err().to_string();
2362 assert!(
2363 full_train_error.contains("fit_partition=full_train"),
2364 "unexpected full-train error: {full_train_error}"
2365 );
2366
2367 let mut augmented_validation_binding = binding();
2368 augmented_validation_binding
2369 .view_policy
2370 .include_augmented_validation = true;
2371 let augmented_error = augmented_validation_binding
2372 .validate()
2373 .unwrap_err()
2374 .to_string();
2375 assert!(
2376 augmented_error.contains("include_augmented_validation=true"),
2377 "unexpected augmented-validation error: {augmented_error}"
2378 );
2379
2380 let mut excluded_binding = binding();
2381 excluded_binding.view_policy.include_excluded = true;
2382 let excluded_error = excluded_binding.validate().unwrap_err().to_string();
2383 assert!(
2384 excluded_error.contains("include_excluded=true"),
2385 "unexpected excluded-row error: {excluded_error}"
2386 );
2387 }
2388
2389 #[test]
2390 fn data_view_policy_requires_explicit_unsafe_flags_for_debug_views() {
2391 let mut binding = binding();
2392 binding.view_policy.fit_partition = DataRequestPartition::FullTrain;
2393 binding.view_policy.include_augmented_validation = true;
2394 binding.view_policy.include_excluded = true;
2395 binding.view_policy.unsafe_flags = BTreeSet::from([
2396 DataViewPolicy::ALLOW_FIT_CV_FULL_TRAIN_VIEW.to_string(),
2397 DataViewPolicy::ALLOW_AUGMENTED_VALIDATION_VIEW.to_string(),
2398 DataViewPolicy::ALLOW_EXCLUDED_ROWS.to_string(),
2399 ]);
2400
2401 binding.validate().unwrap();
2402 }
2403
2404 #[test]
2405 fn validates_external_data_envelope_subset() {
2406 let envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2407 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2408 ))
2409 .unwrap();
2410
2411 assert_eq!(
2412 envelope.schema_version,
2413 EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION
2414 );
2415 binding().validate_envelope(&envelope).unwrap();
2416 assert!(envelope.data_content_fingerprint.is_none());
2417 assert!(envelope.target_content_fingerprint.is_none());
2418 }
2419
2420 fn external_test_predict_cohort() -> PredictCohort {
2421 let records = vec![
2422 crate::relation::SampleRelation::new(
2423 crate::ids::ObservationId::new("obs:holdout:1").unwrap(),
2424 SampleId::new("sample:holdout:1").unwrap(),
2425 ),
2426 crate::relation::SampleRelation::new(
2427 crate::ids::ObservationId::new("obs:holdout:2").unwrap(),
2428 SampleId::new("sample:holdout:2").unwrap(),
2429 ),
2430 ];
2431 PredictCohort::from_relations(
2432 PredictCohortRole::ExternalTest,
2433 SampleRelationSet { records },
2434 vec!["classification:y".to_string()],
2435 "c".repeat(64),
2436 Some("d".repeat(64)),
2437 )
2438 .unwrap()
2439 }
2440
2441 fn cv_fold_set(sample_ids: [&str; 2]) -> FoldSet {
2442 let sample_ids = sample_ids
2443 .into_iter()
2444 .map(|sample_id| SampleId::new(sample_id).unwrap())
2445 .collect::<Vec<_>>();
2446 FoldSet {
2447 id: "foldset:cv".to_string(),
2448 sample_ids: sample_ids.clone(),
2449 folds: vec![
2450 FoldAssignment {
2451 fold_id: FoldId::new("fold:0").unwrap(),
2452 train_sample_ids: vec![sample_ids[1].clone()],
2453 validation_sample_ids: vec![sample_ids[0].clone()],
2454 metadata: BTreeMap::new(),
2455 },
2456 FoldAssignment {
2457 fold_id: FoldId::new("fold:1").unwrap(),
2458 train_sample_ids: vec![sample_ids[0].clone()],
2459 validation_sample_ids: vec![sample_ids[1].clone()],
2460 metadata: BTreeMap::new(),
2461 },
2462 ],
2463 sample_groups: BTreeMap::new(),
2464 partition_mode: FoldPartitionMode::Partition,
2465 }
2466 }
2467
2468 #[test]
2469 fn external_data_envelope_v2_closes_predict_cohort_identity() {
2470 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2471 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2472 ))
2473 .unwrap();
2474 let cohort = external_test_predict_cohort();
2475 envelope.schema_version = EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V2;
2476 envelope.predict_cohort = Some(cohort.clone());
2477 envelope.validate().unwrap();
2478
2479 let mut v1 = envelope.clone();
2480 v1.schema_version = EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V1;
2481 assert!(v1
2482 .validate()
2483 .unwrap_err()
2484 .to_string()
2485 .contains("V1 cannot carry predict_cohort"));
2486
2487 let mut fingerprint_drift = cohort;
2488 fingerprint_drift.cohort_fingerprint = "0".repeat(64);
2489 envelope.predict_cohort = Some(fingerprint_drift);
2490 assert!(envelope
2491 .validate()
2492 .unwrap_err()
2493 .to_string()
2494 .contains("predict cohort fingerprint"));
2495 }
2496
2497 #[test]
2498 fn predict_cohort_refuses_noncanonical_identity_and_inference_targets() {
2499 let mut cohort = external_test_predict_cohort();
2500 cohort.physical_sample_ids.reverse();
2501 assert!(cohort
2502 .validate()
2503 .unwrap_err()
2504 .to_string()
2505 .contains("strictly sorted and unique"));
2506
2507 let mut inference = external_test_predict_cohort();
2508 inference.role = PredictCohortRole::Inference;
2509 assert!(inference
2510 .validate()
2511 .unwrap_err()
2512 .to_string()
2513 .contains("inference predict cohort must not carry target_content_fingerprint"));
2514 }
2515
2516 #[test]
2517 fn predict_cohort_constructor_derives_closed_identity_from_relations() {
2518 let mut first = crate::relation::SampleRelation::new(
2519 crate::ids::ObservationId::new("obs:holdout:one").unwrap(),
2520 SampleId::new("sample:physical:two").unwrap(),
2521 );
2522 first.origin_sample_id = Some(SampleId::new("sample:origin:one").unwrap());
2523 let second = crate::relation::SampleRelation::new(
2524 crate::ids::ObservationId::new("obs:holdout:two").unwrap(),
2525 SampleId::new("sample:physical:one").unwrap(),
2526 );
2527 let cohort = PredictCohort::from_relations(
2528 PredictCohortRole::Inference,
2529 SampleRelationSet {
2530 records: vec![first, second],
2531 },
2532 vec!["classification:y".to_string()],
2533 "e".repeat(64),
2534 None,
2535 )
2536 .unwrap();
2537
2538 assert_eq!(
2539 cohort.physical_sample_ids,
2540 vec![
2541 SampleId::new("sample:physical:one").unwrap(),
2542 SampleId::new("sample:physical:two").unwrap(),
2543 ]
2544 );
2545 assert_eq!(
2546 cohort.origin_sample_ids,
2547 vec![
2548 SampleId::new("sample:origin:one").unwrap(),
2549 SampleId::new("sample:physical:one").unwrap(),
2550 ]
2551 );
2552 cohort.validate().unwrap();
2553 }
2554
2555 #[test]
2556 fn external_test_predict_cohort_is_disjoint_from_cv_identity_closure() {
2557 let cohort = external_test_predict_cohort();
2558 cohort
2559 .validate_against_cv_fold_set(&cv_fold_set(["sample:cv:1", "sample:cv:2"]))
2560 .unwrap();
2561 let error = cohort
2562 .validate_against_cv_fold_set(&cv_fold_set(["sample:holdout:1", "sample:cv:2"]))
2563 .unwrap_err()
2564 .to_string();
2565 assert!(
2566 error.contains("overlaps CV fold sample or origin"),
2567 "{error}"
2568 );
2569 }
2570
2571 #[test]
2572 fn external_test_predict_cohort_refuses_cv_origin_alias() {
2573 let cohort = external_test_predict_cohort();
2574 let mut cv_relation = crate::relation::SampleRelation::new(
2575 crate::ids::ObservationId::new("obs:cv:1").unwrap(),
2576 SampleId::new("sample:cv:1").unwrap(),
2577 );
2578 cv_relation.origin_sample_id = Some(SampleId::new("sample:holdout:1").unwrap());
2579 let cv_relations = SampleRelationSet {
2580 records: vec![cv_relation],
2581 };
2582 let error = cohort
2583 .validate_against_cv_relations(&cv_relations)
2584 .unwrap_err()
2585 .to_string();
2586 assert!(
2587 error.contains("overlaps CV relation identity closure"),
2588 "{error}"
2589 );
2590 }
2591
2592 #[test]
2593 fn in_memory_provider_exposes_predict_cohort_only_to_predict() {
2594 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2595 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2596 ))
2597 .unwrap();
2598 let cohort = external_test_predict_cohort();
2599 envelope.schema_version = EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION_V2;
2600 envelope.predict_cohort = Some(cohort.clone());
2601 let mut provider =
2602 InMemoryDataProvider::new(ControllerId::new("controller:data.v2").unwrap());
2603 provider.register_envelope(envelope).unwrap();
2604
2605 assert_eq!(
2606 provider.predict_cohort(&binding(), Phase::Predict).unwrap(),
2607 Some(cohort)
2608 );
2609 let error = provider
2610 .predict_cohort(&binding(), Phase::Refit)
2611 .unwrap_err()
2612 .to_string();
2613 assert!(error.contains("only during PREDICT"), "{error}");
2614 }
2615
2616 #[test]
2617 fn external_data_envelope_content_identity_is_additive_and_validated() {
2618 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2619 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2620 ))
2621 .unwrap();
2622 envelope.data_content_fingerprint = Some("a".repeat(64));
2623 envelope.target_content_fingerprint = Some("b".repeat(64));
2624 envelope.validate().unwrap();
2625
2626 envelope.data_content_fingerprint = Some("not-a-fingerprint".to_string());
2627 assert!(envelope
2628 .validate()
2629 .unwrap_err()
2630 .to_string()
2631 .contains("data content fingerprint"));
2632 }
2633
2634 #[test]
2635 fn in_memory_provider_attests_complete_training_identity() {
2636 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2637 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2638 ))
2639 .unwrap();
2640 envelope.data_content_fingerprint = Some("a".repeat(64));
2641 envelope.target_content_fingerprint = Some("b".repeat(64));
2642 let provider = InMemoryDataProvider::with_envelope(
2643 ControllerId::new("controller:data.provider").unwrap(),
2644 envelope,
2645 )
2646 .unwrap();
2647
2648 let identity = provider
2649 .training_data_identity(&binding())
2650 .unwrap()
2651 .expect("content-aware envelope must attest training identity");
2652 identity.validate().unwrap();
2653 assert_eq!(identity.requirement_key, "model:base.x");
2654 assert_eq!(identity.data_content_fingerprint, "a".repeat(64));
2655 assert_eq!(identity.target_content_fingerprint, "b".repeat(64));
2656 }
2657
2658 #[test]
2659 fn validates_multisource_repetition_envelope_fixture() {
2660 let envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2661 "../tests/fixtures/package/data/coordinator_data_plan_envelope_multisource_repetitions.json"
2662 ))
2663 .unwrap();
2664
2665 envelope.validate().unwrap();
2666 let relations = envelope.coordinator_relations.as_ref().unwrap();
2667 assert_eq!(relations.records.len(), 8);
2668 let source_counts = relations.records.iter().fold(
2669 BTreeMap::<String, usize>::new(),
2670 |mut counts, record| {
2671 if record.unit_level == EntityUnitLevel::Observation {
2672 *counts
2673 .entry(record.source_id.clone().expect("source_id"))
2674 .or_default() += 1;
2675 }
2676 counts
2677 },
2678 );
2679 assert_eq!(source_counts["A"], 2);
2680 assert_eq!(source_counts["B"], 3);
2681 assert_eq!(source_counts["C"], 2);
2682 let combo = relations
2683 .records
2684 .iter()
2685 .find(|record| record.unit_level == EntityUnitLevel::Combo)
2686 .expect("relation-backed combo row");
2687 assert_eq!(combo.sample_id.as_str(), "sample:1");
2688 assert_eq!(
2689 combo.origin_sample_id.as_ref().unwrap().as_str(),
2690 combo.sample_id.as_str()
2691 );
2692 assert_eq!(combo.component_observation_ids.len(), 3);
2693 for source_id in ["A", "B", "C"] {
2694 assert!(combo
2695 .component_observation_ids
2696 .iter()
2697 .any(|observation_id| observation_id.as_str().contains(source_id)));
2698 }
2699 assert_eq!(
2700 relations
2701 .sample_for_observation(
2702 &crate::ids::ObservationId::new("obs.s1.combo.A0.B0.C0").unwrap()
2703 )
2704 .unwrap()
2705 .as_str(),
2706 "sample:1"
2707 );
2708 }
2709
2710 #[cfg(dag_ml_workspace_contract_fixtures)]
2711 #[test]
2712 fn published_external_data_envelope_schema_declares_current_version() {
2713 let schema: serde_json::Value = serde_json::from_str(include_str!(
2714 "../../../docs/contracts/coordinator_data_plan_envelope.schema.json"
2715 ))
2716 .unwrap();
2717
2718 assert_eq!(
2719 schema["properties"]["schema_version"]["const"].as_u64(),
2720 Some(EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION as u64)
2721 );
2722 assert!(schema["required"]
2723 .as_array()
2724 .unwrap()
2725 .iter()
2726 .any(|field| field.as_str() == Some("schema_version")));
2727 }
2728
2729 #[test]
2730 fn refuses_unsupported_external_data_envelope_schema_version() {
2731 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2732 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2733 ))
2734 .unwrap();
2735 envelope.schema_version = EXTERNAL_DATA_PLAN_ENVELOPE_SCHEMA_VERSION + 1;
2736
2737 assert!(binding().validate_envelope(&envelope).is_err());
2738 }
2739
2740 #[test]
2741 fn refuses_envelope_fingerprint_mismatch() {
2742 let mut envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2743 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2744 ))
2745 .unwrap();
2746 envelope.plan_fingerprint = "0".repeat(64);
2747
2748 assert!(binding().validate_envelope(&envelope).is_err());
2749 }
2750
2751 #[test]
2752 fn in_memory_provider_materializes_validated_data_handles() {
2753 let envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2754 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2755 ))
2756 .unwrap();
2757 let provider = InMemoryDataProvider::with_envelope(
2758 ControllerId::new("controller:data.provider").unwrap(),
2759 envelope,
2760 )
2761 .unwrap();
2762
2763 let handle = provider
2764 .materialize(&DataMaterializationRequest {
2765 run_id: RunId::new("run:data").unwrap(),
2766 node_id: NodeId::new("model:base").unwrap(),
2767 input_name: "x".to_string(),
2768 phase: Phase::FitCv,
2769 variant_id: Some(VariantId::new("variant:base").unwrap()),
2770 fold_id: Some(FoldId::new("fold:0").unwrap()),
2771 binding: binding(),
2772 predict_cohort: None,
2773 })
2774 .unwrap();
2775
2776 let record = provider.handle_record(handle.handle).unwrap();
2777 assert_eq!(record.input_name, "x");
2778 assert_eq!(record.relation_record_count, Some(4));
2779 assert_eq!(provider.handle_records().len(), 1);
2780 }
2781
2782 #[test]
2783 fn in_memory_provider_registration_is_idempotent_for_same_envelope() {
2784 let envelope: ExternalDataPlanEnvelope = serde_json::from_str(include_str!(
2785 "../tests/fixtures/package/data/coordinator_data_plan_envelope_sample12.json"
2786 ))
2787 .unwrap();
2788 let mut provider =
2789 InMemoryDataProvider::new(ControllerId::new("controller:data.provider").unwrap());
2790
2791 provider.register_envelope(envelope.clone()).unwrap();
2792 provider.register_envelope(envelope).unwrap();
2793 }
2794
2795 #[test]
2796 fn in_memory_provider_refuses_unknown_envelope() {
2797 let provider =
2798 InMemoryDataProvider::new(ControllerId::new("controller:data.provider").unwrap());
2799
2800 assert!(provider
2801 .materialize(&DataMaterializationRequest {
2802 run_id: RunId::new("run:data").unwrap(),
2803 node_id: NodeId::new("model:base").unwrap(),
2804 input_name: "x".to_string(),
2805 phase: Phase::FitCv,
2806 variant_id: None,
2807 fold_id: None,
2808 binding: binding(),
2809 predict_cohort: None,
2810 })
2811 .is_err());
2812 }
2813
2814 fn cartesian_combination() -> CombinationPlan {
2815 CombinationPlan {
2816 mode: CombinationMode::Cartesian,
2817 component_source_ids: vec!["source:a".to_string(), "source:b".to_string()],
2818 component_unit_ids: Vec::new(),
2819 match_key: None,
2820 reference_source_id: None,
2821 seed: None,
2822 cap: None,
2823 budget: Some(32),
2824 missing_source_policy: Some(RepresentationMissingSourcePolicy::Strict),
2825 metadata: BTreeMap::new(),
2826 }
2827 }
2828
2829 fn compatibility_report() -> RepresentationCompatibilityReport {
2830 RepresentationCompatibilityReport {
2831 policy: RepresentationMissingSourcePolicy::Mask,
2832 outcome: RepresentationCompatibilityOutcome::CompatibleWithFallback,
2833 fallback_used: Some("mask".to_string()),
2834 warning_severity: Some(RepresentationCompatibilitySeverity::Warning),
2835 affected_source_count: 1,
2836 affected_repetition_count: 2,
2837 affected_sample_count: 3,
2838 train_relation_fingerprint: Some("c".repeat(64)),
2839 predict_relation_fingerprint: Some("d".repeat(64)),
2840 train_unit_count: Some(6),
2841 predict_unit_count: Some(4),
2842 fixed_width_required: true,
2843 final_reducer_stabilizes_output: true,
2844 cartesian_combo_count_changed: true,
2845 late_fusion_branch_delta: true,
2846 messages: vec!["mask fallback applied for missing source".to_string()],
2847 metadata: BTreeMap::new(),
2848 }
2849 }
2850
2851 #[derive(serde::Deserialize)]
2852 #[serde(deny_unknown_fields)]
2853 struct D9GoldenFixture {
2854 schema_version: u32,
2855 golden_scenarios: Vec<D9GoldenScenario>,
2856 }
2857
2858 #[derive(serde::Deserialize)]
2859 #[serde(deny_unknown_fields)]
2860 struct D9GoldenScenario {
2861 scenario_id: String,
2862 flow: Vec<String>,
2863 mock_phase_path: Vec<String>,
2864 representation_replay_manifest: RepresentationReplayManifest,
2865 assertions: Vec<String>,
2866 }
2867
2868 #[test]
2869 fn d9_golden_multisource_repetition_manifests_validate() {
2870 let fixture: D9GoldenFixture = serde_json::from_str(include_str!(
2871 "../tests/fixtures/package/data/d9_golden_multisource_scenarios.json"
2872 ))
2873 .unwrap();
2874 assert_eq!(fixture.schema_version, 1);
2875 assert_eq!(fixture.golden_scenarios.len(), 7);
2876
2877 let mut scenario_ids = BTreeSet::new();
2878 let mut has_same_repetition_replay = false;
2879 let mut has_changed_repetition_replay = false;
2880 let mut has_combo_meta_fit_influence = false;
2881 for scenario in &fixture.golden_scenarios {
2882 assert!(
2883 scenario_ids.insert(scenario.scenario_id.as_str()),
2884 "duplicate D9 scenario {}",
2885 scenario.scenario_id
2886 );
2887 assert!(!scenario.flow.is_empty());
2888 assert_eq!(scenario.mock_phase_path, ["fit_cv", "refit", "predict"]);
2889 assert!(!scenario.assertions.is_empty());
2890
2891 let manifest = &scenario.representation_replay_manifest;
2892 manifest.validate().unwrap();
2893 assert_eq!(
2894 manifest.final_output_unit_level,
2895 Some(EntityUnitLevel::PhysicalSample),
2896 "{} must publish sample-level outputs",
2897 scenario.scenario_id
2898 );
2899 if manifest.output_unit_level == EntityUnitLevel::Combo {
2900 assert!(
2901 manifest.final_reduction_id.is_some(),
2902 "{} must declare combo-to-sample reduction",
2903 scenario.scenario_id
2904 );
2905 assert!(
2906 !manifest.combo_selection.is_empty(),
2907 "{} must retain relation-backed combo identities",
2908 scenario.scenario_id
2909 );
2910 }
2911
2912 if let (Some(train), Some(predict)) = (
2913 &manifest.train_compatibility,
2914 &manifest.predict_compatibility,
2915 ) {
2916 has_same_repetition_replay |= train.train_unit_count == predict.predict_unit_count
2917 && train.train_relation_fingerprint == predict.predict_relation_fingerprint;
2918 has_changed_repetition_replay |= train.train_unit_count
2919 != predict.predict_unit_count
2920 || train.train_relation_fingerprint != predict.predict_relation_fingerprint;
2921 }
2922
2923 if scenario.scenario_id == "d9.combo_meta_post.relation_backed_adapters" {
2924 has_combo_meta_fit_influence = manifest
2925 .metadata
2926 .get("fit_influence_policy")
2927 .is_some_and(|value| value == "equal_sample_influence");
2928 }
2929 }
2930
2931 assert!(scenario_ids.contains("d9.per_source_aggregate.source_models.sample_reducer"));
2932 assert!(scenario_ids.contains("d9.late_fusion_by_source.prediction_join.meta_model"));
2933 assert!(scenario_ids.contains("d9.cartesian_full.model.combo_to_sample_reducer"));
2934 assert!(scenario_ids.contains("d9.cartesian_mc.deterministic_replay"));
2935 assert!(scenario_ids.contains("d9.stack_fixed.strict_cardinality"));
2936 assert!(scenario_ids.contains("d9.stack_padded_masked.missing_repetition"));
2937 assert!(scenario_ids.contains("d9.combo_meta_post.relation_backed_adapters"));
2938 assert!(has_same_repetition_replay);
2939 assert!(has_changed_repetition_replay);
2940 assert!(has_combo_meta_fit_influence);
2941 }
2942
2943 #[test]
2944 fn representation_plan_validates_cartesian_and_monte_carlo_contracts() {
2945 let cartesian = RepresentationPlan::CartesianProduct(CartesianProductRepresentation {
2946 combination_plan: cartesian_combination(),
2947 output_unit_level: EntityUnitLevel::Combo,
2948 cardinality: RepresentationCardinality::ManyToMany,
2949 preserve_provenance: true,
2950 });
2951 cartesian.validate().unwrap();
2952
2953 let monte_carlo =
2954 RepresentationPlan::MonteCarloCartesian(MonteCarloCartesianRepresentation {
2955 combination_plan: CombinationPlan {
2956 mode: CombinationMode::SampleK,
2957 component_source_ids: vec!["source:a".to_string(), "source:b".to_string()],
2958 component_unit_ids: Vec::new(),
2959 match_key: None,
2960 reference_source_id: None,
2961 seed: Some(42),
2962 cap: Some(8),
2963 budget: None,
2964 missing_source_policy: Some(RepresentationMissingSourcePolicy::Warn),
2965 metadata: BTreeMap::new(),
2966 },
2967 output_unit_level: EntityUnitLevel::Observation,
2968 cardinality: RepresentationCardinality::BoundedMany,
2969 preserve_provenance: true,
2970 });
2971 monte_carlo.validate().unwrap();
2972
2973 let mut bad = cartesian_combination();
2974 bad.mode = CombinationMode::SampleK;
2975 bad.seed = Some(7);
2976 bad.cap = Some(0);
2977 assert!(bad.validate().is_err());
2978 }
2979
2980 #[test]
2981 fn stack_representations_validate_cardinality_and_mask_policy() {
2982 let fixed = RepresentationPlan::StackFixed(StackFixedRepresentation {
2983 output_unit_level: EntityUnitLevel::SourceSample,
2984 cardinality: RepresentationCardinality::OneToMany,
2985 expected_cardinality: 3,
2986 component_source_ids: vec!["source:a".to_string(), "source:b".to_string()],
2987 });
2988 fixed.validate().unwrap();
2989
2990 let padded = RepresentationPlan::StackPaddedMasked(StackPaddedMaskedRepresentation {
2991 output_unit_level: EntityUnitLevel::SourceSample,
2992 cardinality: RepresentationCardinality::BoundedMany,
2993 expected_cardinality: 4,
2994 missing_source_policy: RepresentationMissingSourcePolicy::Mask,
2995 requires_missing_masks: true,
2996 component_source_ids: vec!["source:a".to_string()],
2997 });
2998 padded.validate().unwrap();
2999
3000 let bad = RepresentationPlan::StackPaddedMasked(StackPaddedMaskedRepresentation {
3001 output_unit_level: EntityUnitLevel::SourceSample,
3002 cardinality: RepresentationCardinality::BoundedMany,
3003 expected_cardinality: 4,
3004 missing_source_policy: RepresentationMissingSourcePolicy::ImputeDeclared,
3005 requires_missing_masks: false,
3006 component_source_ids: Vec::new(),
3007 });
3008 assert!(bad.validate().is_err());
3009 }
3010
3011 #[test]
3012 fn representation_compatibility_report_enforces_missingness_policy() {
3013 compatibility_report().validate().unwrap();
3014
3015 let strict = RepresentationCompatibilityReport {
3016 policy: RepresentationMissingSourcePolicy::Strict,
3017 outcome: RepresentationCompatibilityOutcome::Incompatible,
3018 fallback_used: None,
3019 warning_severity: None,
3020 affected_source_count: 1,
3021 affected_repetition_count: 0,
3022 affected_sample_count: 1,
3023 train_relation_fingerprint: None,
3024 predict_relation_fingerprint: None,
3025 train_unit_count: None,
3026 predict_unit_count: None,
3027 fixed_width_required: false,
3028 final_reducer_stabilizes_output: false,
3029 cartesian_combo_count_changed: false,
3030 late_fusion_branch_delta: false,
3031 messages: Vec::new(),
3032 metadata: BTreeMap::new(),
3033 };
3034 strict.validate().unwrap();
3035
3036 let mut bad_non_strict = compatibility_report();
3037 bad_non_strict.fallback_used = None;
3038 assert!(bad_non_strict.validate().is_err());
3039
3040 let mut bad_fixed_width = compatibility_report();
3041 bad_fixed_width.policy = RepresentationMissingSourcePolicy::ImputeDeclared;
3042 bad_fixed_width.fallback_used = Some("impute_declared".to_string());
3043 assert!(bad_fixed_width.validate().is_err());
3044
3045 let mut bad_cartesian = compatibility_report();
3046 bad_cartesian.final_reducer_stabilizes_output = false;
3047 assert!(bad_cartesian.validate().is_err());
3048
3049 let bad_relation_drift = RepresentationCompatibilityReport {
3050 policy: RepresentationMissingSourcePolicy::Strict,
3051 outcome: RepresentationCompatibilityOutcome::Compatible,
3052 fallback_used: None,
3053 warning_severity: None,
3054 affected_source_count: 0,
3055 affected_repetition_count: 0,
3056 affected_sample_count: 0,
3057 train_relation_fingerprint: Some("a".repeat(64)),
3058 predict_relation_fingerprint: Some("b".repeat(64)),
3059 train_unit_count: Some(3),
3060 predict_unit_count: Some(3),
3061 fixed_width_required: false,
3062 final_reducer_stabilizes_output: true,
3063 cartesian_combo_count_changed: false,
3064 late_fusion_branch_delta: false,
3065 messages: Vec::new(),
3066 metadata: BTreeMap::new(),
3067 };
3068 let error = bad_relation_drift.validate().unwrap_err().to_string();
3069 assert!(
3070 error.contains("relation fingerprint mismatch requires affected units"),
3071 "unexpected D9 relation drift error: {error}"
3072 );
3073
3074 let mut bad_unit_drift = bad_relation_drift;
3075 bad_unit_drift.predict_relation_fingerprint =
3076 bad_unit_drift.train_relation_fingerprint.clone();
3077 bad_unit_drift.predict_unit_count = Some(2);
3078 let error = bad_unit_drift.validate().unwrap_err().to_string();
3079 assert!(
3080 error.contains("unit count mismatch requires affected units"),
3081 "unexpected D9 unit-count drift error: {error}"
3082 );
3083 }
3084
3085 #[test]
3086 fn representation_replay_manifest_round_trips_and_validates() {
3087 let plan = RepresentationPlan::CartesianProduct(CartesianProductRepresentation {
3088 combination_plan: cartesian_combination(),
3089 output_unit_level: EntityUnitLevel::Combo,
3090 cardinality: RepresentationCardinality::ManyToMany,
3091 preserve_provenance: true,
3092 });
3093 let manifest = RepresentationReplayManifest {
3094 manifest_id: "repr:combo.ab".to_string(),
3095 representation_plan: plan,
3096 combination_plan: Some(cartesian_combination()),
3097 output_unit_level: EntityUnitLevel::Combo,
3098 output_representation: Some("combo_observation".to_string()),
3099 relation_fingerprint: Some("a".repeat(64)),
3100 feature_schema_fingerprint: Some("b".repeat(64)),
3101 final_reduction_id: Some("reduction:combo_to_sample".to_string()),
3102 sample_observation_mapping: vec![
3103 RepresentationSampleObservationMapping {
3104 physical_sample_id: "sample:1".to_string(),
3105 source_id: "source:a".to_string(),
3106 observation_ids: vec!["obs:a.1".to_string(), "obs:a.2".to_string()],
3107 },
3108 RepresentationSampleObservationMapping {
3109 physical_sample_id: "sample:1".to_string(),
3110 source_id: "source:b".to_string(),
3111 observation_ids: vec!["obs:b.1".to_string()],
3112 },
3113 ],
3114 combo_selection: vec![RepresentationComboSelectionRecord {
3115 combo_unit_id: "combo:sample1:a1:b1".to_string(),
3116 physical_sample_id: "sample:1".to_string(),
3117 component_observation_ids: vec!["obs:a.1".to_string(), "obs:b.1".to_string()],
3118 seed: Some(42),
3119 }],
3120 qc_policy_refs: vec!["qc:default".to_string()],
3121 outlier_policy_refs: vec!["outlier:none".to_string()],
3122 missing_source_policy: Some(RepresentationMissingSourcePolicy::Strict),
3123 missing_repetition_policy: Some(RepresentationMissingSourcePolicy::Warn),
3124 prediction_representation: Some("sample_prediction".to_string()),
3125 final_output_unit_level: Some(EntityUnitLevel::PhysicalSample),
3126 train_compatibility: Some(RepresentationCompatibilityReport {
3127 policy: RepresentationMissingSourcePolicy::Strict,
3128 outcome: RepresentationCompatibilityOutcome::Compatible,
3129 fallback_used: None,
3130 warning_severity: None,
3131 affected_source_count: 0,
3132 affected_repetition_count: 0,
3133 affected_sample_count: 0,
3134 train_relation_fingerprint: Some("a".repeat(64)),
3135 predict_relation_fingerprint: None,
3136 train_unit_count: Some(1),
3137 predict_unit_count: Some(1),
3138 fixed_width_required: false,
3139 final_reducer_stabilizes_output: true,
3140 cartesian_combo_count_changed: false,
3141 late_fusion_branch_delta: false,
3142 messages: Vec::new(),
3143 metadata: BTreeMap::new(),
3144 }),
3145 predict_compatibility: Some(compatibility_report()),
3146 metadata: BTreeMap::new(),
3147 };
3148
3149 manifest.validate().unwrap();
3150 let encoded = serde_json::to_string(&manifest).unwrap();
3151 let decoded: RepresentationReplayManifest = serde_json::from_str(&encoded).unwrap();
3152 assert_eq!(decoded, manifest);
3153 }
3154}