dag-ml-core 0.3.23

//! Durable, identity-keyed wiring for the native split-conformal kernel.
//!
//! This module deliberately accepts point predictions and truth only.  The
//! scheduler/controller boundary remains unchanged: a host supplies the
//! ordinary PREDICT result and DAG-ML validates its stable sample identities,
//! calibrates with [`crate::conformal`], and can apply the persisted record on
//! another ordinary PREDICT result.

use std::collections::BTreeSet;

use serde::{Deserialize, Serialize};

use crate::canonical::parse_typed_json;
use crate::conformal::{
    apply_split_absolute_residual, finite_sample_conformal_rank, split_absolute_residual_quantiles,
    ConformalMultiTargetPolicy, ConformalSmallSamplePolicy, RegressionConformalInterval,
    SplitConformalQuantile,
};
use crate::error::{DagMlError, Result};
use crate::ids::SampleId;
use crate::oof::PredictionBlock;
use crate::phase::Phase;
use crate::replay::{TrainingReplayOutcome, TrainingReplayRequest};
use crate::runtime::NativePredictorDescriptorV1;
use crate::training::PortablePredictorPackage;

/// V1 did not bind calibration to the training/replay provenance closure.  It
/// is deliberately not accepted: callers must migrate to this closed V2 form.
pub const CONFORMAL_RUNTIME_SCHEMA_VERSION: u32 = 2;

/// Stable, presentation-only projection of one validated single-target
/// conformal PREDICT replay.  It contains no calibration algorithm input and
/// no mutable model handle: Core and Studio may transport or render it, but
/// must never recalculate its bounds.
pub const CONFORMAL_PRESENTATION_SCHEMA_VERSION: u32 = 1;

/// Multi-target, Archive-bound presentation contract. V1 remains the stable
/// scalar UI projection; V2 is an additive native transport surface.
pub const CONFORMAL_PRESENTATION_SCHEMA_VERSION_V2: u32 = 2;

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationInterval {
    pub coverage: f64,
    /// `None` represents a deliberately unbounded interval, never an
    /// infinity/sentinel endpoint.
    pub lower: Vec<Option<f64>>,
    pub upper: Vec<Option<f64>>,
    /// Exact native split-conformal radius for this coverage, or `None` for
    /// the declared unbounded small-sample policy.
    pub qhat: Option<f64>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationV1 {
    pub schema_version: u32,
    pub package_fingerprint: String,
    pub replay_outcome_fingerprint: String,
    pub binding_id: String,
    pub target_name: String,
    pub sample_ids: Vec<SampleId>,
    pub point_predictions: Vec<f64>,
    pub intervals: Vec<ConformalPresentationInterval>,
    pub calibration_fingerprint: String,
    pub presentation_fingerprint: String,
}

/// Dimensions copied from the validated point block and native predictor.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationDimensionsV2 {
    pub sample_count: u64,
    pub target_count: u32,
}

/// Persisted split-conformal guarantee. These fields are copied from the
/// calibrated package and are never derived by a presentation consumer.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationGuaranteeV2 {
    pub calibration_sample_count: u64,
    pub multi_target_policy: ConformalMultiTargetPolicy,
    pub small_sample_policy: ConformalSmallSamplePolicy,
    pub quantiles: Vec<SplitConformalQuantile>,
}

/// Content-bound native identity for the exact model which produced the point
/// predictions. The model digest and descriptor are independently retained so
/// a host cannot substitute another N4MM with compatible dimensions.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationPredictorV2 {
    pub model_artifact_fingerprint: String,
    pub predictor_binding_fingerprint: String,
    pub predictor_descriptor_fingerprint: String,
}

/// Closed multi-target presentation of one already-calculated PREDICT replay.
///
/// The complete point and interval blocks are retained so validation can reuse
/// DAG-ML's authoritative sample-order, target-order and interval-closure
/// checks. `archive_sha256` is supplied by the aggregate which validated the
/// enclosing archive bytes; [`Self::validate_against_package`] binds every
/// remaining field to the parsed Package V2.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalPresentationV2 {
    pub schema_version: u32,
    pub archive_sha256: String,
    pub package_fingerprint: String,
    pub replay_outcome_fingerprint: String,
    pub binding_id: String,
    pub predictor: ConformalPresentationPredictorV2,
    pub dimensions: ConformalPresentationDimensionsV2,
    pub target_names: Vec<String>,
    pub sample_ids: Vec<SampleId>,
    pub point_prediction: PredictionBlock,
    pub interval_block: ConformalIntervalBlock,
    pub guarantee: ConformalPresentationGuaranteeV2,
    pub calibration_fingerprint: String,
    pub presentation_fingerprint: String,
}

/// Relation-derived calibration cohort. Physical and origin identities are
/// both retained so a relation-expanded training cohort cannot be bypassed by
/// presenting only one namespace. The attachment boundary derives and checks
/// these fields from an authoritative [`crate::relation::SampleRelationSet`].
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalCalibrationCohort {
    pub role: String,
    pub physical_sample_ids: Vec<SampleId>,
    pub origin_sample_ids: Vec<SampleId>,
    pub target_names: Vec<String>,
    pub manifest_fingerprint: String,
}

/// Complete, canonical provenance closure supplied by the replay boundary.
/// These are not optional hints: attached calibration checks every member
/// against the exact source outcome and replay before it is persisted.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalCalibrationContext {
    pub predictor_binding_fingerprint: String,
    pub source_training_outcome_fingerprint: String,
    pub calibration_replay_outcome_fingerprint: String,
    pub data_identities_fingerprint: String,
    pub fold_set_fingerprint: String,
    pub training_influence_fingerprint: String,
    pub relation_fingerprint: String,
    pub calibration_cohort: ConformalCalibrationCohort,
    pub context_fingerprint: String,
}

/// Closed, self-fingerprinted split-conformal state retained beside a bundle.
/// `sample_ids` is the calibration order, not an interchangeable set: this
/// makes accidental positional joins fail before residuals are calculated.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalCalibration {
    pub schema_version: u32,
    pub binding_id: String,
    pub target_names: Vec<String>,
    pub sample_ids: Vec<SampleId>,
    pub coverages: Vec<f64>,
    pub multi_target_policy: ConformalMultiTargetPolicy,
    pub small_sample_policy: ConformalSmallSamplePolicy,
    pub quantiles: Vec<SplitConformalQuantile>,
    pub context: ConformalCalibrationContext,
    pub calibration_fingerprint: String,
}

/// Typed reference retained by portable execution bundles.  It contains no
/// host object or duplicate algorithm state; the complete state stays in the
/// matching `TrainingOutcome`.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalCalibrationRef {
    pub schema_version: u32,
    pub binding_id: String,
    pub calibration_fingerprint: String,
}

