lodestone_types/

lifecycle_completion.rs

//! External content-key and aggregate-then-delete contracts.
//!
//! This module freezes current-format authority and failure semantics. It does
//! not encrypt an object, contact a resolver, retire raw data, register a public
//! operation, or make either lifecycle capability runtime reachable.

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::num::NonZeroU128;

use serde::de::{DeserializeOwned, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use crate::{
    ClusterRevision, CollectionIndexRef, FieldPathDigestV1, ManifestGeneration, PhysicalObjectId, PolicyGeneration,
    RootSetRef, SchemaGeneration, SeqNo, ShardId, SourceCutV1, SourceIncarnation,
};

/// Current aggregate-then-delete contract and object-codec format.
pub const LIFECYCLE_COMPLETION_FORMAT_VERSION: u16 = 2;
/// Current external-key binding and encrypted-object metadata format.
pub const EXTERNAL_KEY_FORMAT_VERSION: u16 = 3;
pub const MAX_LIFECYCLE_CONTRACT_BYTES: usize = 1_048_576;
pub const MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES: usize = 4 * 1_024;
pub const MAX_ROLLUP_DIMENSIONS: usize = 32;
pub const MAX_ROLLUP_MEASURES: usize = 64;
pub const MAX_ROLLUP_COVERAGES: usize = 4_096;
pub const MAX_ROLLUP_SHARDS: usize = 512;
pub const MAX_ROLLUP_OUTPUT_ROWS: usize = 262_144;
pub const MAX_ROLLUP_OUTPUT_BYTES: usize = 16 * 1_024 * 1_024;
pub const MAX_ROLLUP_SCALAR_BYTES: usize = 4_096;
pub const MAX_ROLLUP_BUCKET_MILLIS: u64 = 366 * 24 * 60 * 60 * 1_000;
pub const MAX_CONTENT_KEY_CACHE_MILLIS: u64 = 300_000;
pub const ROLLUP_QUERY_CONTRACT_DIGEST_V1: [u8; 32] = [
    0x7c, 0x3e, 0x41, 0x11, 0x81, 0xaa, 0x97, 0x39, 0x2d, 0x63, 0x22, 0xcd, 0xd8, 0xf0, 0x02, 0x07, 0xea, 0x5f, 0x06,
    0xa7, 0x90, 0x6b, 0x28, 0x81, 0x42, 0x89, 0x9c, 0xb8, 0x02, 0x17, 0x57, 0x80,
];

fn digest_is_nonzero(value: &[u8; 32]) -> bool {
    value.iter().any(|byte| *byte != 0)
}

fn require_digest(value: &[u8; 32], name: &'static str) -> Result<(), LifecycleContractError> {
    if digest_is_nonzero(value) {
        Ok(())
    } else {
        Err(LifecycleContractError::Violation(name))
    }
}

macro_rules! semantic_digest {
    ($name:ident, $description:literal) => {
        #[doc = $description]
        #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
        #[serde(transparent)]
        pub struct $name([u8; 32]);

        impl $name {
            pub fn new(value: [u8; 32]) -> Result<Self, LifecycleContractError> {
                require_digest(&value, concat!(stringify!($name), " cannot be zero"))?;
                Ok(Self(value))
            }

            #[must_use]
            pub const fn get(self) -> [u8; 32] {
                self.0
            }

            fn validate(self) -> Result<(), LifecycleContractError> {
                require_digest(&self.0, concat!(stringify!($name), " cannot be zero"))
            }
        }
    };
}

macro_rules! nonzero_decimal_identity {
    ($name:ident, $description:literal) => {
        #[doc = $description]
        #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
        pub struct $name(NonZeroU128);

        impl $name {
            pub fn new(value: u128) -> Result<Self, LifecycleContractError> {
                NonZeroU128::new(value)
                    .map(Self)
                    .ok_or(LifecycleContractError::ZeroIdentity(stringify!($name)))
            }

            #[must_use]
            pub const fn get(self) -> u128 {
                self.0.get()
            }
        }

        impl Display for $name {
            fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                Display::fmt(&self.get(), formatter)
            }
        }

        impl Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
            where
                S: Serializer,
            {
                serializer.serialize_str(&self.get().to_string())
            }
        }

        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: Deserializer<'de>,
            {
                struct CanonicalDecimalVisitor;

                impl Visitor<'_> for CanonicalDecimalVisitor {
                    type Value = $name;

                    fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                        formatter.write_str("a canonical nonzero u128 decimal string")
                    }

                    fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
                    where
                        E: serde::de::Error,
                    {
                        let bytes = value.as_bytes();
                        if bytes.is_empty()
                            || !matches!(bytes[0], b'1'..=b'9')
                            || !bytes[1..].iter().all(u8::is_ascii_digit)
                        {
                            return Err(E::custom("expected a canonical nonzero u128 decimal string"));
                        }
                        let parsed = value
                            .parse::<u128>()
                            .map_err(|_| E::custom("identity exceeds u128"))?;
                        NonZeroU128::new(parsed)
                            .map($name)
                            .ok_or_else(|| E::custom("identity cannot be zero"))
                    }
                }

                deserializer.deserialize_str(CanonicalDecimalVisitor)
            }
        }
    };
}

semantic_digest!(
    ExternalKeyContextDigestV1,
    "Digest of the exact non-secret external-key context."
);
semantic_digest!(
    RollupQueryContractDigestV1,
    "Digest of the admitted query shapes proven by the rollup planner."
);
semantic_digest!(
    RollupCoverageDigestV1,
    "Digest of the exact coverage/root state proven safe for routing."
);
nonzero_decimal_identity!(ContentKeyIdV1, "Never-reused identity of one content-encryption key.");
nonzero_decimal_identity!(
    ExternalKeyBindingIdV1,
    "Never-reused identity of one opaque index-authorized external-key binding."
);
nonzero_decimal_identity!(
    RollupPolicyIdV1,
    "Never-reused identity of one aggregate-then-delete policy."
);

/// Only the current authenticated document-object algorithm is admitted.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DocumentEncryptionAlgorithmV1 {
    Aes256Gcm,
}

/// Exact non-secret context authenticated by the external resolver.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ExternalKeyContextV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub content_key_id: ContentKeyIdV1,
    pub content_key_generation: u64,
    pub binding_id: ExternalKeyBindingIdV1,
    pub binding_generation: u64,
    pub access_generation: u64,
    pub object_context_digest: [u8; 32],
}

impl ExternalKeyContextV1 {
    pub fn validate(self) -> Result<(), LifecycleContractError> {
        validate_external_key_format(self.format_version)?;
        if self.content_key_generation == 0 || self.binding_generation == 0 || self.access_generation == 0 {
            return Err(LifecycleContractError::Violation(
                "external-key generations cannot be zero",
            ));
        }
        require_digest(
            &self.object_context_digest,
            "external-key object context cannot be zero",
        )
    }

    pub fn digest(self) -> Result<ExternalKeyContextDigestV1, LifecycleContractError> {
        self.validate()?;
        let bytes = serde_json::to_vec(&self).map_err(LifecycleContractError::serde)?;
        ExternalKeyContextDigestV1::new(*blake3::hash(&bytes).as_bytes())
    }
}

