lodestone_types/

derived_overload.rs

//! P13.456A cache, adaptive-promotion, and overload semantic contracts.
//!
//! These types are deliberately not wired to a public handler. They freeze the
//! current-format inputs and refusal rules that P13.4B, P13.5A, and P13.6A must
//! implement without creating another durability or mapping authority.

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

use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::{
    CollectionId, CollectionIndexRef, CollectionSecurityGeneration, Epoch, IndexMode, IngestAuthorityV1, LogicalMillis,
    ManifestGeneration, PolicyGeneration, RootCutId, SchemaGeneration, SegmentId, SeqNo, ShardId, ShardKey,
};

pub const DERIVED_OVERLOAD_FORMAT_VERSION: u16 = 3;
pub const MAX_DERIVED_CONTRACT_BYTES: usize = 1_048_576;
pub const MAX_CACHE_SHARDS: usize = 512;
pub const MAX_CACHE_RESULT_WINDOW: u32 = 100_000;
pub const MAX_CACHE_ENTRY_BYTES: u64 = 16 * 1_024 * 1_024;
pub const MAX_CACHE_COLLECTION_BYTES: u64 = 512 * 1_024 * 1_024;
pub const MAX_CACHE_COLLECTION_ENTRIES: u32 = 4_096;
pub const MAX_PROMOTED_FIELDS: u16 = 64;
pub const MAX_HEAT_RECORDS: u32 = 16_384;
pub const MAX_DISTINCT_QUERIES_PER_HEAT_RECORD: u32 = 1_000_000;
pub const MAX_PROMOTION_EXTRA_BYTES: u64 = 1_073_741_824;
pub const MAX_PROMOTED_COLUMN_SIDECAR_BYTES: usize = 64 * 1_024 * 1_024;
pub const MAX_PROMOTED_COLUMN_ROWS: u32 = 1_000_000;
pub const PROMOTED_COLUMN_SIDECAR_NAME: &str = "field-promotion.v1.json";
pub const MAX_OVERLOAD_BACKLOG_BYTES: u64 = 1_073_741_824;
pub const MAX_OVERLOAD_BACKLOG_AGE_MILLIS: u64 = 300_000;
pub const MAX_OVERLOAD_COLLECTION_SHARE_BYTES: u64 = 268_435_456;
pub const MAX_OVERLOAD_RECOVERY_OPS_PER_PASS: u32 = 10_000;
pub const MAX_PENDING_OVERLAY_DOCUMENTS: u64 = 1_000_000;

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

fn require_nonzero_digest(value: &[u8; 32], name: &'static str) -> Result<(), DerivedContractError> {
    if digest_is_nonzero(value) {
        Ok(())
    } else {
        Err(DerivedContractError::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, DerivedContractError> {
                require_nonzero_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<(), DerivedContractError> {
                require_nonzero_digest(&self.0, concat!(stringify!($name), " cannot be zero"))
            }
        }
    };
}

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

        impl $name {
            pub const fn new(value: u64) -> Result<Self, DerivedContractError> {
                if value == 0 {
                    Err(DerivedContractError::Violation(concat!(
                        stringify!($name),
                        " cannot be zero"
                    )))
                } else {
                    Ok(Self(value))
                }
            }

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

            fn validate(self) -> Result<(), DerivedContractError> {
                Self::new(self.0).map(|_| ())
            }
        }
    };
}

semantic_digest!(AuthorizedScopeDigestV1, "Digest of the exact authorized query scope.");
semantic_digest!(NormalizedPlanDigestV1, "Digest of one canonical normalized query plan.");
semantic_digest!(OutputSemanticsDigestV1, "Digest of result-affecting output semantics.");
semantic_digest!(FeatureFlagsDigestV1, "Digest of result-affecting feature flags.");
semantic_digest!(SnapshotDigestV1, "Digest of one exact PIT snapshot binding.");
semantic_digest!(FieldPathDigestV1, "Opaque digest of one logical field path.");
semantic_generation!(
    CollectionResetGenerationV1,
    "Collection-wide result-cache reset generation."
);
semantic_generation!(
    DeletionOverlayGenerationV1,
    "Current privacy/deletion overlay generation."
);
semantic_generation!(KeyAccessGenerationV1, "Current key-access authorization generation.");
semantic_generation!(PromotionDecisionGenerationV1, "Physical promotion decision generation.");

/// Computes the privacy-safe identity used by heat, decisions, and physical
/// layout metadata. Raw field paths remain transient caller input.
pub fn field_path_digest_v1(path: &str) -> Result<FieldPathDigestV1, DerivedContractError> {
    if !crate::is_canonical_field_path(path) {
        return Err(DerivedContractError::Violation(
            "field path is not canonical and cannot be promoted",
        ));
    }
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/p13/field-path/v1\0");
    hasher.update(path.as_bytes());
    FieldPathDigestV1::new(*hasher.finalize().as_bytes())
}

/// One shard generation in an exact result-cache identity vector.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CacheShardGenerationV1 {
    pub shard_id: ShardId,
    pub manifest_generation: ManifestGeneration,
}

/// The visibility class used to compute a result.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheFreshnessV1 {
    Committed,
    Realtime,
}

/// A realtime request may cache only its immutable committed sub-result.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheReuseScopeV1 {
    CompleteCommittedResult,
    CommittedSubresultOfRealtime,
}

/// Exact root binding used by the query. No unresolved clock is accepted.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CacheSnapshotBindingV1 {
    Current,
    Pit { snapshot_digest: SnapshotDigestV1 },
    RetainedHistory { root_cut_id: RootCutId },
}

impl CacheSnapshotBindingV1 {
    fn validate(self) -> Result<(), DerivedContractError> {
        match self {
            Self::Current | Self::RetainedHistory { .. } => Ok(()),
            Self::Pit { snapshot_digest } => snapshot_digest.validate(),
        }
    }
}

/// Complete semantic identity of one reusable result or committed sub-result.
#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ResultCacheKeyV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub collection_security_generation: CollectionSecurityGeneration,
    pub authorized_scope_digest: AuthorizedScopeDigestV1,
    pub policy_generation: PolicyGeneration,
    pub collection_reset_generation: CollectionResetGenerationV1,
    pub deletion_overlay_generation: DeletionOverlayGenerationV1,
    pub key_access_generation: KeyAccessGenerationV1,
    pub normalized_plan_digest: NormalizedPlanDigestV1,
    pub output_semantics_digest: OutputSemanticsDigestV1,
    pub expected_shard_count: u16,
    pub shard_generations: Vec<CacheShardGenerationV1>,
    pub mapping_generation: SchemaGeneration,
    pub freshness: CacheFreshnessV1,
    pub reuse_scope: CacheReuseScopeV1,
    pub from: u32,
    pub limit: u32,
    pub feature_flags_digest: FeatureFlagsDigestV1,
    pub snapshot: CacheSnapshotBindingV1,
}