/// Identity-preserving interval result for one replayed point block.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalIntervalBlock {
    pub schema_version: u32,
    pub binding_id: String,
    pub sample_ids: Vec<SampleId>,
    pub intervals: Vec<RegressionConformalInterval>,
    pub calibration_fingerprint: String,
    pub point_prediction_fingerprint: String,
}

/// Truth supplied by the data layer for a calibration replay.  It carries the
/// same stable physical sample ids as the point block so a host can never
/// smuggle a positional `y_true` matrix across a reordered replay.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConformalCalibrationTruth {
    pub sample_ids: Vec<SampleId>,
    pub values: Vec<Vec<f64>>,
}

impl ConformalIntervalBlock {
    pub fn validate(&self) -> Result<()> {
        if self.schema_version != CONFORMAL_RUNTIME_SCHEMA_VERSION
            || self.binding_id.trim().is_empty()
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal interval block has an unsupported version or empty binding id"
                    .to_string(),
            ));
        }
        validate_unique_samples(&self.sample_ids)?;
        if self.intervals.is_empty()
            || self
                .intervals
                .iter()
                .any(|interval| interval.cells.len() != self.sample_ids.len())
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal interval block does not cover its exact sample ids".to_string(),
            ));
        }
        validate_sha256(&self.calibration_fingerprint)?;
        validate_sha256(&self.point_prediction_fingerprint)
    }
}

impl ConformalPresentationV1 {
    pub fn from_json(json: &str) -> Result<Self> {
        let raw_fingerprint = parse_typed_json(json)
            .and_then(|value| value.fingerprint_without("presentation_fingerprint"))
            .map_err(|error| {
                DagMlError::RuntimeValidation(format!(
                    "conformal presentation is outside strict TCV1 JSON: {error}"
                ))
            })?;
        let presentation: Self = serde_json::from_str(json)?;
        if presentation.presentation_fingerprint != raw_fingerprint {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation fingerprint does not match original TCV1 JSON".to_string(),
            ));
        }
        presentation.validate()?;
        Ok(presentation)
    }

    pub fn compute_fingerprint(&self) -> Result<String> {
        let json = serde_json::to_string(self)?;
        parse_typed_json(&json)
            .and_then(|value| value.fingerprint_without("presentation_fingerprint"))
            .map_err(|error| {
                DagMlError::RuntimeValidation(format!(
                    "conformal presentation is outside strict TCV1 JSON: {error}"
                ))
            })
    }

    pub fn validate(&self) -> Result<()> {
        if self.schema_version != CONFORMAL_PRESENTATION_SCHEMA_VERSION
            || self.binding_id.trim().is_empty()
            || self.target_name.trim().is_empty()
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation has an unsupported version or empty binding metadata"
                    .to_string(),
            ));
        }
        for fingerprint in [
            &self.package_fingerprint,
            &self.replay_outcome_fingerprint,
            &self.calibration_fingerprint,
            &self.presentation_fingerprint,
        ] {
            validate_sha256(fingerprint)?;
        }
        validate_unique_samples(&self.sample_ids)?;
        if self.sample_ids.is_empty()
            || self.point_predictions.len() != self.sample_ids.len()
            || self
                .point_predictions
                .iter()
                .any(|value| !value.is_finite())
            || self.intervals.is_empty()
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation does not exactly cover finite point predictions"
                    .to_string(),
            ));
        }
        let mut prior_coverage = None;
        for interval in &self.intervals {
            if !(interval.coverage.is_finite()
                && 0.0 < interval.coverage
                && interval.coverage < 1.0)
                || prior_coverage.is_some_and(|prior| prior >= interval.coverage)
                || interval.lower.len() != self.sample_ids.len()
                || interval.upper.len() != self.sample_ids.len()
                || interval
                    .qhat
                    .is_some_and(|value| !value.is_finite() || value < 0.0)
            {
                return Err(DagMlError::RuntimeValidation(
                    "conformal presentation has invalid coverage or interval cardinality"
                        .to_string(),
                ));
            }
            for ((point, lower), upper) in self
                .point_predictions
                .iter()
                .zip(&interval.lower)
                .zip(&interval.upper)
            {
                match (lower, upper) {
                    (Some(lower), Some(upper))
                        if lower.is_finite()
                            && upper.is_finite()
                            && lower <= point
                            && point <= upper => {}
                    (None, None) if interval.qhat.is_none() => {}
                    _ => {
                        return Err(DagMlError::RuntimeValidation(
                            "conformal presentation interval endpoints are inconsistent"
                                .to_string(),
                        ));
                    }
                }
            }
            prior_coverage = Some(interval.coverage);
        }
        if self.presentation_fingerprint != self.compute_fingerprint()? {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation fingerprint does not match TCV1 content".to_string(),
            ));
        }
        Ok(())
    }
}