/// Opaque external content-key binding. Core never interprets its bytes.
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct IndexExternalKeyBindingV1 {
    pub format_version: u16,
    pub binding_id: ExternalKeyBindingIdV1,
    pub collection_index_ref: CollectionIndexRef,
    pub content_key_id: ContentKeyIdV1,
    pub content_key_generation: u64,
    pub binding_generation: u64,
    pub access_generation: u64,
    pub opaque_binding: Vec<u8>,
    pub context: ExternalKeyContextV1,
    pub context_digest: ExternalKeyContextDigestV1,
    pub created_at_revision: ClusterRevision,
}

impl fmt::Debug for IndexExternalKeyBindingV1 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("IndexExternalKeyBindingV1")
            .field("format_version", &self.format_version)
            .field("binding_id", &self.binding_id)
            .field("collection_index_ref", &self.collection_index_ref)
            .field("content_key_id", &self.content_key_id)
            .field("content_key_generation", &self.content_key_generation)
            .field("binding_generation", &self.binding_generation)
            .field("access_generation", &self.access_generation)
            .field("opaque_binding", &"[redacted]")
            .field("context", &self.context)
            .field("context_digest", &self.context_digest)
            .field("created_at_revision", &self.created_at_revision)
            .finish()
    }
}

impl IndexExternalKeyBindingV1 {
    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_external_key_format(self.format_version)?;
        if self.content_key_generation == 0
            || self.binding_generation == 0
            || self.access_generation == 0
            || self.created_at_revision.get() == 0
        {
            return Err(LifecycleContractError::Violation(
                "external-key binding requires nonzero generations",
            ));
        }
        if self.opaque_binding.is_empty()
            || self.opaque_binding.len() > MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES
            || self.opaque_binding.iter().all(|byte| *byte == 0)
        {
            return Err(LifecycleContractError::Budget {
                field: "opaque external-key binding bytes",
                observed: self.opaque_binding.len() as u64,
                maximum: MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES as u64,
            });
        }
        self.context.validate()?;
        self.context_digest.validate()?;
        if self.context.collection_index_ref != self.collection_index_ref
            || self.context.content_key_id != self.content_key_id
            || self.context.content_key_generation != self.content_key_generation
            || self.context.binding_id != self.binding_id
            || self.context.binding_generation != self.binding_generation
            || self.context.access_generation != self.access_generation
            || self.context.digest()? != self.context_digest
        {
            return Err(LifecycleContractError::Violation(
                "external-key binding and context identities differ",
            ));
        }
        Ok(())
    }
}

/// Immutable document-store metadata for one encrypted object. Mutable
/// resolver binding and access generations intentionally do not appear here:
/// rotating them must not require rewriting immutable ciphertext.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EncryptedObjectKeyRefV1 {
    pub format_version: u16,
    pub physical_object: PhysicalObjectId,
    pub algorithm: DocumentEncryptionAlgorithmV1,
    pub content_key_id: ContentKeyIdV1,
    pub content_key_generation: u64,
    pub nonce: [u8; 12],
    pub authenticated_context_digest: [u8; 32],
    pub ciphertext_bytes: u64,
}

impl EncryptedObjectKeyRefV1 {
    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_external_key_format(self.format_version)?;
        self.physical_object
            .validate()
            .map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
        if self.content_key_generation == 0 || self.ciphertext_bytes < 16 {
            return Err(LifecycleContractError::Violation(
                "encrypted objects require a nonzero key generation and an authentication tag",
            ));
        }
        if self.nonce != object_nonce_v1(self.content_key_id, self.content_key_generation, &self.physical_object)? {
            return Err(LifecycleContractError::Violation(
                "encrypted-object nonce is not derived from its never-reused physical identity",
            ));
        }
        let expected = object_authenticated_context_digest_v1(
            self.algorithm,
            self.content_key_id,
            self.content_key_generation,
            &self.physical_object,
            self.ciphertext_bytes,
        )?;
        if self.authenticated_context_digest != expected {
            return Err(LifecycleContractError::Violation(
                "encrypted-object authenticated context does not exact-match metadata",
            ));
        }
        Ok(())
    }
}

/// Derives a deterministic 96-bit nonce from a never-reused physical object
/// identity and content-key generation. Re-encrypting another payload under
/// the same physical identity is forbidden by immutable-object publication.
pub fn object_nonce_v1(
    content_key_id: ContentKeyIdV1,
    content_key_generation: u64,
    physical_object: &PhysicalObjectId,
) -> Result<[u8; 12], LifecycleContractError> {
    if content_key_generation == 0 {
        return Err(LifecycleContractError::Violation(
            "content-key generation cannot be zero",
        ));
    }
    physical_object
        .validate()
        .map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/p13/object-nonce/v1\0");
    hasher.update(&content_key_id.get().to_be_bytes());
    hasher.update(&content_key_generation.to_be_bytes());
    hasher.update(&serde_json::to_vec(physical_object).map_err(LifecycleContractError::serde)?);
    let digest = hasher.finalize();
    let mut nonce = [0_u8; 12];
    nonce.copy_from_slice(&digest.as_bytes()[..12]);
    Ok(nonce)
}