impl ResultCacheKeyV1 {
    pub fn validate(&self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.collection_security_generation.get() == 0
            || self.policy_generation.get() == 0
            || self.mapping_generation.get() == 0
        {
            return Err(DerivedContractError::Violation(
                "cache generations and authorization revisions must be nonzero",
            ));
        }
        self.authorized_scope_digest.validate()?;
        self.collection_reset_generation.validate()?;
        self.deletion_overlay_generation.validate()?;
        self.key_access_generation.validate()?;
        self.normalized_plan_digest.validate()?;
        self.output_semantics_digest.validate()?;
        self.feature_flags_digest.validate()?;
        if self.shard_generations.is_empty() || self.shard_generations.len() > MAX_CACHE_SHARDS {
            return Err(DerivedContractError::Budget {
                field: "cache shard generations",
                observed: self.shard_generations.len() as u64,
                maximum: MAX_CACHE_SHARDS as u64,
            });
        }
        if usize::from(self.expected_shard_count) != self.shard_generations.len() {
            return Err(DerivedContractError::Violation(
                "cache shard vector does not contain the complete expected shard count",
            ));
        }
        let mut previous = None;
        for shard in &self.shard_generations {
            if shard.manifest_generation.get() == 0 || previous.is_some_and(|prior| prior >= shard.shard_id) {
                return Err(DerivedContractError::Violation(
                    "cache shard generations must be nonzero, unique, and strictly ordered",
                ));
            }
            previous = Some(shard.shard_id);
        }
        if self.limit == 0
            || self
                .from
                .checked_add(self.limit)
                .is_none_or(|window| window > MAX_CACHE_RESULT_WINDOW)
        {
            return Err(DerivedContractError::Violation(
                "cache result window is outside the bounded contract",
            ));
        }
        match (self.freshness, self.reuse_scope) {
            (CacheFreshnessV1::Committed, CacheReuseScopeV1::CompleteCommittedResult)
            | (CacheFreshnessV1::Realtime, CacheReuseScopeV1::CommittedSubresultOfRealtime) => {}
            _ => {
                return Err(DerivedContractError::Violation(
                    "realtime requests may cache only an immutable committed sub-result",
                ));
            }
        }
        self.snapshot.validate()
    }
}

/// Per-collection cache admission and eviction budget.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ResultCacheBudgetV1 {
    pub format_version: u16,
    pub collection_id: CollectionId,
    pub maximum_entry_bytes: u64,
    pub maximum_collection_bytes: u64,
    pub maximum_collection_entries: u32,
}

impl ResultCacheBudgetV1 {
    pub fn validate(self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.maximum_entry_bytes == 0
            || self.maximum_entry_bytes > MAX_CACHE_ENTRY_BYTES
            || self.maximum_collection_bytes < self.maximum_entry_bytes
            || self.maximum_collection_bytes > MAX_CACHE_COLLECTION_BYTES
            || self.maximum_collection_entries == 0
            || self.maximum_collection_entries > MAX_CACHE_COLLECTION_ENTRIES
        {
            return Err(DerivedContractError::Violation(
                "result-cache budget is outside the bounded contract",
            ));
        }
        Ok(())
    }
}

/// Numeric-only status; it cannot carry query text, document data, or cached bytes.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ResultCacheMetricsV1 {
    pub format_version: u16,
    pub collection_id: CollectionId,
    pub hits: u64,
    pub misses: u64,
    pub admissions: u64,
    pub bypasses: u64,
    pub evictions: u64,
    pub collection_bytes: u64,
    pub collection_entries: u32,
}

impl ResultCacheMetricsV1 {
    pub fn validate(self, budget: ResultCacheBudgetV1) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        budget.validate()?;
        if self.collection_id != budget.collection_id
            || self.collection_bytes > budget.maximum_collection_bytes
            || self.collection_entries > budget.maximum_collection_entries
        {
            return Err(DerivedContractError::Violation(
                "result-cache metrics exceed the recorded collection budget",
            ));
        }
        Ok(())
    }
}

/// Privacy classification applied before promotion is considered.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PromotionPrivacyV1 {
    Ordinary,
    Sensitive,
    Restricted,
}

/// Closed physical type set for a promoted field representation.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PromotedFieldTypeV1 {
    Keyword,
    I64,
    U64,
    F64,
    DateMillis,
    Boolean,
}

/// Bounded, privacy-safe aggregate heat. Raw field names and query text are absent by construction.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FieldHeatAggregateV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub mapping_generation: SchemaGeneration,
    pub field_path_digest: FieldPathDigestV1,
    pub window_start_logical_millis: LogicalMillis,
    pub window_end_logical_millis: LogicalMillis,
    pub filter_uses: u64,
    pub sort_uses: u64,
    pub aggregation_uses: u64,
    pub distinct_query_count: u32,
    pub estimated_cardinality: u64,
    pub estimated_extra_bytes: u64,
    pub privacy: PromotionPrivacyV1,
    pub field_type: PromotedFieldTypeV1,
    pub observed_type_conflict: bool,
    pub path_policy_eligible: bool,
}

impl FieldHeatAggregateV1 {
    pub fn validate(&self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.mapping_generation.get() == 0
            || self.window_start_logical_millis.get() >= self.window_end_logical_millis.get()
            || self.distinct_query_count == 0
            || self.distinct_query_count > MAX_DISTINCT_QUERIES_PER_HEAT_RECORD
            || self.estimated_extra_bytes > MAX_PROMOTION_EXTRA_BYTES
        {
            return Err(DerivedContractError::Violation(
                "field heat aggregate is outside the bounded contract",
            ));
        }
        self.field_path_digest.validate()?;
        let total_uses = self
            .filter_uses
            .checked_add(self.sort_uses)
            .and_then(|uses| uses.checked_add(self.aggregation_uses))
            .ok_or(DerivedContractError::ArithmeticOverflow)?;
        if total_uses == 0 || u64::from(self.distinct_query_count) > total_uses {
            return Err(DerivedContractError::Violation(
                "field heat requires enough physical operations for its distinct-query count",
            ));
        }
        Ok(())
    }
}

/// Policy limits for physical field promotion. Logical mapping semantics remain authoritative.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FieldPromotionPolicyV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub mapping_generation: SchemaGeneration,
    pub policy_generation: PolicyGeneration,
    pub enabled: bool,
    pub minimum_query_count: u32,
    pub maximum_promoted_fields: u16,
    pub maximum_cardinality: u64,
    pub maximum_extra_bytes: u64,
    pub allow_sensitive: bool,
}

impl FieldPromotionPolicyV1 {
    pub fn validate(self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.mapping_generation.get() == 0
            || self.policy_generation.get() == 0
            || self.minimum_query_count == 0
            || self.minimum_query_count > MAX_DISTINCT_QUERIES_PER_HEAT_RECORD
            || self.maximum_promoted_fields == 0
            || self.maximum_promoted_fields > MAX_PROMOTED_FIELDS
            || self.maximum_cardinality == 0
            || self.maximum_extra_bytes == 0
            || self.maximum_extra_bytes > MAX_PROMOTION_EXTRA_BYTES
        {
            return Err(DerivedContractError::Violation(
                "field-promotion policy is outside the bounded contract",
            ));
        }
        Ok(())
    }