impl ConformalPresentationV2 {
    /// Parse and validate the self-contained transport shape. Consumers which
    /// possess the source package must use [`Self::from_json_for_package`] to
    /// additionally verify every provenance cross-link.
    pub fn from_json(json: &str) -> Result<Self> {
        let raw_fingerprint = parse_typed_json(json)
            .and_then(|value| value.fingerprint_without("presentation_fingerprint"))
            .map_err(|error| {
                DagMlError::RuntimeValidation(format!(
                    "conformal presentation V2 is outside strict TCV1 JSON: {error}"
                ))
            })?;
        let presentation: Self = serde_json::from_str(json)?;
        if presentation.presentation_fingerprint != raw_fingerprint {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 fingerprint does not match original TCV1 JSON"
                    .to_string(),
            ));
        }
        presentation.validate()?;
        Ok(presentation)
    }

    /// Parse a presentation and bind it to the exact validated Package V2 and
    /// native descriptors inspected from its model bytes.
    pub fn from_json_for_package(
        json: &str,
        package: &PortablePredictorPackage,
        native_predictors: &[NativePredictorDescriptorV1],
    ) -> Result<Self> {
        let presentation = Self::from_json(json)?;
        presentation.validate_against_package(package, native_predictors)?;
        Ok(presentation)
    }

    pub fn compute_fingerprint(&self) -> Result<String> {
        let json = serde_json::to_string(self)?;
        parse_typed_json(&json)
            .and_then(|value| value.fingerprint_without("presentation_fingerprint"))
            .map_err(|error| {
                DagMlError::RuntimeValidation(format!(
                    "conformal presentation V2 is outside strict TCV1 JSON: {error}"
                ))
            })
    }

    /// Validate the self-contained ordered identities, dimensions, guarantee,
    /// interval closure and TCV1 fingerprint.
    pub fn validate(&self) -> Result<()> {
        if self.schema_version != CONFORMAL_PRESENTATION_SCHEMA_VERSION_V2 {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 has an unsupported schema version".to_string(),
            ));
        }
        for fingerprint in [
            &self.archive_sha256,
            &self.package_fingerprint,
            &self.replay_outcome_fingerprint,
            &self.predictor.model_artifact_fingerprint,
            &self.predictor.predictor_binding_fingerprint,
            &self.predictor.predictor_descriptor_fingerprint,
            &self.calibration_fingerprint,
            &self.presentation_fingerprint,
        ] {
            validate_sha256(fingerprint)?;
        }
        let target_count = u32::try_from(self.target_names.len()).map_err(|_| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 target count exceeds u32".to_string(),
            )
        })?;
        let sample_count = u64::try_from(self.sample_ids.len()).map_err(|_| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 sample count exceeds u64".to_string(),
            )
        })?;
        if self.target_names.is_empty()
            || self.dimensions.target_count != target_count
            || self.dimensions.sample_count != sample_count
            || self.sample_ids != self.point_prediction.sample_ids
            || self.target_names != self.point_prediction.target_names
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 dimensions or ordered identities do not match"
                    .to_string(),
            ));
        }
        self.point_prediction.validate_content()?;
        self.interval_block.validate()?;
        if self.guarantee.calibration_sample_count == 0
            || self.interval_block.binding_id != self.binding_id
            || self.interval_block.sample_ids != self.sample_ids
            || self.interval_block.calibration_fingerprint != self.calibration_fingerprint
            || self.interval_block.point_prediction_fingerprint
                != point_prediction_fingerprint_for_runtime(&self.point_prediction)?
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 interval identities do not match".to_string(),
            ));
        }
        let expected_intervals = apply_split_absolute_residual(
            &self.point_prediction.values,
            &self.guarantee.quantiles,
            self.guarantee.multi_target_policy,
        )
        .map_err(|error| {
            DagMlError::RuntimeValidation(format!(
                "conformal presentation V2 has an invalid guarantee: {error}"
            ))
        })?;
        if expected_intervals != self.interval_block.intervals {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 intervals do not close over its persisted guarantee"
                    .to_string(),
            ));
        }
        if self.presentation_fingerprint != self.compute_fingerprint()? {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 fingerprint does not match TCV1 content".to_string(),
            ));
        }
        Ok(())
    }

    /// Validate all presentation fields against the authoritative package and
    /// descriptors. Interval closure delegates to the existing conformal
    /// runtime; this method performs no calibration or interval calculation.
    pub fn validate_against_package(
        &self,
        package: &PortablePredictorPackage,
        native_predictors: &[NativePredictorDescriptorV1],
    ) -> Result<()> {
        package.validate()?;
        self.validate()?;
        if self.package_fingerprint != package.package_fingerprint {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 does not match its package fingerprint".to_string(),
            ));
        }
        let calibration = package.conformal_calibration.as_ref().ok_or_else(|| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 requires package calibration state".to_string(),
            )
        })?;
        let binding = package
            .output_bindings
            .iter()
            .find(|binding| binding.binding_id == calibration.binding_id)
            .ok_or_else(|| {
                DagMlError::RuntimeValidation(
                    "conformal presentation V2 calibration binding is absent".to_string(),
                )
            })?;
        let (model_fingerprint, descriptor) =
            presentation_native_predictor(package, binding, native_predictors)?;
        if self.binding_id != calibration.binding_id
            || self.predictor.predictor_binding_fingerprint != binding.binding_fingerprint
            || self.predictor.predictor_binding_fingerprint
                != calibration.context.predictor_binding_fingerprint
            || self.predictor.model_artifact_fingerprint != model_fingerprint
            || self.predictor.predictor_descriptor_fingerprint != descriptor.descriptor_fingerprint
            || self.calibration_fingerprint != calibration.calibration_fingerprint
            || self.target_names != binding.target_names
            || self.target_names != calibration.target_names
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 model, binding, targets or calibration do not cross-link"
                    .to_string(),
            ));
        }
        let target_count = i32::try_from(self.target_names.len()).map_err(|_| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 target count exceeds i32".to_string(),
            )
        })?;
        if descriptor.dimensions.n_targets != target_count {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 target count does not match native predictor"
                    .to_string(),
            ));
        }
        if self.guarantee.calibration_sample_count
            != u64::try_from(calibration.sample_ids.len()).map_err(|_| {
                DagMlError::RuntimeValidation(
                    "conformal presentation V2 calibration sample count exceeds u64".to_string(),
                )
            })?
            || self.guarantee.multi_target_policy != calibration.multi_target_policy
            || self.guarantee.small_sample_policy != calibration.small_sample_policy
            || self.guarantee.quantiles != calibration.quantiles
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation V2 guarantee does not match calibration state".to_string(),
            ));
        }
        self.interval_block
            .validate_against(calibration, &self.point_prediction)?;
        Ok(())
    }
}

fn presentation_native_predictor<'a>(
    package: &'a PortablePredictorPackage,
    binding: &crate::training::OutputBinding,
    native_predictors: &'a [NativePredictorDescriptorV1],
) -> Result<(&'a str, &'a NativePredictorDescriptorV1)> {
    let records = package
        .execution_bundle
        .refit_artifacts
        .iter()
        .filter(|record| record.node_id == binding.node_id && record.artifact.kind == "n4m_model")
        .collect::<Vec<_>>();
    let [record] = records.as_slice() else {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 requires exactly one native model artifact for its binding"
                .to_string(),
        ));
    };
    let model_fingerprint = record
        .artifact
        .content_fingerprint
        .as_deref()
        .ok_or_else(|| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 native model has no content fingerprint".to_string(),
            )
        })?;
    let descriptors = native_predictors
        .iter()
        .filter(|descriptor| {
            descriptor.artifact_sha256 == model_fingerprint
                && descriptor.owner_controller == record.controller_id
        })
        .collect::<Vec<_>>();
    let [descriptor] = descriptors.as_slice() else {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 requires exactly one byte-attested predictor descriptor"
                .to_string(),
        ));
    };
    descriptor.validate()?;
    if record
        .artifact
        .native_predictor_descriptor
        .as_ref()
        .is_some_and(|embedded| embedded != *descriptor)
    {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 byte-attested descriptor differs from package metadata"
                .to_string(),
        ));
    }
    Ok((model_fingerprint, descriptor))
}