pub fn object_authenticated_context_digest_v1(
    algorithm: DocumentEncryptionAlgorithmV1,
    content_key_id: ContentKeyIdV1,
    content_key_generation: u64,
    physical_object: &PhysicalObjectId,
    ciphertext_bytes: u64,
) -> Result<[u8; 32], LifecycleContractError> {
    if content_key_generation == 0 || ciphertext_bytes < 16 {
        return Err(LifecycleContractError::Violation(
            "authenticated context requires a key generation and ciphertext tag",
        ));
    }
    physical_object
        .validate()
        .map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
    let binding = (
        EXTERNAL_KEY_FORMAT_VERSION,
        algorithm,
        content_key_id,
        content_key_generation,
        physical_object,
        ciphertext_bytes,
    );
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/p13/object-aad/v1\0");
    hasher.update(&serde_json::to_vec(&binding).map_err(LifecycleContractError::serde)?);
    Ok(*hasher.finalize().as_bytes())
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RollupTimeZoneV1 {
    Utc,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupMeasureV1 {
    Count,
    Sum { field_path_digest: FieldPathDigestV1 },
    Min { field_path_digest: FieldPathDigestV1 },
    Max { field_path_digest: FieldPathDigestV1 },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupPolicyV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub policy_id: RollupPolicyIdV1,
    pub policy_generation: PolicyGeneration,
    /// Explicit acknowledgement that incompatible detail queries will fail
    /// after this policy retires their raw coverage.
    pub detail_retirement_acknowledged: bool,
    pub time_zone: RollupTimeZoneV1,
    pub timestamp_field_digest: FieldPathDigestV1,
    pub bucket_width_millis: u64,
    pub bucket_alignment_millis: u64,
    pub dimensions: Vec<FieldPathDigestV1>,
    pub measures: Vec<RollupMeasureV1>,
}

impl RollupPolicyV1 {
    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_format(self.format_version)?;
        if self.policy_generation.get() == 0
            || !self.detail_retirement_acknowledged
            || self.bucket_width_millis == 0
            || self.bucket_width_millis > MAX_ROLLUP_BUCKET_MILLIS
            || self.bucket_alignment_millis >= self.bucket_width_millis
        {
            return Err(LifecycleContractError::Violation(
                "rollup generation or bucket geometry is outside the current contract",
            ));
        }
        require_digest(
            &self.timestamp_field_digest.get(),
            "rollup timestamp-field digest cannot be zero",
        )?;
        if self.dimensions.len() > MAX_ROLLUP_DIMENSIONS
            || self.dimensions.windows(2).any(|pair| pair[0] >= pair[1])
            || self.measures.is_empty()
            || self.measures.len() > MAX_ROLLUP_MEASURES
            || self.measures.windows(2).any(|pair| pair[0] >= pair[1])
        {
            return Err(LifecycleContractError::Violation(
                "rollup dimensions and measures must be bounded, unique, and canonically ordered",
            ));
        }
        for dimension in &self.dimensions {
            require_digest(&dimension.get(), "rollup dimension digest cannot be zero")?;
        }
        for measure in &self.measures {
            match measure {
                RollupMeasureV1::Count => {}
                RollupMeasureV1::Sum { field_path_digest }
                | RollupMeasureV1::Min { field_path_digest }
                | RollupMeasureV1::Max { field_path_digest } => {
                    require_digest(&field_path_digest.get(), "rollup measure digest cannot be zero")?;
                }
            }
        }
        Ok(())
    }

    pub fn boundary_aligned(&self, value: u64) -> bool {
        value >= self.bucket_alignment_millis
            && (value - self.bucket_alignment_millis).is_multiple_of(self.bucket_width_millis)
    }
}

/// Immutable aggregate-then-delete input fence. This is embedded in the sole
/// leased maintenance task and deliberately excludes staged output and
/// resumable progress.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupWorkPlanV1 {
    pub format_version: u16,
    pub policy: RollupPolicyV1,
    pub schema_generation: SchemaGeneration,
    pub source_incarnation: Option<SourceIncarnation>,
    pub source_cut: Option<SourceCutV1>,
    pub source_generations: Vec<RollupSourceGenerationV1>,
    pub range_start_millis: u64,
    pub range_end_millis: u64,
    pub late_data_watermark_millis: u64,
}

impl RollupWorkPlanV1 {
    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_format(self.format_version)?;
        self.policy.validate()?;
        if self.schema_generation.get() == 0
            || self.range_start_millis >= self.range_end_millis
            || !self.policy.boundary_aligned(self.range_start_millis)
            || !self.policy.boundary_aligned(self.range_end_millis)
            || self.late_data_watermark_millis < self.range_end_millis
        {
            return Err(LifecycleContractError::Violation(
                "rollup work plan schema, aligned range, or late-data watermark is invalid",
            ));
        }
        if self.source_generations.is_empty()
            || self.source_generations.len() > MAX_ROLLUP_SHARDS
            || self
                .source_generations
                .windows(2)
                .any(|pair| pair[0].shard_id >= pair[1].shard_id)
            || self
                .source_generations
                .iter()
                .any(|source| source.manifest_generation.get() == 0)
        {
            return Err(LifecycleContractError::Violation(
                "rollup source generations must be complete, bounded, and canonically ordered",
            ));
        }
        match (&self.source_incarnation, &self.source_cut) {
            (None, None) => Ok(()),
            (Some(incarnation), Some(cut)) if *incarnation == cut.source_incarnation => cut
                .validate()
                .map_err(|error| LifecycleContractError::Nested(error.to_string())),
            _ => Err(LifecycleContractError::Violation(
                "replayable-source rollup work requires an exact matching source cut",
            )),
        }
    }

    pub fn digest(&self) -> Result<[u8; 32], LifecycleContractError> {
        self.validate()?;
        let bytes = serde_json::to_vec(self).map_err(LifecycleContractError::serde)?;
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/p13/rollup-work-plan/v1\0");
        hasher.update(&bytes);
        Ok(*hasher.finalize().as_bytes())
    }
}

/// Canonical dimension value retained in a rollup row. Floating-point values
/// are intentionally excluded so grouping and replay remain byte-exact.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupScalarV1 {
    Null,
    Bool(bool),
    I64(i64),
    U64(u64),
    String(String),
}

impl RollupScalarV1 {
    fn validate(&self) -> Result<(), LifecycleContractError> {
        if matches!(self, Self::String(value) if value.len() > MAX_ROLLUP_SCALAR_BYTES) {
            return Err(LifecycleContractError::Budget {
                field: "rollup dimension scalar bytes",
                observed: match self {
                    Self::String(value) => value.len() as u64,
                    _ => 0,
                },
                maximum: MAX_ROLLUP_SCALAR_BYTES as u64,
            });
        }
        Ok(())
    }
}

/// One measure cell in policy order. The tag prevents a count from being
/// confused with an arithmetically equal sum/min/max during replay.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupAggregateValueV1 {
    Count(u64),
    Sum(i64),
    Min(i64),
    Max(i64),
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupOutputRowV1 {
    pub bucket_start_millis: u64,
    pub dimensions: Vec<RollupScalarV1>,
    pub measures: Vec<RollupAggregateValueV1>,
}

/// Current-format deterministic aggregate object. It binds the complete work
/// plan rather than becoming a second policy, source, or coverage authority.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupOutputV1 {
    pub format_version: u16,
    pub work_plan_digest: [u8; 32],
    pub rows: Vec<RollupOutputRowV1>,
    pub input_documents: u64,
    pub input_bytes: u64,
}