    /// Returns a physical-only decision; denial never changes logical mapping behavior.
    pub fn admits(
        self,
        heat: &FieldHeatAggregateV1,
        currently_promoted_fields: u16,
        currently_promoted_extra_bytes: u64,
    ) -> Result<bool, DerivedContractError> {
        self.validate()?;
        heat.validate()?;
        if heat.collection_index_ref != self.collection_index_ref || heat.mapping_generation != self.mapping_generation
        {
            return Err(DerivedContractError::Violation(
                "field heat does not belong to the promotion policy scope and mapping generation",
            ));
        }
        if currently_promoted_fields > self.maximum_promoted_fields
            || currently_promoted_extra_bytes > self.maximum_extra_bytes
        {
            return Err(DerivedContractError::Violation(
                "current promoted layout exceeds the governing policy",
            ));
        }
        let projected_extra_bytes = currently_promoted_extra_bytes
            .checked_add(heat.estimated_extra_bytes)
            .ok_or(DerivedContractError::ArithmeticOverflow)?;
        Ok(self.enabled
            && heat.distinct_query_count >= self.minimum_query_count
            && heat.estimated_cardinality <= self.maximum_cardinality
            && projected_extra_bytes <= self.maximum_extra_bytes
            && currently_promoted_fields < self.maximum_promoted_fields
            && !heat.observed_type_conflict
            && heat.path_policy_eligible
            && heat.privacy != PromotionPrivacyV1::Restricted
            && (heat.privacy != PromotionPrivacyV1::Sensitive || self.allow_sensitive))
    }
}

/// Per-segment physical layout. Readers must branch per segment and preserve logical equivalence.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PromotedFieldLayoutV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub segment_id: SegmentId,
    pub field_path_digest: FieldPathDigestV1,
    pub mapping_generation: SchemaGeneration,
    pub decision_generation: PromotionDecisionGenerationV1,
    pub field_type: PromotedFieldTypeV1,
}

impl PromotedFieldLayoutV1 {
    pub fn validate(self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        self.field_path_digest.validate()?;
        self.decision_generation.validate()?;
        if self.segment_id.get() == 0 || self.mapping_generation.get() == 0 {
            return Err(DerivedContractError::Violation(
                "promoted layout generations must be nonzero",
            ));
        }
        Ok(())
    }
}

/// Numeric/digest-only promotion status; it contains no raw field path or query text.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FieldPromotionStatusV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub mapping_generation: SchemaGeneration,
    pub policy_generation: PolicyGeneration,
    pub last_decision_generation: Option<PromotionDecisionGenerationV1>,
    pub heat_records: u32,
    pub promoted_fields: u16,
    pub extra_bytes: u64,
    pub enabled: bool,
}

impl FieldPromotionStatusV1 {
    pub fn validate(self, policy: FieldPromotionPolicyV1) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        policy.validate()?;
        if self.collection_index_ref != policy.collection_index_ref
            || self.mapping_generation != policy.mapping_generation
            || self.policy_generation != policy.policy_generation
            || self.enabled != policy.enabled
            || self.heat_records > MAX_HEAT_RECORDS
            || self.promoted_fields > policy.maximum_promoted_fields
            || self.extra_bytes > policy.maximum_extra_bytes
            || (self.promoted_fields > 0 && self.last_decision_generation.is_none())
            || (self.promoted_fields == 0 && self.extra_bytes != 0)
        {
            return Err(DerivedContractError::Violation(
                "field-promotion status is inconsistent with the governing policy",
            ));
        }
        if let Some(generation) = self.last_decision_generation {
            generation.validate()?;
        }
        Ok(())
    }
}

/// Canonical scalar stored in a derived promoted column. Floating-point
/// values use normalized IEEE bits so equality and ordering remain explicit.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum PromotedFieldValueV1 {
    Keyword(String),
    I64(i64),
    U64(u64),
    F64Bits(u64),
    DateMillis(u64),
    Boolean(bool),
}

impl PromotedFieldValueV1 {
    pub fn validate_for_type(&self, field_type: PromotedFieldTypeV1) -> Result<(), DerivedContractError> {
        let matches_type = matches!(
            (self, field_type),
            (Self::Keyword(_), PromotedFieldTypeV1::Keyword)
                | (Self::I64(_), PromotedFieldTypeV1::I64)
                | (Self::U64(_), PromotedFieldTypeV1::U64)
                | (Self::F64Bits(_), PromotedFieldTypeV1::F64)
                | (Self::DateMillis(_), PromotedFieldTypeV1::DateMillis)
                | (Self::Boolean(_), PromotedFieldTypeV1::Boolean)
        );
        if !matches_type {
            return Err(DerivedContractError::Violation(
                "promoted column value does not match its closed physical type",
            ));
        }
        match self {
            Self::Keyword(value) if value.len() > crate::MAX_MAPPING_FIELD_NAME_BYTES.saturating_mul(64) => Err(
                DerivedContractError::Violation("promoted keyword value exceeds the bounded scalar representation"),
            ),
            Self::F64Bits(bits) if !f64::from_bits(*bits).is_finite() => Err(DerivedContractError::Violation(
                "promoted floating-point value must be finite",
            )),
            Self::F64Bits(bits) if f64::from_bits(*bits) == 0.0 && *bits != 0.0_f64.to_bits() => Err(
                DerivedContractError::Violation("promoted floating-point zero must use canonical positive-zero bits"),
            ),
            _ => Ok(()),
        }
    }
}

/// One present scalar at an exact physical document ordinal. Missing/null
/// values are omitted rather than assigned a competing null interpretation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PromotedFieldRowV1 {
    pub document_ordinal: u32,
    pub value: PromotedFieldValueV1,
}

/// One immutable physical column for a field digest in a segment.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PromotedFieldColumnV1 {
    pub layout: PromotedFieldLayoutV1,
    pub rows: Vec<PromotedFieldRowV1>,
}

impl PromotedFieldColumnV1 {
    pub fn validate(&self, document_count: u32) -> Result<(), DerivedContractError> {
        self.layout.validate()?;
        if self.rows.len() > document_count as usize || self.rows.len() > MAX_PROMOTED_COLUMN_ROWS as usize {
            return Err(DerivedContractError::Budget {
                field: "promoted column rows",
                observed: self.rows.len() as u64,
                maximum: u64::from(document_count.min(MAX_PROMOTED_COLUMN_ROWS)),
            });
        }
        let mut previous = None;
        for row in &self.rows {
            if row.document_ordinal >= document_count || previous.is_some_and(|prior| prior >= row.document_ordinal) {
                return Err(DerivedContractError::Violation(
                    "promoted column rows must be unique, ordered, and inside the segment",
                ));
            }
            row.value.validate_for_type(self.layout.field_type)?;
            previous = Some(row.document_ordinal);
        }
        Ok(())
    }
}

/// Current immutable promoted-column sidecar. It is a child of the ordinary
/// segment/root graph and has no independent publication authority.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PromotedColumnSidecarV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub segment_id: SegmentId,
    pub mapping_generation: SchemaGeneration,
    pub document_count: u32,
    pub columns: Vec<PromotedFieldColumnV1>,
}

