use std::collections::{BTreeMap, BTreeSet};
use serde::{Deserialize, Serialize};
use crate::bundle::{bundle_prediction_requirement_key, ExecutionBundle, RefitArtifactRecord};
use crate::canonical::parse_typed_json;
use crate::controller::{
ArtifactPolicy, ControllerCapability, ControllerFitScope, ControllerManifest,
ControllerRegistry,
};
use crate::criteria::TrainingLossRoleReference;
use crate::data::{data_binding_requirement_key, DataBinding, ExternalDataPlanEnvelope};
use crate::error::{DagMlError, Result};
use crate::graph::{GraphSpec, NodeKind, PortKind};
use crate::ids::{ArtifactId, BundleId, FoldId, GroupId, NodeId, SampleId};
use crate::phase::Phase;
use crate::plan::{build_execution_plan, CampaignSpec, ExecutionPlan};
use crate::policy::PredictionLevel;
use crate::relation::{EntityUnitLevel, SampleRelationSet};
use crate::selection::{RefitStrategy, SelectionPolicy};
pub const TRAINING_REQUEST_SCHEMA_VERSION: u32 = 1;
pub const TRAINING_REQUEST_SCHEMA_ID: &str =
"https://github.com/GBeurier/dag-ml/schemas/training_request.v1.schema.json";
pub const CACHE_NAMESPACE_SCHEMA_VERSION: u32 = 1;
pub const CACHE_NAMESPACE_SCHEMA_ID: &str =
"https://github.com/GBeurier/dag-ml/schemas/cache_namespace.v1.schema.json";
pub const PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION: u32 = 1;
pub const PORTABLE_PREDICTOR_PACKAGE_SCHEMA_ID: &str =
"https://github.com/GBeurier/dag-ml/schemas/portable_predictor_package.v1.schema.json";
pub const OUTPUT_BINDING_SCHEMA_VERSION: u32 = 1;
pub const TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION: u32 = 1;
pub const PARAMETER_PATCH_SCHEMA_VERSION: u32 = 1;
pub const PARAMETER_PROJECTION_SCHEMA_VERSION: u32 = 1;
type InfluenceCoordinate = (TrainingInfluenceKind, String, Option<NodeId>);
type ExpectedInfluenceCoordinates = BTreeMap<InfluenceCoordinate, BTreeSet<SampleId>>;
type InfluenceCapabilitySlot = (NodeId, TrainingInfluenceKind, Phase, Option<FoldId>);
fn deserialize_required_nullable<'de, D, T>(
deserializer: D,
) -> std::result::Result<Option<T>, D::Error>
where
D: serde::Deserializer<'de>,
T: Deserialize<'de>,
{
Option::<T>::deserialize(deserializer)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PredictionKind {
RegressionPoint,
ClassLabel,
ClassProbability,
DecisionScore,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PredictionSource {
FinalRefit,
CvEnsemble,
FoldMember,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OutputOrder {
TargetOrder,
TargetMajorClassMinor,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingOutputRequest {
pub output_id: String,
pub node_id: NodeId,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub port_name: Option<String>,
pub prediction_level: PredictionLevel,
#[serde(deserialize_with = "deserialize_required_nullable")]
pub unit_level: Option<EntityUnitLevel>,
pub prediction_kind: PredictionKind,
pub target_names: Vec<String>,
pub target_units: Vec<Option<String>>,
pub class_labels: Vec<Vec<String>>,
pub output_order: OutputOrder,
pub target_space: String,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ResolvedTrainingOutput {
pub output_id: String,
pub node_id: NodeId,
pub port_name: String,
pub prediction_level: PredictionLevel,
#[serde(default)]
pub unit_level: Option<EntityUnitLevel>,
pub prediction_kind: PredictionKind,
pub target_names: Vec<String>,
pub target_units: Vec<Option<String>>,
pub class_labels: Vec<Vec<String>>,
pub output_order: OutputOrder,
pub target_space: String,
}
impl TrainingOutputRequest {
pub fn validate(&self) -> Result<()> {
validate_identifier_text("training output_id", &self.output_id)?;
validate_output_unit_level(self.prediction_level, self.unit_level)?;
validate_output_shape(
self.prediction_kind,
self.output_order,
&self.target_names,
&self.target_units,
&self.class_labels,
&self.target_space,
)?;
if self
.port_name
.as_ref()
.is_some_and(|name| name.trim().is_empty())
{
return contract_error(format!(
"training output `{}` has an empty port_name",
self.output_id
));
}
Ok(())
}
pub fn resolve(&self, graph: &GraphSpec) -> Result<ResolvedTrainingOutput> {
self.validate()?;
let node = graph
.nodes
.iter()
.find(|node| node.id == self.node_id)
.ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"training output `{}` references unknown node `{}`",
self.output_id, self.node_id
))
})?;
let prediction_ports = node
.ports
.outputs
.iter()
.filter(|port| port.kind == PortKind::Prediction)
.collect::<Vec<_>>();
let port_name = match self.port_name.as_deref() {
Some(requested) => {
let port = node
.ports
.outputs
.iter()
.find(|port| port.name == requested)
.ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"training output `{}` references absent port `{}.{requested}`",
self.output_id, self.node_id
))
})?;
if port.kind != PortKind::Prediction {
return contract_error(format!(
"training output `{}` port `{}.{requested}` is not a prediction port",
self.output_id, self.node_id
));
}
requested.to_string()
}
None => match prediction_ports.as_slice() {
[] => {
return contract_error(format!(
"training output `{}` node `{}` exposes no prediction output",
self.output_id, self.node_id
));
}
[only] => only.name.clone(),
_ => {
return contract_error(format!(
"training output `{}` node `{}` exposes multiple prediction outputs; port_name is required",
self.output_id, self.node_id
));
}
},
};
Ok(ResolvedTrainingOutput {
output_id: self.output_id.clone(),
node_id: self.node_id.clone(),
port_name,
prediction_level: self.prediction_level,
unit_level: self.unit_level,
prediction_kind: self.prediction_kind,
target_names: self.target_names.clone(),
target_units: self.target_units.clone(),
class_labels: self.class_labels.clone(),
output_order: self.output_order,
target_space: self.target_space.clone(),
})
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TrainingSchedulerKind {
Sequential,
Parallel,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TrainingSchedulerBackend {
Threads,
Processes,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingSchedulerOptions {
pub kind: TrainingSchedulerKind,
#[serde(default)]
pub backend: Option<TrainingSchedulerBackend>,
pub workers: u32,
}
impl TrainingSchedulerOptions {
fn validate(&self) -> Result<()> {
match (self.kind, self.backend, self.workers) {
(TrainingSchedulerKind::Sequential, None, 1) => Ok(()),
(TrainingSchedulerKind::Sequential, Some(_), _) => contract_error(
"sequential training scheduler forbids a parallel backend".to_string(),
),
(TrainingSchedulerKind::Sequential, None, _) => {
contract_error("sequential training scheduler requires workers=1".to_string())
}
(TrainingSchedulerKind::Parallel, None, _) => contract_error(
"parallel training scheduler requires an explicit backend".to_string(),
),
(TrainingSchedulerKind::Parallel, Some(_), 0 | 1) => {
contract_error("parallel training scheduler requires workers>=2".to_string())
}
(TrainingSchedulerKind::Parallel, Some(_), _) => Ok(()),
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingResourceLimits {
pub cpu_threads: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub memory_bytes: Option<u64>,
pub gpu_devices: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wall_time_ms: Option<u64>,
}
impl TrainingResourceLimits {
fn validate(&self, scheduler: &TrainingSchedulerOptions) -> Result<()> {
if self.cpu_threads == 0 {
return contract_error("training resources require cpu_threads>=1".to_string());
}
if scheduler.workers > self.cpu_threads {
return contract_error(format!(
"training scheduler workers={} exceeds cpu_threads={}",
scheduler.workers, self.cpu_threads
));
}
if self.memory_bytes == Some(0) {
return contract_error("training memory_bytes must be positive".to_string());
}
if self.wall_time_ms == Some(0) {
return contract_error("training wall_time_ms must be positive".to_string());
}
validate_sorted_unique_text("training gpu_devices", &self.gpu_devices, false)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CvArtifactRetention {
Discard,
MetadataOnly,
Retain,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PredictionCacheRetention {
Discard,
Retain,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FittedArtifactMode {
PortableRequired,
AllowHostSidecar,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingArtifactOptions {
pub cv_artifacts: CvArtifactRetention,
pub prediction_caches: PredictionCacheRetention,
pub fitted_artifacts: FittedArtifactMode,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingOptions {
pub refit: bool,
#[serde(deserialize_with = "deserialize_required_nullable")]
pub refit_strategy: Option<RefitStrategy>,
pub seed: u64,
pub selection: SelectionPolicy,
pub selection_output_id: String,
pub outputs: Vec<TrainingOutputRequest>,
pub scheduler: TrainingSchedulerOptions,
pub resources: TrainingResourceLimits,
pub artifacts: TrainingArtifactOptions,
}
impl TrainingOptions {
fn validate(&self, graph: &GraphSpec) -> Result<Vec<ResolvedTrainingOutput>> {
match (self.refit, self.refit_strategy) {
(true, None) => {
return contract_error(
"training refit=true requires an explicit refit_strategy".to_string(),
);
}
(false, Some(_)) => {
return contract_error("training refit=false forbids refit_strategy".to_string());
}
_ => {}
}
self.selection.validate()?;
validate_identifier_text("training selection_output_id", &self.selection_output_id)?;
self.scheduler.validate()?;
self.resources.validate(&self.scheduler)?;
if !self.refit && self.artifacts.prediction_caches != PredictionCacheRetention::Retain {
return contract_error(
"training refit=false requires retained prediction caches for REFIT replay"
.to_string(),
);
}
if self.outputs.is_empty() {
return contract_error("training options require at least one output".to_string());
}
let mut previous_id: Option<&str> = None;
let mut coordinates = BTreeSet::new();
let mut resolved = Vec::with_capacity(self.outputs.len());
for output in &self.outputs {
if previous_id.is_some_and(|previous| previous >= output.output_id.as_str()) {
return contract_error(
"training outputs must be strictly sorted by output_id".to_string(),
);
}
previous_id = Some(output.output_id.as_str());
let output = output.resolve(graph)?;
if !coordinates.insert((output.node_id.clone(), output.port_name.clone())) {
return contract_error(format!(
"training outputs bind `{}.{}` more than once",
output.node_id, output.port_name
));
}
resolved.push(output);
}
if !resolved
.iter()
.any(|output| output.output_id == self.selection_output_id)
{
return contract_error(format!(
"training selection_output_id `{}` does not identify a declared output",
self.selection_output_id
));
}
Ok(resolved)
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingDataIdentity {
pub requirement_key: String,
pub schema_fingerprint: String,
pub plan_fingerprint: String,
pub relation_fingerprint: String,
pub data_content_fingerprint: String,
pub target_content_fingerprint: String,
pub identity_fingerprint: String,
}
impl TrainingDataIdentity {
pub fn from_binding_envelope(
binding: &DataBinding,
envelope: &ExternalDataPlanEnvelope,
) -> Result<Self> {
binding.validate_envelope(envelope)?;
let requirement_key = data_binding_requirement_key(&binding.node_id, &binding.input_name);
let relation_fingerprint = envelope.relation_fingerprint.clone().ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"external data envelope for `{requirement_key}` cannot attest training without a relation fingerprint"
))
})?;
let data_content_fingerprint =
envelope.data_content_fingerprint.clone().ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"external data envelope for `{requirement_key}` cannot attest training without a data content fingerprint"
))
})?;
let target_content_fingerprint =
envelope.target_content_fingerprint.clone().ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"external data envelope for `{requirement_key}` cannot attest training without a target content fingerprint"
))
})?;
let mut identity = Self {
requirement_key,
schema_fingerprint: envelope.schema_fingerprint.clone(),
plan_fingerprint: envelope.plan_fingerprint.clone(),
relation_fingerprint,
data_content_fingerprint,
target_content_fingerprint,
identity_fingerprint: zero_fingerprint(),
};
identity.identity_fingerprint = identity.compute_fingerprint()?;
identity.validate()?;
Ok(identity)
}
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "identity_fingerprint", "training data identity")
}
pub fn validate(&self) -> Result<()> {
validate_non_empty("training data requirement_key", &self.requirement_key)?;
for (label, value) in [
("training data schema", &self.schema_fingerprint),
("training data plan", &self.plan_fingerprint),
("training data relation", &self.relation_fingerprint),
("training data content", &self.data_content_fingerprint),
("training target content", &self.target_content_fingerprint),
("training data identity", &self.identity_fingerprint),
] {
validate_sha256(label, value)?;
}
if self.identity_fingerprint != self.compute_fingerprint()? {
return contract_error(format!(
"training data identity `{}` fingerprint does not match TCV1 content",
self.requirement_key
));
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ParameterNamespace {
Operator,
Fit,
Control,
Structural,
}
impl ParameterNamespace {
pub const fn plan_root(self) -> &'static str {
match self {
Self::Operator => "params",
Self::Fit => "fit_params",
Self::Control => "control_params",
Self::Structural => "structural_params",
}
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ParameterPatch {
pub schema_version: u32,
pub node_id: NodeId,
pub namespace: ParameterNamespace,
pub path: Vec<String>,
pub value: serde_json::Value,
}
impl ParameterPatch {
pub fn validate(&self) -> Result<()> {
if self.schema_version != PARAMETER_PATCH_SCHEMA_VERSION {
return unsupported_version(
"parameter patch",
self.schema_version,
PARAMETER_PATCH_SCHEMA_VERSION,
);
}
if self.path.is_empty() {
return contract_error(format!(
"parameter patch for `{}` has an empty path",
self.node_id
));
}
for segment in &self.path {
if segment.trim().is_empty() || segment == "-" {
return contract_error(format!(
"parameter patch for `{}` has an invalid path segment `{segment}`",
self.node_id
));
}
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NodePatchPolicy {
pub node_id: NodeId,
pub allowed_namespaces: BTreeSet<ParameterNamespace>,
}
impl NodePatchPolicy {
fn validate(&self) -> Result<()> {
if self.allowed_namespaces.is_empty() {
return contract_error(format!(
"node patch policy `{}` allows no namespaces",
self.node_id
));
}
Ok(())
}
}
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NamespacedNodeParameters {
#[serde(default)]
pub params: BTreeMap<String, serde_json::Value>,
#[serde(default)]
pub fit_params: BTreeMap<String, serde_json::Value>,
#[serde(default)]
pub control_params: BTreeMap<String, serde_json::Value>,
#[serde(default)]
pub structural_params: BTreeMap<String, serde_json::Value>,
}
impl NamespacedNodeParameters {
fn namespace_mut(
&mut self,
namespace: ParameterNamespace,
) -> &mut BTreeMap<String, serde_json::Value> {
match namespace {
ParameterNamespace::Operator => &mut self.params,
ParameterNamespace::Fit => &mut self.fit_params,
ParameterNamespace::Control => &mut self.control_params,
ParameterNamespace::Structural => &mut self.structural_params,
}
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ParameterProjection {
pub schema_version: u32,
pub nodes: BTreeMap<NodeId, NamespacedNodeParameters>,
pub requires_recompile: bool,
pub structural_patch_count: u32,
pub patches_fingerprint: String,
pub projection_fingerprint: String,
}
impl ParameterProjection {
pub fn from_json(json: &str) -> Result<Self> {
let raw_fingerprint = strict_tcv1_fingerprint_without(
json,
"projection_fingerprint",
"parameter projection",
)?;
let projection: Self = serde_json::from_str(json)?;
if projection.projection_fingerprint != raw_fingerprint {
return contract_error(
"parameter projection fingerprint does not match original TCV1 JSON".to_string(),
);
}
projection.validate()?;
Ok(projection)
}
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "projection_fingerprint", "parameter projection")
}
pub fn validate(&self) -> Result<()> {
if self.schema_version != PARAMETER_PROJECTION_SCHEMA_VERSION {
return unsupported_version(
"parameter projection",
self.schema_version,
PARAMETER_PROJECTION_SCHEMA_VERSION,
);
}
validate_sha256("parameter patches", &self.patches_fingerprint)?;
validate_sha256("parameter projection", &self.projection_fingerprint)?;
if self.requires_recompile != (self.structural_patch_count > 0) {
return contract_error(
"parameter projection requires_recompile must equal structural_patch_count>0"
.to_string(),
);
}
if self.projection_fingerprint != self.compute_fingerprint()? {
return contract_error(
"parameter projection fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
}
pub fn project_parameter_patches(
plan: &ExecutionPlan,
patches: &[ParameterPatch],
policies: &[NodePatchPolicy],
) -> Result<ParameterProjection> {
plan.validate()?;
validate_canonical_patches(patches)?;
let policy_map = validate_patch_policies(plan, policies)?;
let patched_nodes = patches
.iter()
.map(|patch| patch.node_id.clone())
.collect::<BTreeSet<_>>();
if policy_map.keys().cloned().collect::<BTreeSet<_>>() != patched_nodes {
return contract_error(
"node patch policies must exactly cover nodes targeted by patches".to_string(),
);
}
let mut nodes = plan
.node_plans
.iter()
.map(|(node_id, node_plan)| {
(
node_id.clone(),
NamespacedNodeParameters {
params: node_plan.params.clone(),
..NamespacedNodeParameters::default()
},
)
})
.collect::<BTreeMap<_, _>>();
let mut structural_patch_count = 0_u32;
for patch in patches {
let policy = policy_map.get(&patch.node_id).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch for `{}` has no node patch policy",
patch.node_id
))
})?;
if !policy.contains(&patch.namespace) {
return contract_error(format!(
"parameter namespace `{:?}` is forbidden for node `{}`",
patch.namespace, patch.node_id
));
}
let node = nodes.get_mut(&patch.node_id).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch references unknown node `{}`",
patch.node_id
))
})?;
deep_set_object_key(
node.namespace_mut(patch.namespace),
&patch.path,
patch.value.clone(),
&patch.node_id,
)?;
if patch.namespace == ParameterNamespace::Structural {
structural_patch_count = structural_patch_count.checked_add(1).ok_or_else(|| {
DagMlError::CampaignValidation(
"parameter projection has too many structural patches".to_string(),
)
})?;
}
}
let patches_fingerprint = tcv1_fingerprint(patches, "parameter patches")?;
let mut projection = ParameterProjection {
schema_version: PARAMETER_PROJECTION_SCHEMA_VERSION,
nodes,
requires_recompile: structural_patch_count > 0,
structural_patch_count,
patches_fingerprint,
projection_fingerprint: zero_fingerprint(),
};
projection.projection_fingerprint = projection.compute_fingerprint()?;
projection.validate()?;
Ok(projection)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TrainingInfluenceKind {
TransformFit,
ModelFit,
HpoSelection,
EarlyStopping,
WeightingResampling,
TrainedMetaAggregation,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ControllerInfluenceRequirement {
pub node_id: NodeId,
pub kind: TrainingInfluenceKind,
pub scope_id: String,
pub phase: Phase,
#[serde(deserialize_with = "deserialize_required_nullable")]
pub fold_id: Option<FoldId>,
pub physical_sample_ids: Vec<SampleId>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingInfluenceEntry {
pub kind: TrainingInfluenceKind,
pub scope_id: String,
#[serde(deserialize_with = "deserialize_required_nullable")]
pub node_id: Option<NodeId>,
pub physical_sample_ids: Vec<SampleId>,
pub origin_sample_ids: Vec<SampleId>,
pub group_ids: Vec<GroupId>,
}
impl TrainingInfluenceEntry {
fn validate(&self) -> Result<()> {
validate_identifier_text("training influence scope_id", &self.scope_id)?;
validate_sorted_unique_ids(
"training influence physical_sample_ids",
&self.physical_sample_ids,
true,
)?;
validate_sorted_unique_ids(
"training influence origin_sample_ids",
&self.origin_sample_ids,
false,
)?;
validate_sorted_unique_ids("training influence group_ids", &self.group_ids, false)
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingInfluenceManifest {
pub schema_version: u32,
pub relation_fingerprint: String,
pub entries: Vec<TrainingInfluenceEntry>,
pub manifest_fingerprint: String,
}
impl TrainingInfluenceManifest {
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "manifest_fingerprint", "training influence manifest")
}
pub fn derive_for_projection(
projection: &TrainingContractProjection,
request: &TrainingRequest,
relations: &SampleRelationSet,
) -> Result<Self> {
projection.validate()?;
relations.validate()?;
let relation_fingerprint = relations.fingerprint()?;
if request
.data_identities
.iter()
.any(|identity| identity.relation_fingerprint != relation_fingerprint)
{
return contract_error(
"training data identities do not all bind the influence relation".to_string(),
);
}
let expected = expected_influence_coordinates(
request,
&projection.plan,
&projection.predictor_node_ids,
)?;
let mut entries = Vec::with_capacity(expected.len());
for ((kind, scope_id, node_id), samples) in expected {
let (origin_sample_ids, group_ids) =
influence_identity_closure_for_samples(&scope_id, &samples, relations)?;
entries.push(TrainingInfluenceEntry {
kind,
scope_id,
node_id,
physical_sample_ids: samples.into_iter().collect(),
origin_sample_ids,
group_ids,
});
}
let mut manifest = Self {
schema_version: TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
relation_fingerprint,
entries,
manifest_fingerprint: zero_fingerprint(),
};
manifest.manifest_fingerprint = manifest.compute_fingerprint()?;
manifest.validate_for_projection(projection, request, relations)?;
Ok(manifest)
}
pub fn validate(&self) -> Result<()> {
if self.schema_version != TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION {
return unsupported_version(
"training influence manifest",
self.schema_version,
TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
);
}
validate_sha256("training influence relation", &self.relation_fingerprint)?;
validate_sha256("training influence manifest", &self.manifest_fingerprint)?;
if self.entries.is_empty() {
return contract_error(
"training influence manifest requires at least one entry".to_string(),
);
}
let mut previous: Option<(TrainingInfluenceKind, &str, Option<&NodeId>)> = None;
for entry in &self.entries {
entry.validate()?;
let key = (entry.kind, entry.scope_id.as_str(), entry.node_id.as_ref());
if previous.as_ref().is_some_and(|previous| previous >= &key) {
return contract_error(
"training influence entries must be strictly canonically sorted".to_string(),
);
}
previous = Some(key);
}
if self.manifest_fingerprint != self.compute_fingerprint()? {
return contract_error(
"training influence manifest fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
pub fn validate_for_projection(
&self,
projection: &TrainingContractProjection,
request: &TrainingRequest,
relations: &SampleRelationSet,
) -> Result<()> {
self.validate()?;
relations.validate()?;
let relation_fingerprint = relations.fingerprint()?;
if self.relation_fingerprint != relation_fingerprint {
return contract_error(
"training influence relation fingerprint does not match relation set".to_string(),
);
}
if request
.data_identities
.iter()
.any(|identity| identity.relation_fingerprint != relation_fingerprint)
{
return contract_error(
"training data identities do not all bind the influence relation".to_string(),
);
}
let expected = expected_influence_coordinates(
request,
&projection.plan,
&projection.predictor_node_ids,
)?;
let mut actual = BTreeSet::new();
for entry in &self.entries {
if let Some(node_id) = &entry.node_id {
if !projection.predictor_node_ids.contains(node_id) {
return contract_error(format!(
"training influence node `{node_id}` is outside predictor closure"
));
}
}
let coordinate = (entry.kind, entry.scope_id.clone(), entry.node_id.clone());
let expected_samples = expected.get(&coordinate).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"training influence contains undeclared coordinate `{:?}/{}/{:?}`",
entry.kind, entry.scope_id, entry.node_id
))
})?;
if entry
.physical_sample_ids
.iter()
.cloned()
.collect::<BTreeSet<_>>()
!= *expected_samples
{
return contract_error(format!(
"training influence coordinate `{:?}/{}/{:?}` does not contain the exact scope samples",
entry.kind, entry.scope_id, entry.node_id
));
}
validate_influence_identity_closure(entry, relations)?;
actual.insert(coordinate);
}
let expected_keys = expected.into_keys().collect::<BTreeSet<_>>();
if actual != expected_keys {
return contract_error(
"training influence entries do not exactly cover capability-derived phase scopes"
.to_string(),
);
}
Ok(())
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingRequest {
pub schema_version: u32,
pub request_id: String,
pub plan_id: String,
pub graph: GraphSpec,
pub campaign: CampaignSpec,
pub controller_manifests: Vec<ControllerManifest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub training_losses: Vec<TrainingLossRoleReference>,
pub data_identities: Vec<TrainingDataIdentity>,
pub parameter_patches: Vec<ParameterPatch>,
pub patch_policies: Vec<NodePatchPolicy>,
pub influence_requirements: Vec<ControllerInfluenceRequirement>,
pub options: TrainingOptions,
pub request_fingerprint: String,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingContractProjection {
pub request_id: String,
pub request_fingerprint: String,
pub plan: ExecutionPlan,
pub outputs: Vec<ResolvedTrainingOutput>,
pub predictor_node_ids: BTreeSet<NodeId>,
pub parameters: ParameterProjection,
}
impl TrainingContractProjection {
pub fn from_json(json: &str) -> Result<Self> {
parse_typed_json(json).map_err(|error| {
DagMlError::RuntimeValidation(format!(
"training contract projection is outside strict TCV1 JSON: {error}"
))
})?;
let mut deserializer = serde_json::Deserializer::from_str(json);
let mut ignored_paths = Vec::new();
let projection: Self = serde_ignored::deserialize(&mut deserializer, |path| {
ignored_paths.push(path.to_string());
})?;
if !ignored_paths.is_empty() {
ignored_paths.sort();
ignored_paths.dedup();
return contract_error(format!(
"training contract projection contains unknown field(s) at: {}",
ignored_paths.join(", ")
));
}
projection.validate()?;
Ok(projection)
}
pub fn validate(&self) -> Result<()> {
validate_identifier_text("training projection request_id", &self.request_id)?;
validate_sha256(
"training projection request_fingerprint",
&self.request_fingerprint,
)?;
self.plan.validate()?;
self.parameters.validate()?;
if self.parameters.nodes.keys().collect::<BTreeSet<_>>()
!= self.plan.node_plans.keys().collect::<BTreeSet<_>>()
{
return contract_error(
"training projection parameter nodes do not exactly match execution plan"
.to_string(),
);
}
if self.outputs.is_empty() {
return contract_error("training projection requires at least one output".to_string());
}
let mut previous_id: Option<&str> = None;
let mut coordinates = BTreeSet::new();
for output in &self.outputs {
if previous_id.is_some_and(|previous| previous >= output.output_id.as_str()) {
return contract_error(
"training projection outputs must be strictly sorted by output_id".to_string(),
);
}
previous_id = Some(output.output_id.as_str());
let requested = TrainingOutputRequest {
output_id: output.output_id.clone(),
node_id: output.node_id.clone(),
port_name: Some(output.port_name.clone()),
prediction_level: output.prediction_level,
unit_level: output.unit_level,
prediction_kind: output.prediction_kind,
target_names: output.target_names.clone(),
target_units: output.target_units.clone(),
class_labels: output.class_labels.clone(),
output_order: output.output_order,
target_space: output.target_space.clone(),
};
if requested.resolve(&self.plan.graph_plan.graph)? != *output {
return contract_error(
"training projection contains a non-canonical resolved output".to_string(),
);
}
if !coordinates.insert((output.node_id.clone(), output.port_name.clone())) {
return contract_error(
"training projection contains duplicate output coordinates".to_string(),
);
}
}
let expected_closure = predictor_closure(
&self.plan,
self.outputs.iter().map(|output| &output.node_id),
)?;
if self.predictor_node_ids != expected_closure {
return contract_error(
"training projection predictor_node_ids do not match output closure".to_string(),
);
}
Ok(())
}
}
impl TrainingRequest {
pub fn from_json(json: &str) -> Result<Self> {
let raw_fingerprint =
strict_tcv1_fingerprint_without(json, "request_fingerprint", "training request")?;
let request: Self = serde_json::from_str(json)?;
if request.request_fingerprint != raw_fingerprint {
return contract_error(
"training request fingerprint does not match original TCV1 JSON".to_string(),
);
}
request.validate()?;
Ok(request)
}
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "request_fingerprint", "training request")
}
pub fn validate(&self) -> Result<()> {
self.project().map(|_| ())
}
pub fn project(&self) -> Result<TrainingContractProjection> {
if self.schema_version != TRAINING_REQUEST_SCHEMA_VERSION {
return unsupported_version(
"training request",
self.schema_version,
TRAINING_REQUEST_SCHEMA_VERSION,
);
}
validate_identifier_text("training request_id", &self.request_id)?;
validate_non_empty("training plan_id", &self.plan_id)?;
self.graph.validate()?;
self.campaign.validate()?;
if self.campaign.root_seed != Some(self.options.seed) {
return contract_error(
"training options seed must exactly match campaign.root_seed".to_string(),
);
}
validate_sha256("training request", &self.request_fingerprint)?;
if self.request_fingerprint != self.compute_fingerprint()? {
return contract_error(
"training request fingerprint does not match TCV1 content".to_string(),
);
}
let outputs = self.options.validate(&self.graph)?;
let mut registry = ControllerRegistry::new();
let mut previous_controller: Option<&str> = None;
for manifest in &self.controller_manifests {
if previous_controller
.is_some_and(|previous| previous >= manifest.controller_id.as_str())
{
return contract_error(
"training controller_manifests must be strictly sorted by controller_id"
.to_string(),
);
}
previous_controller = Some(manifest.controller_id.as_str());
registry.register(manifest.clone())?;
}
let mut plan = build_execution_plan(
self.plan_id.clone(),
self.graph.clone(),
self.campaign.clone(),
®istry,
)?;
validate_output_controllers(&plan, &outputs)?;
validate_selection_output(&plan, &self.options, &outputs)?;
let predictor_node_ids =
predictor_closure(&plan, outputs.iter().map(|output| &output.node_id))?;
attach_training_losses(&self.training_losses, &predictor_node_ids, &mut plan)?;
plan.validate()?;
validate_training_data_identities(self, &plan)?;
let parameters =
project_parameter_patches(&plan, &self.parameter_patches, &self.patch_policies)?;
validate_scheduler_capabilities(&self.options.scheduler, &plan, &predictor_node_ids)?;
validate_artifact_mode(&self.options.artifacts, &plan, &predictor_node_ids)?;
validate_influence_requirements(self, &plan, &predictor_node_ids)?;
let projection = TrainingContractProjection {
request_id: self.request_id.clone(),
request_fingerprint: self.request_fingerprint.clone(),
plan,
outputs,
predictor_node_ids,
parameters,
};
projection.validate()?;
Ok(projection)
}
}
fn attach_training_losses(
roles: &[TrainingLossRoleReference],
predictor_node_ids: &BTreeSet<NodeId>,
plan: &mut ExecutionPlan,
) -> Result<()> {
let mut previous_key: Option<(NodeId, Option<String>, BTreeSet<Phase>)> = None;
for role in roles {
role.validate()?;
let key = (
role.node_id.clone(),
role.output_id.clone(),
role.phases.clone(),
);
if previous_key
.as_ref()
.is_some_and(|previous| previous >= &key)
{
return contract_error(
"training losses must be strictly sorted by node_id, output_id and phases"
.to_string(),
);
}
previous_key = Some(key);
if !predictor_node_ids.contains(&role.node_id) {
return contract_error(format!(
"training loss node `{}` is outside the predictor closure",
role.node_id
));
}
let node_plan = plan.node_plans.get_mut(&role.node_id).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"training loss references unknown node `{}`",
role.node_id
))
})?;
node_plan.training_losses.push(role.clone());
}
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CacheNamespace {
pub schema_version: u32,
pub prediction_requirement_key: String,
pub data_requirement_key: String,
pub producer_node_id: NodeId,
pub source_port_name: String,
pub consumer_node_id: NodeId,
pub target_port_name: String,
pub phase: Phase,
pub params_fingerprint: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub training_loss_fingerprint: Option<String>,
pub data_identity_fingerprint: String,
pub fold_id: FoldId,
pub trial_id: String,
pub seed: u64,
pub namespace_fingerprint: String,
}
impl CacheNamespace {
#[allow(clippy::too_many_arguments)]
pub fn new(
prediction_requirement_key: String,
data_requirement_key: String,
producer_node_id: NodeId,
source_port_name: String,
consumer_node_id: NodeId,
target_port_name: String,
params_fingerprint: String,
training_loss_fingerprint: Option<String>,
data_identity_fingerprint: String,
fold_id: FoldId,
trial_id: String,
seed: u64,
) -> Result<Self> {
let mut namespace = Self {
schema_version: CACHE_NAMESPACE_SCHEMA_VERSION,
prediction_requirement_key,
data_requirement_key,
producer_node_id,
source_port_name,
consumer_node_id,
target_port_name,
phase: Phase::FitCv,
params_fingerprint,
training_loss_fingerprint,
data_identity_fingerprint,
fold_id,
trial_id,
seed,
namespace_fingerprint: zero_fingerprint(),
};
namespace.namespace_fingerprint = namespace.compute_fingerprint()?;
namespace.validate()?;
Ok(namespace)
}
pub fn from_json(json: &str) -> Result<Self> {
let raw_fingerprint =
strict_tcv1_fingerprint_without(json, "namespace_fingerprint", "cache namespace")?;
let namespace: Self = serde_json::from_str(json)?;
if namespace.namespace_fingerprint != raw_fingerprint {
return contract_error(
"cache namespace fingerprint does not match original TCV1 JSON".to_string(),
);
}
namespace.validate()?;
Ok(namespace)
}
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "namespace_fingerprint", "cache namespace")
}
pub fn validate(&self) -> Result<()> {
if self.schema_version != CACHE_NAMESPACE_SCHEMA_VERSION {
return unsupported_version(
"cache namespace",
self.schema_version,
CACHE_NAMESPACE_SCHEMA_VERSION,
);
}
validate_non_empty(
"cache namespace prediction_requirement_key",
&self.prediction_requirement_key,
)?;
validate_non_empty(
"cache namespace data_requirement_key",
&self.data_requirement_key,
)?;
validate_non_empty("cache namespace source_port_name", &self.source_port_name)?;
validate_non_empty("cache namespace target_port_name", &self.target_port_name)?;
if self.phase != Phase::FitCv {
return contract_error(
"cache namespace V1 is fold-scoped and permits only FIT_CV".to_string(),
);
}
let expected_requirement_key = bundle_prediction_requirement_key(
&self.producer_node_id,
&self.source_port_name,
&self.consumer_node_id,
&self.target_port_name,
);
if self.prediction_requirement_key != expected_requirement_key {
return contract_error(
"cache namespace requirement_key does not match producer/source/consumer/target coordinates"
.to_string(),
);
}
validate_identifier_text("cache namespace trial_id", &self.trial_id)?;
for (label, fingerprint) in [
("cache params", &self.params_fingerprint),
("cache data identity", &self.data_identity_fingerprint),
("cache namespace", &self.namespace_fingerprint),
] {
validate_sha256(label, fingerprint)?;
}
if let Some(fingerprint) = &self.training_loss_fingerprint {
validate_sha256("cache training loss", fingerprint)?;
}
if self.namespace_fingerprint != self.compute_fingerprint()? {
return contract_error(
"cache namespace fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
pub fn validate_for_identity(&self, identity: &TrainingDataIdentity) -> Result<()> {
self.validate()?;
identity.validate()?;
if self.data_requirement_key != identity.requirement_key
|| self.data_identity_fingerprint != identity.identity_fingerprint
{
return contract_error(
"cache namespace does not bind the complete training data identity".to_string(),
);
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct OutputBinding {
pub schema_version: u32,
pub binding_id: String,
pub node_id: NodeId,
pub port_name: String,
pub prediction_level: PredictionLevel,
#[serde(default)]
pub unit_level: Option<EntityUnitLevel>,
pub prediction_kind: PredictionKind,
pub prediction_source: PredictionSource,
#[serde(default)]
pub refit_strategy: Option<RefitStrategy>,
pub aggregation_fingerprint: String,
pub target_names: Vec<String>,
pub target_units: Vec<Option<String>>,
pub class_labels: Vec<Vec<String>>,
pub output_order: OutputOrder,
pub target_space: String,
pub binding_fingerprint: String,
}
impl OutputBinding {
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "binding_fingerprint", "output binding")
}
pub fn validate(&self, graph: &GraphSpec) -> Result<()> {
if self.schema_version != OUTPUT_BINDING_SCHEMA_VERSION {
return unsupported_version(
"output binding",
self.schema_version,
OUTPUT_BINDING_SCHEMA_VERSION,
);
}
validate_identifier_text("output binding_id", &self.binding_id)?;
validate_non_empty("output binding port_name", &self.port_name)?;
validate_sha256("output aggregation", &self.aggregation_fingerprint)?;
validate_sha256("output binding", &self.binding_fingerprint)?;
validate_output_unit_level(self.prediction_level, self.unit_level)?;
validate_output_shape(
self.prediction_kind,
self.output_order,
&self.target_names,
&self.target_units,
&self.class_labels,
&self.target_space,
)?;
match (self.prediction_source, self.refit_strategy) {
(PredictionSource::FinalRefit, None) => {
return contract_error(
"final_refit output binding requires refit_strategy".to_string(),
);
}
(PredictionSource::CvEnsemble | PredictionSource::FoldMember, Some(_)) => {
return contract_error(
"non-final output binding forbids refit_strategy".to_string(),
);
}
_ => {}
}
let request = TrainingOutputRequest {
output_id: self.binding_id.clone(),
node_id: self.node_id.clone(),
port_name: Some(self.port_name.clone()),
prediction_level: self.prediction_level,
unit_level: self.unit_level,
prediction_kind: self.prediction_kind,
target_names: self.target_names.clone(),
target_units: self.target_units.clone(),
class_labels: self.class_labels.clone(),
output_order: self.output_order,
target_space: self.target_space.clone(),
};
request.resolve(graph)?;
if self.binding_fingerprint != self.compute_fingerprint()? {
return contract_error(
"output binding fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PredictorTemplate {
pub graph: GraphSpec,
pub campaign: CampaignSpec,
pub controller_manifests: BTreeMap<crate::ids::ControllerId, ControllerManifest>,
pub template_fingerprint: String,
}
impl PredictorTemplate {
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "template_fingerprint", "predictor template")
}
pub fn validate(&self) -> Result<()> {
self.graph.validate()?;
self.campaign.validate()?;
for (controller_id, manifest) in &self.controller_manifests {
if controller_id != &manifest.controller_id {
return contract_error(format!(
"predictor template controller key `{controller_id}` does not match manifest `{}`",
manifest.controller_id
));
}
manifest.validate()?;
}
validate_sha256("predictor template", &self.template_fingerprint)?;
if self.template_fingerprint != self.compute_fingerprint()? {
return contract_error(
"predictor template fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TrainingOutcomeRef {
pub outcome_id: String,
pub outcome_fingerprint: String,
pub training_request_fingerprint: String,
pub effective_plan_fingerprint: String,
pub execution_bundle_id: BundleId,
pub execution_bundle_fingerprint: String,
pub output_binding_fingerprints: Vec<String>,
pub training_influence_fingerprint: String,
pub data_identities_fingerprint: String,
}
impl TrainingOutcomeRef {
pub(crate) fn validate(&self) -> Result<()> {
validate_identifier_text("training outcome_id", &self.outcome_id)?;
for (label, fingerprint) in [
("training outcome", &self.outcome_fingerprint),
(
"training outcome request",
&self.training_request_fingerprint,
),
(
"training outcome effective plan",
&self.effective_plan_fingerprint,
),
(
"training outcome influence",
&self.training_influence_fingerprint,
),
(
"training outcome execution bundle",
&self.execution_bundle_fingerprint,
),
(
"training outcome data identities",
&self.data_identities_fingerprint,
),
] {
validate_sha256(label, fingerprint)?;
}
if self.output_binding_fingerprints.is_empty() {
return contract_error(
"training outcome reference requires output binding fingerprints".to_string(),
);
}
for fingerprint in &self.output_binding_fingerprints {
validate_sha256("training outcome output binding", fingerprint)?;
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ArtifactLoadMode {
NativePortable,
HostSidecar,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PackageArtifactBinding {
pub artifact_id: ArtifactId,
pub load_mode: ArtifactLoadMode,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PortablePredictorPackage {
pub schema_version: u32,
pub package_id: String,
pub template: PredictorTemplate,
pub training_request_fingerprint: String,
pub training_outcome: TrainingOutcomeRef,
pub effective_plan: ExecutionPlan,
pub execution_bundle: ExecutionBundle,
pub output_bindings: Vec<OutputBinding>,
pub predictor_node_ids: Vec<NodeId>,
pub training_influence: TrainingInfluenceManifest,
pub data_identities: Vec<TrainingDataIdentity>,
pub fitted_artifact_mode: FittedArtifactMode,
pub artifact_bindings: Vec<PackageArtifactBinding>,
pub package_fingerprint: String,
}
impl PortablePredictorPackage {
pub fn compute_fingerprint(&self) -> Result<String> {
tcv1_fingerprint_without(self, "package_fingerprint", "portable predictor package")
}
pub fn from_json(json: &str) -> Result<Self> {
let raw_fingerprint = strict_tcv1_fingerprint_without(
json,
"package_fingerprint",
"portable predictor package",
)?;
let package: Self = serde_json::from_str(json)?;
if package.package_fingerprint != raw_fingerprint {
return contract_error(
"portable predictor package fingerprint does not match original TCV1 JSON"
.to_string(),
);
}
package.validate()?;
Ok(package)
}
pub fn validate(&self) -> Result<()> {
if self.schema_version != PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION {
return unsupported_version(
"portable predictor package",
self.schema_version,
PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION,
);
}
validate_identifier_text("portable predictor package_id", &self.package_id)?;
validate_sha256(
"portable predictor training request",
&self.training_request_fingerprint,
)?;
validate_sha256("portable predictor package", &self.package_fingerprint)?;
self.training_outcome.validate()?;
if self.training_request_fingerprint != self.training_outcome.training_request_fingerprint {
return contract_error(
"portable predictor request fingerprint is not cross-linked by outcome reference"
.to_string(),
);
}
self.template.validate()?;
self.effective_plan.validate()?;
if self.template.graph != self.effective_plan.graph_plan.graph
|| self.template.campaign != self.effective_plan.campaign
|| self.template.controller_manifests != self.effective_plan.controller_manifests
{
return contract_error(
"portable predictor template does not exactly match effective plan".to_string(),
);
}
self.execution_bundle
.validate_against_plan(&self.effective_plan)?;
let effective_plan_fingerprint =
tcv1_fingerprint(&self.effective_plan, "portable predictor effective plan")?;
if effective_plan_fingerprint != self.training_outcome.effective_plan_fingerprint {
return contract_error(
"portable predictor effective plan fingerprint is not cross-linked by outcome reference"
.to_string(),
);
}
if self.execution_bundle.bundle_id != self.training_outcome.execution_bundle_id {
return contract_error(
"portable predictor bundle id is not cross-linked by outcome reference".to_string(),
);
}
let execution_bundle_fingerprint = tcv1_fingerprint(
&self.execution_bundle,
"portable predictor execution bundle",
)?;
if execution_bundle_fingerprint != self.training_outcome.execution_bundle_fingerprint {
return contract_error(
"portable predictor execution bundle content is not cross-linked by outcome reference"
.to_string(),
);
}
self.training_influence.validate()?;
if self.training_influence.manifest_fingerprint
!= self.training_outcome.training_influence_fingerprint
{
return contract_error(
"portable predictor influence is not cross-linked by outcome reference".to_string(),
);
}
if self.output_bindings.is_empty() {
return contract_error(
"portable predictor package requires at least one output binding".to_string(),
);
}
let mut previous_binding: Option<&str> = None;
let mut output_nodes = Vec::new();
let mut coordinates = BTreeSet::new();
for binding in &self.output_bindings {
if previous_binding.is_some_and(|previous| previous >= binding.binding_id.as_str()) {
return contract_error(
"portable predictor output bindings must be sorted by binding_id".to_string(),
);
}
previous_binding = Some(binding.binding_id.as_str());
binding.validate(&self.effective_plan.graph_plan.graph)?;
if !coordinates.insert((binding.node_id.clone(), binding.port_name.clone())) {
return contract_error(format!(
"portable predictor binds `{}.{}` more than once",
binding.node_id, binding.port_name
));
}
if binding.prediction_source == PredictionSource::FinalRefit
&& self.execution_bundle.refit_artifacts.is_empty()
{
return contract_error(
"final_refit output binding requires refit artifacts".to_string(),
);
}
output_nodes.push(&binding.node_id);
}
if self
.output_bindings
.iter()
.map(|binding| binding.binding_fingerprint.clone())
.collect::<Vec<_>>()
!= self.training_outcome.output_binding_fingerprints
{
return contract_error(
"portable predictor output bindings are not cross-linked by outcome reference"
.to_string(),
);
}
let expected_closure = predictor_closure(&self.effective_plan, output_nodes)?;
validate_sorted_unique_ids(
"portable predictor_node_ids",
&self.predictor_node_ids,
true,
)?;
if self
.predictor_node_ids
.iter()
.cloned()
.collect::<BTreeSet<_>>()
!= expected_closure
{
return contract_error(
"portable predictor_node_ids do not exactly match output closure".to_string(),
);
}
if self.training_influence.entries.iter().any(|entry| {
entry
.node_id
.as_ref()
.is_some_and(|node_id| !expected_closure.contains(node_id))
}) {
return contract_error(
"portable predictor influence references a node outside predictor closure"
.to_string(),
);
}
validate_package_base_influence(
&self.training_influence,
&self.effective_plan,
&expected_closure,
)?;
validate_package_data_identities(self)?;
let data_identities_fingerprint =
tcv1_fingerprint(&self.data_identities, "portable predictor data identities")?;
if data_identities_fingerprint != self.training_outcome.data_identities_fingerprint {
return contract_error(
"portable predictor data identity content is not cross-linked by outcome reference"
.to_string(),
);
}
if self.data_identities.iter().any(|identity| {
identity.relation_fingerprint != self.training_influence.relation_fingerprint
}) {
return contract_error(
"portable predictor data identities and training influence bind different relations"
.to_string(),
);
}
validate_package_artifact_bindings(self)?;
if !crate::training_runtime::closure_predict_replayable(
&self.effective_plan,
&expected_closure,
&self.execution_bundle,
)? {
return contract_error(
"portable predictor package is not PREDICT-replayable: its predictor closure does not support PREDICT with self-contained retained artifacts".to_string(),
);
}
let value = serde_json::to_value(self)?;
if contains_runtime_handle(&value) {
return contract_error(
"portable predictor package must not contain runtime handles".to_string(),
);
}
if self.package_fingerprint != self.compute_fingerprint()? {
return contract_error(
"portable predictor package fingerprint does not match TCV1 content".to_string(),
);
}
Ok(())
}
pub fn load_with<H>(
self,
mut resolver: impl FnMut(&RefitArtifactRecord) -> Result<H>,
) -> Result<LoadedPredictor<H>> {
self.validate()?;
let mut artifacts = BTreeMap::new();
let sidecar_ids = self
.artifact_bindings
.iter()
.filter(|binding| binding.load_mode == ArtifactLoadMode::HostSidecar)
.map(|binding| &binding.artifact_id)
.collect::<BTreeSet<_>>();
for record in self
.execution_bundle
.refit_artifacts
.iter()
.filter(|record| sidecar_ids.contains(&record.artifact.id))
{
let handle = resolver(record)?;
artifacts.insert(record.artifact.id.clone(), handle);
}
LoadedPredictor::new(self, artifacts)
}
}
pub struct LoadedPredictor<H> {
package: PortablePredictorPackage,
artifacts: BTreeMap<ArtifactId, H>,
}
impl<H> LoadedPredictor<H> {
pub fn new(
package: PortablePredictorPackage,
artifacts: BTreeMap<ArtifactId, H>,
) -> Result<Self> {
package.validate()?;
let expected = package
.artifact_bindings
.iter()
.filter(|binding| binding.load_mode == ArtifactLoadMode::HostSidecar)
.map(|binding| binding.artifact_id.clone())
.collect::<BTreeSet<_>>();
let actual = artifacts.keys().cloned().collect::<BTreeSet<_>>();
if actual != expected {
return contract_error(
"loaded predictor sidecar artifacts do not exactly match package references"
.to_string(),
);
}
Ok(Self { package, artifacts })
}
pub fn package(&self) -> &PortablePredictorPackage {
&self.package
}
pub fn artifact(&self, artifact_id: &ArtifactId) -> Option<&H> {
self.artifacts.get(artifact_id)
}
pub fn into_parts(self) -> (PortablePredictorPackage, BTreeMap<ArtifactId, H>) {
(self.package, self.artifacts)
}
}
fn validate_output_unit_level(
prediction_level: PredictionLevel,
unit_level: Option<EntityUnitLevel>,
) -> Result<()> {
match prediction_level {
PredictionLevel::Sample if unit_level != Some(EntityUnitLevel::PhysicalSample) => {
contract_error("sample-level output requires unit_level=physical_sample".to_string())
}
PredictionLevel::Target | PredictionLevel::Group if unit_level.is_some() => {
contract_error("target/group output requires unit_level=null".to_string())
}
_ => Ok(()),
}
}
fn validate_output_shape(
prediction_kind: PredictionKind,
output_order: OutputOrder,
target_names: &[String],
target_units: &[Option<String>],
class_labels: &[Vec<String>],
target_space: &str,
) -> Result<()> {
validate_non_empty("output target_space", target_space)?;
validate_unique_text("output target_names", target_names, true)?;
if target_units.len() != target_names.len() || class_labels.len() != target_names.len() {
return contract_error(
"output target_units and class_labels must have one entry per target".to_string(),
);
}
for unit in target_units.iter().flatten() {
validate_non_empty("output target unit", unit)?;
}
let class_output = prediction_kind == PredictionKind::ClassProbability;
for labels in class_labels {
validate_unique_text("output class labels", labels, class_output)?;
}
if prediction_kind == PredictionKind::RegressionPoint
&& class_labels.iter().any(|labels| !labels.is_empty())
{
return contract_error("regression output class label arrays must be empty".to_string());
}
match (prediction_kind, output_order) {
(PredictionKind::ClassProbability, OutputOrder::TargetMajorClassMinor) => Ok(()),
(PredictionKind::ClassProbability, _) => contract_error(
"class_probability output requires target_major_class_minor order".to_string(),
),
(_, OutputOrder::TargetOrder) => Ok(()),
_ => contract_error("non-probability output requires target_order".to_string()),
}
}
fn validate_canonical_patches(patches: &[ParameterPatch]) -> Result<()> {
let mut previous: Option<(&NodeId, ParameterNamespace, &[String])> = None;
for patch in patches {
patch.validate()?;
let key = (&patch.node_id, patch.namespace, patch.path.as_slice());
if let Some(previous_key) = previous.as_ref() {
if previous_key >= &key {
return contract_error(
"parameter patches must be strictly sorted by (node_id, namespace, path)"
.to_string(),
);
}
if previous_key.0 == key.0
&& previous_key.1 == key.1
&& (key.2.starts_with(previous_key.2) || previous_key.2.starts_with(key.2))
{
return contract_error(format!(
"parameter patches for `{}` contain a conflicting parent/child path",
patch.node_id
));
}
}
previous = Some(key);
}
Ok(())
}
fn validate_patch_policies(
plan: &ExecutionPlan,
policies: &[NodePatchPolicy],
) -> Result<BTreeMap<NodeId, BTreeSet<ParameterNamespace>>> {
let mut previous: Option<&NodeId> = None;
let mut result = BTreeMap::new();
for policy in policies {
policy.validate()?;
if previous.is_some_and(|previous| previous >= &policy.node_id) {
return contract_error(
"node patch policies must be strictly sorted by node_id".to_string(),
);
}
previous = Some(&policy.node_id);
if !plan.node_plans.contains_key(&policy.node_id) {
return contract_error(format!(
"node patch policy references unknown node `{}`",
policy.node_id
));
}
result.insert(policy.node_id.clone(), policy.allowed_namespaces.clone());
}
Ok(result)
}
fn deep_set_object_key(
root: &mut BTreeMap<String, serde_json::Value>,
path: &[String],
value: serde_json::Value,
node_id: &NodeId,
) -> Result<()> {
if path.len() == 1 {
root.insert(path[0].clone(), value);
return Ok(());
}
let first = root.get_mut(&path[0]).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch for `{node_id}` is missing intermediate path `{}`",
path[0]
))
})?;
let mut cursor = first;
for segment in &path[1..path.len() - 1] {
let object = cursor.as_object_mut().ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch for `{node_id}` crosses a scalar or array at `{segment}`"
))
})?;
cursor = object.get_mut(segment).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch for `{node_id}` is missing intermediate path `{segment}`"
))
})?;
}
let object = cursor.as_object_mut().ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"parameter patch for `{node_id}` crosses a scalar or array before final key"
))
})?;
object.insert(path[path.len() - 1].clone(), value);
Ok(())
}
fn predictor_closure<'a>(
plan: &ExecutionPlan,
roots: impl IntoIterator<Item = &'a NodeId>,
) -> Result<BTreeSet<NodeId>> {
let mut pending = roots.into_iter().cloned().collect::<Vec<_>>();
let mut closure = BTreeSet::new();
while let Some(node_id) = pending.pop() {
if !closure.insert(node_id.clone()) {
continue;
}
let node = plan.node_plans.get(&node_id).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"predictor closure references unknown node `{node_id}`"
))
})?;
pending.extend(node.input_nodes.iter().cloned());
}
Ok(closure)
}
fn base_influence_kind(plan: &ExecutionPlan, node_id: &NodeId) -> Option<TrainingInfluenceKind> {
let node_plan = &plan.node_plans[node_id];
if matches!(
node_plan.fit_scope,
ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly
) {
return None;
}
let oof_consumers = plan
.graph_plan
.graph
.edges
.iter()
.filter(|edge| edge.contract.requires_oof)
.map(|edge| edge.target.node_id.clone())
.collect::<BTreeSet<_>>();
Some(
if oof_consumers.contains(node_id)
|| node_plan
.controller_capabilities
.contains(&ControllerCapability::TrainsAggregation)
{
TrainingInfluenceKind::TrainedMetaAggregation
} else if node_plan.kind == NodeKind::Model {
TrainingInfluenceKind::ModelFit
} else if node_plan.kind == NodeKind::Tuner {
TrainingInfluenceKind::HpoSelection
} else {
TrainingInfluenceKind::TransformFit
},
)
}
fn capability_influence_kinds(
plan: &ExecutionPlan,
node_id: &NodeId,
) -> BTreeSet<TrainingInfluenceKind> {
let capabilities = &plan.node_plans[node_id].controller_capabilities;
let mut kinds = BTreeSet::new();
if capabilities.contains(&ControllerCapability::PerformsInternalTuning)
&& base_influence_kind(plan, node_id) != Some(TrainingInfluenceKind::HpoSelection)
{
kinds.insert(TrainingInfluenceKind::HpoSelection);
}
if capabilities.contains(&ControllerCapability::UsesEarlyStopping) {
kinds.insert(TrainingInfluenceKind::EarlyStopping);
}
if capabilities.contains(&ControllerCapability::UsesTrainingWeights) {
kinds.insert(TrainingInfluenceKind::WeightingResampling);
}
kinds
}
fn expected_influence_coordinates(
request: &TrainingRequest,
plan: &ExecutionPlan,
closure: &BTreeSet<NodeId>,
) -> Result<ExpectedInfluenceCoordinates> {
let fold_set = plan.fold_set.as_ref().ok_or_else(|| {
DagMlError::CampaignValidation(
"training influence scopes require an explicit fold_set".to_string(),
)
})?;
let all_samples = fold_set.sample_ids.iter().cloned().collect::<BTreeSet<_>>();
let mut expected = BTreeMap::new();
for node_id in closure {
let node_plan = &plan.node_plans[node_id];
let Some(base_kind) = base_influence_kind(plan, node_id) else {
continue;
};
let mut scopes = Vec::<(String, BTreeSet<SampleId>)>::new();
if node_plan.supported_phases.contains(&Phase::FitCv) {
match node_plan.fit_scope {
ControllerFitScope::FoldTrain => {
scopes.extend(fold_set.folds.iter().map(|fold| {
(
format!("fit_cv:{}", fold.fold_id),
fold.train_sample_ids.iter().cloned().collect(),
)
}));
}
ControllerFitScope::FullTrain => {
scopes.push(("fit_cv:full".to_string(), all_samples.clone()));
}
ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly => {}
}
}
if request.options.refit && node_plan.supported_phases.contains(&Phase::Refit) {
scopes.push(("refit:full".to_string(), all_samples.clone()));
}
for (scope_id, samples) in scopes {
expected.insert((base_kind, scope_id, Some(node_id.clone())), samples);
}
}
for requirement in &request.influence_requirements {
let key = (
requirement.kind,
requirement.scope_id.clone(),
Some(requirement.node_id.clone()),
);
if expected
.insert(
key,
requirement.physical_sample_ids.iter().cloned().collect(),
)
.is_some()
{
return contract_error(format!(
"controller influence scope `{}` collides with a derived influence coordinate",
requirement.scope_id
));
}
}
expected.insert(
(
TrainingInfluenceKind::HpoSelection,
format!("select:{}", request.options.selection.id),
None,
),
all_samples,
);
Ok(expected)
}
fn validate_influence_requirements(
request: &TrainingRequest,
plan: &ExecutionPlan,
closure: &BTreeSet<NodeId>,
) -> Result<()> {
let fold_set = plan.fold_set.as_ref().ok_or_else(|| {
DagMlError::CampaignValidation(
"controller influence requirements need an explicit fold_set".to_string(),
)
})?;
let all_samples = fold_set.sample_ids.iter().cloned().collect::<BTreeSet<_>>();
let mut expected_slots = BTreeMap::<InfluenceCapabilitySlot, BTreeSet<SampleId>>::new();
for node_id in closure {
let node_plan = &plan.node_plans[node_id];
if base_influence_kind(plan, node_id).is_none() {
continue;
}
let kinds = capability_influence_kinds(plan, node_id);
if node_plan.supported_phases.contains(&Phase::FitCv) {
match node_plan.fit_scope {
ControllerFitScope::FoldTrain => {
for fold in &fold_set.folds {
for kind in &kinds {
expected_slots.insert(
(
node_id.clone(),
*kind,
Phase::FitCv,
Some(fold.fold_id.clone()),
),
fold.train_sample_ids.iter().cloned().collect(),
);
}
}
}
ControllerFitScope::FullTrain => {
for kind in &kinds {
expected_slots.insert(
(node_id.clone(), *kind, Phase::FitCv, None),
all_samples.clone(),
);
}
}
ControllerFitScope::Stateless | ControllerFitScope::InferenceOnly => {}
}
}
if request.options.refit && node_plan.supported_phases.contains(&Phase::Refit) {
for kind in kinds {
expected_slots.insert(
(node_id.clone(), kind, Phase::Refit, None),
all_samples.clone(),
);
}
}
}
let mut actual_slots = BTreeSet::new();
let mut previous: Option<(TrainingInfluenceKind, &str, &NodeId)> = None;
for requirement in &request.influence_requirements {
validate_identifier_text(
"controller influence requirement scope_id",
&requirement.scope_id,
)?;
validate_sorted_unique_ids(
"controller influence requirement physical_sample_ids",
&requirement.physical_sample_ids,
true,
)?;
let key = (
requirement.kind,
requirement.scope_id.as_str(),
&requirement.node_id,
);
if previous.as_ref().is_some_and(|previous| previous >= &key) {
return contract_error(
"controller influence requirements must be strictly sorted by (kind, scope_id, node_id)"
.to_string(),
);
}
previous = Some(key);
if !closure.contains(&requirement.node_id) {
return contract_error(format!(
"controller influence requirement node `{}` is outside predictor closure",
requirement.node_id
));
}
if !matches!(requirement.phase, Phase::FitCv | Phase::Refit) {
return contract_error(format!(
"controller influence scope `{}` uses non-training phase {:?}",
requirement.scope_id, requirement.phase
));
}
let slot = (
requirement.node_id.clone(),
requirement.kind,
requirement.phase,
requirement.fold_id.clone(),
);
let eligible_samples = expected_slots.get(&slot).ok_or_else(|| {
DagMlError::CampaignValidation(format!(
"controller influence scope `{}` is not required by active controller capabilities",
requirement.scope_id
))
})?;
let actual_samples = requirement
.physical_sample_ids
.iter()
.cloned()
.collect::<BTreeSet<_>>();
if !actual_samples.is_subset(eligible_samples) {
let outer_validation_overlap = requirement
.fold_id
.as_ref()
.and_then(|fold_id| fold_set.folds.iter().find(|fold| &fold.fold_id == fold_id))
.is_some_and(|fold| {
fold.validation_sample_ids
.iter()
.any(|sample_id| actual_samples.contains(sample_id))
});
if outer_validation_overlap {
return contract_error(format!(
"controller influence scope `{}` leaks outer validation samples",
requirement.scope_id
));
}
return contract_error(format!(
"controller influence scope `{}` uses samples outside its training cohort",
requirement.scope_id
));
}
match requirement.kind {
TrainingInfluenceKind::WeightingResampling if actual_samples != *eligible_samples => {
return contract_error(format!(
"weighting influence scope `{}` must cover its complete fit cohort",
requirement.scope_id
));
}
TrainingInfluenceKind::EarlyStopping
if actual_samples.len() >= eligible_samples.len() =>
{
return contract_error(format!(
"early-stopping influence scope `{}` must be a strict training-cohort subset",
requirement.scope_id
));
}
_ => {}
}
if !actual_slots.insert(slot) {
return contract_error(format!(
"controller influence capability slot is declared more than once at `{}`",
requirement.scope_id
));
}
}
if actual_slots != expected_slots.into_keys().collect::<BTreeSet<_>>() {
return contract_error(
"controller influence requirements do not exactly cover active capability scopes"
.to_string(),
);
}
Ok(())
}
fn validate_influence_identity_closure(
entry: &TrainingInfluenceEntry,
relations: &SampleRelationSet,
) -> Result<()> {
let physical = entry.physical_sample_ids.iter().collect::<BTreeSet<_>>();
let mut found = BTreeSet::new();
let mut origins = BTreeSet::new();
let mut groups = BTreeSet::new();
for relation in &relations.records {
if physical.contains(&relation.sample_id) {
found.insert(&relation.sample_id);
if let Some(origin) = &relation.origin_sample_id {
origins.insert(origin.clone());
}
if let Some(group) = &relation.group_id {
groups.insert(group.clone());
}
}
}
if found.len() != physical.len() {
return contract_error(format!(
"training influence `{}` contains physical samples absent from relation set",
entry.scope_id
));
}
if entry
.origin_sample_ids
.iter()
.cloned()
.collect::<BTreeSet<_>>()
!= origins
{
return contract_error(format!(
"training influence `{}` origin closure does not match relation set",
entry.scope_id
));
}
if entry.group_ids.iter().cloned().collect::<BTreeSet<_>>() != groups {
return contract_error(format!(
"training influence `{}` group closure does not match relation set",
entry.scope_id
));
}
Ok(())
}
fn validate_output_controllers(
plan: &ExecutionPlan,
outputs: &[ResolvedTrainingOutput],
) -> Result<()> {
for output in outputs {
let node = &plan.node_plans[&output.node_id];
if !node
.controller_capabilities
.contains(&ControllerCapability::EmitsPredictions)
{
return contract_error(format!(
"training output node `{}` controller does not declare emits_predictions",
output.node_id
));
}
}
Ok(())
}
fn validate_selection_output(
plan: &ExecutionPlan,
options: &TrainingOptions,
outputs: &[ResolvedTrainingOutput],
) -> Result<()> {
let output = outputs
.iter()
.find(|output| output.output_id == options.selection_output_id)
.expect("TrainingOptions::validate resolved selection_output_id");
let node_plan = &plan.node_plans[&output.node_id];
if !node_plan.supported_phases.contains(&Phase::FitCv) {
return contract_error(format!(
"training selection output `{}` is not scorable in FIT_CV",
output.output_id
));
}
let graph_node = plan
.graph_plan
.graph
.nodes
.iter()
.find(|node| node.id == output.node_id)
.expect("execution plan output node exists in graph");
let binds_declared_prediction_port = graph_node
.ports
.outputs
.iter()
.any(|port| port.kind == PortKind::Prediction && port.name == output.port_name);
if !binds_declared_prediction_port {
return contract_error(format!(
"training selection output `{}` port `{}.{}` is not a declared prediction port",
output.output_id, output.node_id, output.port_name
));
}
let campaign_metric_level = plan.campaign.aggregation_policy.selection_metric_level;
if output.prediction_level != campaign_metric_level {
return contract_error(format!(
"training selection output `{}` prediction level does not match campaign selection_metric_level",
output.output_id
));
}
if options
.selection
.required_metric_level
.is_some_and(|level| level != campaign_metric_level)
{
return contract_error(format!(
"training selection output `{}` prediction level does not match selection.required_metric_level",
output.output_id
));
}
let metric_name = options.selection.metric.name.as_str();
let objective = options.selection.metric.objective;
crate::metrics::RegressionMetricKind::resolve_for_prediction_kind(
metric_name,
objective,
output.prediction_kind,
)?;
Ok(())
}
fn validate_scheduler_capabilities(
scheduler: &TrainingSchedulerOptions,
plan: &ExecutionPlan,
closure: &BTreeSet<NodeId>,
) -> Result<()> {
let Some(backend) = scheduler.backend else {
return Ok(());
};
for node_id in closure {
let capabilities = &plan.node_plans[node_id].controller_capabilities;
match backend {
TrainingSchedulerBackend::Threads => {
if !capabilities.contains(&ControllerCapability::ThreadSafe) {
return contract_error(format!(
"parallel thread scheduler requires thread_safe controller for `{node_id}`"
));
}
if capabilities.contains(&ControllerCapability::NeedsPythonGil) {
return contract_error(format!(
"parallel thread scheduler refuses needs_python_gil controller for `{node_id}`"
));
}
}
TrainingSchedulerBackend::Processes => {
if !capabilities.contains(&ControllerCapability::ProcessSafe) {
return contract_error(format!(
"parallel process scheduler requires process_safe controller for `{node_id}`"
));
}
}
}
}
Ok(())
}
fn validate_artifact_mode(
artifacts: &TrainingArtifactOptions,
plan: &ExecutionPlan,
closure: &BTreeSet<NodeId>,
) -> Result<()> {
if artifacts.fitted_artifacts != FittedArtifactMode::PortableRequired {
return Ok(());
}
for node_id in closure {
let node = &plan.node_plans[node_id];
if node
.controller_capabilities
.contains(&ControllerCapability::EmitsArtifacts)
&& node.artifact_policy == ArtifactPolicy::HostOnly
{
return contract_error(format!(
"portable_required training artifacts refuse host_only controller for `{node_id}`"
));
}
}
Ok(())
}
fn validate_training_data_identities(
request: &TrainingRequest,
plan: &ExecutionPlan,
) -> Result<()> {
let mut expected = BTreeMap::new();
for binding in plan.campaign.data_bindings.values().flatten() {
let key = data_binding_requirement_key(&binding.node_id, &binding.input_name);
if let Some(previous) = expected.insert(key.clone(), binding) {
let previous_coordinates = (&previous.node_id, previous.input_name.as_str());
let coordinates = (&binding.node_id, binding.input_name.as_str());
let detail = if previous_coordinates == coordinates {
"duplicate coordinates"
} else {
"distinct coordinates collide under the V1 node.input spelling"
};
return contract_error(format!(
"training data bindings render duplicate requirement key `{key}`: {detail}"
));
}
}
let mut actual = BTreeMap::new();
let mut previous: Option<&str> = None;
for identity in &request.data_identities {
identity.validate()?;
if previous.is_some_and(|previous| previous >= identity.requirement_key.as_str()) {
return contract_error(
"training data identities must be strictly sorted by requirement_key".to_string(),
);
}
previous = Some(identity.requirement_key.as_str());
actual.insert(identity.requirement_key.as_str(), identity);
}
if actual.keys().copied().collect::<BTreeSet<_>>()
!= expected.keys().map(String::as_str).collect::<BTreeSet<_>>()
{
return contract_error(
"training data identities must exactly cover campaign data bindings".to_string(),
);
}
for (key, binding) in expected {
let identity = actual[&key.as_str()];
if identity.schema_fingerprint != binding.schema_fingerprint
|| identity.plan_fingerprint != binding.plan_fingerprint
|| binding.relation_fingerprint.as_deref()
!= Some(identity.relation_fingerprint.as_str())
{
return contract_error(format!(
"training data identity `{key}` does not match data binding fingerprints"
));
}
}
Ok(())
}
fn validate_package_data_identities(package: &PortablePredictorPackage) -> Result<()> {
let expected = package
.execution_bundle
.data_requirements
.iter()
.map(|requirement| (requirement.key(), requirement))
.collect::<BTreeMap<_, _>>();
let mut actual = BTreeMap::new();
let mut previous: Option<&str> = None;
for identity in &package.data_identities {
identity.validate()?;
if previous.is_some_and(|previous| previous >= identity.requirement_key.as_str()) {
return contract_error(
"portable predictor data identities must be sorted by requirement_key".to_string(),
);
}
previous = Some(identity.requirement_key.as_str());
actual.insert(identity.requirement_key.clone(), identity);
}
if actual.keys().collect::<BTreeSet<_>>() != expected.keys().collect::<BTreeSet<_>>() {
return contract_error(
"portable predictor data identities do not exactly match bundle requirements"
.to_string(),
);
}
for (key, requirement) in expected {
let identity = actual[&key];
if identity.schema_fingerprint != requirement.schema_fingerprint
|| identity.plan_fingerprint != requirement.plan_fingerprint
|| requirement.relation_fingerprint.as_deref()
!= Some(identity.relation_fingerprint.as_str())
{
return contract_error(format!(
"portable predictor data identity `{key}` does not match bundle fingerprints"
));
}
}
Ok(())
}
fn influence_identity_closure_for_samples(
scope_id: &str,
samples: &BTreeSet<SampleId>,
relations: &SampleRelationSet,
) -> Result<(Vec<SampleId>, Vec<GroupId>)> {
let mut found = BTreeSet::new();
let mut origins = BTreeSet::new();
let mut groups = BTreeSet::new();
for relation in &relations.records {
if samples.contains(&relation.sample_id) {
found.insert(&relation.sample_id);
if let Some(origin) = &relation.origin_sample_id {
origins.insert(origin.clone());
}
if let Some(group) = &relation.group_id {
groups.insert(group.clone());
}
}
}
if found.len() != samples.len() {
return contract_error(format!(
"training influence `{scope_id}` contains physical samples absent from relation set"
));
}
Ok((origins.into_iter().collect(), groups.into_iter().collect()))
}
fn validate_package_base_influence(
influence: &TrainingInfluenceManifest,
plan: &ExecutionPlan,
closure: &BTreeSet<NodeId>,
) -> Result<()> {
let expected = closure
.iter()
.filter(|node_id| {
plan.node_plans[*node_id]
.supported_phases
.contains(&Phase::FitCv)
})
.filter_map(|node_id| base_influence_kind(plan, node_id).map(|kind| (node_id, kind)))
.collect::<BTreeMap<_, _>>();
let base_kinds = [
TrainingInfluenceKind::TransformFit,
TrainingInfluenceKind::ModelFit,
TrainingInfluenceKind::HpoSelection,
TrainingInfluenceKind::TrainedMetaAggregation,
]
.into_iter()
.collect::<BTreeSet<_>>();
let mut actual = BTreeMap::<&NodeId, Vec<TrainingInfluenceKind>>::new();
for entry in &influence.entries {
if let Some(node_id) = entry.node_id.as_ref() {
if base_kinds.contains(&entry.kind) {
actual.entry(node_id).or_default().push(entry.kind);
}
}
}
if actual.keys().copied().collect::<BTreeSet<_>>()
!= expected.keys().copied().collect::<BTreeSet<_>>()
{
return contract_error(
"portable predictor base-influence nodes do not exactly match predictor closure"
.to_string(),
);
}
for (node_id, expected_kind) in expected {
let kinds = &actual[node_id];
if kinds.is_empty() || kinds.iter().any(|kind| *kind != expected_kind) {
return contract_error(format!(
"portable predictor influence node `{node_id}` entries do not all have expected kind `{:?}`",
expected_kind
));
}
}
Ok(())
}
fn validate_package_artifact_bindings(package: &PortablePredictorPackage) -> Result<()> {
let mut previous: Option<&ArtifactId> = None;
for binding in &package.artifact_bindings {
if previous.is_some_and(|previous| previous >= &binding.artifact_id) {
return contract_error(
"portable predictor artifact bindings must be strictly sorted by artifact_id"
.to_string(),
);
}
previous = Some(&binding.artifact_id);
}
let expected = package
.execution_bundle
.refit_artifacts
.iter()
.map(|record| record.artifact.id.clone())
.collect::<BTreeSet<_>>();
let actual = package
.artifact_bindings
.iter()
.map(|binding| binding.artifact_id.clone())
.collect::<BTreeSet<_>>();
if actual != expected {
return contract_error(
"portable predictor artifact bindings do not exactly match bundle artifacts"
.to_string(),
);
}
for binding in &package.artifact_bindings {
let record = package
.execution_bundle
.refit_artifacts
.iter()
.find(|record| record.artifact.id == binding.artifact_id)
.expect("artifact id sets were checked above");
let node_plan = &package.effective_plan.node_plans[&record.node_id];
match binding.load_mode {
ArtifactLoadMode::NativePortable => {
record.artifact.validate_portable()?;
if node_plan.artifact_policy == ArtifactPolicy::HostOnly {
return contract_error(format!(
"host_only artifact `{}` cannot be classified native_portable",
binding.artifact_id
));
}
}
ArtifactLoadMode::HostSidecar => {
if package.fitted_artifact_mode != FittedArtifactMode::AllowHostSidecar {
return contract_error(format!(
"portable_required package forbids host sidecar artifact `{}`",
binding.artifact_id
));
}
}
}
}
Ok(())
}
pub(crate) fn contains_runtime_handle(value: &serde_json::Value) -> bool {
match value {
serde_json::Value::Array(values) => values.iter().any(contains_runtime_handle),
serde_json::Value::Object(values) => {
values.keys().any(|key| {
let key = key.to_ascii_lowercase();
key == "handle" || key.ends_with("_handle") || key.ends_with("_handles")
}) || values.values().any(contains_runtime_handle)
}
_ => false,
}
}
fn tcv1_fingerprint<T: Serialize + ?Sized>(value: &T, label: &str) -> Result<String> {
let json = serde_json::to_string(value)?;
parse_typed_json(&json)
.and_then(|value| value.fingerprint())
.map_err(|error| DagMlError::RuntimeValidation(format!("{label} is outside TCV1: {error}")))
}
fn tcv1_fingerprint_without<T: Serialize>(value: &T, field: &str, label: &str) -> Result<String> {
let json = serde_json::to_string(value)?;
parse_typed_json(&json)
.and_then(|value| value.fingerprint_without(field))
.map_err(|error| DagMlError::RuntimeValidation(format!("{label} is outside TCV1: {error}")))
}
fn strict_tcv1_fingerprint_without(json: &str, field: &str, label: &str) -> Result<String> {
parse_typed_json(json)
.and_then(|value| value.fingerprint_without(field))
.map_err(|error| {
DagMlError::RuntimeValidation(format!("{label} is outside strict TCV1: {error}"))
})
}
fn validate_sha256(label: &str, value: &str) -> Result<()> {
if value.len() != 64
|| !value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
return contract_error(format!(
"{label} fingerprint must be 64 lowercase hexadecimal characters"
));
}
Ok(())
}
fn zero_fingerprint() -> String {
"0".repeat(64)
}
fn validate_identifier_text(label: &str, value: &str) -> Result<()> {
if value.is_empty()
|| value.len() > 128
|| !value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':'))
{
return contract_error(format!("{label} is not a valid DAG-ML identifier"));
}
Ok(())
}
fn validate_non_empty(label: &str, value: &str) -> Result<()> {
if value.trim().is_empty() {
return contract_error(format!("{label} must be non-empty"));
}
Ok(())
}
fn validate_sorted_unique_text(
label: &str,
values: &[String],
require_non_empty: bool,
) -> Result<()> {
if require_non_empty && values.is_empty() {
return contract_error(format!("{label} must be non-empty"));
}
let mut previous: Option<&str> = None;
for value in values {
validate_non_empty(label, value)?;
if previous.is_some_and(|previous| previous >= value.as_str()) {
return contract_error(format!("{label} must be strictly sorted and unique"));
}
previous = Some(value.as_str());
}
Ok(())
}
fn validate_unique_text(label: &str, values: &[String], require_non_empty: bool) -> Result<()> {
if require_non_empty && values.is_empty() {
return contract_error(format!("{label} must be non-empty"));
}
let mut seen = BTreeSet::new();
for value in values {
validate_non_empty(label, value)?;
if !seen.insert(value.as_str()) {
return contract_error(format!("{label} must be unique"));
}
}
Ok(())
}
fn validate_sorted_unique_ids<T: Ord + std::fmt::Display>(
label: &str,
values: &[T],
require_non_empty: bool,
) -> Result<()> {
if require_non_empty && values.is_empty() {
return contract_error(format!("{label} must be non-empty"));
}
if values.windows(2).any(|pair| pair[0] >= pair[1]) {
return contract_error(format!("{label} must be strictly sorted and unique"));
}
Ok(())
}
fn unsupported_version<T>(label: &str, actual: u32, expected: u32) -> Result<T> {
contract_error(format!(
"{label} uses unsupported schema_version {actual}, expected {expected}"
))
}
fn contract_error<T>(message: String) -> Result<T> {
Err(DagMlError::CampaignValidation(message))
}
#[cfg(test)]
mod tests {
use serde_json::{json, Value};
use super::*;
use crate::ids::{ControllerId, ObservationId};
use crate::relation::SampleRelation;
use crate::selection::{MetricObjective, SelectionMetric};
fn manifests() -> Vec<ControllerManifest> {
let all: Vec<ControllerManifest> =
serde_json::from_str(include_str!("../../../examples/controller_manifests.json"))
.unwrap();
let mut selected = all
.into_iter()
.filter(|manifest| {
matches!(
manifest.operator_kind,
NodeKind::Transform | NodeKind::Model
)
})
.collect::<Vec<_>>();
selected.sort_by(|left, right| left.controller_id.cmp(&right.controller_id));
selected
}
fn data_identity(campaign: &CampaignSpec) -> TrainingDataIdentity {
let binding = &campaign.data_bindings[&NodeId::new("model:base").unwrap()][0];
let mut identity = TrainingDataIdentity {
requirement_key: "model:base.x".to_string(),
schema_fingerprint: binding.schema_fingerprint.clone(),
plan_fingerprint: binding.plan_fingerprint.clone(),
relation_fingerprint: binding.relation_fingerprint.clone().unwrap(),
data_content_fingerprint: "1".repeat(64),
target_content_fingerprint: "2".repeat(64),
identity_fingerprint: zero_fingerprint(),
};
identity.identity_fingerprint = identity.compute_fingerprint().unwrap();
identity
}
fn request() -> TrainingRequest {
let graph: GraphSpec =
serde_json::from_str(include_str!("../../../examples/minimal_graph.json")).unwrap();
let campaign: CampaignSpec = serde_json::from_str(include_str!(
"../../../examples/campaign_oof_generation.json"
))
.unwrap();
let mut request = TrainingRequest {
schema_version: TRAINING_REQUEST_SCHEMA_VERSION,
request_id: "training:request.test".to_string(),
plan_id: "plan:training.test".to_string(),
graph,
data_identities: vec![data_identity(&campaign)],
campaign,
controller_manifests: manifests(),
training_losses: Vec::new(),
parameter_patches: Vec::new(),
patch_policies: Vec::new(),
influence_requirements: Vec::new(),
options: TrainingOptions {
refit: true,
refit_strategy: Some(RefitStrategy::RefitOne),
seed: 12345,
selection: SelectionPolicy {
id: "selection:rmse".to_string(),
metric: SelectionMetric {
name: "rmse".to_string(),
objective: MetricObjective::Minimize,
},
required_metric_level: None,
require_finite: true,
evaluation_scope: None,
refit_slot_plan: None,
stacking_fit_contract: None,
reduction_id: None,
},
selection_output_id: "output:prediction".to_string(),
outputs: vec![TrainingOutputRequest {
output_id: "output:prediction".to_string(),
node_id: NodeId::new("model:base").unwrap(),
port_name: None,
prediction_level: PredictionLevel::Sample,
unit_level: Some(EntityUnitLevel::PhysicalSample),
prediction_kind: PredictionKind::RegressionPoint,
target_names: vec!["protein".to_string()],
target_units: vec![Some("percent".to_string())],
class_labels: vec![Vec::new()],
output_order: OutputOrder::TargetOrder,
target_space: "raw".to_string(),
}],
scheduler: TrainingSchedulerOptions {
kind: TrainingSchedulerKind::Sequential,
backend: None,
workers: 1,
},
resources: TrainingResourceLimits {
cpu_threads: 1,
memory_bytes: Some(1024),
gpu_devices: Vec::new(),
wall_time_ms: Some(10_000),
},
artifacts: TrainingArtifactOptions {
cv_artifacts: CvArtifactRetention::MetadataOnly,
prediction_caches: PredictionCacheRetention::Retain,
fitted_artifacts: FittedArtifactMode::AllowHostSidecar,
},
},
request_fingerprint: zero_fingerprint(),
};
request.request_fingerprint = request.compute_fingerprint().unwrap();
request
}
fn resign_request(request: &mut TrainingRequest) {
request.request_fingerprint = zero_fingerprint();
request.request_fingerprint = request.compute_fingerprint().unwrap();
}
fn custom_training_loss_role() -> TrainingLossRoleReference {
let fixture: Value = serde_json::from_str(include_str!(
"../../../examples/fixtures/criteria/criteria_contracts.v1.json"
))
.unwrap();
let mut role: TrainingLossRoleReference =
serde_json::from_value(fixture["valid"]["training_loss_role"].clone()).unwrap();
role.node_id = NodeId::new("model:base").unwrap();
role
}
#[test]
fn training_request_projects_identically_for_refit_on_and_off() {
let request = request();
let refit = request.project().unwrap();
assert_eq!(refit.outputs[0].port_name, "oof");
assert!(!refit.parameters.requires_recompile);
assert_eq!(
refit.predictor_node_ids,
BTreeSet::from([
NodeId::new("model:base").unwrap(),
NodeId::new("transform:snv").unwrap(),
])
);
let mut no_refit = request;
no_refit.options.refit = false;
no_refit.options.refit_strategy = None;
resign_request(&mut no_refit);
let projection = no_refit.project().unwrap();
assert_eq!(projection.outputs, refit.outputs);
assert_eq!(projection.plan, refit.plan);
}
#[test]
fn training_request_resolves_custom_loss_into_node_plan() {
let mut request = request();
request.training_losses = vec![custom_training_loss_role()];
let model_manifest = request
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap();
model_manifest.capabilities.extend([
ControllerCapability::NeedsPythonGil,
ControllerCapability::SupportsConfigurableLoss,
ControllerCapability::SupportsCustomLoss,
ControllerCapability::SupportsDifferentiableLoss,
]);
resign_request(&mut request);
let projection = request.project().unwrap();
let node_plan = &projection.plan.node_plans[&NodeId::new("model:base").unwrap()];
assert_eq!(node_plan.training_losses, request.training_losses);
assert!(node_plan
.training_loss_fingerprint(Phase::FitCv)
.unwrap()
.is_some());
assert!(node_plan
.training_loss_fingerprint(Phase::Refit)
.unwrap()
.is_some());
}
#[test]
fn training_request_rejects_loss_without_controller_capability() {
let mut request = request();
request.training_losses = vec![custom_training_loss_role()];
resign_request(&mut request);
let error = request.project().unwrap_err().to_string();
assert!(
error.contains("configurable loss"),
"unexpected error: {error}"
);
}
#[test]
fn training_request_rejects_duplicate_rendered_data_requirement_keys() {
let mut request = request();
let node_id = NodeId::new("model:base").unwrap();
let duplicate = request.campaign.data_bindings[&node_id][0].clone();
request
.campaign
.data_bindings
.get_mut(&node_id)
.unwrap()
.push(duplicate);
resign_request(&mut request);
let error = request.project().unwrap_err();
assert!(error
.to_string()
.contains("render duplicate requirement key `model:base.x`"));
}
#[test]
fn training_options_reject_unknown_fields_and_binary64_integer_substitution() {
let value = serde_json::to_value(request()).unwrap();
let mut unknown = value.clone();
unknown["options"]["mystery"] = json!(true);
let error = serde_json::from_value::<TrainingRequest>(unknown).unwrap_err();
assert!(error.to_string().contains("unknown field"));
let mut binary64_seed = value;
binary64_seed["options"]["seed"] = json!(12345.0);
assert!(serde_json::from_value::<TrainingRequest>(binary64_seed).is_err());
}
#[test]
fn training_request_from_json_requires_explicit_patch_collections() {
let request_json = serde_json::to_value(request()).unwrap();
for field in ["parameter_patches", "patch_policies"] {
let mut missing = request_json.clone();
missing.as_object_mut().unwrap().remove(field);
let error = TrainingRequest::from_json(&serde_json::to_string(&missing).unwrap())
.expect_err("required patch collection omission must fail closed");
assert!(error.to_string().contains(field), "{field}: {error}");
}
}
#[test]
fn required_nullable_fields_reject_omission_but_accept_explicit_null() {
let base = request();
let output = serde_json::to_value(&base.options.outputs[0]).unwrap();
assert!(output.get("unit_level").is_some());
let mut missing = output.clone();
missing.as_object_mut().unwrap().remove("unit_level");
let error = serde_json::from_value::<TrainingOutputRequest>(missing).unwrap_err();
assert!(
error.to_string().contains("missing field") && error.to_string().contains("unit_level"),
"unit_level omission: {error}"
);
let mut null_unit = output;
null_unit["unit_level"] = json!(null);
assert!(serde_json::from_value::<TrainingOutputRequest>(null_unit)
.unwrap()
.unit_level
.is_none());
let resources = serde_json::to_value(&base.options.resources).unwrap();
assert_eq!(resources["gpu_devices"], json!([]));
let mut missing = resources.clone();
missing.as_object_mut().unwrap().remove("gpu_devices");
let error = serde_json::from_value::<TrainingResourceLimits>(missing).unwrap_err();
assert!(
error.to_string().contains("missing field")
&& error.to_string().contains("gpu_devices"),
"gpu_devices omission: {error}"
);
assert!(serde_json::from_value::<TrainingResourceLimits>(resources)
.unwrap()
.gpu_devices
.is_empty());
let options = serde_json::to_value(&base.options).unwrap();
assert!(options.get("refit_strategy").is_some());
let mut missing = options.clone();
missing.as_object_mut().unwrap().remove("refit_strategy");
let error = serde_json::from_value::<TrainingOptions>(missing).unwrap_err();
assert!(
error.to_string().contains("missing field")
&& error.to_string().contains("refit_strategy"),
"refit_strategy omission: {error}"
);
let mut null_strategy = options;
null_strategy["refit_strategy"] = json!(null);
assert!(serde_json::from_value::<TrainingOptions>(null_strategy)
.unwrap()
.refit_strategy
.is_none());
let requirement = serde_json::to_value(ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind: TrainingInfluenceKind::EarlyStopping,
scope_id: "early:fold:0".to_string(),
phase: Phase::FitCv,
fold_id: Some(FoldId::new("fold:0").unwrap()),
physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
})
.unwrap();
assert!(requirement.get("fold_id").is_some());
let mut missing = requirement.clone();
missing.as_object_mut().unwrap().remove("fold_id");
let error = serde_json::from_value::<ControllerInfluenceRequirement>(missing).unwrap_err();
assert!(
error.to_string().contains("missing field") && error.to_string().contains("fold_id"),
"fold_id omission: {error}"
);
let mut null_fold = requirement;
null_fold["fold_id"] = json!(null);
assert!(
serde_json::from_value::<ControllerInfluenceRequirement>(null_fold)
.unwrap()
.fold_id
.is_none()
);
let entry = serde_json::to_value(TrainingInfluenceEntry {
kind: TrainingInfluenceKind::ModelFit,
scope_id: "model:base:fold:0".to_string(),
node_id: Some(NodeId::new("model:base").unwrap()),
physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
origin_sample_ids: Vec::new(),
group_ids: Vec::new(),
})
.unwrap();
assert!(entry.get("node_id").is_some());
let mut missing = entry.clone();
missing.as_object_mut().unwrap().remove("node_id");
let error = serde_json::from_value::<TrainingInfluenceEntry>(missing).unwrap_err();
assert!(
error.to_string().contains("missing field") && error.to_string().contains("node_id"),
"node_id omission: {error}"
);
let mut null_node = entry;
null_node["node_id"] = json!(null);
assert!(serde_json::from_value::<TrainingInfluenceEntry>(null_node)
.unwrap()
.node_id
.is_none());
}
#[test]
fn output_resolution_rejects_no_output_ambiguity_and_non_prediction_ports() {
let request = request();
let output = &request.options.outputs[0];
let mut no_output = request.graph.clone();
no_output.nodes[1].ports.outputs.clear();
assert!(output
.resolve(&no_output)
.unwrap_err()
.to_string()
.contains("no prediction output"));
let mut ambiguous = request.graph.clone();
let mut second = ambiguous.nodes[1].ports.outputs[0].clone();
second.name = "probability".to_string();
ambiguous.nodes[1].ports.outputs.push(second);
assert!(output
.resolve(&ambiguous)
.unwrap_err()
.to_string()
.contains("multiple prediction outputs"));
let mut explicit = output.clone();
explicit.port_name = Some("x".to_string());
assert!(explicit.resolve(&request.graph).is_err());
}
#[test]
fn output_target_and_class_orders_are_semantic_not_lexically_sorted() {
let graph = request().graph;
let mut binding = OutputBinding {
schema_version: OUTPUT_BINDING_SCHEMA_VERSION,
binding_id: "output:ordered".to_string(),
node_id: NodeId::new("model:base").unwrap(),
port_name: "oof".to_string(),
prediction_level: PredictionLevel::Sample,
unit_level: Some(EntityUnitLevel::PhysicalSample),
prediction_kind: PredictionKind::RegressionPoint,
prediction_source: PredictionSource::FinalRefit,
refit_strategy: Some(RefitStrategy::RefitOne),
aggregation_fingerprint: "6".repeat(64),
target_names: vec!["z_target".to_string(), "a_target".to_string()],
target_units: vec![Some("z_unit".to_string()), Some("a_unit".to_string())],
class_labels: vec![Vec::new(), Vec::new()],
output_order: OutputOrder::TargetOrder,
target_space: "raw".to_string(),
binding_fingerprint: zero_fingerprint(),
};
binding.binding_fingerprint = binding.compute_fingerprint().unwrap();
binding.validate(&graph).unwrap();
let original = binding.binding_fingerprint.clone();
binding.target_names.swap(0, 1);
binding.binding_fingerprint = zero_fingerprint();
binding.binding_fingerprint = binding.compute_fingerprint().unwrap();
binding.validate(&graph).unwrap();
assert_ne!(original, binding.binding_fingerprint);
}
#[test]
fn output_unit_levels_and_class_vocabularies_match_w0_contract() {
let request = request();
let graph = &request.graph;
let mut output = request.options.outputs[0].clone();
output.unit_level = None;
assert!(output
.resolve(graph)
.unwrap_err()
.to_string()
.contains("physical_sample"));
output.prediction_level = PredictionLevel::Target;
output.unit_level = Some(EntityUnitLevel::PhysicalSample);
assert!(output
.resolve(graph)
.unwrap_err()
.to_string()
.contains("unit_level=null"));
output.unit_level = None;
let target_wire = serde_json::to_value(&output).unwrap();
assert_eq!(target_wire.get("unit_level"), Some(&Value::Null));
output.resolve(graph).unwrap();
output.prediction_level = PredictionLevel::Sample;
output.unit_level = Some(EntityUnitLevel::PhysicalSample);
output.prediction_kind = PredictionKind::ClassLabel;
output.class_labels = vec![Vec::new()];
output.output_order = OutputOrder::TargetOrder;
output.resolve(graph).unwrap();
output.class_labels = vec![vec!["low".to_string(), "high".to_string()]];
output.resolve(graph).unwrap();
output.prediction_kind = PredictionKind::DecisionScore;
output.resolve(graph).unwrap();
output.class_labels = vec![Vec::new()];
output.resolve(graph).unwrap();
output.prediction_kind = PredictionKind::RegressionPoint;
output.class_labels = vec![vec!["not-a-regression-class".to_string()]];
assert!(output.resolve(graph).is_err());
output.prediction_kind = PredictionKind::ClassProbability;
output.output_order = OutputOrder::TargetMajorClassMinor;
output.class_labels = vec![Vec::new()];
assert!(output.resolve(graph).is_err());
output.class_labels = vec![vec!["low".to_string(), "high".to_string()]];
output.resolve(graph).unwrap();
}
#[test]
fn selection_output_metric_matrix_is_explicit() {
let mut level_mismatch = request();
level_mismatch.options.outputs[0].prediction_level = PredictionLevel::Target;
level_mismatch.options.outputs[0].unit_level = None;
assert!(level_mismatch
.options
.selection
.required_metric_level
.is_none());
resign_request(&mut level_mismatch);
assert!(level_mismatch
.validate()
.unwrap_err()
.to_string()
.contains("campaign selection_metric_level"));
let mut regression = request();
regression.options.selection.metric.objective = MetricObjective::Maximize;
resign_request(&mut regression);
assert!(regression
.validate()
.unwrap_err()
.to_string()
.contains("not supported for RegressionPoint"));
let mut class_label = request();
class_label.options.outputs[0].prediction_kind = PredictionKind::ClassLabel;
class_label.options.outputs[0].class_labels =
vec![vec!["low".to_string(), "high".to_string()]];
class_label.options.selection.metric.name = "accuracy".to_string();
class_label.options.selection.metric.objective = MetricObjective::Maximize;
resign_request(&mut class_label);
class_label.validate().unwrap();
let mut probability = class_label.clone();
probability.options.outputs[0].prediction_kind = PredictionKind::ClassProbability;
probability.options.outputs[0].output_order = OutputOrder::TargetMajorClassMinor;
resign_request(&mut probability);
assert!(probability
.validate()
.unwrap_err()
.to_string()
.contains("not supported for ClassProbability"));
let mut decision = class_label;
decision.options.outputs[0].prediction_kind = PredictionKind::DecisionScore;
resign_request(&mut decision);
assert!(decision
.validate()
.unwrap_err()
.to_string()
.contains("not supported for DecisionScore"));
}
fn request_with_nested_params() -> TrainingRequest {
let mut request = request();
let model = request
.graph
.nodes
.iter_mut()
.find(|node| node.id.as_str() == "model:base")
.unwrap();
model.params.insert(
"nested".to_string(),
json!({"depth": {"alpha": 1}, "array": [1, 2]}),
);
resign_request(&mut request);
request
}
fn patch(namespace: ParameterNamespace, path: &[&str], value: Value) -> ParameterPatch {
ParameterPatch {
schema_version: PARAMETER_PATCH_SCHEMA_VERSION,
node_id: NodeId::new("model:base").unwrap(),
namespace,
path: path.iter().map(|part| (*part).to_string()).collect(),
value,
}
}
fn patch_policy(namespaces: &[ParameterNamespace]) -> NodePatchPolicy {
NodePatchPolicy {
node_id: NodeId::new("model:base").unwrap(),
allowed_namespaces: namespaces.iter().copied().collect(),
}
}
#[test]
fn namespaced_deep_patch_is_isolated_bijective_and_structural() {
let plan = request_with_nested_params().project().unwrap().plan;
let original = plan.node_plans[&NodeId::new("model:base").unwrap()]
.params
.clone();
let patches = vec![
patch(
ParameterNamespace::Operator,
&["nested", "depth", "alpha"],
json!(2),
),
patch(ParameterNamespace::Fit, &["epochs"], json!(12)),
patch(ParameterNamespace::Structural, &["topology"], json!("wide")),
];
let projection = project_parameter_patches(
&plan,
&patches,
&[patch_policy(&[
ParameterNamespace::Operator,
ParameterNamespace::Fit,
ParameterNamespace::Structural,
])],
)
.unwrap();
let node = &projection.nodes[&NodeId::new("model:base").unwrap()];
assert_eq!(node.params["nested"]["depth"]["alpha"], json!(2));
assert_eq!(node.fit_params["epochs"], json!(12));
assert_eq!(node.structural_params["topology"], json!("wide"));
assert!(projection.requires_recompile);
assert_eq!(
plan.node_plans[&NodeId::new("model:base").unwrap()].params,
original
);
assert_eq!(ParameterNamespace::Operator.plan_root(), "params");
assert_eq!(ParameterNamespace::Fit.plan_root(), "fit_params");
assert_eq!(ParameterNamespace::Control.plan_root(), "control_params");
assert_eq!(
ParameterNamespace::Structural.plan_root(),
"structural_params"
);
let mut dishonest = projection.clone();
dishonest.requires_recompile = false;
dishonest.projection_fingerprint = zero_fingerprint();
dishonest.projection_fingerprint = dishonest.compute_fingerprint().unwrap();
assert!(dishonest.validate().is_err());
}
#[test]
fn patch_projection_rejects_namespace_order_duplicates_parent_child_and_arrays() {
let plan = request_with_nested_params().project().unwrap().plan;
let operator = patch_policy(&[ParameterNamespace::Operator]);
let forbidden = patch(ParameterNamespace::Fit, &["epochs"], json!(3));
assert!(
project_parameter_patches(&plan, &[forbidden], std::slice::from_ref(&operator))
.unwrap_err()
.to_string()
.contains("forbidden")
);
let duplicate = patch(ParameterNamespace::Operator, &["x"], json!(1));
assert!(project_parameter_patches(
&plan,
&[duplicate.clone(), duplicate],
std::slice::from_ref(&operator)
)
.is_err());
let parent = patch(
ParameterNamespace::Operator,
&["nested", "depth"],
json!({"alpha": 2}),
);
let child = patch(
ParameterNamespace::Operator,
&["nested", "depth", "alpha"],
json!(3),
);
assert!(project_parameter_patches(
&plan,
&[parent, child],
std::slice::from_ref(&operator),
)
.unwrap_err()
.to_string()
.contains("parent/child"));
let array = patch(
ParameterNamespace::Operator,
&["nested", "array", "0"],
json!(9),
);
assert!(
project_parameter_patches(&plan, &[array], std::slice::from_ref(&operator)).is_err()
);
let out_of_order = vec![
patch(ParameterNamespace::Operator, &["z"], json!(1)),
patch(ParameterNamespace::Operator, &["a"], json!(1)),
];
assert!(project_parameter_patches(&plan, &out_of_order, &[operator]).is_err());
assert!(project_parameter_patches(
&plan,
&[],
&[patch_policy(&[ParameterNamespace::Operator])]
)
.is_err());
}
#[test]
fn patch_tcv1_distinguishes_integer_and_binary64() {
let integer = vec![patch(ParameterNamespace::Operator, &["x"], json!(2))];
let binary64 = vec![patch(ParameterNamespace::Operator, &["x"], json!(2.0))];
assert_ne!(
tcv1_fingerprint(&integer, "integer patch").unwrap(),
tcv1_fingerprint(&binary64, "binary64 patch").unwrap()
);
}
fn cache_namespace() -> CacheNamespace {
let mut namespace = CacheNamespace {
schema_version: CACHE_NAMESPACE_SCHEMA_VERSION,
prediction_requirement_key: bundle_prediction_requirement_key(
&NodeId::new("model:base").unwrap(),
"oof",
&NodeId::new("model:meta").unwrap(),
"stacked",
),
data_requirement_key: "model:base.x".to_string(),
producer_node_id: NodeId::new("model:base").unwrap(),
source_port_name: "oof".to_string(),
consumer_node_id: NodeId::new("model:meta").unwrap(),
target_port_name: "stacked".to_string(),
phase: Phase::FitCv,
params_fingerprint: "a".repeat(64),
training_loss_fingerprint: None,
data_identity_fingerprint: "b".repeat(64),
fold_id: FoldId::new("fold:0").unwrap(),
trial_id: "trial:0".to_string(),
seed: 7,
namespace_fingerprint: zero_fingerprint(),
};
namespace.namespace_fingerprint = namespace.compute_fingerprint().unwrap();
namespace
}
#[test]
fn cache_namespace_is_candidate_dataset_fold_trial_and_seed_specific() {
let identity = request().data_identities.remove(0);
let mut base = cache_namespace();
base.data_identity_fingerprint = identity.identity_fingerprint.clone();
base.namespace_fingerprint = zero_fingerprint();
base.namespace_fingerprint = base.compute_fingerprint().unwrap();
base.validate_for_identity(&identity).unwrap();
for mutation in 0..6 {
let mut changed = base.clone();
match mutation {
0 => changed.params_fingerprint = "d".repeat(64),
1 => changed.training_loss_fingerprint = Some("e".repeat(64)),
2 => changed.data_identity_fingerprint = "f".repeat(64),
3 => changed.fold_id = FoldId::new("fold:1").unwrap(),
4 => changed.trial_id = "trial:1".to_string(),
_ => changed.seed += 1,
}
changed.namespace_fingerprint = zero_fingerprint();
changed.namespace_fingerprint = changed.compute_fingerprint().unwrap();
changed.validate().unwrap();
assert_ne!(base.namespace_fingerprint, changed.namespace_fingerprint);
}
let mut value = serde_json::to_value(&base).unwrap();
value["seed"] = json!(7.0);
assert!(serde_json::from_value::<CacheNamespace>(value).is_err());
let mut relation_changed = identity.clone();
relation_changed.relation_fingerprint = "d".repeat(64);
relation_changed.identity_fingerprint = zero_fingerprint();
relation_changed.identity_fingerprint = relation_changed.compute_fingerprint().unwrap();
assert!(base.validate_for_identity(&relation_changed).is_err());
let mut other_dataset_namespace = base.clone();
other_dataset_namespace.data_identity_fingerprint =
relation_changed.identity_fingerprint.clone();
other_dataset_namespace.namespace_fingerprint = zero_fingerprint();
other_dataset_namespace.namespace_fingerprint =
other_dataset_namespace.compute_fingerprint().unwrap();
assert_ne!(
base.namespace_fingerprint,
other_dataset_namespace.namespace_fingerprint
);
let mut other_output = base.clone();
other_output.source_port_name = "probability".to_string();
other_output.prediction_requirement_key = bundle_prediction_requirement_key(
&other_output.producer_node_id,
&other_output.source_port_name,
&other_output.consumer_node_id,
&other_output.target_port_name,
);
other_output.namespace_fingerprint = zero_fingerprint();
other_output.namespace_fingerprint = other_output.compute_fingerprint().unwrap();
other_output.validate().unwrap();
assert_ne!(
base.namespace_fingerprint,
other_output.namespace_fingerprint
);
let mut wrong_phase = base.clone();
wrong_phase.phase = Phase::Refit;
wrong_phase.namespace_fingerprint = zero_fingerprint();
wrong_phase.namespace_fingerprint = wrong_phase.compute_fingerprint().unwrap();
assert!(wrong_phase.validate().is_err());
}
fn relations() -> SampleRelationSet {
let records = (1..=4)
.map(|index| {
let mut relation = SampleRelation::new(
ObservationId::new(format!("observation:{index}")).unwrap(),
SampleId::new(format!("sample:{index}")).unwrap(),
);
relation.group_id =
Some(GroupId::new(if index <= 2 { "group:0" } else { "group:1" }).unwrap());
relation
})
.collect();
SampleRelationSet { records }
}
fn request_for_relations(relations: &SampleRelationSet) -> TrainingRequest {
let mut request = request();
let fingerprint = relations.fingerprint().unwrap();
request
.campaign
.data_bindings
.get_mut(&NodeId::new("model:base").unwrap())
.unwrap()[0]
.relation_fingerprint = Some(fingerprint.clone());
request.data_identities[0].relation_fingerprint = fingerprint;
request.data_identities[0].identity_fingerprint = zero_fingerprint();
request.data_identities[0].identity_fingerprint =
request.data_identities[0].compute_fingerprint().unwrap();
resign_request(&mut request);
request
}
fn influence_manifest(
request: &TrainingRequest,
projection: &TrainingContractProjection,
relations: &SampleRelationSet,
) -> TrainingInfluenceManifest {
let expected = expected_influence_coordinates(
request,
&projection.plan,
&projection.predictor_node_ids,
)
.unwrap();
let entries = expected
.into_iter()
.map(|((kind, scope_id, node_id), samples)| {
let groups = relations
.records
.iter()
.filter(|relation| samples.contains(&relation.sample_id))
.filter_map(|relation| relation.group_id.clone())
.collect::<BTreeSet<_>>()
.into_iter()
.collect();
TrainingInfluenceEntry {
kind,
scope_id,
node_id,
physical_sample_ids: samples.into_iter().collect(),
origin_sample_ids: Vec::new(),
group_ids: groups,
}
})
.collect();
let mut manifest = TrainingInfluenceManifest {
schema_version: TRAINING_INFLUENCE_MANIFEST_SCHEMA_VERSION,
relation_fingerprint: relations.fingerprint().unwrap(),
entries,
manifest_fingerprint: zero_fingerprint(),
};
manifest.manifest_fingerprint = manifest.compute_fingerprint().unwrap();
manifest
}
fn resign_manifest(manifest: &mut TrainingInfluenceManifest) {
manifest.manifest_fingerprint = zero_fingerprint();
manifest.manifest_fingerprint = manifest.compute_fingerprint().unwrap();
}
fn early_stopping_requirements() -> Vec<ControllerInfluenceRequirement> {
vec![
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind: TrainingInfluenceKind::EarlyStopping,
scope_id: "early:fold:0".to_string(),
phase: Phase::FitCv,
fold_id: Some(FoldId::new("fold:0").unwrap()),
physical_sample_ids: vec![SampleId::new("sample:3").unwrap()],
},
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind: TrainingInfluenceKind::EarlyStopping,
scope_id: "early:fold:1".to_string(),
phase: Phase::FitCv,
fold_id: Some(FoldId::new("fold:1").unwrap()),
physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
},
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind: TrainingInfluenceKind::EarlyStopping,
scope_id: "early:refit".to_string(),
phase: Phase::Refit,
fold_id: None,
physical_sample_ids: vec![SampleId::new("sample:1").unwrap()],
},
]
}
fn full_scope_requirements(
kind: TrainingInfluenceKind,
prefix: &str,
) -> Vec<ControllerInfluenceRequirement> {
vec![
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind,
scope_id: format!("{prefix}:fold:0"),
phase: Phase::FitCv,
fold_id: Some(FoldId::new("fold:0").unwrap()),
physical_sample_ids: vec![
SampleId::new("sample:3").unwrap(),
SampleId::new("sample:4").unwrap(),
],
},
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind,
scope_id: format!("{prefix}:fold:1"),
phase: Phase::FitCv,
fold_id: Some(FoldId::new("fold:1").unwrap()),
physical_sample_ids: vec![
SampleId::new("sample:1").unwrap(),
SampleId::new("sample:2").unwrap(),
],
},
ControllerInfluenceRequirement {
node_id: NodeId::new("model:base").unwrap(),
kind,
scope_id: format!("{prefix}:refit"),
phase: Phase::Refit,
fold_id: None,
physical_sample_ids: (1..=4)
.map(|index| SampleId::new(format!("sample:{index}")).unwrap())
.collect(),
},
]
}
#[test]
fn influence_evidence_is_capability_complete_and_relation_closed() {
let relations = relations();
let request = request_for_relations(&relations);
let projection = request.project().unwrap();
let manifest = influence_manifest(&request, &projection, &relations);
assert_eq!(manifest.entries.len(), 7);
manifest
.validate_for_projection(&projection, &request, &relations)
.unwrap();
let mut missing = manifest.clone();
missing.entries.remove(1);
resign_manifest(&mut missing);
assert!(missing
.validate_for_projection(&projection, &request, &relations)
.unwrap_err()
.to_string()
.contains("phase scopes"));
let mut wrong_group = manifest;
wrong_group.entries[0].group_ids.pop();
resign_manifest(&mut wrong_group);
assert!(wrong_group
.validate_for_projection(&projection, &request, &relations)
.unwrap_err()
.to_string()
.contains("group closure"));
}
#[test]
fn influence_capabilities_require_every_fold_and_refit_scope_and_refuse_extra() {
let relations = relations();
let mut request = request_for_relations(&relations);
let model_manifest = request
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap();
model_manifest
.capabilities
.insert(ControllerCapability::UsesEarlyStopping);
request.influence_requirements = early_stopping_requirements();
resign_request(&mut request);
let projection = request.project().unwrap();
let mut manifest = influence_manifest(&request, &projection, &relations);
assert_eq!(
manifest
.entries
.iter()
.filter(|entry| entry.kind == TrainingInfluenceKind::EarlyStopping)
.count(),
3
);
let removed = manifest
.entries
.iter()
.position(|entry| entry.kind == TrainingInfluenceKind::EarlyStopping)
.unwrap();
manifest.entries.remove(removed);
resign_manifest(&mut manifest);
assert!(manifest
.validate_for_projection(&projection, &request, &relations)
.unwrap_err()
.to_string()
.contains("phase scopes"));
let mut leaked_request = request.clone();
leaked_request.influence_requirements[0].physical_sample_ids =
vec![SampleId::new("sample:1").unwrap()];
resign_request(&mut leaked_request);
assert!(leaked_request
.project()
.unwrap_err()
.to_string()
.contains("outer validation"));
let base_request = request_for_relations(&relations);
let base_projection = base_request.project().unwrap();
let mut extra = influence_manifest(&base_request, &base_projection, &relations);
let model_entry = extra
.entries
.iter()
.find(|entry| {
entry.kind == TrainingInfluenceKind::ModelFit
&& entry.scope_id.starts_with("fit_cv:")
})
.unwrap()
.clone();
extra.entries.push(TrainingInfluenceEntry {
kind: TrainingInfluenceKind::EarlyStopping,
..model_entry
});
extra.entries.sort_by(|left, right| {
(left.kind, left.scope_id.as_str(), left.node_id.as_ref()).cmp(&(
right.kind,
right.scope_id.as_str(),
right.node_id.as_ref(),
))
});
resign_manifest(&mut extra);
assert!(extra
.validate_for_projection(&base_projection, &base_request, &relations)
.unwrap_err()
.to_string()
.contains("undeclared coordinate"));
}
#[test]
fn influence_requirement_cannot_claim_a_capability_the_controller_lacks() {
let mut request = request();
request.influence_requirements = early_stopping_requirements();
resign_request(&mut request);
assert!(request
.project()
.unwrap_err()
.to_string()
.contains("not required by active controller capabilities"));
}
#[test]
fn influence_capability_matrix_covers_weights_internal_tuning_and_trained_aggregation() {
let relations = relations();
for (capability, kind, prefix) in [
(
ControllerCapability::UsesTrainingWeights,
TrainingInfluenceKind::WeightingResampling,
"weighting",
),
(
ControllerCapability::PerformsInternalTuning,
TrainingInfluenceKind::HpoSelection,
"internal_hpo",
),
] {
let mut request = request_for_relations(&relations);
let model = request
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap();
model.capabilities.insert(capability);
if capability == ControllerCapability::UsesTrainingWeights {
model
.capabilities
.insert(ControllerCapability::SupportsSampleWeights);
}
request.influence_requirements = full_scope_requirements(kind, prefix);
resign_request(&mut request);
let projection = request.project().unwrap();
let mut manifest = influence_manifest(&request, &projection, &relations);
assert_eq!(
manifest
.entries
.iter()
.filter(|entry| entry.kind == kind && entry.node_id.is_some())
.count(),
3
);
manifest
.validate_for_projection(&projection, &request, &relations)
.unwrap();
let removed = manifest
.entries
.iter()
.position(|entry| entry.kind == kind && entry.node_id.is_some())
.unwrap();
manifest.entries.remove(removed);
resign_manifest(&mut manifest);
assert!(manifest
.validate_for_projection(&projection, &request, &relations)
.unwrap_err()
.to_string()
.contains("phase scopes"));
}
let mut aggregation = request_for_relations(&relations);
let model = aggregation
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap();
model
.capabilities
.insert(ControllerCapability::TrainsAggregation);
resign_request(&mut aggregation);
assert!(aggregation.validate().is_err());
let model = aggregation
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap();
model
.capabilities
.insert(ControllerCapability::AggregatesPredictions);
resign_request(&mut aggregation);
let projection = aggregation.project().unwrap();
let mut manifest = influence_manifest(&aggregation, &projection, &relations);
assert!(manifest.entries.iter().any(|entry| {
entry
.node_id
.as_ref()
.is_some_and(|node_id| node_id.as_str() == "model:base")
&& entry.kind == TrainingInfluenceKind::TrainedMetaAggregation
}));
assert!(!manifest.entries.iter().any(|entry| {
entry
.node_id
.as_ref()
.is_some_and(|node_id| node_id.as_str() == "model:base")
&& entry.kind == TrainingInfluenceKind::ModelFit
}));
manifest
.validate_for_projection(&projection, &aggregation, &relations)
.unwrap();
let removed = manifest
.entries
.iter()
.position(|entry| {
entry
.node_id
.as_ref()
.is_some_and(|node_id| node_id.as_str() == "model:base")
&& entry.kind == TrainingInfluenceKind::TrainedMetaAggregation
})
.unwrap();
manifest.entries.remove(removed);
resign_manifest(&mut manifest);
assert!(manifest
.validate_for_projection(&projection, &aggregation, &relations)
.is_err());
}
#[test]
fn parallel_scheduler_is_bound_to_thread_or_process_capabilities() {
let mut threaded = request();
threaded.options.scheduler = TrainingSchedulerOptions {
kind: TrainingSchedulerKind::Parallel,
backend: Some(TrainingSchedulerBackend::Threads),
workers: 2,
};
threaded.options.resources.cpu_threads = 2;
resign_request(&mut threaded);
threaded.validate().unwrap();
let mut unsafe_threads = threaded.clone();
unsafe_threads
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap()
.capabilities
.remove(&ControllerCapability::ThreadSafe);
resign_request(&mut unsafe_threads);
assert!(unsafe_threads
.validate()
.unwrap_err()
.to_string()
.contains("thread_safe"));
let mut gil_threads = threaded.clone();
gil_threads
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap()
.capabilities
.insert(ControllerCapability::NeedsPythonGil);
resign_request(&mut gil_threads);
assert!(gil_threads
.validate()
.unwrap_err()
.to_string()
.contains("needs_python_gil"));
gil_threads.options.scheduler.backend = Some(TrainingSchedulerBackend::Processes);
resign_request(&mut gil_threads);
gil_threads.validate().unwrap();
}
#[test]
fn portable_required_artifact_mode_rejects_host_only_controller() {
let mut request = request();
request.options.artifacts.fitted_artifacts = FittedArtifactMode::PortableRequired;
request
.controller_manifests
.iter_mut()
.find(|manifest| manifest.operator_kind == NodeKind::Model)
.unwrap()
.artifact_policy = ArtifactPolicy::HostOnly;
resign_request(&mut request);
assert!(request
.validate()
.unwrap_err()
.to_string()
.contains("host_only"));
request.options.artifacts.fitted_artifacts = FittedArtifactMode::AllowHostSidecar;
resign_request(&mut request);
request.validate().unwrap();
}
fn package() -> PortablePredictorPackage {
let outcome: Value = serde_json::from_str(include_str!(
"../../../examples/fixtures/estimator/training_outcome_refit.v1.json"
))
.unwrap();
let effective_plan: ExecutionPlan =
serde_json::from_value(outcome["effective_plan"].clone()).unwrap();
let execution_bundle: ExecutionBundle =
serde_json::from_value(outcome["execution_bundle"].clone()).unwrap();
let output_bindings = outcome["outputs"]
.as_array()
.unwrap()
.iter()
.map(|output| serde_json::from_value(output["binding"].clone()).unwrap())
.collect::<Vec<OutputBinding>>();
let training_influence: TrainingInfluenceManifest =
serde_json::from_value(outcome["training_influence"].clone()).unwrap();
let mut template = PredictorTemplate {
graph: effective_plan.graph_plan.graph.clone(),
campaign: effective_plan.campaign.clone(),
controller_manifests: effective_plan.controller_manifests.clone(),
template_fingerprint: zero_fingerprint(),
};
template.template_fingerprint = template.compute_fingerprint().unwrap();
let mut data_identities = execution_bundle
.data_requirements
.iter()
.map(|requirement| {
let mut identity = TrainingDataIdentity {
requirement_key: requirement.key(),
schema_fingerprint: requirement.schema_fingerprint.clone(),
plan_fingerprint: requirement.plan_fingerprint.clone(),
relation_fingerprint: requirement.relation_fingerprint.clone().unwrap(),
data_content_fingerprint: "3".repeat(64),
target_content_fingerprint: "4".repeat(64),
identity_fingerprint: zero_fingerprint(),
};
identity.identity_fingerprint = identity.compute_fingerprint().unwrap();
identity
})
.collect::<Vec<_>>();
data_identities.sort_by(|left, right| left.requirement_key.cmp(&right.requirement_key));
let closure = predictor_closure(
&effective_plan,
output_bindings.iter().map(|binding| &binding.node_id),
)
.unwrap();
let mut artifact_bindings = execution_bundle
.refit_artifacts
.iter()
.map(|record| PackageArtifactBinding {
artifact_id: record.artifact.id.clone(),
load_mode: ArtifactLoadMode::HostSidecar,
})
.collect::<Vec<_>>();
artifact_bindings.sort_by(|left, right| left.artifact_id.cmp(&right.artifact_id));
let output_binding_fingerprints = output_bindings
.iter()
.map(|binding| binding.binding_fingerprint.clone())
.collect::<Vec<_>>();
let execution_bundle_fingerprint =
tcv1_fingerprint(&execution_bundle, "test execution bundle").unwrap();
let data_identities_fingerprint =
tcv1_fingerprint(&data_identities, "test data identities").unwrap();
let mut package = PortablePredictorPackage {
schema_version: PORTABLE_PREDICTOR_PACKAGE_SCHEMA_VERSION,
package_id: "predictor:package.test".to_string(),
template,
training_request_fingerprint: "5".repeat(64),
training_outcome: TrainingOutcomeRef {
outcome_id: outcome["outcome_id"].as_str().unwrap().to_string(),
outcome_fingerprint: outcome["outcome_fingerprint"].as_str().unwrap().to_string(),
training_request_fingerprint: "5".repeat(64),
effective_plan_fingerprint: outcome["effective_plan_fingerprint"]
.as_str()
.unwrap()
.to_string(),
execution_bundle_id: execution_bundle.bundle_id.clone(),
execution_bundle_fingerprint,
output_binding_fingerprints,
training_influence_fingerprint: training_influence.manifest_fingerprint.clone(),
data_identities_fingerprint,
},
effective_plan,
execution_bundle,
output_bindings,
predictor_node_ids: closure.into_iter().collect(),
training_influence,
data_identities,
fitted_artifact_mode: FittedArtifactMode::AllowHostSidecar,
artifact_bindings,
package_fingerprint: zero_fingerprint(),
};
package.package_fingerprint = package.compute_fingerprint().unwrap();
package
}
#[test]
fn portable_package_round_trips_loads_sidecar_and_rejects_tamper_and_future() {
let package = package();
package.validate().unwrap();
let json = serde_json::to_string(&package).unwrap();
PortablePredictorPackage::from_json(&json).unwrap();
let loaded = package
.clone()
.load_with(|record| Ok(format!("handle:{}", record.artifact.id)))
.unwrap();
let first = &package.artifact_bindings[0].artifact_id;
assert_eq!(loaded.artifact(first).unwrap(), &format!("handle:{first}"));
let mut missing_handles = BTreeMap::new();
missing_handles.insert(first.clone(), "only-one".to_string());
assert!(LoadedPredictor::new(package.clone(), missing_handles).is_err());
let mut tampered = package.clone();
tampered.output_bindings[0].target_space = "tampered".to_string();
assert!(tampered.validate().is_err());
let mut future = package.clone();
future.schema_version += 1;
future.package_fingerprint = zero_fingerprint();
future.package_fingerprint = future.compute_fingerprint().unwrap();
assert!(future.validate().is_err());
let mut binary64 = serde_json::to_value(package).unwrap();
binary64["effective_plan"]["campaign"]["root_seed"] = json!(12345.0);
assert!(serde_json::from_value::<PortablePredictorPackage>(binary64).is_err());
}
#[test]
fn portable_package_strict_parser_rejects_duplicate_and_nfc_colliding_keys() {
let json = serde_json::to_string(&package()).unwrap();
let duplicate = json.replacen(
"\"schema_version\":1",
"\"schema_version\":1,\"schema_version\":1",
1,
);
assert!(PortablePredictorPackage::from_json(&duplicate)
.unwrap_err()
.to_string()
.contains("duplicate JSON object key"));
let collision = json.replacen(
"\"metadata\":{}",
"\"metadata\":{\"é\":1,\"e\\u0301\":2}",
1,
);
assert!(PortablePredictorPackage::from_json(&collision)
.unwrap_err()
.to_string()
.contains("NFC-colliding"));
}
#[test]
fn portable_package_rejects_refingerprinted_crosslink_and_relation_drift() {
let mut plan_drift = package();
plan_drift.training_outcome.effective_plan_fingerprint = "f".repeat(64);
plan_drift.package_fingerprint = zero_fingerprint();
plan_drift.package_fingerprint = plan_drift.compute_fingerprint().unwrap();
assert!(plan_drift
.validate()
.unwrap_err()
.to_string()
.contains("effective plan fingerprint"));
let mut binding_drift = package();
binding_drift.output_bindings[0].target_space = "other".to_string();
binding_drift.output_bindings[0].binding_fingerprint = zero_fingerprint();
binding_drift.output_bindings[0].binding_fingerprint = binding_drift.output_bindings[0]
.compute_fingerprint()
.unwrap();
binding_drift.package_fingerprint = zero_fingerprint();
binding_drift.package_fingerprint = binding_drift.compute_fingerprint().unwrap();
assert!(binding_drift
.validate()
.unwrap_err()
.to_string()
.contains("output bindings are not cross-linked"));
let mut relation_drift = package();
relation_drift.data_identities[0].relation_fingerprint = "e".repeat(64);
relation_drift.data_identities[0].identity_fingerprint = zero_fingerprint();
relation_drift.data_identities[0].identity_fingerprint = relation_drift.data_identities[0]
.compute_fingerprint()
.unwrap();
relation_drift.package_fingerprint = zero_fingerprint();
relation_drift.package_fingerprint = relation_drift.compute_fingerprint().unwrap();
assert!(relation_drift
.validate()
.unwrap_err()
.to_string()
.contains("bundle fingerprints"));
let mut content_drift = package();
content_drift.data_identities[0].data_content_fingerprint = "d".repeat(64);
content_drift.data_identities[0].identity_fingerprint = zero_fingerprint();
content_drift.data_identities[0].identity_fingerprint = content_drift.data_identities[0]
.compute_fingerprint()
.unwrap();
content_drift.package_fingerprint = zero_fingerprint();
content_drift.package_fingerprint = content_drift.compute_fingerprint().unwrap();
assert!(content_drift
.validate()
.unwrap_err()
.to_string()
.contains("data identity content"));
let mut bundle_drift = package();
bundle_drift
.execution_bundle
.metadata
.insert("same_id_drift".to_string(), json!(true));
bundle_drift.package_fingerprint = zero_fingerprint();
bundle_drift.package_fingerprint = bundle_drift.compute_fingerprint().unwrap();
assert!(bundle_drift
.validate()
.unwrap_err()
.to_string()
.contains("execution bundle content"));
}
#[test]
fn portable_required_package_has_no_host_sidecar_subset() {
let mut package = package();
package.fitted_artifact_mode = FittedArtifactMode::PortableRequired;
for binding in &mut package.artifact_bindings {
binding.load_mode = ArtifactLoadMode::NativePortable;
}
package.package_fingerprint = zero_fingerprint();
package.package_fingerprint = package.compute_fingerprint().unwrap();
package.validate().unwrap();
let loaded = LoadedPredictor::<String>::new(package, BTreeMap::new()).unwrap();
assert!(loaded.artifacts.is_empty());
}
#[test]
fn package_refuses_runtime_handle_shape_even_when_nested_in_metadata() {
for payload in [
json!({"handle": 9, "owner_controller": "controller:model.mock"}),
json!({"nested": {"model_handle": 9}}),
json!({"nested": [{"runtime_handles": [9]}]}),
] {
let mut package = package();
package
.execution_bundle
.metadata
.insert("forbidden".to_string(), payload);
package.training_outcome.execution_bundle_fingerprint =
tcv1_fingerprint(&package.execution_bundle, "runtime-handle test bundle").unwrap();
package.package_fingerprint = zero_fingerprint();
package.package_fingerprint = package.compute_fingerprint().unwrap();
assert!(package
.validate()
.unwrap_err()
.to_string()
.contains("runtime handles"));
}
}
#[test]
fn portable_w0_output_binding_and_influence_fingerprints_match_production_tcv1() {
let package = package();
for binding in &package.output_bindings {
assert_eq!(
binding.binding_fingerprint,
binding.compute_fingerprint().unwrap()
);
}
assert_eq!(
package.training_influence.manifest_fingerprint,
package.training_influence.compute_fingerprint().unwrap()
);
}
#[test]
fn portable_package_accepts_multi_scope_base_influence_per_node() {
let mut package = package();
let base_kinds = [
TrainingInfluenceKind::TransformFit,
TrainingInfluenceKind::ModelFit,
TrainingInfluenceKind::HpoSelection,
TrainingInfluenceKind::TrainedMetaAggregation,
]
.into_iter()
.collect::<BTreeSet<_>>();
let mut entries = Vec::new();
for entry in package.training_influence.entries.clone() {
if entry.node_id.is_some() && base_kinds.contains(&entry.kind) {
for suffix in ["fit_cv:fold:0", "fit_cv:fold:1", "refit:full"] {
let mut scoped = entry.clone();
scoped.scope_id = format!("{suffix}:{}", entry.scope_id);
entries.push(scoped);
}
} else {
entries.push(entry);
}
}
entries.sort_by(|left, right| {
(left.kind, &left.scope_id, &left.node_id).cmp(&(
right.kind,
&right.scope_id,
&right.node_id,
))
});
package.training_influence.entries = entries;
package.training_influence.manifest_fingerprint = zero_fingerprint();
package.training_influence.manifest_fingerprint =
package.training_influence.compute_fingerprint().unwrap();
package.training_outcome.training_influence_fingerprint =
package.training_influence.manifest_fingerprint.clone();
package.package_fingerprint = zero_fingerprint();
package.package_fingerprint = package.compute_fingerprint().unwrap();
package.validate().unwrap();
}
#[test]
fn controller_id_import_remains_the_same_public_type() {
let id = ControllerId::new("controller:model.mock").unwrap();
assert!(package().template.controller_manifests.contains_key(&id));
}
#[test]
fn committed_w1_fixtures_match_rust_and_independent_tcv1_oracle() {
let refit_json =
include_str!("../../../examples/fixtures/training/training_request_refit.v1.json");
let refit = TrainingRequest::from_json(refit_json).unwrap();
let no_refit = TrainingRequest::from_json(include_str!(
"../../../examples/fixtures/training/training_request_no_refit.v1.json"
))
.unwrap();
let active_influence = TrainingRequest::from_json(include_str!(
"../../../examples/fixtures/training/training_request_active_influence.v1.json"
))
.unwrap();
let package_request = TrainingRequest::from_json(include_str!(
"../../../examples/fixtures/training/training_request_package_refit.v1.json"
))
.unwrap();
assert!(refit.options.refit);
assert!(!no_refit.options.refit);
assert_eq!(active_influence.influence_requirements.len(), 6);
let package_json =
include_str!("../../../examples/fixtures/training/portable_predictor_package.v1.json");
let package = PortablePredictorPackage::from_json(package_json).unwrap();
assert_eq!(
package.training_request_fingerprint,
package_request.request_fingerprint
);
assert_eq!(package.data_identities, package_request.data_identities);
let namespace = CacheNamespace::from_json(include_str!(
"../../../examples/fixtures/training/cache_namespace_fit_cv.v1.json"
))
.unwrap();
let identity = package
.data_identities
.iter()
.find(|identity| identity.requirement_key == namespace.data_requirement_key)
.unwrap();
namespace.validate_for_identity(identity).unwrap();
let projection: ParameterProjection = serde_json::from_str(include_str!(
"../../../examples/fixtures/training/parameter_projection_empty.v1.json"
))
.unwrap();
projection.validate().unwrap();
let negatives: serde_json::Value = serde_json::from_str(include_str!(
"../../../examples/fixtures/training/negative_cases.v1.json"
))
.unwrap();
for case in negatives["cases"].as_array().unwrap() {
let document = serde_json::to_string(&case["document"]).unwrap();
let error = match case["contract"].as_str().unwrap() {
"cache_namespace" => CacheNamespace::from_json(&document).unwrap_err(),
"portable_predictor_package" => {
PortablePredictorPackage::from_json(&document).unwrap_err()
}
"training_outcome" => {
crate::training_runtime::TrainingOutcome::from_json(&document).unwrap_err()
}
"training_request" => TrainingRequest::from_json(&document).unwrap_err(),
other => panic!("unknown negative contract {other}"),
};
assert!(
error
.to_string()
.contains(case["expected_error"].as_str().unwrap()),
"{}: {error}",
case["id"]
);
}
}
#[test]
fn projection_strict_parsers_reject_duplicate_and_nfc_colliding_keys() {
let parameter_json =
include_str!("../../../examples/fixtures/training/parameter_projection_empty.v1.json");
ParameterProjection::from_json(parameter_json).unwrap();
let duplicate = parameter_json.replacen(
"\"schema_version\": 1",
"\"schema_version\": 1, \"schema_version\": 1",
1,
);
assert!(ParameterProjection::from_json(&duplicate)
.unwrap_err()
.to_string()
.contains("duplicate JSON object key"));
let collision = parameter_json.replacen('{', "{\"é\":1,\"e\\u0301\":2,", 1);
assert!(ParameterProjection::from_json(&collision)
.unwrap_err()
.to_string()
.contains("NFC-colliding"));
let request = request();
let projection = request.project().unwrap();
let projection_json = serde_json::to_string(&projection).unwrap();
TrainingContractProjection::from_json(&projection_json).unwrap();
let duplicate = projection_json.replacen(
"\"request_id\":",
"\"request_id\":\"duplicate\",\"request_id\":",
1,
);
assert!(TrainingContractProjection::from_json(&duplicate)
.unwrap_err()
.to_string()
.contains("duplicate JSON object key"));
let collision = projection_json.replacen('{', "{\"é\":1,\"e\\u0301\":2,", 1);
assert!(TrainingContractProjection::from_json(&collision)
.unwrap_err()
.to_string()
.contains("NFC-colliding"));
for path in [
&["plan", "graph_plan", "graph"][..],
&["plan", "campaign"][..],
] {
let mut unknown: serde_json::Value = serde_json::from_str(&projection_json).unwrap();
let mut parent = &mut unknown;
for segment in path {
parent = &mut parent[*segment];
}
parent["unknown_projection_field"] = json!(true);
let error =
TrainingContractProjection::from_json(&serde_json::to_string(&unknown).unwrap())
.unwrap_err();
let expected_path = format!("{}.unknown_projection_field", path.join("."));
assert!(error.to_string().contains("unknown field"), "{error}");
assert!(error.to_string().contains(&expected_path), "{error}");
}
}
}