/// Project an already-validated Package V2 PREDICT replay without performing
/// any conformal calculation. The aggregate supplies the SHA-256 of the exact
/// archive bytes and descriptors inspected from their native model members.
pub fn build_conformal_presentation_v2(
    archive_sha256: &str,
    package: &PortablePredictorPackage,
    request: &TrainingReplayRequest,
    replay: &TrainingReplayOutcome,
    native_predictors: &[NativePredictorDescriptorV1],
) -> Result<ConformalPresentationV2> {
    validate_sha256(archive_sha256)?;
    package.validate()?;
    request.validate()?;
    replay.validate_against_package(package, request)?;
    if replay.phase != Phase::Predict {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 requires a PREDICT replay".to_string(),
        ));
    }
    let calibration = package.conformal_calibration.as_ref().ok_or_else(|| {
        DagMlError::RuntimeValidation(
            "conformal presentation V2 requires package calibration state".to_string(),
        )
    })?;
    let output = replay
        .outputs
        .iter()
        .find(|output| output.binding.binding_id == calibration.binding_id)
        .ok_or_else(|| {
            DagMlError::RuntimeValidation(
                "conformal presentation V2 replay is missing the calibrated binding".to_string(),
            )
        })?;
    let [point_prediction] = output.predictions.as_slice() else {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 requires exactly one point prediction block".to_string(),
        ));
    };
    let interval_blocks = replay
        .conformal_intervals
        .iter()
        .filter(|block| block.binding_id == calibration.binding_id)
        .collect::<Vec<_>>();
    let [interval_block] = interval_blocks.as_slice() else {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation V2 requires exactly one interval block".to_string(),
        ));
    };
    interval_block.validate_against(calibration, point_prediction)?;
    let (model_artifact_fingerprint, descriptor) =
        presentation_native_predictor(package, &output.binding, native_predictors)?;
    let mut presentation = ConformalPresentationV2 {
        schema_version: CONFORMAL_PRESENTATION_SCHEMA_VERSION_V2,
        archive_sha256: archive_sha256.to_string(),
        package_fingerprint: package.package_fingerprint.clone(),
        replay_outcome_fingerprint: replay.outcome_fingerprint.clone(),
        binding_id: calibration.binding_id.clone(),
        predictor: ConformalPresentationPredictorV2 {
            model_artifact_fingerprint: model_artifact_fingerprint.to_string(),
            predictor_binding_fingerprint: output.binding.binding_fingerprint.clone(),
            predictor_descriptor_fingerprint: descriptor.descriptor_fingerprint.clone(),
        },
        dimensions: ConformalPresentationDimensionsV2 {
            sample_count: u64::try_from(point_prediction.sample_ids.len()).map_err(|_| {
                DagMlError::RuntimeValidation(
                    "conformal presentation V2 sample count exceeds u64".to_string(),
                )
            })?,
            target_count: u32::try_from(output.binding.target_names.len()).map_err(|_| {
                DagMlError::RuntimeValidation(
                    "conformal presentation V2 target count exceeds u32".to_string(),
                )
            })?,
        },
        target_names: output.binding.target_names.clone(),
        sample_ids: point_prediction.sample_ids.clone(),
        point_prediction: point_prediction.clone(),
        interval_block: (*interval_block).clone(),
        guarantee: ConformalPresentationGuaranteeV2 {
            calibration_sample_count: u64::try_from(calibration.sample_ids.len()).map_err(
                |_| {
                    DagMlError::RuntimeValidation(
                        "conformal presentation V2 calibration sample count exceeds u64"
                            .to_string(),
                    )
                },
            )?,
            multi_target_policy: calibration.multi_target_policy,
            small_sample_policy: calibration.small_sample_policy,
            quantiles: calibration.quantiles.clone(),
        },
        calibration_fingerprint: calibration.calibration_fingerprint.clone(),
        presentation_fingerprint: "0".repeat(64),
    };
    presentation.presentation_fingerprint = presentation.compute_fingerprint()?;
    presentation.validate_against_package(package, native_predictors)?;
    Ok(presentation)
}

/// Project a verified Package V2 and PREDICT replay into the exact scalar
/// representation consumed by the shared UI.  Multi-target output is refused
/// instead of selecting or reshaping a target implicitly.
pub fn build_conformal_presentation_v1(
    package: &PortablePredictorPackage,
    request: &TrainingReplayRequest,
    replay: &TrainingReplayOutcome,
) -> Result<ConformalPresentationV1> {
    package.validate()?;
    request.validate()?;
    replay.validate_against_package(package, request)?;
    if replay.phase != Phase::Predict {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation requires a PREDICT replay".to_string(),
        ));
    }
    let calibration = package.conformal_calibration.as_ref().ok_or_else(|| {
        DagMlError::RuntimeValidation(
            "conformal presentation requires package calibration state".to_string(),
        )
    })?;
    if calibration.target_names.len() != 1 {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation refuses multi-target output".to_string(),
        ));
    }
    let output = replay
        .outputs
        .iter()
        .find(|output| output.binding.binding_id == calibration.binding_id)
        .ok_or_else(|| {
            DagMlError::RuntimeValidation(
                "conformal presentation replay is missing the calibrated binding".to_string(),
            )
        })?;
    if output.binding.target_names != calibration.target_names || output.predictions.len() != 1 {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation requires exactly one matching scalar point block".to_string(),
        ));
    }
    let point = &output.predictions[0];
    point.validate_content()?;
    if point
        .values
        .iter()
        .any(|row| row.len() != 1 || !row[0].is_finite())
    {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation requires finite single-target point predictions".to_string(),
        ));
    }
    let intervals = replay
        .conformal_intervals
        .iter()
        .filter(|interval| interval.binding_id == calibration.binding_id)
        .collect::<Vec<_>>();
    if intervals.len() != 1 {
        return Err(DagMlError::RuntimeValidation(
            "conformal presentation requires exactly one calibrated interval block".to_string(),
        ));
    }
    let interval_block = intervals[0];
    interval_block.validate_against(calibration, point)?;
    let mut presentation_intervals = Vec::with_capacity(interval_block.intervals.len());
    for interval in &interval_block.intervals {
        let quantile = calibration
            .quantiles
            .iter()
            .find(|quantile| quantile.coverage == interval.coverage)
            .ok_or_else(|| {
                DagMlError::RuntimeValidation(
                    "conformal presentation interval coverage is absent from calibration"
                        .to_string(),
                )
            })?;
        if quantile.radii.len() != 1 || interval.cells.iter().any(|row| row.len() != 1) {
            return Err(DagMlError::RuntimeValidation(
                "conformal presentation requires scalar calibration radii and cells".to_string(),
            ));
        }
        let qhat = match quantile.radii[0] {
            crate::conformal::ConformalRadius::Finite(value)
                if value.is_finite() && value >= 0.0 =>
            {
                Some(value)
            }
            crate::conformal::ConformalRadius::Unbounded => None,
            _ => {
                return Err(DagMlError::RuntimeValidation(
                    "conformal presentation calibration radius is invalid".to_string(),
                ));
            }
        };
        let (lower, upper) = interval.cells.iter().map(|row| row[0].endpoints()).unzip();
        presentation_intervals.push(ConformalPresentationInterval {
            coverage: interval.coverage,
            lower,
            upper,
            qhat,
        });
    }
    presentation_intervals.sort_by(|left, right| left.coverage.total_cmp(&right.coverage));
    let mut presentation = ConformalPresentationV1 {
        schema_version: CONFORMAL_PRESENTATION_SCHEMA_VERSION,
        package_fingerprint: package.package_fingerprint.clone(),
        replay_outcome_fingerprint: replay.outcome_fingerprint.clone(),
        binding_id: calibration.binding_id.clone(),
        target_name: calibration.target_names[0].clone(),
        sample_ids: point.sample_ids.clone(),
        point_predictions: point.values.iter().map(|row| row[0]).collect(),
        intervals: presentation_intervals,
        calibration_fingerprint: calibration.calibration_fingerprint.clone(),
        presentation_fingerprint: "0".repeat(64),
    };
    presentation.presentation_fingerprint = presentation.compute_fingerprint()?;
    presentation.validate()?;
    Ok(presentation)
}