impl PromotedColumnSidecarV1 {
    pub fn validate(&self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.segment_id.get() == 0
            || self.mapping_generation.get() == 0
            || self.document_count > MAX_PROMOTED_COLUMN_ROWS
            || self.columns.is_empty()
            || self.columns.len() > usize::from(MAX_PROMOTED_FIELDS)
        {
            return Err(DerivedContractError::Violation(
                "promoted column sidecar is outside the bounded segment contract",
            ));
        }
        let mut previous_digest = None;
        let mut decision_generations = std::collections::BTreeSet::new();
        for column in &self.columns {
            let layout = column.layout;
            if layout.collection_index_ref != self.collection_index_ref
                || layout.segment_id != self.segment_id
                || layout.mapping_generation != self.mapping_generation
                || previous_digest.is_some_and(|previous| previous >= layout.field_path_digest.get())
            {
                return Err(DerivedContractError::Violation(
                    "promoted sidecar columns must match the segment and be strictly digest ordered",
                ));
            }
            if !decision_generations.insert(layout.decision_generation.get()) {
                return Err(DerivedContractError::Violation(
                    "promoted sidecar decision generations must be unique",
                ));
            }
            column.validate(self.document_count)?;
            previous_digest = Some(layout.field_path_digest.get());
        }
        Ok(())
    }