impl RollupOutputV1 {
    pub fn validate(&self, plan: &RollupWorkPlanV1) -> Result<(), LifecycleContractError> {
        validate_format(self.format_version)?;
        plan.validate()?;
        if self.work_plan_digest != plan.digest()?
            || self.rows.is_empty()
            || self.rows.len() > MAX_ROLLUP_OUTPUT_ROWS
            || u64::try_from(self.rows.len()).is_err()
            || u64::try_from(self.rows.len()).is_ok_and(|rows| rows > self.input_documents)
            || self.input_documents == 0
            || self.input_bytes == 0
        {
            return Err(LifecycleContractError::Violation(
                "rollup output does not bind the exact bounded work plan and input",
            ));
        }
        let mut previous_key: Option<(u64, &[RollupScalarV1])> = None;
        let count_position = plan
            .policy
            .measures
            .iter()
            .position(|measure| *measure == RollupMeasureV1::Count);
        let mut counted_documents = 0_u64;
        for row in &self.rows {
            if row.bucket_start_millis < plan.range_start_millis
                || row.bucket_start_millis >= plan.range_end_millis
                || !plan.policy.boundary_aligned(row.bucket_start_millis)
                || row.dimensions.len() != plan.policy.dimensions.len()
                || row.measures.len() != plan.policy.measures.len()
                || row.dimensions.iter().any(|value| value.validate().is_err())
                || row.measures.iter().zip(&plan.policy.measures).any(|(value, measure)| {
                    !matches!(
                        (value, measure),
                        (RollupAggregateValueV1::Count(_), RollupMeasureV1::Count)
                            | (RollupAggregateValueV1::Sum(_), RollupMeasureV1::Sum { .. })
                            | (RollupAggregateValueV1::Min(_), RollupMeasureV1::Min { .. })
                            | (RollupAggregateValueV1::Max(_), RollupMeasureV1::Max { .. })
                    )
                })
                || previous_key.is_some_and(|prior| prior >= (row.bucket_start_millis, &row.dimensions))
            {
                return Err(LifecycleContractError::Violation(
                    "rollup rows must match policy shape and be canonically ordered",
                ));
            }
            if let Some(position) = count_position {
                let RollupAggregateValueV1::Count(count) = row.measures[position] else {
                    return Err(LifecycleContractError::Violation(
                        "rollup count measure does not match the current policy",
                    ));
                };
                if count == 0 {
                    return Err(LifecycleContractError::Violation(
                        "rollup groups cannot contain an empty count",
                    ));
                }
                counted_documents = counted_documents
                    .checked_add(count)
                    .ok_or(LifecycleContractError::Violation("rollup count total overflowed"))?;
            }
            previous_key = Some((row.bucket_start_millis, &row.dimensions));
        }
        if count_position.is_some() && counted_documents != self.input_documents {
            return Err(LifecycleContractError::Violation(
                "rollup count total does not match the exact input document count",
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupSourceGenerationV1 {
    pub shard_id: ShardId,
    pub manifest_generation: ManifestGeneration,
    pub correction_through: SeqNo,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupRoutingProofV1 {
    pub query_contract_digest: RollupQueryContractDigestV1,
    pub coverage_digest: RollupCoverageDigestV1,
    pub proven_at_revision: ClusterRevision,
}

impl RollupRoutingProofV1 {
    pub fn validate(self) -> Result<(), LifecycleContractError> {
        self.query_contract_digest.validate()?;
        self.coverage_digest.validate()?;
        if self.query_contract_digest != rollup_query_contract_digest_v1()? || self.proven_at_revision.get() == 0 {
            return Err(LifecycleContractError::Violation(
                "rollup routing proof requires the current exact query contract and a nonzero catalog revision",
            ));
        }
        Ok(())
    }
}

/// Digest of the sole current planner semantics that may authorize raw-data
/// retirement. A nonzero caller-selected digest is not sufficient evidence.
pub fn rollup_query_contract_digest_v1() -> Result<RollupQueryContractDigestV1, LifecycleContractError> {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/p13/rollup-query-contract/v1\0");
    hasher.update(
        b"utc-fixed-buckets;closed-count-sum-min-max;half-open-gap-free-non-overlapping;exact-match-or-raw-unavailable",
    );
    let observed = *hasher.finalize().as_bytes();
    if observed != ROLLUP_QUERY_CONTRACT_DIGEST_V1 {
        return Err(LifecycleContractError::Violation(
            "rollup query semantics changed without a current-format digest revision",
        ));
    }
    RollupQueryContractDigestV1::new(observed)
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RawRetirementStateV1 {
    RawPresent,
    RoutingProven,
    RawRetired,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupCoverageV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub policy: RollupPolicyV1,
    pub schema_generation: SchemaGeneration,
    pub source_incarnation: Option<SourceIncarnation>,
    pub source_cut: Option<SourceCutV1>,
    pub source_generations: Vec<RollupSourceGenerationV1>,
    pub range_start_millis: u64,
    pub range_end_millis: u64,
    pub late_data_watermark_millis: u64,
    pub output_root: RootSetRef,
    pub raw_retirement_state: RawRetirementStateV1,
    pub routing_proof: Option<RollupRoutingProofV1>,
}

impl RollupCoverageV1 {
    pub fn coverage_digest(&self) -> Result<RollupCoverageDigestV1, LifecycleContractError> {
        let binding = (
            self.format_version,
            self.collection_index_ref,
            &self.policy,
            self.schema_generation,
            self.source_incarnation,
            &self.source_cut,
            &self.source_generations,
            self.range_start_millis,
            self.range_end_millis,
            self.late_data_watermark_millis,
            &self.output_root,
        );
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/p13/rollup-coverage/v1\0");
        hasher.update(&serde_json::to_vec(&binding).map_err(LifecycleContractError::serde)?);
        RollupCoverageDigestV1::new(*hasher.finalize().as_bytes())
    }

    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_format(self.format_version)?;
        self.policy.validate()?;
        if self.policy.collection_index_ref != self.collection_index_ref
            || self.schema_generation.get() == 0
            || self.range_start_millis >= self.range_end_millis
            || !self.policy.boundary_aligned(self.range_start_millis)
            || !self.policy.boundary_aligned(self.range_end_millis)
            || self.late_data_watermark_millis < self.range_end_millis
        {
            return Err(LifecycleContractError::Violation(
                "rollup coverage scope, schema, aligned range, or late-data watermark is invalid",
            ));
        }
        if self.source_generations.is_empty()
            || self.source_generations.len() > MAX_ROLLUP_SHARDS
            || self
                .source_generations
                .windows(2)
                .any(|pair| pair[0].shard_id >= pair[1].shard_id)
            || self
                .source_generations
                .iter()
                .any(|source| source.manifest_generation.get() == 0)
        {
            return Err(LifecycleContractError::Violation(
                "rollup source generations must be complete, bounded, and canonically ordered",
            ));
        }
        match (&self.source_incarnation, &self.source_cut) {
            (None, None) => {}
            (Some(incarnation), Some(cut)) if *incarnation == cut.source_incarnation => cut
                .validate()
                .map_err(|error| LifecycleContractError::Nested(error.to_string()))?,
            _ => {
                return Err(LifecycleContractError::Violation(
                    "replayable-source rollup coverage requires an exact matching source cut",
                ));
            }
        }
        self.output_root
            .validate()
            .map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
        if self.output_root.item_count != 1
            || self.output_root.page_count != 1
            || self.output_root.schema_generations.as_slice() != [self.schema_generation]
        {
            return Err(LifecycleContractError::Violation(
                "current rollup coverage requires one output object for the exact schema generation",
            ));
        }
        match (self.raw_retirement_state, self.routing_proof) {
            (RawRetirementStateV1::RawPresent, None) => {}
            (RawRetirementStateV1::RoutingProven | RawRetirementStateV1::RawRetired, Some(proof)) => {
                proof.validate()?;
                if proof.coverage_digest != self.coverage_digest()? {
                    return Err(LifecycleContractError::Violation(
                        "raw-retirement proof does not bind the exact current coverage and output root",
                    ));
                }
            }
            _ => {
                return Err(LifecycleContractError::Violation(
                    "raw retirement requires an exact planner routing proof",
                ));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupCoverageLedgerV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub revision: u64,
    pub coverages: Vec<RollupCoverageV1>,
}

impl RollupCoverageLedgerV1 {
    pub fn validate(&self) -> Result<(), LifecycleContractError> {
        validate_format(self.format_version)?;
        if self.revision == 0 || self.coverages.len() > MAX_ROLLUP_COVERAGES {
            return Err(LifecycleContractError::Budget {
                field: "rollup coverage rows or revision",
                observed: self.coverages.len() as u64,
                maximum: MAX_ROLLUP_COVERAGES as u64,
            });
        }
        let mut prior_end = None;
        for coverage in &self.coverages {
            coverage.validate()?;
            if coverage.collection_index_ref != self.collection_index_ref
                || prior_end.is_some_and(|end| end > coverage.range_start_millis)
                || coverage
                    .routing_proof
                    .is_some_and(|proof| proof.proven_at_revision.get() > self.revision)
            {
                return Err(LifecycleContractError::Violation(
                    "rollup coverages must be non-overlapping and canonically ordered for one index",
                ));
            }
            prior_end = Some(coverage.range_end_millis);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RollupQueryShapeV1 {
    pub range_start_millis: u64,
    pub range_end_millis: u64,
    pub time_zone: RollupTimeZoneV1,
    pub bucket_width_millis: u64,
    pub bucket_alignment_millis: u64,
    pub dimensions: Vec<FieldPathDigestV1>,
    pub measures: Vec<RollupMeasureV1>,
}

impl RollupQueryShapeV1 {
    fn matches(&self, policy: &RollupPolicyV1) -> bool {
        self.time_zone == policy.time_zone
            && self.bucket_width_millis == policy.bucket_width_millis
            && self.bucket_alignment_millis == policy.bucket_alignment_millis
            && self.dimensions == policy.dimensions
            && self.measures == policy.measures
            && policy.boundary_aligned(self.range_start_millis)
            && policy.boundary_aligned(self.range_end_millis)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RollupRouteKindV1 {
    Raw,
    Rollup,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RollupRouteSliceV1 {
    pub range_start_millis: u64,
    pub range_end_millis: u64,
    pub kind: RollupRouteKindV1,
    /// Exact coverage/output-root identity for a rollup slice. Raw slices do
    /// not carry a coverage identity.
    pub rollup_coverage_digest: Option<RollupCoverageDigestV1>,
}

/// Builds an exact, gap-free, non-overlapping route. A query whose semantics
/// do not match already-retired raw coverage refuses instead of approximating.
pub fn route_rollup_query_v1(
    ledger: &RollupCoverageLedgerV1,
    query: &RollupQueryShapeV1,
) -> Result<Vec<RollupRouteSliceV1>, LifecycleContractError> {
    ledger.validate()?;
    if query.range_start_millis >= query.range_end_millis {
        return Err(LifecycleContractError::Violation(
            "rollup query range is empty or reversed",
        ));
    }
    let mut slices = Vec::new();
    let mut cursor = query.range_start_millis;
    for coverage in &ledger.coverages {
        if coverage.range_end_millis <= cursor || coverage.range_start_millis >= query.range_end_millis {
            continue;
        }
        if cursor < coverage.range_start_millis {
            push_route_slice(
                &mut slices,
                cursor,
                coverage.range_start_millis.min(query.range_end_millis),
                RollupRouteKindV1::Raw,
                None,
            );
            cursor = coverage.range_start_millis.min(query.range_end_millis);
        }
        if cursor >= query.range_end_millis {
            break;
        }
        let end = coverage.range_end_millis.min(query.range_end_millis);
        let exact_rollup = query.matches(&coverage.policy);
        if !exact_rollup && coverage.raw_retirement_state == RawRetirementStateV1::RawRetired {
            return Err(LifecycleContractError::DetailRetiredByPolicy);
        }
        let rollup_coverage_digest = exact_rollup.then(|| coverage.coverage_digest()).transpose()?;
        push_route_slice(
            &mut slices,
            cursor.max(coverage.range_start_millis),
            end,
            if exact_rollup {
                RollupRouteKindV1::Rollup
            } else {
                RollupRouteKindV1::Raw
            },
            rollup_coverage_digest,
        );
        cursor = end;
    }
    if cursor < query.range_end_millis {
        push_route_slice(
            &mut slices,
            cursor,
            query.range_end_millis,
            RollupRouteKindV1::Raw,
            None,
        );
    }
    if slices.is_empty()
        || slices[0].range_start_millis != query.range_start_millis
        || slices
            .last()
            .is_none_or(|slice| slice.range_end_millis != query.range_end_millis)
        || slices
            .windows(2)
            .any(|pair| pair[0].range_end_millis != pair[1].range_start_millis)
        || slices
            .iter()
            .any(|slice| matches!(slice.kind, RollupRouteKindV1::Raw) != slice.rollup_coverage_digest.is_none())
    {
        return Err(LifecycleContractError::Violation(
            "rollup routing did not cover the query exactly once",
        ));
    }
    Ok(slices)
}

fn push_route_slice(
    slices: &mut Vec<RollupRouteSliceV1>,
    start: u64,
    end: u64,
    kind: RollupRouteKindV1,
    rollup_coverage_digest: Option<RollupCoverageDigestV1>,
) {
    if start >= end {
        return;
    }
    if let Some(previous) = slices.last_mut()
        && previous.kind == kind
        && previous.range_end_millis == start
        && previous.rollup_coverage_digest == rollup_coverage_digest
    {
        previous.range_end_millis = end;
        return;
    }
    slices.push(RollupRouteSliceV1 {
        range_start_millis: start,
        range_end_millis: end,
        kind,
        rollup_coverage_digest,
    });
}

pub fn encode_encrypted_object_key_ref_v1(value: &EncryptedObjectKeyRefV1) -> Result<Vec<u8>, LifecycleContractError> {
    encode_current(value, EncryptedObjectKeyRefV1::validate)
}

pub fn decode_encrypted_object_key_ref_v1(bytes: &[u8]) -> Result<EncryptedObjectKeyRefV1, LifecycleContractError> {
    decode_current(bytes, EncryptedObjectKeyRefV1::validate)
}

pub fn encode_index_external_key_binding_v1(
    value: &IndexExternalKeyBindingV1,
) -> Result<Vec<u8>, LifecycleContractError> {
    encode_current(value, IndexExternalKeyBindingV1::validate)
}

pub fn decode_index_external_key_binding_v1(bytes: &[u8]) -> Result<IndexExternalKeyBindingV1, LifecycleContractError> {
    decode_current(bytes, IndexExternalKeyBindingV1::validate)
}

pub fn encode_rollup_policy_v1(value: &RollupPolicyV1) -> Result<Vec<u8>, LifecycleContractError> {
    encode_current(value, RollupPolicyV1::validate)
}

pub fn decode_rollup_policy_v1(bytes: &[u8]) -> Result<RollupPolicyV1, LifecycleContractError> {
    decode_current(bytes, RollupPolicyV1::validate)
}

pub fn encode_rollup_output_v1(
    value: &RollupOutputV1,
    plan: &RollupWorkPlanV1,
) -> Result<Vec<u8>, LifecycleContractError> {
    value.validate(plan)?;
    let encoded = serde_json::to_vec(value).map_err(LifecycleContractError::serde)?;
    if encoded.len() > MAX_ROLLUP_OUTPUT_BYTES {
        return Err(LifecycleContractError::Budget {
            field: "rollup output bytes",
            observed: encoded.len() as u64,
            maximum: MAX_ROLLUP_OUTPUT_BYTES as u64,
        });
    }
    Ok(encoded)
}

pub fn decode_rollup_output_v1(
    bytes: &[u8],
    plan: &RollupWorkPlanV1,
) -> Result<RollupOutputV1, LifecycleContractError> {
    if bytes.is_empty() || bytes.len() > MAX_ROLLUP_OUTPUT_BYTES {
        return Err(LifecycleContractError::Budget {
            field: "rollup output bytes",
            observed: bytes.len() as u64,
            maximum: MAX_ROLLUP_OUTPUT_BYTES as u64,
        });
    }
    let value: RollupOutputV1 = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
    value.validate(plan)?;
    let canonical = serde_json::to_vec(&value).map_err(LifecycleContractError::serde)?;
    if canonical != bytes {
        return Err(LifecycleContractError::Violation(
            "rollup output is not the canonical current-format encoding",
        ));
    }
    Ok(value)
}

pub fn encode_rollup_coverage_ledger_v1(value: &RollupCoverageLedgerV1) -> Result<Vec<u8>, LifecycleContractError> {
    encode_current(value, RollupCoverageLedgerV1::validate)
}

pub fn decode_rollup_coverage_ledger_v1(bytes: &[u8]) -> Result<RollupCoverageLedgerV1, LifecycleContractError> {
    decode_current(bytes, RollupCoverageLedgerV1::validate)
}

fn encode_current<T>(
    value: &T,
    validate: impl FnOnce(&T) -> Result<(), LifecycleContractError>,
) -> Result<Vec<u8>, LifecycleContractError>
where
    T: Serialize,
{
    validate(value)?;
    let bytes = serde_json::to_vec(value).map_err(LifecycleContractError::serde)?;
    if bytes.len() > MAX_LIFECYCLE_CONTRACT_BYTES {
        return Err(LifecycleContractError::Budget {
            field: "lifecycle contract bytes",
            observed: bytes.len() as u64,
            maximum: MAX_LIFECYCLE_CONTRACT_BYTES as u64,
        });
    }
    Ok(bytes)
}

fn decode_current<T>(
    bytes: &[u8],
    validate: impl FnOnce(&T) -> Result<(), LifecycleContractError>,
) -> Result<T, LifecycleContractError>
where
    T: DeserializeOwned + Serialize,
{
    if bytes.len() > MAX_LIFECYCLE_CONTRACT_BYTES {
        return Err(LifecycleContractError::Budget {
            field: "lifecycle contract bytes",
            observed: bytes.len() as u64,
            maximum: MAX_LIFECYCLE_CONTRACT_BYTES as u64,
        });
    }
    let probe: serde_json::Value = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
    let _version = probe
        .get("format_version")
        .ok_or(LifecycleContractError::MissingFormatVersion)?
        .as_u64()
        .and_then(|value| u16::try_from(value).ok())
        .ok_or_else(|| LifecycleContractError::Serde("format_version must be a u16".to_owned()))?;
    let value: T = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
    validate(&value)?;
    if serde_json::to_vec(&value).map_err(LifecycleContractError::serde)? != bytes {
        return Err(LifecycleContractError::NonCanonical);
    }
    Ok(value)
}

fn validate_format(version: u16) -> Result<(), LifecycleContractError> {
    if version == LIFECYCLE_COMPLETION_FORMAT_VERSION {
        Ok(())
    } else {
        Err(LifecycleContractError::UnsupportedFormat(version))
    }
}

fn validate_external_key_format(version: u16) -> Result<(), LifecycleContractError> {
    if version == EXTERNAL_KEY_FORMAT_VERSION {
        Ok(())
    } else {
        Err(LifecycleContractError::UnsupportedFormat(version))
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum LifecycleContractError {
    UnsupportedFormat(u16),
    MissingFormatVersion,
    NonCanonical,
    ZeroIdentity(&'static str),
    Budget {
        field: &'static str,
        observed: u64,
        maximum: u64,
    },
    Violation(&'static str),
    DetailRetiredByPolicy,
    Nested(String),
    Serde(String),
}

impl LifecycleContractError {
    fn serde(error: serde_json::Error) -> Self {
        Self::Serde(error.to_string())
    }
}

impl Display for LifecycleContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedFormat(version) => write!(formatter, "unsupported lifecycle format {version}"),
            Self::MissingFormatVersion => formatter.write_str("lifecycle format version is missing"),
            Self::NonCanonical => formatter.write_str("lifecycle bytes are not canonical current format"),
            Self::ZeroIdentity(name) => write!(formatter, "{name} cannot be zero"),
            Self::Budget {
                field,
                observed,
                maximum,
            } => write!(formatter, "{field} {observed} exceeds {maximum}"),
            Self::Violation(message) => formatter.write_str(message),
            Self::DetailRetiredByPolicy => {
                formatter.write_str("detail_retired_by_policy: raw detail is retired and this query is outside the declared rollup contract")
            }
            Self::Nested(message) => write!(formatter, "nested lifecycle authority refused input: {message}"),
            Self::Serde(message) => write!(formatter, "lifecycle codec refused input: {message}"),
        }
    }
}

impl Error for LifecycleContractError {}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{BackendAuthorityId, CollectionId, IndexId, ObjectIncarnation, ObjectKey};

    fn scope() -> Result<CollectionIndexRef, Box<dyn Error>> {
        Ok(CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(1)?))
    }

    fn physical(key: &str) -> Result<PhysicalObjectId, Box<dyn Error>> {
        Ok(PhysicalObjectId {
            backend_authority: BackendAuthorityId::new("backend-1")?,
            object_incarnation: ObjectIncarnation::new(1)?,
            object_key: ObjectKey::new(key)?,
        })
    }

    fn binding() -> Result<IndexExternalKeyBindingV1, Box<dyn Error>> {
        let context = ExternalKeyContextV1 {
            format_version: EXTERNAL_KEY_FORMAT_VERSION,
            collection_index_ref: scope()?,
            content_key_id: ContentKeyIdV1::new(1)?,
            content_key_generation: 1,
            binding_id: ExternalKeyBindingIdV1::new(1)?,
            binding_generation: 1,
            access_generation: 1,
            object_context_digest: [8; 32],
        };
        Ok(IndexExternalKeyBindingV1 {
            format_version: EXTERNAL_KEY_FORMAT_VERSION,
            binding_id: context.binding_id,
            collection_index_ref: context.collection_index_ref,
            content_key_id: context.content_key_id,
            content_key_generation: context.content_key_generation,
            binding_generation: context.binding_generation,
            access_generation: context.access_generation,
            opaque_binding: vec![7; 64],
            context,
            context_digest: context.digest()?,
            created_at_revision: ClusterRevision::new(1),
        })
    }

    #[test]
    fn encrypted_object_and_binding_are_exact_current_format() -> Result<(), Box<dyn Error>> {
        let physical_object = physical("encrypted/object-1")?;
        let content_key_id = ContentKeyIdV1::new(1)?;
        let object = EncryptedObjectKeyRefV1 {
            format_version: EXTERNAL_KEY_FORMAT_VERSION,
            nonce: object_nonce_v1(content_key_id, 1, &physical_object)?,
            authenticated_context_digest: object_authenticated_context_digest_v1(
                DocumentEncryptionAlgorithmV1::Aes256Gcm,
                content_key_id,
                1,
                &physical_object,
                128,
            )?,
            physical_object,
            algorithm: DocumentEncryptionAlgorithmV1::Aes256Gcm,
            content_key_id,
            content_key_generation: 1,
            ciphertext_bytes: 128,
        };
        let bytes = encode_encrypted_object_key_ref_v1(&object)?;
        assert_eq!(decode_encrypted_object_key_ref_v1(&bytes)?, object);
        let binding = binding()?;
        let bytes = encode_index_external_key_binding_v1(&binding)?;
        assert_eq!(decode_index_external_key_binding_v1(&bytes)?, binding);
        Ok(())
    }

    #[test]
    fn rollup_route_is_exact_and_retired_raw_refuses_mismatch() -> Result<(), Box<dyn Error>> {
        let dimension = FieldPathDigestV1::new([1; 32])?;
        let policy = RollupPolicyV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            collection_index_ref: scope()?,
            policy_id: RollupPolicyIdV1::new(1)?,
            policy_generation: PolicyGeneration::new(1),
            detail_retirement_acknowledged: true,
            time_zone: RollupTimeZoneV1::Utc,
            timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
            bucket_width_millis: 60_000,
            bucket_alignment_millis: 0,
            dimensions: vec![dimension],
            measures: vec![RollupMeasureV1::Count],
        };
        let policy_bytes = encode_rollup_policy_v1(&policy)?;
        assert_eq!(decode_rollup_policy_v1(&policy_bytes)?, policy);
        let mut unacknowledged = policy.clone();
        unacknowledged.detail_retirement_acknowledged = false;
        assert!(matches!(
            unacknowledged.validate(),
            Err(LifecycleContractError::Violation(_))
        ));
        let root = RootSetRef {
            format_version: crate::ROOT_SET_REF_FORMAT_VERSION,
            root: physical("rollup/root")?,
            content_digest: [9; 32],
            item_count: 1,
            page_count: 1,
            schema_generations: vec![SchemaGeneration::new(1)],
        };
        let mut coverage = RollupCoverageV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            collection_index_ref: scope()?,
            policy: policy.clone(),
            schema_generation: SchemaGeneration::new(1),
            source_incarnation: None,
            source_cut: None,
            source_generations: vec![RollupSourceGenerationV1 {
                shard_id: ShardId::new(0),
                manifest_generation: ManifestGeneration::new(1),
                correction_through: SeqNo::new(10),
            }],
            range_start_millis: 60_000,
            range_end_millis: 180_000,
            late_data_watermark_millis: 180_000,
            output_root: root,
            raw_retirement_state: RawRetirementStateV1::RawPresent,
            routing_proof: None,
        };
        let mut invalid_output_root = coverage.clone();
        invalid_output_root.output_root.item_count = 2;
        assert!(matches!(
            invalid_output_root.validate(),
            Err(LifecycleContractError::Violation(_))
        ));
        coverage.raw_retirement_state = RawRetirementStateV1::RawRetired;
        coverage.routing_proof = Some(RollupRoutingProofV1 {
            query_contract_digest: rollup_query_contract_digest_v1()?,
            coverage_digest: coverage.coverage_digest()?,
            proven_at_revision: ClusterRevision::new(2),
        });
        let ledger = RollupCoverageLedgerV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            collection_index_ref: scope()?,
            revision: 2,
            coverages: vec![coverage],
        };
        let bytes = encode_rollup_coverage_ledger_v1(&ledger)?;
        assert_eq!(decode_rollup_coverage_ledger_v1(&bytes)?, ledger);
        let coverage_digest = ledger.coverages[0].coverage_digest()?;
        let query = RollupQueryShapeV1 {
            range_start_millis: 0,
            range_end_millis: 240_000,
            time_zone: RollupTimeZoneV1::Utc,
            bucket_width_millis: 60_000,
            bucket_alignment_millis: 0,
            dimensions: vec![dimension],
            measures: vec![RollupMeasureV1::Count],
        };
        assert_eq!(
            route_rollup_query_v1(&ledger, &query)?,
            vec![
                RollupRouteSliceV1 {
                    range_start_millis: 0,
                    range_end_millis: 60_000,
                    kind: RollupRouteKindV1::Raw,
                    rollup_coverage_digest: None,
                },
                RollupRouteSliceV1 {
                    range_start_millis: 60_000,
                    range_end_millis: 180_000,
                    kind: RollupRouteKindV1::Rollup,
                    rollup_coverage_digest: Some(coverage_digest),
                },
                RollupRouteSliceV1 {
                    range_start_millis: 180_000,
                    range_end_millis: 240_000,
                    kind: RollupRouteKindV1::Raw,
                    rollup_coverage_digest: None,
                },
            ]
        );
        let mismatch = RollupQueryShapeV1 {
            dimensions: Vec::new(),
            ..query
        };
        assert_eq!(
            route_rollup_query_v1(&ledger, &mismatch),
            Err(LifecycleContractError::DetailRetiredByPolicy)
        );
        let mut stale_proof = ledger.clone();
        stale_proof.coverages[0].routing_proof = Some(RollupRoutingProofV1 {
            query_contract_digest: RollupQueryContractDigestV1::new([5; 32])?,
            coverage_digest: stale_proof.coverages[0].coverage_digest()?,
            proven_at_revision: ClusterRevision::new(2),
        });
        assert!(matches!(
            stale_proof.validate(),
            Err(LifecycleContractError::Violation(_))
        ));
        let mut future_proof = ledger.clone();
        future_proof.coverages[0]
            .routing_proof
            .as_mut()
            .ok_or("routing proof missing")?
            .proven_at_revision = ClusterRevision::new(3);
        assert!(matches!(
            future_proof.validate(),
            Err(LifecycleContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn rollup_output_refuses_empty_or_miscounted_groups() -> Result<(), Box<dyn Error>> {
        let plan = RollupWorkPlanV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            policy: RollupPolicyV1 {
                format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
                collection_index_ref: scope()?,
                policy_id: RollupPolicyIdV1::new(1)?,
                policy_generation: PolicyGeneration::new(1),
                detail_retirement_acknowledged: true,
                time_zone: RollupTimeZoneV1::Utc,
                timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
                bucket_width_millis: 1_000,
                bucket_alignment_millis: 0,
                dimensions: Vec::new(),
                measures: vec![RollupMeasureV1::Count],
            },
            schema_generation: SchemaGeneration::new(1),
            source_incarnation: None,
            source_cut: None,
            source_generations: vec![RollupSourceGenerationV1 {
                shard_id: ShardId::new(0),
                manifest_generation: ManifestGeneration::new(1),
                correction_through: SeqNo::new(2),
            }],
            range_start_millis: 0,
            range_end_millis: 1_000,
            late_data_watermark_millis: 1_000,
        };
        let valid = RollupOutputV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            work_plan_digest: plan.digest()?,
            rows: vec![RollupOutputRowV1 {
                bucket_start_millis: 0,
                dimensions: Vec::new(),
                measures: vec![RollupAggregateValueV1::Count(2)],
            }],
            input_documents: 2,
            input_bytes: 64,
        };
        encode_rollup_output_v1(&valid, &plan)?;
        let mut empty = valid.clone();
        empty.rows.clear();
        assert!(encode_rollup_output_v1(&empty, &plan).is_err());
        let mut miscounted = valid;
        miscounted.rows[0].measures[0] = RollupAggregateValueV1::Count(1);
        assert!(encode_rollup_output_v1(&miscounted, &plan).is_err());
        Ok(())
    }

    #[test]
    fn adjacent_rollup_roots_remain_distinct_route_slices() -> Result<(), Box<dyn Error>> {
        let dimension = FieldPathDigestV1::new([1; 32])?;
        let policy = RollupPolicyV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            collection_index_ref: scope()?,
            policy_id: RollupPolicyIdV1::new(1)?,
            policy_generation: PolicyGeneration::new(1),
            detail_retirement_acknowledged: true,
            time_zone: RollupTimeZoneV1::Utc,
            timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
            bucket_width_millis: 60_000,
            bucket_alignment_millis: 0,
            dimensions: vec![dimension],
            measures: vec![RollupMeasureV1::Count],
        };
        let root = |name: &str| -> Result<RootSetRef, Box<dyn Error>> {
            Ok(RootSetRef {
                format_version: crate::ROOT_SET_REF_FORMAT_VERSION,
                root: physical(name)?,
                content_digest: if name.ends_with('1') { [8; 32] } else { [9; 32] },
                item_count: 1,
                page_count: 1,
                schema_generations: vec![SchemaGeneration::new(1)],
            })
        };
        let coverage = |start, end, output_root| -> Result<RollupCoverageV1, Box<dyn Error>> {
            Ok(RollupCoverageV1 {
                format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
                collection_index_ref: scope()?,
                policy: policy.clone(),
                schema_generation: SchemaGeneration::new(1),
                source_incarnation: None,
                source_cut: None,
                source_generations: vec![RollupSourceGenerationV1 {
                    shard_id: ShardId::new(0),
                    manifest_generation: ManifestGeneration::new(1),
                    correction_through: SeqNo::new(10),
                }],
                range_start_millis: start,
                range_end_millis: end,
                late_data_watermark_millis: end,
                output_root,
                raw_retirement_state: RawRetirementStateV1::RawPresent,
                routing_proof: None,
            })
        };
        let first = coverage(60_000, 120_000, root("rollup/root-1")?)?;
        let second = coverage(120_000, 180_000, root("rollup/root-2")?)?;
        let first_digest = first.coverage_digest()?;
        let second_digest = second.coverage_digest()?;
        assert_ne!(first_digest, second_digest);
        let ledger = RollupCoverageLedgerV1 {
            format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
            collection_index_ref: scope()?,
            revision: 1,
            coverages: vec![first, second],
        };
        let query = RollupQueryShapeV1 {
            range_start_millis: 60_000,
            range_end_millis: 180_000,
            time_zone: RollupTimeZoneV1::Utc,
            bucket_width_millis: 60_000,
            bucket_alignment_millis: 0,
            dimensions: vec![dimension],
            measures: vec![RollupMeasureV1::Count],
        };
        assert_eq!(
            route_rollup_query_v1(&ledger, &query)?,
            vec![
                RollupRouteSliceV1 {
                    range_start_millis: 60_000,
                    range_end_millis: 120_000,
                    kind: RollupRouteKindV1::Rollup,
                    rollup_coverage_digest: Some(first_digest),
                },
                RollupRouteSliceV1 {
                    range_start_millis: 120_000,
                    range_end_millis: 180_000,
                    kind: RollupRouteKindV1::Rollup,
                    rollup_coverage_digest: Some(second_digest),
                },
            ]
        );
        Ok(())
    }

    #[test]
    fn codec_refuses_old_future_missing_unknown_duplicate_noncanonical_and_oversized() -> Result<(), Box<dyn Error>> {
        let bytes = encode_index_external_key_binding_v1(&binding()?)?;
        let mut old: serde_json::Value = serde_json::from_slice(&bytes)?;
        let old_version = EXTERNAL_KEY_FORMAT_VERSION - 1;
        old["format_version"] = serde_json::Value::from(old_version);
        assert_eq!(
            decode_index_external_key_binding_v1(&serde_json::to_vec(&old)?),
            Err(LifecycleContractError::UnsupportedFormat(old_version))
        );
        let mut future: serde_json::Value = serde_json::from_slice(&bytes)?;
        let future_version = EXTERNAL_KEY_FORMAT_VERSION + 1;
        future["format_version"] = serde_json::Value::from(future_version);
        assert_eq!(
            decode_index_external_key_binding_v1(&serde_json::to_vec(&future)?),
            Err(LifecycleContractError::UnsupportedFormat(future_version))
        );
        assert_eq!(
            decode_index_external_key_binding_v1(br#"{"binding_id":"1"}"#),
            Err(LifecycleContractError::MissingFormatVersion)
        );
        let mut unknown: serde_json::Value = serde_json::from_slice(&bytes)?;
        unknown["unknown_binding"] = serde_json::json!(true);
        assert!(matches!(
            decode_index_external_key_binding_v1(&serde_json::to_vec(&unknown)?),
            Err(LifecycleContractError::Serde(_))
        ));
        let version_prefix = format!("{{\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},");
        let duplicate_prefix = format!(
            "{{\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},"
        );
        let duplicate = String::from_utf8(bytes.clone())?.replacen(&version_prefix, &duplicate_prefix, 1);
        assert!(matches!(
            decode_index_external_key_binding_v1(duplicate.as_bytes()),
            Err(LifecycleContractError::Serde(_))
        ));
        let mut whitespace = bytes;
        whitespace.push(b'\n');
        assert_eq!(
            decode_index_external_key_binding_v1(&whitespace),
            Err(LifecycleContractError::NonCanonical)
        );
        assert!(matches!(
            decode_index_external_key_binding_v1(&vec![b' '; MAX_LIFECYCLE_CONTRACT_BYTES + 1]),
            Err(LifecycleContractError::Budget { .. })
        ));
        let mut mismatched = binding()?;
        mismatched.binding_id = ExternalKeyBindingIdV1::new(2)?;
        assert!(matches!(
            mismatched.validate(),
            Err(LifecycleContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn serialized_binding_contains_only_opaque_non_plaintext_material() -> Result<(), Box<dyn Error>> {
        let value = serde_json::to_value(binding()?)?;
        assert!(value.get("opaque_binding").is_some());
        assert!(value.get("plaintext_key").is_none());
        assert!(value.get("content_key_bytes").is_none());
        assert_eq!(value.as_object().map(serde_json::Map::len), Some(11));
        Ok(())
    }
}