impl ConformalCalibration {
    #[allow(clippy::too_many_arguments)]
    pub fn calibrate_with_truth(
        binding_id: impl Into<String>,
        target_names: Vec<String>,
        predictions: &PredictionBlock,
        truth: &ConformalCalibrationTruth,
        context: ConformalCalibrationContext,
        coverages: Vec<f64>,
        multi_target_policy: ConformalMultiTargetPolicy,
        small_sample_policy: ConformalSmallSamplePolicy,
    ) -> Result<Self> {
        predictions.validate_content()?;
        validate_identity_aligned_truth(predictions, truth)?;
        context.validate_for_truth(truth, &target_names)?;
        if target_names.len() != predictions.values[0].len()
            || (!predictions.target_names.is_empty() && predictions.target_names != target_names)
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal target order does not match the point prediction binding".to_string(),
            ));
        }
        let residuals = predictions
            .values
            .iter()
            .zip(&truth.values)
            .map(|(prediction, actual)| {
                prediction
                    .iter()
                    .zip(actual)
                    .map(|(point, value)| (point - value).abs())
                    .collect::<Vec<_>>()
            })
            .collect::<Vec<_>>();
        let quantiles = split_absolute_residual_quantiles(
            &residuals,
            &coverages,
            multi_target_policy,
            small_sample_policy,
        )
        .map_err(|error| {
            DagMlError::RuntimeValidation(format!("conformal calibration failed: {error}"))
        })?;
        let calibration = Self {
            schema_version: CONFORMAL_RUNTIME_SCHEMA_VERSION,
            binding_id: binding_id.into(),
            target_names,
            sample_ids: predictions.sample_ids.clone(),
            coverages,
            multi_target_policy,
            small_sample_policy,
            quantiles,
            context,
            calibration_fingerprint: String::new(),
        };
        stabilize_calibration_for_tcv1(calibration)
    }

    pub fn reference(&self) -> Result<ConformalCalibrationRef> {
        self.validate()?;
        Ok(ConformalCalibrationRef {
            schema_version: CONFORMAL_RUNTIME_SCHEMA_VERSION,
            binding_id: self.binding_id.clone(),
            calibration_fingerprint: self.calibration_fingerprint.clone(),
        })
    }

    pub fn compute_fingerprint(&self) -> Result<String> {
        fingerprint_without(self, "calibration_fingerprint", "conformal calibration")
    }

    pub fn from_json(json: &str) -> Result<Self> {
        let raw = parse_typed_json(json)
            .and_then(|value| value.fingerprint_without("calibration_fingerprint"))
            .map_err(|error| {
                DagMlError::RuntimeValidation(format!(
                    "conformal calibration is not strict TCV1 JSON: {error}"
                ))
            })?;
        let calibration: Self = serde_json::from_str(json)?;
        if calibration.calibration_fingerprint != raw {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration fingerprint does not match original TCV1 JSON".to_string(),
            ));
        }
        calibration.validate()?;
        Ok(calibration)
    }

    pub fn validate(&self) -> Result<()> {
        if self.schema_version != CONFORMAL_RUNTIME_SCHEMA_VERSION {
            return Err(DagMlError::RuntimeValidation(format!(
                "conformal calibration has unsupported schema_version {}",
                self.schema_version
            )));
        }
        if self.binding_id.trim().is_empty() || self.target_names.is_empty() {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration requires a binding id and target names".to_string(),
            ));
        }
        validate_unique_samples(&self.sample_ids)?;
        self.context.validate_for_calibration(self)?;
        if self.coverages.is_empty() || self.quantiles.len() != self.coverages.len() {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration coverages and quantiles must have equal non-zero length"
                    .to_string(),
            ));
        }
        if self
            .quantiles
            .iter()
            .zip(&self.coverages)
            .any(|(quantile, coverage)| quantile.coverage.to_bits() != coverage.to_bits())
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration quantile coverage order does not match coverages"
                    .to_string(),
            ));
        }
        let sample_count = u64::try_from(self.sample_ids.len()).map_err(|_| {
            DagMlError::RuntimeValidation(
                "conformal calibration sample count exceeds u64".to_string(),
            )
        })?;
        for (index, (coverage, quantile)) in self.coverages.iter().zip(&self.quantiles).enumerate()
        {
            let expected =
                finite_sample_conformal_rank(sample_count, *coverage).map_err(|error| {
                    DagMlError::RuntimeValidation(format!(
                        "invalid conformal rank at coverage {index}: {error}"
                    ))
                })?;
            if quantile.rank != expected {
                return Err(DagMlError::RuntimeValidation(format!(
                    "conformal quantile rank at coverage {index} does not match sample count and coverage"
                )));
            }
        }
        // The kernel validates coverage ordering, radius shape, and nestedness
        // before application; applying to one finite dummy row is a compact
        // validation that does not introduce another conformal algorithm.
        apply_split_absolute_residual(
            &[vec![0.0; self.target_names.len()]],
            &self.quantiles,
            self.multi_target_policy,
        )
        .map_err(|error| {
            DagMlError::RuntimeValidation(format!("invalid conformal quantiles: {error}"))
        })?;
        validate_sha256(&self.calibration_fingerprint)?;
        if self.calibration_fingerprint != self.compute_fingerprint()? {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration fingerprint does not match TCV1 content".to_string(),
            ));
        }
        Ok(())
    }

    pub fn apply(&self, predictions: &PredictionBlock) -> Result<ConformalIntervalBlock> {
        self.validate()?;
        predictions.validate_content()?;
        if predictions.target_names != self.target_names {
            return Err(DagMlError::RuntimeValidation(
                "conformal application target order does not match calibration".to_string(),
            ));
        }
        let intervals = apply_split_absolute_residual(
            &predictions.values,
            &self.quantiles,
            self.multi_target_policy,
        )
        .map_err(|error| {
            DagMlError::RuntimeValidation(format!("conformal application failed: {error}"))
        })?;
        Ok(ConformalIntervalBlock {
            schema_version: CONFORMAL_RUNTIME_SCHEMA_VERSION,
            binding_id: self.binding_id.clone(),
            sample_ids: predictions.sample_ids.clone(),
            intervals,
            calibration_fingerprint: self.calibration_fingerprint.clone(),
            point_prediction_fingerprint: point_prediction_fingerprint_for_runtime(predictions)?,
        })
    }
}