    #[must_use]
    pub fn column(&self, digest: FieldPathDigestV1) -> Option<&PromotedFieldColumnV1> {
        self.columns
            .binary_search_by_key(&digest.get(), |column| column.layout.field_path_digest.get())
            .ok()
            .and_then(|index| self.columns.get(index))
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OverloadModeV1 {
    Shed,
    Buffer,
}

/// Bounded overload behavior. A buffer is a projection of the existing registered WAL.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct OverloadPolicyV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub policy_generation: PolicyGeneration,
    pub mode: OverloadModeV1,
    pub maximum_backlog_bytes: u64,
    pub maximum_backlog_age_millis: u64,
    pub maximum_collection_share_bytes: u64,
    pub maximum_recovery_ops_per_pass: u32,
    pub maximum_pending_overlay_documents: u64,
}

impl OverloadPolicyV1 {
    pub fn validate(self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        if self.policy_generation.get() == 0
            || self.maximum_backlog_bytes == 0
            || self.maximum_backlog_bytes > MAX_OVERLOAD_BACKLOG_BYTES
            || self.maximum_backlog_age_millis == 0
            || self.maximum_backlog_age_millis > MAX_OVERLOAD_BACKLOG_AGE_MILLIS
            || self.maximum_collection_share_bytes == 0
            || self.maximum_collection_share_bytes > self.maximum_backlog_bytes
            || self.maximum_collection_share_bytes > MAX_OVERLOAD_COLLECTION_SHARE_BYTES
            || self.maximum_recovery_ops_per_pass == 0
            || self.maximum_recovery_ops_per_pass > MAX_OVERLOAD_RECOVERY_OPS_PER_PASS
            || self.maximum_pending_overlay_documents == 0
            || self.maximum_pending_overlay_documents > MAX_PENDING_OVERLAY_DOCUMENTS
        {
            return Err(DerivedContractError::Violation(
                "overload policy is outside the bounded contract",
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OverloadDispositionV1 {
    Shed,
    BufferRegisteredWal,
    PauseReplayableSource,
    RefuseStateMode,
    RefuseMissingMutableOverlay,
}

/// Selects policy without changing the existing acknowledgment authority.
pub fn overload_disposition_v1(
    policy: OverloadPolicyV1,
    index_mode: IndexMode,
    ingest_authority: &IngestAuthorityV1,
    exact_mutable_overlay_available: bool,
) -> Result<OverloadDispositionV1, DerivedContractError> {
    policy.validate()?;
    // INVARIANT: state plus replayable-source is an intentional overload
    // refusal, but the nested authority still has to be valid current format.
    // Validate that nested shape in a source-compatible mode before selecting
    // the state refusal below.
    let validation_mode =
        if index_mode == IndexMode::State && matches!(ingest_authority, IngestAuthorityV1::ReplayableSource { .. }) {
            IndexMode::AppendLog
        } else {
            index_mode
        };
    ingest_authority
        .validate_for_mode(validation_mode)
        .map_err(DerivedContractError::source)?;
    if policy.mode == OverloadModeV1::Shed {
        return Ok(OverloadDispositionV1::Shed);
    }
    if index_mode == IndexMode::State {
        return Ok(OverloadDispositionV1::RefuseStateMode);
    }
    if matches!(ingest_authority, IngestAuthorityV1::ReplayableSource { .. }) {
        return Ok(OverloadDispositionV1::PauseReplayableSource);
    }
    match index_mode {
        IndexMode::AppendLog => Ok(OverloadDispositionV1::BufferRegisteredWal),
        IndexMode::Mutable if exact_mutable_overlay_available => Ok(OverloadDispositionV1::BufferRegisteredWal),
        IndexMode::Mutable => Ok(OverloadDispositionV1::RefuseMissingMutableOverlay),
        IndexMode::State => Ok(OverloadDispositionV1::RefuseStateMode),
    }
}

/// Exact collector-level pending-mutation coverage derived from registered WAL records.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PendingMutationOverlayV1 {
    pub format_version: u16,
    pub shard_key: ShardKey,
    pub writer_epoch: Epoch,
    pub registered_through: SeqNo,
    pub indexed_through: SeqNo,
    pub get_overlay_through: SeqNo,
    pub conditional_write_through: SeqNo,
    pub collector_visibility_through: SeqNo,
    pub subscription_through: SeqNo,
    pub pending_documents: u64,
    pub pending_tombstones: u64,
    pub pending_bytes: u64,
    pub oldest_age_millis: u64,
}

impl PendingMutationOverlayV1 {
    pub fn validate(self) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        let durable = self.registered_through.get();
        let has_no_pending_work = self.pending_documents == 0
            && self.pending_tombstones == 0
            && self.pending_bytes == 0
            && self.oldest_age_millis == 0;
        if self.writer_epoch.get() == 0
            || self.indexed_through.get() > durable
            || self.get_overlay_through.get() != durable
            || self.conditional_write_through.get() != durable
            || self.collector_visibility_through.get() != durable
            || self.subscription_through.get() != durable
            || self.pending_documents > MAX_PENDING_OVERLAY_DOCUMENTS
            || self.pending_tombstones > self.pending_documents
            || self.pending_bytes > MAX_OVERLOAD_BACKLOG_BYTES
            || self.oldest_age_millis > MAX_OVERLOAD_BACKLOG_AGE_MILLIS
            || (self.indexed_through == self.registered_through && !has_no_pending_work)
        {
            return Err(DerivedContractError::Violation(
                "pending mutation overlay does not exactly cover the registered WAL cut",
            ));
        }
        Ok(())
    }

    pub fn lag_operations(self) -> Result<u64, DerivedContractError> {
        self.validate()?;
        Ok(self.registered_through.get() - self.indexed_through.get())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OverloadShedReasonV1 {
    BacklogBytes,
    BacklogAge,
    CollectionShare,
    RecoveryWork,
    PendingOverlayDocuments,
}

/// Deterministic cap evaluation order used by every writer replica.
pub fn overload_shed_reason_v1(
    policy: OverloadPolicyV1,
    backlog_bytes: u64,
    backlog_age_millis: u64,
    collection_share_bytes: u64,
    recovery_ops_this_pass: u32,
    pending_overlay_documents: u64,
) -> Result<Option<OverloadShedReasonV1>, DerivedContractError> {
    policy.validate()?;
    Ok(if backlog_bytes > policy.maximum_backlog_bytes {
        Some(OverloadShedReasonV1::BacklogBytes)
    } else if backlog_age_millis > policy.maximum_backlog_age_millis {
        Some(OverloadShedReasonV1::BacklogAge)
    } else if collection_share_bytes > policy.maximum_collection_share_bytes {
        Some(OverloadShedReasonV1::CollectionShare)
    } else if recovery_ops_this_pass > policy.maximum_recovery_ops_per_pass {
        Some(OverloadShedReasonV1::RecoveryWork)
    } else if pending_overlay_documents > policy.maximum_pending_overlay_documents {
        Some(OverloadShedReasonV1::PendingOverlayDocuments)
    } else {
        None
    })
}

/// Honest numeric overload status for one exact writer incarnation.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct OverloadStatusV1 {
    pub format_version: u16,
    pub shard_key: ShardKey,
    pub writer_epoch: Epoch,
    pub configured_mode: OverloadModeV1,
    pub effective_disposition: OverloadDispositionV1,
    pub registered_through: SeqNo,
    pub indexed_through: SeqNo,
    pub lag_operations: u64,
    pub backlog_bytes: u64,
    pub oldest_age_millis: u64,
    pub collection_share_bytes: u64,
    pub recovery_ops_this_pass: u32,
    pub pending_overlay_documents: u64,
    pub shed_reason: Option<OverloadShedReasonV1>,
}

impl OverloadStatusV1 {
    pub fn validate(
        self,
        policy: OverloadPolicyV1,
        index_mode: IndexMode,
        ingest_authority: &IngestAuthorityV1,
        exact_mutable_overlay_available: bool,
    ) -> Result<(), DerivedContractError> {
        validate_format(self.format_version)?;
        policy.validate()?;
        if self.writer_epoch.get() == 0
            || self.shard_key.collection_index_ref() != policy.collection_index_ref
            || self.configured_mode != policy.mode
            || self.indexed_through.get() > self.registered_through.get()
            || self.lag_operations != self.registered_through.get() - self.indexed_through.get()
            || self.collection_share_bytes > self.backlog_bytes
        {
            return Err(DerivedContractError::Violation(
                "overload status watermarks or policy binding are inconsistent",
            ));
        }
        let expected_reason = overload_shed_reason_v1(
            policy,
            self.backlog_bytes,
            self.oldest_age_millis,
            self.collection_share_bytes,
            self.recovery_ops_this_pass,
            self.pending_overlay_documents,
        )?;
        let configured_disposition =
            overload_disposition_v1(policy, index_mode, ingest_authority, exact_mutable_overlay_available)?;
        let expected_disposition = if expected_reason.is_some() {
            OverloadDispositionV1::Shed
        } else {
            configured_disposition
        };
        if self.shed_reason != expected_reason || self.effective_disposition != expected_disposition {
            return Err(DerivedContractError::Violation(
                "overload status does not report the deterministic cap disposition",
            ));
        }
        Ok(())
    }
}

pub fn encode_result_cache_key_v1(value: &ResultCacheKeyV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_result_cache_key_v1(bytes: &[u8]) -> Result<ResultCacheKeyV1, DerivedContractError> {
    decode_current(bytes, ResultCacheKeyV1::validate)
}

pub fn encode_result_cache_budget_v1(value: &ResultCacheBudgetV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_result_cache_budget_v1(bytes: &[u8]) -> Result<ResultCacheBudgetV1, DerivedContractError> {
    decode_current(bytes, |value: &ResultCacheBudgetV1| (*value).validate())
}

pub fn encode_result_cache_metrics_v1(
    value: &ResultCacheMetricsV1,
    budget: ResultCacheBudgetV1,
) -> Result<Vec<u8>, DerivedContractError> {
    value.validate(budget)?;
    encode_bounded(value)
}

pub fn decode_result_cache_metrics_v1(
    bytes: &[u8],
    budget: ResultCacheBudgetV1,
) -> Result<ResultCacheMetricsV1, DerivedContractError> {
    decode_current(bytes, |value: &ResultCacheMetricsV1| (*value).validate(budget))
}

pub fn encode_field_heat_v1(value: &FieldHeatAggregateV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_field_heat_v1(bytes: &[u8]) -> Result<FieldHeatAggregateV1, DerivedContractError> {
    decode_current(bytes, FieldHeatAggregateV1::validate)
}

pub fn encode_field_promotion_policy_v1(value: &FieldPromotionPolicyV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_field_promotion_policy_v1(bytes: &[u8]) -> Result<FieldPromotionPolicyV1, DerivedContractError> {
    decode_current(bytes, |value: &FieldPromotionPolicyV1| (*value).validate())
}

pub fn encode_promoted_field_layout_v1(value: &PromotedFieldLayoutV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_promoted_field_layout_v1(bytes: &[u8]) -> Result<PromotedFieldLayoutV1, DerivedContractError> {
    decode_current(bytes, |value: &PromotedFieldLayoutV1| (*value).validate())
}

pub fn encode_field_promotion_status_v1(
    value: &FieldPromotionStatusV1,
    policy: FieldPromotionPolicyV1,
) -> Result<Vec<u8>, DerivedContractError> {
    value.validate(policy)?;
    encode_bounded(value)
}

pub fn decode_field_promotion_status_v1(
    bytes: &[u8],
    policy: FieldPromotionPolicyV1,
) -> Result<FieldPromotionStatusV1, DerivedContractError> {
    decode_current(bytes, |value: &FieldPromotionStatusV1| (*value).validate(policy))
}

pub fn encode_promoted_column_sidecar_v1(value: &PromotedColumnSidecarV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    let bytes = serde_json::to_vec(value).map_err(DerivedContractError::serde)?;
    if bytes.len() > MAX_PROMOTED_COLUMN_SIDECAR_BYTES {
        return Err(DerivedContractError::Budget {
            field: "promoted column sidecar bytes",
            observed: bytes.len() as u64,
            maximum: MAX_PROMOTED_COLUMN_SIDECAR_BYTES as u64,
        });
    }
    Ok(bytes)
}

pub fn decode_promoted_column_sidecar_v1(bytes: &[u8]) -> Result<PromotedColumnSidecarV1, DerivedContractError> {
    if bytes.len() > MAX_PROMOTED_COLUMN_SIDECAR_BYTES {
        return Err(DerivedContractError::Budget {
            field: "promoted column sidecar bytes",
            observed: bytes.len() as u64,
            maximum: MAX_PROMOTED_COLUMN_SIDECAR_BYTES as u64,
        });
    }
    #[derive(Deserialize)]
    struct FormatProbe {
        format_version: Option<u16>,
    }
    let probe: FormatProbe = serde_json::from_slice(bytes).map_err(DerivedContractError::serde)?;
    validate_format(probe.format_version.ok_or(DerivedContractError::MissingFormatVersion)?)?;
    let decoded: PromotedColumnSidecarV1 = serde_json::from_slice(bytes).map_err(DerivedContractError::serde)?;
    decoded.validate()?;
    if serde_json::to_vec(&decoded).map_err(DerivedContractError::serde)? != bytes {
        return Err(DerivedContractError::NonCanonical);
    }
    Ok(decoded)
}

pub fn encode_overload_policy_v1(value: &OverloadPolicyV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_overload_policy_v1(bytes: &[u8]) -> Result<OverloadPolicyV1, DerivedContractError> {
    decode_current(bytes, |value: &OverloadPolicyV1| (*value).validate())
}

pub fn encode_overload_status_v1(
    value: &OverloadStatusV1,
    policy: OverloadPolicyV1,
    index_mode: IndexMode,
    ingest_authority: &IngestAuthorityV1,
    exact_mutable_overlay_available: bool,
) -> Result<Vec<u8>, DerivedContractError> {
    value.validate(policy, index_mode, ingest_authority, exact_mutable_overlay_available)?;
    encode_bounded(value)
}

pub fn decode_overload_status_v1(
    bytes: &[u8],
    policy: OverloadPolicyV1,
    index_mode: IndexMode,
    ingest_authority: &IngestAuthorityV1,
    exact_mutable_overlay_available: bool,
) -> Result<OverloadStatusV1, DerivedContractError> {
    decode_current(bytes, |value: &OverloadStatusV1| {
        (*value).validate(policy, index_mode, ingest_authority, exact_mutable_overlay_available)
    })
}

pub fn encode_pending_overlay_v1(value: &PendingMutationOverlayV1) -> Result<Vec<u8>, DerivedContractError> {
    value.validate()?;
    encode_bounded(value)
}

pub fn decode_pending_overlay_v1(bytes: &[u8]) -> Result<PendingMutationOverlayV1, DerivedContractError> {
    decode_current(bytes, |value: &PendingMutationOverlayV1| (*value).validate())
}

fn encode_bounded<T: Serialize>(value: &T) -> Result<Vec<u8>, DerivedContractError> {
    let bytes = serde_json::to_vec(value).map_err(DerivedContractError::serde)?;
    if bytes.len() > MAX_DERIVED_CONTRACT_BYTES {
        return Err(DerivedContractError::Budget {
            field: "derived contract bytes",
            observed: bytes.len() as u64,
            maximum: MAX_DERIVED_CONTRACT_BYTES as u64,
        });
    }
    Ok(bytes)
}

fn decode_current<T>(
    bytes: &[u8],
    validate: impl FnOnce(&T) -> Result<(), DerivedContractError>,
) -> Result<T, DerivedContractError>
where
    T: DeserializeOwned + Serialize,
{
    #[derive(Deserialize)]
    struct FormatProbe {
        format_version: Option<u16>,
    }

    if bytes.len() > MAX_DERIVED_CONTRACT_BYTES {
        return Err(DerivedContractError::Budget {
            field: "derived contract bytes",
            observed: bytes.len() as u64,
            maximum: MAX_DERIVED_CONTRACT_BYTES as u64,
        });
    }
    let probe: FormatProbe = serde_json::from_slice(bytes).map_err(DerivedContractError::serde)?;
    let version = probe.format_version.ok_or(DerivedContractError::MissingFormatVersion)?;
    validate_format(version)?;
    let decoded: T = serde_json::from_slice(bytes).map_err(DerivedContractError::serde)?;
    validate(&decoded)?;
    if serde_json::to_vec(&decoded).map_err(DerivedContractError::serde)? != bytes {
        return Err(DerivedContractError::NonCanonical);
    }
    Ok(decoded)
}

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

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

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

    fn source(error: impl Display) -> Self {
        Self::Source(error.to_string())
    }
}

impl Display for DerivedContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedFormat(version) => write!(formatter, "unsupported derived/overload format {version}"),
            Self::MissingFormatVersion => formatter.write_str("derived/overload format version is missing"),
            Self::NonCanonical => formatter.write_str("derived/overload bytes are not canonical current format"),
            Self::Budget {
                field,
                observed,
                maximum,
            } => write!(formatter, "{field} {observed} exceeds {maximum}"),
            Self::Violation(message) => formatter.write_str(message),
            Self::ArithmeticOverflow => formatter.write_str("derived/overload arithmetic overflowed"),
            Self::Source(message) => write!(formatter, "derived/overload source authority refused input: {message}"),
            Self::Serde(message) => write!(formatter, "derived/overload codec refused input: {message}"),
        }
    }
}

impl Error for DerivedContractError {}

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

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

    fn cache_key() -> Result<ResultCacheKeyV1, Box<dyn Error>> {
        Ok(ResultCacheKeyV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            collection_security_generation: CollectionSecurityGeneration::new(1),
            authorized_scope_digest: AuthorizedScopeDigestV1::new([1; 32])?,
            policy_generation: PolicyGeneration::new(1),
            collection_reset_generation: CollectionResetGenerationV1::new(1)?,
            deletion_overlay_generation: DeletionOverlayGenerationV1::new(1)?,
            key_access_generation: KeyAccessGenerationV1::new(1)?,
            normalized_plan_digest: NormalizedPlanDigestV1::new([2; 32])?,
            output_semantics_digest: OutputSemanticsDigestV1::new([3; 32])?,
            expected_shard_count: 1,
            shard_generations: vec![CacheShardGenerationV1 {
                shard_id: ShardId::new(0),
                manifest_generation: ManifestGeneration::new(1),
            }],
            mapping_generation: SchemaGeneration::new(1),
            freshness: CacheFreshnessV1::Committed,
            reuse_scope: CacheReuseScopeV1::CompleteCommittedResult,
            from: 0,
            limit: 10,
            feature_flags_digest: FeatureFlagsDigestV1::new([4; 32])?,
            snapshot: CacheSnapshotBindingV1::Current,
        })
    }

    #[test]
    fn cache_key_is_exact_and_realtime_is_subresult_only() -> Result<(), Box<dyn Error>> {
        let key = cache_key()?;
        let bytes = encode_result_cache_key_v1(&key)?;
        assert_eq!(decode_result_cache_key_v1(&bytes)?, key);
        let mut stale = key.clone();
        stale.freshness = CacheFreshnessV1::Realtime;
        assert!(matches!(stale.validate(), Err(DerivedContractError::Violation(_))));
        stale.reuse_scope = CacheReuseScopeV1::CommittedSubresultOfRealtime;
        stale.validate()?;
        let budget = ResultCacheBudgetV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_id: CollectionId::new(1)?,
            maximum_entry_bytes: 1_024,
            maximum_collection_bytes: 4_096,
            maximum_collection_entries: 8,
        };
        let budget_bytes = encode_result_cache_budget_v1(&budget)?;
        assert_eq!(decode_result_cache_budget_v1(&budget_bytes)?, budget);
        let metrics = ResultCacheMetricsV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_id: CollectionId::new(1)?,
            hits: 1,
            misses: 1,
            admissions: 1,
            bypasses: 1,
            evictions: 0,
            collection_bytes: 1_024,
            collection_entries: 1,
        };
        let metrics_bytes = encode_result_cache_metrics_v1(&metrics, budget)?;
        assert_eq!(decode_result_cache_metrics_v1(&metrics_bytes, budget)?, metrics);
        Ok(())
    }

    #[test]
    fn promotion_is_physical_bounded_and_privacy_constrained() -> Result<(), Box<dyn Error>> {
        let heat = FieldHeatAggregateV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            mapping_generation: SchemaGeneration::new(1),
            field_path_digest: FieldPathDigestV1::new([7; 32])?,
            window_start_logical_millis: LogicalMillis::new(1),
            window_end_logical_millis: LogicalMillis::new(2),
            filter_uses: 4,
            sort_uses: 1,
            aggregation_uses: 1,
            distinct_query_count: 4,
            estimated_cardinality: 100,
            estimated_extra_bytes: 1_024,
            privacy: PromotionPrivacyV1::Ordinary,
            field_type: PromotedFieldTypeV1::Keyword,
            observed_type_conflict: false,
            path_policy_eligible: true,
        };
        let policy = FieldPromotionPolicyV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            mapping_generation: SchemaGeneration::new(1),
            policy_generation: PolicyGeneration::new(1),
            enabled: true,
            minimum_query_count: 3,
            maximum_promoted_fields: 4,
            maximum_cardinality: 1_000,
            maximum_extra_bytes: 2_048,
            allow_sensitive: false,
        };
        let policy_bytes = encode_field_promotion_policy_v1(&policy)?;
        assert_eq!(decode_field_promotion_policy_v1(&policy_bytes)?, policy);
        assert!(policy.admits(&heat, 0, 0)?);
        let mut restricted = heat.clone();
        restricted.privacy = PromotionPrivacyV1::Restricted;
        assert!(!policy.admits(&restricted, 0, 0)?);
        let mut conflict = heat.clone();
        conflict.observed_type_conflict = true;
        assert!(!policy.admits(&conflict, 0, 0)?);
        assert!(!policy.admits(&heat, policy.maximum_promoted_fields, 0)?);
        assert!(!policy.admits(&heat, 0, policy.maximum_extra_bytes)?);
        let mut stale_heat = heat.clone();
        stale_heat.mapping_generation = SchemaGeneration::new(2);
        assert!(matches!(
            policy.admits(&stale_heat, 0, 0),
            Err(DerivedContractError::Violation(_))
        ));
        let mut impossible_heat = heat.clone();
        impossible_heat.distinct_query_count = 7;
        assert!(matches!(
            impossible_heat.validate(),
            Err(DerivedContractError::Violation(_))
        ));
        let layout = PromotedFieldLayoutV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            segment_id: SegmentId::new(1),
            field_path_digest: heat.field_path_digest,
            mapping_generation: heat.mapping_generation,
            decision_generation: PromotionDecisionGenerationV1::new(1)?,
            field_type: heat.field_type,
        };
        let layout_bytes = encode_promoted_field_layout_v1(&layout)?;
        assert_eq!(decode_promoted_field_layout_v1(&layout_bytes)?, layout);
        let status = FieldPromotionStatusV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            mapping_generation: SchemaGeneration::new(1),
            policy_generation: PolicyGeneration::new(1),
            last_decision_generation: Some(PromotionDecisionGenerationV1::new(1)?),
            heat_records: 1,
            promoted_fields: 1,
            extra_bytes: 1_024,
            enabled: true,
        };
        let status_bytes = encode_field_promotion_status_v1(&status, policy)?;
        assert_eq!(decode_field_promotion_status_v1(&status_bytes, policy)?, status);
        let mut stale_status = status;
        stale_status.mapping_generation = SchemaGeneration::new(2);
        assert!(matches!(
            stale_status.validate(policy),
            Err(DerivedContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn promoted_column_sidecar_is_strict_current_and_segment_bound() -> Result<(), Box<dyn Error>> {
        let digest = field_path_digest_v1("service.latency")?;
        assert_eq!(digest, field_path_digest_v1("service.latency")?);
        assert_ne!(digest, field_path_digest_v1("service.duration")?);
        let sidecar = PromotedColumnSidecarV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            segment_id: SegmentId::new(4),
            mapping_generation: SchemaGeneration::new(2),
            document_count: 3,
            columns: vec![PromotedFieldColumnV1 {
                layout: PromotedFieldLayoutV1 {
                    format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
                    collection_index_ref: scope()?,
                    segment_id: SegmentId::new(4),
                    field_path_digest: digest,
                    mapping_generation: SchemaGeneration::new(2),
                    decision_generation: PromotionDecisionGenerationV1::new(3)?,
                    field_type: PromotedFieldTypeV1::U64,
                },
                rows: vec![
                    PromotedFieldRowV1 {
                        document_ordinal: 0,
                        value: PromotedFieldValueV1::U64(10),
                    },
                    PromotedFieldRowV1 {
                        document_ordinal: 2,
                        value: PromotedFieldValueV1::U64(20),
                    },
                ],
            }],
        };
        let bytes = encode_promoted_column_sidecar_v1(&sidecar)?;
        assert_eq!(decode_promoted_column_sidecar_v1(&bytes)?, sidecar);

        let mut future = serde_json::from_slice::<serde_json::Value>(&bytes)?;
        future["format_version"] = serde_json::json!(DERIVED_OVERLOAD_FORMAT_VERSION.saturating_add(1));
        assert!(matches!(
            decode_promoted_column_sidecar_v1(&serde_json::to_vec(&future)?),
            Err(DerivedContractError::UnsupportedFormat(version))
                if version == DERIVED_OVERLOAD_FORMAT_VERSION + 1
        ));
        let mut unknown = serde_json::from_slice::<serde_json::Value>(&bytes)?;
        unknown["unknown"] = serde_json::json!(true);
        assert!(matches!(
            decode_promoted_column_sidecar_v1(&serde_json::to_vec(&unknown)?),
            Err(DerivedContractError::Serde(_))
        ));
        let current_version = format!("\"format_version\":{DERIVED_OVERLOAD_FORMAT_VERSION}");
        let duplicate_version = format!("{current_version},{current_version}");
        let duplicate = String::from_utf8(bytes.clone())?.replacen(&current_version, &duplicate_version, 1);
        assert!(matches!(
            decode_promoted_column_sidecar_v1(duplicate.as_bytes()),
            Err(DerivedContractError::Serde(_))
        ));
        let mut wrong_type = sidecar.clone();
        wrong_type.columns[0].rows[0].value = PromotedFieldValueV1::Keyword("10".to_owned());
        assert!(matches!(
            encode_promoted_column_sidecar_v1(&wrong_type),
            Err(DerivedContractError::Violation(_))
        ));
        let mut negative_zero = sidecar.clone();
        negative_zero.columns[0].layout.field_type = PromotedFieldTypeV1::F64;
        negative_zero.columns[0].rows[0].value = PromotedFieldValueV1::F64Bits((-0.0_f64).to_bits());
        negative_zero.columns[0].rows[1].value = PromotedFieldValueV1::F64Bits(1.0_f64.to_bits());
        assert!(matches!(
            encode_promoted_column_sidecar_v1(&negative_zero),
            Err(DerivedContractError::Violation(_))
        ));
        let mut empty = sidecar.clone();
        empty.columns.clear();
        assert!(matches!(
            encode_promoted_column_sidecar_v1(&empty),
            Err(DerivedContractError::Violation(_))
        ));
        let mut duplicate_generation = sidecar.clone();
        let mut duplicate_column = duplicate_generation.columns[0].clone();
        duplicate_column.layout.field_path_digest = field_path_digest_v1("service.duration")?;
        duplicate_generation.columns.push(duplicate_column);
        assert!(matches!(
            encode_promoted_column_sidecar_v1(&duplicate_generation),
            Err(DerivedContractError::Violation(_))
        ));
        let mut unordered = sidecar;
        unordered.columns[0].rows.swap(0, 1);
        assert!(matches!(
            encode_promoted_column_sidecar_v1(&unordered),
            Err(DerivedContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn overload_buffer_never_becomes_a_second_authority() -> Result<(), Box<dyn Error>> {
        let policy = OverloadPolicyV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            collection_index_ref: scope()?,
            policy_generation: PolicyGeneration::new(1),
            mode: OverloadModeV1::Buffer,
            maximum_backlog_bytes: 1_024,
            maximum_backlog_age_millis: 100,
            maximum_collection_share_bytes: 512,
            maximum_recovery_ops_per_pass: 100,
            maximum_pending_overlay_documents: 100,
        };
        let policy_bytes = encode_overload_policy_v1(&policy)?;
        assert_eq!(decode_overload_policy_v1(&policy_bytes)?, policy);
        let push_authority = IngestAuthorityV1::push_with_wal();
        assert_eq!(
            overload_disposition_v1(policy, IndexMode::AppendLog, &push_authority, false)?,
            OverloadDispositionV1::BufferRegisteredWal
        );
        assert_eq!(
            overload_disposition_v1(policy, IndexMode::Mutable, &push_authority, false)?,
            OverloadDispositionV1::RefuseMissingMutableOverlay
        );
        assert_eq!(
            overload_disposition_v1(policy, IndexMode::State, &push_authority, true)?,
            OverloadDispositionV1::RefuseStateMode
        );
        let invalid_authority = IngestAuthorityV1::PushWithWal { format_version: 3 };
        assert!(matches!(
            overload_disposition_v1(policy, IndexMode::AppendLog, &invalid_authority, false),
            Err(DerivedContractError::Source(_))
        ));
        let overlay = PendingMutationOverlayV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            shard_key: ShardKey::new(scope()?, ShardId::new(0)),
            writer_epoch: Epoch::new(1),
            registered_through: SeqNo::new(7),
            indexed_through: SeqNo::new(3),
            get_overlay_through: SeqNo::new(7),
            conditional_write_through: SeqNo::new(7),
            collector_visibility_through: SeqNo::new(7),
            subscription_through: SeqNo::new(7),
            pending_documents: 4,
            pending_tombstones: 2,
            pending_bytes: 128,
            oldest_age_millis: 10,
        };
        overlay.validate()?;
        assert_eq!(overlay.lag_operations()?, 4);
        let mut inconsistent_zero_lag = overlay;
        inconsistent_zero_lag.indexed_through = inconsistent_zero_lag.registered_through;
        assert!(matches!(
            inconsistent_zero_lag.validate(),
            Err(DerivedContractError::Violation(_))
        ));
        assert_eq!(
            overload_shed_reason_v1(policy, 100, 10, 100, 101, 4)?,
            Some(OverloadShedReasonV1::RecoveryWork)
        );
        let status = OverloadStatusV1 {
            format_version: DERIVED_OVERLOAD_FORMAT_VERSION,
            shard_key: ShardKey::new(scope()?, ShardId::new(0)),
            writer_epoch: Epoch::new(1),
            configured_mode: OverloadModeV1::Buffer,
            effective_disposition: OverloadDispositionV1::BufferRegisteredWal,
            registered_through: SeqNo::new(7),
            indexed_through: SeqNo::new(3),
            lag_operations: 4,
            backlog_bytes: 100,
            oldest_age_millis: 10,
            collection_share_bytes: 100,
            recovery_ops_this_pass: 10,
            pending_overlay_documents: 4,
            shed_reason: None,
        };
        let status_bytes = encode_overload_status_v1(&status, policy, IndexMode::AppendLog, &push_authority, false)?;
        assert_eq!(
            decode_overload_status_v1(&status_bytes, policy, IndexMode::AppendLog, &push_authority, false,)?,
            status
        );
        let mut false_state_status = status;
        false_state_status.effective_disposition = OverloadDispositionV1::BufferRegisteredWal;
        assert!(matches!(
            false_state_status.validate(policy, IndexMode::State, &push_authority, true),
            Err(DerivedContractError::Violation(_))
        ));
        let mut impossible_share = status;
        impossible_share.collection_share_bytes = impossible_share.backlog_bytes + 1;
        assert!(matches!(
            impossible_share.validate(policy, IndexMode::AppendLog, &push_authority, false),
            Err(DerivedContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn codec_refuses_future_unknown_duplicate_and_noncanonical() -> Result<(), Box<dyn Error>> {
        let bytes = encode_result_cache_key_v1(&cache_key()?)?;
        let mut future: serde_json::Value = serde_json::from_slice(&bytes)?;
        future["format_version"] = serde_json::Value::from(DERIVED_OVERLOAD_FORMAT_VERSION.saturating_add(1));
        assert_eq!(
            decode_result_cache_key_v1(&serde_json::to_vec(&future)?),
            Err(DerivedContractError::UnsupportedFormat(
                DERIVED_OVERLOAD_FORMAT_VERSION + 1
            ))
        );
        let mut unknown: serde_json::Value = serde_json::from_slice(&bytes)?;
        unknown["unknown_cache_key"] = serde_json::Value::Bool(true);
        assert!(matches!(
            decode_result_cache_key_v1(&serde_json::to_vec(&unknown)?),
            Err(DerivedContractError::Serde(_))
        ));
        let text = String::from_utf8(bytes)?;
        let current_version = format!("\"format_version\":{DERIVED_OVERLOAD_FORMAT_VERSION}");
        let duplicate_version = format!("{current_version},{current_version}");
        let duplicate = text.replacen(&current_version, &duplicate_version, 1);
        assert!(matches!(
            decode_result_cache_key_v1(duplicate.as_bytes()),
            Err(DerivedContractError::Serde(_))
        ));
        let spaced = format!(" {text}");
        assert_eq!(
            decode_result_cache_key_v1(spaced.as_bytes()),
            Err(DerivedContractError::NonCanonical)
        );
        Ok(())
    }
}