impl ConformalCalibrationContext {
    pub fn compute_fingerprint(&self) -> Result<String> {
        fingerprint_without(self, "context_fingerprint", "conformal calibration context")
    }

    pub fn validate_for_truth(
        &self,
        truth: &ConformalCalibrationTruth,
        target_names: &[String],
    ) -> Result<()> {
        self.validate()?;
        if self.calibration_cohort.physical_sample_ids != truth.sample_ids
            || self.calibration_cohort.target_names != target_names
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration cohort must exactly bind truth sample ids and targets"
                    .to_string(),
            ));
        }
        Ok(())
    }

    pub fn validate(&self) -> Result<()> {
        for value in [
            &self.predictor_binding_fingerprint,
            &self.source_training_outcome_fingerprint,
            &self.calibration_replay_outcome_fingerprint,
            &self.data_identities_fingerprint,
            &self.fold_set_fingerprint,
            &self.training_influence_fingerprint,
            &self.relation_fingerprint,
            &self.context_fingerprint,
        ] {
            validate_sha256(value)?;
        }
        self.calibration_cohort.validate()?;
        if self.context_fingerprint != self.compute_fingerprint()? {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration context fingerprint does not match TCV1 content".to_string(),
            ));
        }
        Ok(())
    }

    fn validate_for_calibration(&self, calibration: &ConformalCalibration) -> Result<()> {
        self.validate_for_truth(
            &ConformalCalibrationTruth {
                sample_ids: calibration.sample_ids.clone(),
                values: vec![vec![0.0]; calibration.sample_ids.len()],
            },
            &calibration.target_names,
        )
    }
}

impl ConformalCalibrationCohort {
    pub fn compute_fingerprint(&self) -> Result<String> {
        fingerprint_without(self, "manifest_fingerprint", "conformal calibration cohort")
    }

    pub fn validate(&self) -> Result<()> {
        validate_sha256(&self.manifest_fingerprint)?;
        if self.role != "calibration" || self.target_names.is_empty() {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration context requires calibration cohort role and targets"
                    .to_string(),
            ));
        }
        validate_unique_samples(&self.physical_sample_ids)?;
        if self.origin_sample_ids.iter().collect::<BTreeSet<_>>().len()
            != self.origin_sample_ids.len()
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration origin sample ids must be unique".to_string(),
            ));
        }
        if self.manifest_fingerprint != self.compute_fingerprint()? {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration cohort fingerprint does not match TCV1 content".to_string(),
            ));
        }
        Ok(())
    }
}

impl ConformalIntervalBlock {
    /// Validate interval closure against the actual point block and quantiles;
    /// a matching hash alone is never treated as sufficient.
    pub fn validate_against(
        &self,
        calibration: &ConformalCalibration,
        predictions: &PredictionBlock,
    ) -> Result<()> {
        self.validate()?;
        calibration.validate()?;
        if self.binding_id != calibration.binding_id
            || self.calibration_fingerprint != calibration.calibration_fingerprint
            || self.sample_ids != predictions.sample_ids
            || self.point_prediction_fingerprint
                != point_prediction_fingerprint_for_runtime(predictions)?
        {
            return Err(DagMlError::RuntimeValidation("conformal interval block is not bound to its calibration and point prediction block".to_string()));
        }
        let expected = calibration.apply(predictions)?;
        if self != &expected {
            return Err(DagMlError::RuntimeValidation(
                "conformal interval bounds do not close over point predictions and quantiles"
                    .to_string(),
            ));
        }
        Ok(())
    }
}

impl ConformalCalibrationRef {
    pub fn validate(&self) -> Result<()> {
        if self.schema_version != CONFORMAL_RUNTIME_SCHEMA_VERSION
            || self.binding_id.trim().is_empty()
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration reference has an unsupported version or empty binding id"
                    .to_string(),
            ));
        }
        validate_sha256(&self.calibration_fingerprint)
    }

    pub fn validate_against(&self, calibration: &ConformalCalibration) -> Result<()> {
        self.validate()?;
        calibration.validate()?;
        if self.schema_version != CONFORMAL_RUNTIME_SCHEMA_VERSION
            || self.binding_id != calibration.binding_id
            || self.calibration_fingerprint != calibration.calibration_fingerprint
        {
            return Err(DagMlError::RuntimeValidation(
                "conformal calibration reference does not match calibration state".to_string(),
            ));
        }
        Ok(())
    }
}

fn validate_identity_aligned_truth(
    predictions: &PredictionBlock,
    truth: &ConformalCalibrationTruth,
) -> Result<()> {
    if predictions.sample_ids != truth.sample_ids
        || predictions.values.len() != truth.values.len()
        || truth.values.is_empty()
        || truth
            .values
            .iter()
            .any(|row| row.len() != predictions.values[0].len())
        || truth
            .values
            .iter()
            .flatten()
            .any(|value| !value.is_finite())
    {
        return Err(DagMlError::RuntimeValidation(
            "conformal truth must be finite and exactly row/target aligned by sample id"
                .to_string(),
        ));
    }
    Ok(())
}

fn validate_unique_samples(sample_ids: &[SampleId]) -> Result<()> {
    if sample_ids.is_empty() || sample_ids.iter().collect::<BTreeSet<_>>().len() != sample_ids.len()
    {
        return Err(DagMlError::RuntimeValidation(
            "conformal calibration requires non-empty unique sample ids".to_string(),
        ));
    }
    Ok(())
}

fn 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(|typed| typed.fingerprint_without(field))
        .map_err(|error| DagMlError::RuntimeValidation(format!("{label} is outside TCV1: {error}")))
}

fn stabilize_calibration_for_tcv1(
    mut calibration: ConformalCalibration,
) -> Result<ConformalCalibration> {
    // TCV1 fingerprints the lexical binary64 token.  A radius produced by
    // native arithmetic can need one serde round-trip before that token is the
    // same one a strict JSON reader will observe.  Sign only that fixed point;
    // otherwise a newly created calibration can reject its own serialized form.
    calibration.calibration_fingerprint = "0".repeat(64);
    for _ in 0..8 {
        let json = serde_json::to_string(&calibration)?;
        let before = parse_typed_json(&json).map_err(|error| {
            DagMlError::RuntimeValidation(format!(
                "conformal calibration is outside TCV1 while normalizing: {error}"
            ))
        })?;
        let mut normalized = serde_json::from_str::<ConformalCalibration>(&json)?;
        normalized.calibration_fingerprint = "0".repeat(64);
        let normalized_json = serde_json::to_string(&normalized)?;
        let after = parse_typed_json(&normalized_json).map_err(|error| {
            DagMlError::RuntimeValidation(format!(
                "conformal calibration is outside TCV1 after normalization: {error}"
            ))
        })?;
        if before != after {
            calibration = normalized;
            continue;
        }
        normalized.calibration_fingerprint = after
            .fingerprint_without("calibration_fingerprint")
            .map_err(|error| {
            DagMlError::RuntimeValidation(format!(
                "conformal calibration TCV1 fingerprint failed after normalization: {error}"
            ))
        })?;
        let signed_json = serde_json::to_string(&normalized)?;
        return ConformalCalibration::from_json(&signed_json);
    }
    Err(DagMlError::RuntimeValidation(
        "conformal calibration TCV1 JSON did not reach a serde canonical fixed point".to_string(),
    ))
}

pub(crate) fn point_prediction_fingerprint_for_runtime(
    predictions: &PredictionBlock,
) -> Result<String> {
    predictions.validate_content()?;
    fingerprint_without(predictions, "prediction_id", "conformal point prediction")
}

fn validate_sha256(value: &str) -> Result<()> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(DagMlError::RuntimeValidation(
            "conformal calibration fingerprint must be lowercase SHA-256".to_string(),
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::conformal::{ConformalRadius, RegressionIntervalCell};
    use crate::ids::NodeId;
    use crate::oof::PredictionPartition;

    fn block(ids: &[&str], values: &[f64]) -> PredictionBlock {
        PredictionBlock {
            prediction_id: None,
            producer_node: NodeId::new("model:regressor").unwrap(),
            producer_port: Some("prediction".to_string()),
            partition: PredictionPartition::Validation,
            fold_id: None,
            sample_ids: ids.iter().map(|id| SampleId::new(*id).unwrap()).collect(),
            values: values.iter().map(|value| vec![*value]).collect(),
            target_names: vec!["y".to_string()],
        }
    }

    fn context(ids: Vec<SampleId>, targets: Vec<String>) -> ConformalCalibrationContext {
        let mut cohort = ConformalCalibrationCohort {
            role: "calibration".to_string(),
            physical_sample_ids: ids.clone(),
            origin_sample_ids: ids,
            target_names: targets,
            manifest_fingerprint: String::new(),
        };
        cohort.manifest_fingerprint = cohort.compute_fingerprint().unwrap();
        let mut context = ConformalCalibrationContext {
            predictor_binding_fingerprint: "1".repeat(64),
            source_training_outcome_fingerprint: "2".repeat(64),
            calibration_replay_outcome_fingerprint: "3".repeat(64),
            data_identities_fingerprint: "4".repeat(64),
            fold_set_fingerprint: "5".repeat(64),
            training_influence_fingerprint: "6".repeat(64),
            relation_fingerprint: "7".repeat(64),
            calibration_cohort: cohort,
            context_fingerprint: String::new(),
        };
        context.context_fingerprint = context.compute_fingerprint().unwrap();
        context
    }

    #[test]
    fn calibration_round_trips_and_application_preserves_replay_ids() {
        let calibration = ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["y".to_string()],
            &block(&["s1", "s2", "s3"], &[1.0, 3.0, 5.0]),
            &ConformalCalibrationTruth {
                sample_ids: vec![
                    SampleId::new("s1").unwrap(),
                    SampleId::new("s2").unwrap(),
                    SampleId::new("s3").unwrap(),
                ],
                values: vec![vec![0.0], vec![2.0], vec![4.0]],
            },
            context(
                vec![
                    SampleId::new("s1").unwrap(),
                    SampleId::new("s2").unwrap(),
                    SampleId::new("s3").unwrap(),
                ],
                vec!["y".to_string()],
            ),
            vec![0.5],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error,
        )
        .unwrap();
        let json = serde_json::to_string(&calibration).unwrap();
        let loaded = ConformalCalibration::from_json(&json).unwrap();
        let replay = block(&["new:2", "new:1"], &[10.0, 20.0]);
        let intervals = loaded.apply(&replay).unwrap();
        assert_eq!(intervals.sample_ids, replay.sample_ids);
        assert_eq!(intervals.intervals.len(), 1);
        let cell = intervals.intervals[0].cells[0][0];
        assert_eq!(cell.endpoints(), (Some(9.0), Some(11.0)));
    }

    #[test]
    fn calibration_preserves_non_binary_coverage_fingerprint() {
        let calibration = ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["y".to_string()],
            &block(&["s1", "s2", "s3", "s4"], &[57.28, 69.52, 82.78, 97.06]),
            &ConformalCalibrationTruth {
                sample_ids: vec![
                    SampleId::new("s1").unwrap(),
                    SampleId::new("s2").unwrap(),
                    SampleId::new("s3").unwrap(),
                    SampleId::new("s4").unwrap(),
                ],
                values: vec![vec![64.0], vec![81.0], vec![100.0], vec![121.0]],
            },
            context(
                vec![
                    SampleId::new("s1").unwrap(),
                    SampleId::new("s2").unwrap(),
                    SampleId::new("s3").unwrap(),
                    SampleId::new("s4").unwrap(),
                ],
                vec!["y".to_string()],
            ),
            vec![0.8],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error,
        );
        let calibration = calibration.unwrap();
        let json = serde_json::to_string(&calibration).unwrap();
        assert!(ConformalCalibration::from_json(&json).is_ok());
    }

    #[test]
    fn calibration_refuses_order_and_tamper() {
        let prediction = block(&["s1", "s2"], &[1.0, 2.0]);
        assert!(ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["y".to_string()],
            &prediction,
            &ConformalCalibrationTruth {
                sample_ids: vec![SampleId::new("s2").unwrap(), SampleId::new("s1").unwrap()],
                values: vec![vec![1.0], vec![0.0]],
            },
            context(prediction.sample_ids.clone(), vec!["y".to_string()]),
            vec![0.5],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error,
        )
        .is_err());
        assert!(ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["wrong".to_string()],
            &prediction,
            &ConformalCalibrationTruth {
                sample_ids: prediction.sample_ids.clone(),
                values: vec![vec![0.0], vec![1.0]]
            },
            context(prediction.sample_ids.clone(), vec!["wrong".to_string()]),
            vec![0.5],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error
        )
        .is_err());
        let calibration = ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["y".to_string()],
            &prediction,
            &ConformalCalibrationTruth {
                sample_ids: prediction.sample_ids.clone(),
                values: vec![vec![0.0], vec![1.0]],
            },
            context(prediction.sample_ids.clone(), vec!["y".to_string()]),
            vec![0.5],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error,
        )
        .unwrap();
        let mut value = serde_json::to_value(calibration).unwrap();
        value["quantiles"][0]["rank"] = serde_json::json!(1);
        let mut resigned: ConformalCalibration = serde_json::from_value(value.clone()).unwrap();
        resigned.calibration_fingerprint = resigned.compute_fingerprint().unwrap();
        value = serde_json::to_value(resigned).unwrap();
        assert!(ConformalCalibration::from_json(&value.to_string()).is_err());
    }

    #[test]
    fn v2_context_is_required_and_interval_bounds_close_over_points() {
        let prediction = block(&["cal:1", "cal:2"], &[3.0, 7.0]);
        let truth = ConformalCalibrationTruth {
            sample_ids: prediction.sample_ids.clone(),
            values: vec![vec![2.0], vec![5.0]],
        };
        let calibration = ConformalCalibration::calibrate_with_truth(
            "output:main",
            vec!["y".to_string()],
            &prediction,
            &truth,
            context(prediction.sample_ids.clone(), vec!["y".to_string()]),
            vec![0.5],
            ConformalMultiTargetPolicy::Marginal,
            ConformalSmallSamplePolicy::Error,
        )
        .unwrap();
        let replay = block(&["replay:1"], &[10.0]);
        let mut intervals = calibration.apply(&replay).unwrap();
        intervals.validate_against(&calibration, &replay).unwrap();
        intervals.intervals[0].coverage = 0.8;
        assert!(intervals.validate_against(&calibration, &replay).is_err());

        let mut v1 = serde_json::to_value(&calibration).unwrap();
        v1["schema_version"] = serde_json::json!(1);
        assert!(ConformalCalibration::from_json(&v1.to_string()).is_err());
        let mut missing_context = serde_json::to_value(&calibration).unwrap();
        missing_context.as_object_mut().unwrap().remove("context");
        assert!(ConformalCalibration::from_json(&missing_context.to_string()).is_err());
    }

    #[test]
    fn presentation_round_trips_and_refuses_resigned_interval_tampering() {
        let mut presentation = ConformalPresentationV1 {
            schema_version: CONFORMAL_PRESENTATION_SCHEMA_VERSION,
            package_fingerprint: "1".repeat(64),
            replay_outcome_fingerprint: "2".repeat(64),
            binding_id: "output:main".to_string(),
            target_name: "y".to_string(),
            sample_ids: vec![SampleId::new("predict:1").unwrap()],
            point_predictions: vec![10.0],
            intervals: vec![ConformalPresentationInterval {
                coverage: 0.8,
                lower: vec![Some(8.0)],
                upper: vec![Some(12.0)],
                qhat: Some(2.0),
            }],
            calibration_fingerprint: "3".repeat(64),
            presentation_fingerprint: "0".repeat(64),
        };
        presentation.presentation_fingerprint = presentation.compute_fingerprint().unwrap();
        let json = serde_json::to_string(&presentation).unwrap();
        assert_eq!(
            ConformalPresentationV1::from_json(&json).unwrap(),
            presentation
        );

        let mut tampered: ConformalPresentationV1 = serde_json::from_str(&json).unwrap();
        tampered.intervals[0].lower[0] = Some(11.0);
        tampered.presentation_fingerprint = tampered.compute_fingerprint().unwrap();
        assert!(
            ConformalPresentationV1::from_json(&serde_json::to_string(&tampered).unwrap()).is_err()
        );
    }

    #[test]
    fn presentation_v2_round_trips_multitarget_guarantee_and_refuses_tampering() {
        let point_prediction = PredictionBlock {
            prediction_id: Some("prediction:production".to_string()),
            producer_node: NodeId::new("model:regressor").unwrap(),
            producer_port: Some("prediction".to_string()),
            partition: PredictionPartition::Final,
            fold_id: None,
            sample_ids: vec![
                SampleId::new("sample:two").unwrap(),
                SampleId::new("sample:one").unwrap(),
            ],
            values: vec![vec![10.0, 100.0], vec![20.0, 200.0]],
            target_names: vec!["protein".to_string(), "moisture".to_string()],
        };
        let quantiles = vec![SplitConformalQuantile {
            coverage: 0.8,
            rank: 4,
            radii: vec![ConformalRadius::Finite(1.0), ConformalRadius::Finite(2.0)],
        }];
        let intervals = apply_split_absolute_residual(
            &point_prediction.values,
            &quantiles,
            ConformalMultiTargetPolicy::Marginal,
        )
        .unwrap();
        let interval_block = ConformalIntervalBlock {
            schema_version: CONFORMAL_RUNTIME_SCHEMA_VERSION,
            binding_id: "output:main".to_string(),
            sample_ids: point_prediction.sample_ids.clone(),
            intervals,
            calibration_fingerprint: "7".repeat(64),
            point_prediction_fingerprint: point_prediction_fingerprint_for_runtime(
                &point_prediction,
            )
            .unwrap(),
        };
        let mut presentation = ConformalPresentationV2 {
            schema_version: CONFORMAL_PRESENTATION_SCHEMA_VERSION_V2,
            archive_sha256: "1".repeat(64),
            package_fingerprint: "2".repeat(64),
            replay_outcome_fingerprint: "3".repeat(64),
            binding_id: "output:main".to_string(),
            predictor: ConformalPresentationPredictorV2 {
                model_artifact_fingerprint: "4".repeat(64),
                predictor_binding_fingerprint: "5".repeat(64),
                predictor_descriptor_fingerprint: "6".repeat(64),
            },
            dimensions: ConformalPresentationDimensionsV2 {
                sample_count: 2,
                target_count: 2,
            },
            target_names: point_prediction.target_names.clone(),
            sample_ids: point_prediction.sample_ids.clone(),
            point_prediction,
            interval_block,
            guarantee: ConformalPresentationGuaranteeV2 {
                calibration_sample_count: 4,
                multi_target_policy: ConformalMultiTargetPolicy::Marginal,
                small_sample_policy: ConformalSmallSamplePolicy::Error,
                quantiles,
            },
            calibration_fingerprint: "7".repeat(64),
            presentation_fingerprint: "0".repeat(64),
        };
        presentation.presentation_fingerprint = presentation.compute_fingerprint().unwrap();
        let json = serde_json::to_string(&presentation).unwrap();
        assert_eq!(
            ConformalPresentationV2::from_json(&json).unwrap(),
            presentation
        );

        let mut reordered = presentation.clone();
        reordered.sample_ids.swap(0, 1);
        reordered.presentation_fingerprint = reordered.compute_fingerprint().unwrap();
        assert!(reordered
            .validate()
            .unwrap_err()
            .to_string()
            .contains("ordered identities"));

        let mut altered_interval = presentation;
        altered_interval.interval_block.intervals[0].cells[0][0] = RegressionIntervalCell::Finite {
            lower: 9.5,
            upper: 11.0,
        };
        altered_interval.presentation_fingerprint = altered_interval.compute_fingerprint().unwrap();
        assert!(altered_interval
            .validate()
            .unwrap_err()
            .to_string()
            .contains("persisted guarantee"));
    }
}