lodestone_types/

lib.rs

//! Foundational, dependency-free types shared by Lodestone components.

use std::collections::{BTreeMap, BTreeSet};
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use serde::de::{self, MapAccess, SeqAccess, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

mod build_identity;
mod derived_overload;
mod identity;
mod lifecycle_completion;
mod local_development;
mod metric;
mod mutation_commit;
mod mutation_lifecycle;
mod operation_locator;
mod performance;
mod profile;
mod relationship;
mod replayable_source;
mod resource_spec;
mod shared_root_history;
mod state_outcomes;
mod workload_class;

pub use build_identity::*;
pub use derived_overload::*;
pub use identity::{
    BackendAuthorityId, BackendAuthorityPair, ClusterIdentity, ClusterStorageId, CollectionId, CollectionIndexRef,
    FenceToken, IdentityError, IndexAddress, IndexId, IndexName, NodeIncarnation, ObjectIncarnation, PrimaryTerm,
    ProtectionSnapshotId, RecoveryAttemptId, RecoveryEra, RecoveryGeneration, ShardKey,
};
pub use lifecycle_completion::*;
pub use local_development::{LocalDevelopmentInstanceId, LocalDevelopmentInstanceIdError};
pub use metric::*;
pub use mutation_commit::*;
pub use mutation_lifecycle::*;
pub use operation_locator::{
    ARTIFACT_CHUNK_ALIGNMENT, ARTIFACT_LOCATOR_FORMAT_VERSION, ArtifactLocatorV1, ArtifactReadRangeV1,
    MAX_ARTIFACT_CHUNK_BYTES, OPERATION_LOCATOR_FORMAT_VERSION, OperationAuthorityKindV1, OperationLocatorError,
    OperationLocatorV1, VerifiedArtifactChunkV1, verify_artifact_chunk_v1,
};
pub use performance::{
    BoundedStageSnapshotV2, PerformanceObservationV2, PerformanceStageReadingV2, PerformanceStageV2,
    bounded_stage_snapshot_v2, record_performance_stage_v2,
};
pub use profile::*;
pub use relationship::*;
pub use replayable_source::*;
pub use resource_spec::*;
pub use shared_root_history::*;
pub use state_outcomes::*;
pub use workload_class::*;

/// AWS partition is part of an S3 bucket's globally stable provider identity.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AwsPartition {
    Aws,
    AwsCn,
    AwsUsGov,
}

/// Provider-observed AWS bucket kind. A directory bucket's AZ ID is durable
/// identity and must exact-match the writer's provider-observed AZ ID.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum AwsBucketKind {
    GeneralPurpose,
    Directory { availability_zone_id: String },
}

impl AwsBucketKind {
    #[must_use]
    pub const fn is_directory(&self) -> bool {
        matches!(self, Self::Directory { .. })
    }

    #[must_use]
    pub fn availability_zone_id(&self) -> Option<&str> {
        match self {
            Self::GeneralPurpose => None,
            Self::Directory { availability_zone_id } => Some(availability_zone_id),
        }
    }
}

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

        impl $name {
            /// Constructs the semantic value.
            #[must_use]
            pub const fn new(value: $representation) -> Self {
                Self(value)
            }

            /// Returns the underlying representation.
            #[must_use]
            pub const fn get(self) -> $representation {
                self.0
            }

            /// Returns the next value, or `None` at the representation limit.
            #[must_use]
            pub const fn checked_next(self) -> Option<Self> {
                match self.0.checked_add(1) {
                    Some(value) => Some(Self(value)),
                    None => None,
                }
            }
        }

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

semantic_number!(
    ShardId,
    u16,
    "A semantic identifier for a shard in deterministic models and protocols."
);
semantic_number!(Epoch, u32, "A monotonically increasing shard-writer fencing epoch.");
semantic_number!(
    SeqNo,
    u64,
    "A per-shard monotonically increasing operation sequence number."
);
semantic_number!(
    ManifestGeneration,
    u64,
    "A monotonically increasing immutable-manifest generation."
);
semantic_number!(
    ContinuousEvaluationGeneration,
    u64,
    "An immutable semantic generation of one continuous temporal evaluation."
);
semantic_number!(
    ContinuousEvaluationRevision,
    u64,
    "The CAS revision of one continuous temporal evaluation's durable progress."
);
semantic_number!(
    SealedRootProofGeneration,
    u64,
    "A monotonically increasing sealed-root proof receipt generation. Zero is never issued, so it is the reserved never-proved value."
);
semantic_number!(
    ManifestGraphProofGeneration,
    u64,
    "A monotonically increasing post-recovery manifest-graph proof generation. Zero is never issued."
);
semantic_number!(
    RoutingHash,
    u32,
    "The stored routing hash used for shard-range selection."
);
semantic_number!(
    IdHash,
    u128,
    "The mutable-index document identity hash used by visibility overlays."
);
semantic_number!(SegmentId, u128, "A stable immutable segment identifier.");
semantic_number!(
    StateIncarnation,
    u64,
    "A fenced lifetime identity for one state primary/projection pair."
);
/// Immutable identity assigned to one vector field for the lifetime of an
/// index. Names may be presented by adapters, but sidecars bind this identity.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct VectorFieldId(u32);

impl VectorFieldId {
    /// Constructs a nonzero immutable vector-field identity.
    pub const fn new(value: u32) -> Result<Self, VectorFieldIdError> {
        if value == 0 {
            return Err(VectorFieldIdError);
        }
        Ok(Self(value))
    }

    /// Returns the underlying representation.
    #[must_use]
    pub const fn get(self) -> u32 {
        self.0
    }
}

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

/// Refusal to construct the reserved zero vector-field identity.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VectorFieldIdError;

impl Display for VectorFieldIdError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str("vector field identity must be nonzero")
    }
}

impl Error for VectorFieldIdError {}

impl<'de> Deserialize<'de> for VectorFieldId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        Self::new(u32::deserialize(deserializer)?).map_err(de::Error::custom)
    }
}
/// Upper-half segment identities are reserved for system-derived immutable
/// maintenance artifacts. Ordinary writer allocation is confined below this
/// boundary so a published maintenance segment cannot poison `max + 1`.
pub const RESERVED_SEGMENT_ID_NAMESPACE_START: u128 = 1_u128 << 127;

#[must_use]
pub const fn is_reserved_segment_id(segment_id: SegmentId) -> bool {
    segment_id.get() >= RESERVED_SEGMENT_ID_NAMESPACE_START
}

/// Returns the next ordinary segment identifier while ignoring system-owned
/// reserved artifacts already present in the same manifest. `None` is the
/// exact exhaustion result at the namespace boundary.
pub fn next_ordinary_segment_id<I>(current: I) -> Option<SegmentId>
where
    I: IntoIterator<Item = SegmentId>,
{
    let maximum = current
        .into_iter()
        .filter(|segment_id| !is_reserved_segment_id(*segment_id))
        .max();
    let next = maximum.map_or(Some(SegmentId::new(1)), SegmentId::checked_next)?;
    (!is_reserved_segment_id(next)).then_some(next)
}
semantic_number!(
    ChunkId,
    u64,
    "A node-local monotonically increasing WAL chunk identifier."
);
semantic_number!(NodeId, u64, "A semantic indexer-node identifier.");
semantic_number!(
    ControllerEpoch,
    u64,
    "A nonzero monotonically increasing placement-controller claim epoch."
);
semantic_number!(
    NodeLifecycleGeneration,
    u64,
    "A monotonically increasing generation of one node lifecycle record."
);
semantic_number!(
    CredentialGeneration,
    u64,
    "A monotonically increasing workload-credential binding generation."
);
semantic_number!(
    BootstrapGeneration,
    u64,
    "A monotonically increasing irreversible cluster-bootstrap generation."
);
semantic_number!(
    MembershipGeneration,
    u64,
    "A monotonically increasing external Raft membership generation."
);
semantic_number!(
    PlacementProgressGeneration,
    u64,
    "A monotonically increasing committed placement reconciliation progress generation."
);
semantic_number!(
    RangeTransitionGeneration,
    u64,
    "A monotonically increasing online split or merge transition generation."
);
semantic_number!(
    AssignmentGeneration,
    u64,
    "A monotonically increasing generation for one exact shard-route assignment."
);
semantic_number!(
    ReadLeaseGeneration,
    u64,
    "A monotonically increasing generation for one replicated generation-read lease."
);
semantic_number!(
    SchemaGeneration,
    u32,
    "A monotonically increasing mapping/schema generation."
);
semantic_number!(
    PolicyGeneration,
    u64,
    "A monotonically increasing authoritative index-policy generation."
);
semantic_number!(
    ResourceRevision,
    u64,
    "A monotonically increasing revision of one index resource specification."
);
semantic_number!(
    CrdGeneration,
    u64,
    "A nonzero Kubernetes metadata generation for one exact custom-resource incarnation."
);
semantic_number!(
    CatalogManagementGeneration,
    u64,
    "A monotonically increasing catalog-owned CAS generation for API/CRD management ownership."
);
semantic_number!(
    IndexRevision,
    u64,
    "A monotonically increasing revision of one immutable index incarnation's catalog record."
);
semantic_number!(
    HistoryLedgerRevision,
    u64,
    "A monotonically increasing CAS revision of one index's retained-history ledger."
);
semantic_number!(
    AdmissionGeneration,
    u64,
    "A monotonically increasing catalog admission fence for one index incarnation."
);
semantic_number!(
    RoutingRevision,
    u64,
    "A monotonically increasing revision of one index's complete routing partition."
);
semantic_number!(
    MemtableGeneration,
    u64,
    "A node-local active or frozen memtable generation."
);
semantic_number!(
    BackendAuthority,
    u64,
    "A stable scheduler partition for one authoritative durability backend."
);
semantic_number!(
    ClusterRevision,
    u64,
    "A monotonically increasing revision of the replicated cluster view."
);
semantic_number!(
    RootRevision,
    u64,
    "A monotonically increasing revision of the authoritative durable root set."
);
semantic_number!(
    LogicalMillis,
    u64,
    "Metastore-owned logical milliseconds used for leases without wall-clock dependence."
);
semantic_number!(
    TaskCheckpointGeneration,
    u64,
    "A monotonically increasing CAS generation for one durable task checkpoint."
);
semantic_number!(
    AdministrationRevision,
    u64,
    "A monotonically increasing revision of the collection-scoped administration catalog."
);
semantic_number!(
    AliasRevision,
    u64,
    "A monotonically increasing revision of one exact alias record."
);
semantic_number!(
    ComponentTemplateRevision,
    u64,
    "A monotonically increasing revision of one component-template incarnation."
);
semantic_number!(
    IndexTemplateRevision,
    u64,
    "A monotonically increasing revision of one index-template incarnation."
);
semantic_number!(
    IngestPipelineRevision,
    u64,
    "A monotonically increasing revision of one catalog-owned ingest pipeline."
);
semantic_number!(
    DataStreamRevision,
    u64,
    "A monotonically increasing revision of one data-stream incarnation."
);
semantic_number!(
    LifecyclePolicyRevision,
    u64,
    "A monotonically increasing revision of one lifecycle-policy incarnation."
);
semantic_number!(
    SecurityPolicyRevision,
    u64,
    "A monotonically increasing revision of the sole coordinated security-policy snapshot."
);
semantic_number!(
    CollectionSecurityGeneration,
    u64,
    "A monotonically increasing authorization generation for one Collection's public-security records."
);
semantic_number!(
    QuotaPolicyGeneration,
    u64,
    "A monotonically increasing quota-policy generation scoped to one Collection."
);
semantic_number!(
    SecurityRoleGeneration,
    u64,
    "A monotonically increasing generation of one Collection-local security role."
);
semantic_number!(
    SecurityPrincipalGeneration,
    u64,
    "A monotonically increasing generation of one Collection-local security principal."
);
semantic_number!(
    SourceAuthorizationGeneration,
    u64,
    "A monotonically increasing generation of one source-authorization projection for an exact Collection index."
);
semantic_number!(
    SecurityDirectSequence,
    u64,
    "A monotonically increasing position in the coordinated security direct-command replay journal."
);
semantic_number!(
    ApiKeyGeneration,
    u64,
    "A monotonically increasing generation of one API-key verifier record."
);

pub const AUTHORIZATION_BINDING_FORMAT_VERSION: u16 = 1;

/// Exact Collection-local authority used by one authenticated public request.
///
/// `ordering_revision` and `public_security_root_digest` retain the global
/// provenance of the decision. Callers must use [`Self::same_local_authority`]
/// for ordinary authorization equality so an unrelated Collection mutation
/// cannot invalidate admitted work.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AuthorizationBindingV1 {
    pub format_version: u16,
    pub recovery_era: RecoveryEra,
    pub public_security_root_digest: [u8; 32],
    pub ordering_revision: SecurityPolicyRevision,
    pub collection_id: CollectionId,
    pub collection_security_generation: CollectionSecurityGeneration,
    pub principal_id: PrincipalId,
    pub principal_generation: SecurityPrincipalGeneration,
    pub api_key_id: ApiKeyId,
    pub api_key_generation: ApiKeyGeneration,
    pub role_generations_digest: [u8; 32],
    pub quota_policy_generation: QuotaPolicyGeneration,
    pub authorized_scope_digest: [u8; 32],
}

impl AuthorizationBindingV1 {
    #[must_use]
    pub fn is_valid(self) -> bool {
        self.format_version == AUTHORIZATION_BINDING_FORMAT_VERSION
            && self.public_security_root_digest != [0; 32]
            && self.ordering_revision.get() != 0
            && self.collection_security_generation.get() != 0
            && self.principal_id.get() != 0
            && self.principal_generation.get() != 0
            && self.api_key_id.get() != 0
            && self.api_key_generation.get() != 0
            && self.role_generations_digest != [0; 32]
            && self.quota_policy_generation.get() != 0
            && self.authorized_scope_digest != [0; 32]
    }

    /// Compares only fences that can change the authority of this exact
    /// Collection/principal/key/scope path. Global root and ordering movement
    /// remain provenance and deliberately do not participate.
    #[must_use]
    pub fn same_local_authority(self, other: Self) -> bool {
        self.same_generation_authority(other)
            && self.role_generations_digest == other.role_generations_digest
            && self.authorized_scope_digest == other.authorized_scope_digest
    }

    /// Compares the local authorization revocation and mutation fences while
    /// permitting a caller to narrow or re-evaluate the admitted action/index
    /// scope. Quota owners compare `quota_policy_generation` separately; a
    /// quota-only change must not revoke authorization or queued resource work.
    #[must_use]
    pub fn same_generation_authority(self, other: Self) -> bool {
        self.is_valid()
            && other.is_valid()
            && self.format_version == other.format_version
            && self.recovery_era == other.recovery_era
            && self.collection_id == other.collection_id
            && self.collection_security_generation == other.collection_security_generation
            && self.principal_id == other.principal_id
            && self.principal_generation == other.principal_generation
            && self.api_key_id == other.api_key_id
            && self.api_key_generation == other.api_key_generation
    }
}
semantic_number!(
    AuditSequence,
    u64,
    "A recovery-backed sequence in the tamper-evident security audit chain."
);
semantic_number!(
    AuditRangeId,
    u64,
    "A recovery-backed identity for one fenced Gateway audit sequence range."
);
semantic_number!(
    QuotaGrantId,
    u64,
    "A recovery-backed identity for one aggregate Gateway quota grant."
);
semantic_number!(
    QuotaWindow,
    u64,
    "A coordinated logical quota-window identity that cannot be supplied by an adapter."
);
semantic_number!(
    ProcessMonotonicMillis,
    u64,
    "Process-monotonic milliseconds used only for local freshness, timeout, and lifetime bounds."
);
semantic_number!(
    UnixMillis,
    u64,
    "Unix milliseconds used only for certificate validity observations."
);

/// Nonzero process-lifetime identity for one exact Gateway incarnation.
///
/// This is separate from `NodeToken`: embedded all-role development
/// composition still needs an unforgeable process fence, while distributed
/// Gateway ownership additionally carries the exact `NodeToken`.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct GatewayIncarnation([u8; 32]);

impl GatewayIncarnation {
    pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
        if value == [0; 32] {
            return Err(IdentityError::Zero {
                kind: "GatewayIncarnation",
            });
        }
        Ok(Self(value))
    }

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

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

impl<'de> Deserialize<'de> for GatewayIncarnation {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(value).map_err(de::Error::custom)
    }
}

/// Exact TLS negotiation profile owned by one public listener.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PublicTlsProfileV1 {
    Http,
    Grpc,
    Native,
    Resp,
}

/// One intentional public Gateway protocol edge.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PublicGatewayListenerV1 {
    #[serde(rename = "api")]
    LodestoneApi,
    ElasticsearchHttp,
    OtlpHttp,
    JaegerHttp,
    OtlpGrpc,
    SubscriptionGrpc,
    SubscriptionWebSocket,
    Resp,
}

impl PublicGatewayListenerV1 {
    #[must_use]
    pub const fn required_tls_profile(self) -> PublicTlsProfileV1 {
        match self {
            Self::LodestoneApi => PublicTlsProfileV1::Native,
            Self::ElasticsearchHttp | Self::OtlpHttp | Self::JaegerHttp | Self::SubscriptionWebSocket => {
                PublicTlsProfileV1::Http
            }
            Self::OtlpGrpc | Self::SubscriptionGrpc => PublicTlsProfileV1::Grpc,
            Self::Resp => PublicTlsProfileV1::Resp,
        }
    }
}

/// One immutable production workload role. A distributed identity owns
/// exactly one role; `roles=all` is a development composition and never a
/// value of this enum.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum NodeRole {
    MetastoreVoter,
    Placement,
    Orchestrator,
    Gateway,
    Indexer,
    Searcher,
    Compactor,
}

/// The exact current incarnation of one durable node identity.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NodeToken {
    pub recovery_era: RecoveryEra,
    pub node_id: NodeId,
    pub node_incarnation: NodeIncarnation,
}

/// Domain-separated identity of one Kubernetes object incarnation.
///
/// A digest follows neither a human name nor a delete/recreate. The caller
/// derives it from the Kubernetes cluster UID, namespace, kind, and object
/// UID under the P7 domain.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct KubernetesObjectUidDigest([u8; 32]);

impl KubernetesObjectUidDigest {
    pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
        if value == [0; 32] {
            return Err(IdentityError::Zero {
                kind: "KubernetesObjectUidDigest",
            });
        }
        Ok(Self(value))
    }

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

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

impl<'de> Deserialize<'de> for KubernetesObjectUidDigest {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(value).map_err(de::Error::custom)
    }
}

impl NodeToken {
    pub fn new(
        recovery_era: RecoveryEra,
        node_id: NodeId,
        node_incarnation: NodeIncarnation,
    ) -> Result<Self, IdentityError> {
        if node_id.get() == 0 {
            return Err(IdentityError::Zero { kind: "NodeId" });
        }
        Ok(Self {
            recovery_era,
            node_id,
            node_incarnation,
        })
    }
}

/// Complete fence for placement-controller-owned mutations.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ControllerToken {
    pub recovery_era: RecoveryEra,
    pub controller_epoch: ControllerEpoch,
}

impl ControllerToken {
    pub fn new(recovery_era: RecoveryEra, controller_epoch: ControllerEpoch) -> Result<Self, IdentityError> {
        if controller_epoch.get() == 0 {
            return Err(IdentityError::Zero {
                kind: "ControllerEpoch",
            });
        }
        Ok(Self {
            recovery_era,
            controller_epoch,
        })
    }
}

/// A stable identifier for a durable background task.
///
/// JSON encodes this as a decimal string because JSON numbers and
/// `serde_json::Value` cannot losslessly carry the complete `u128` domain.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct TaskId(u128);

impl TaskId {
    #[must_use]
    pub const fn new(value: u128) -> Self {
        Self(value)
    }

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

    #[must_use]
    pub const fn checked_next(self) -> Option<Self> {
        match self.0.checked_add(1) {
            Some(value) => Some(Self(value)),
            None => None,
        }
    }
}

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

impl Serialize for TaskId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(self)
    }
}

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

        impl Visitor<'_> for TaskIdVisitor {
            type Value = TaskId;

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

            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
            where
                E: de::Error,
            {
                value
                    .parse::<u128>()
                    .map(TaskId)
                    .map_err(|_| E::custom("task id is outside the u128 domain"))
            }
        }

        deserializer.deserialize_str(TaskIdVisitor)
    }
}

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

        impl $name {
            /// Constructs the semantic identity.
            #[must_use]
            pub const fn new(value: u128) -> Self {
                Self(TaskId::new(value))
            }

            /// Returns the underlying recovery-backed allocation value.
            #[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.0, formatter)
            }
        }
    };
}

semantic_decimal_identity!(
    ComponentTemplateId,
    "A recovery-backed identity for one component-template incarnation."
);
semantic_decimal_identity!(
    IndexTemplateId,
    "A recovery-backed identity for one index-template incarnation."
);
semantic_decimal_identity!(
    IngestPipelineId,
    "A recovery-backed identity for one ingest-pipeline incarnation."
);
semantic_decimal_identity!(
    DataStreamId,
    "A recovery-backed identity for one data-stream incarnation."
);
semantic_decimal_identity!(
    LifecyclePolicyId,
    "A recovery-backed identity for one lifecycle-policy incarnation."
);
semantic_decimal_identity!(
    ContinuousEvaluationId,
    "A recovery-backed identity for one Collection-owned continuous temporal evaluation."
);
semantic_decimal_identity!(
    ExternalKeyResolverCapabilityId,
    "An opaque capability identity authorizing one external content-key resolver class."
);
semantic_decimal_identity!(
    PrincipalId,
    "A recovery-backed, never-reused security-principal identity."
);
semantic_decimal_identity!(ApiKeyId, "A recovery-backed, never-reused API-key identity.");
semantic_decimal_identity!(BackupId, "A recovery-backed, never-reused portable-backup identity.");

/// Stable idempotency identity of one administration action.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct AdministrationActionId([u8; 32]);

impl AdministrationActionId {
    /// Constructs a nonzero action identity.
    pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
        if value == [0; 32] {
            return Err(IdentityError::Zero {
                kind: "AdministrationActionId",
            });
        }
        Ok(Self(value))
    }

    /// Returns the stable action bytes.
    #[must_use]
    pub const fn get(self) -> [u8; 32] {
        self.0
    }

    /// Borrows the stable action bytes.
    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }
}

impl<'de> Deserialize<'de> for AdministrationActionId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(value).map_err(de::Error::custom)
    }
}

/// Stable idempotency identity of one security-policy or audit action.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct SecurityActionId([u8; 32]);

impl SecurityActionId {
    /// Constructs a nonzero action identity.
    pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
        if value == [0; 32] {
            return Err(IdentityError::Zero {
                kind: "SecurityActionId",
            });
        }
        Ok(Self(value))
    }

    /// Returns the stable action bytes.
    #[must_use]
    pub const fn get(self) -> [u8; 32] {
        self.0
    }

    /// Borrows the stable action bytes.
    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }
}

impl<'de> Deserialize<'de> for SecurityActionId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(value).map_err(de::Error::custom)
    }
}

/// Stable idempotency identity of one portable-backup lifecycle action.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct BackupActionId([u8; 32]);

impl BackupActionId {
    /// Constructs a nonzero action identity.
    pub fn new(value: [u8; 32]) -> Result<Self, IdentityError> {
        if value == [0; 32] {
            return Err(IdentityError::Zero { kind: "BackupActionId" });
        }
        Ok(Self(value))
    }

    /// Returns the stable action bytes.
    #[must_use]
    pub const fn get(self) -> [u8; 32] {
        self.0
    }

    /// Borrows the stable action bytes.
    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }
}

impl<'de> Deserialize<'de> for BackupActionId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(value).map_err(de::Error::custom)
    }
}

/// The exclusive upper bound of a shard routing-hash range.
///
/// This is a `u64` because the final range ends at `2^32`, one past the
/// largest value representable by [`RoutingHash`].
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct RoutingRangeEnd(u64);

impl RoutingRangeEnd {
    /// The exclusive upper bound of the complete u32 routing space.
    pub const FULL: Self = Self(u32::MAX as u64 + 1);

    /// Constructs a checked exclusive routing-range bound.
    pub fn new(value: u64) -> Result<Self, RoutingRangeError> {
        if value == 0 || value > Self::FULL.0 {
            Err(RoutingRangeError::InvalidEnd(value))
        } else {
            Ok(Self(value))
        }
    }

    /// Returns the underlying exclusive bound.
    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }
}

/// A distribution-ready shard identity and its current fencing epoch.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ShardDescriptor {
    pub shard_id: ShardId,
    pub epoch: Epoch,
    pub hash_start: RoutingHash,
    pub hash_end: RoutingRangeEnd,
}

impl ShardDescriptor {
    /// Constructs a non-empty hash-range descriptor.
    pub fn new(
        shard_id: ShardId,
        epoch: Epoch,
        hash_start: RoutingHash,
        hash_end: RoutingRangeEnd,
    ) -> Result<Self, RoutingRangeError> {
        if u64::from(hash_start.get()) >= hash_end.get() {
            return Err(RoutingRangeError::EmptyOrReversed {
                start: hash_start,
                end: hash_end,
            });
        }
        Ok(Self {
            shard_id,
            epoch,
            hash_start,
            hash_end,
        })
    }

    /// Returns whether the stored routing hash belongs to this descriptor.
    #[must_use]
    pub const fn contains(self, routing_hash: RoutingHash) -> bool {
        let value = routing_hash.get() as u64;
        value >= self.hash_start.get() as u64 && value < self.hash_end.get()
    }
}

/// Invalid routing-range metadata.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RoutingRangeError {
    InvalidEnd(u64),
    EmptyOrReversed { start: RoutingHash, end: RoutingRangeEnd },
}

impl Display for RoutingRangeError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidEnd(end) => write!(
                formatter,
                "routing range end {end} is outside 1..={}",
                RoutingRangeEnd::FULL.get()
            ),
            Self::EmptyOrReversed { start, end } => write!(
                formatter,
                "routing range start {start} must be below exclusive end {}",
                end.get()
            ),
        }
    }
}

impl Error for RoutingRangeError {}

/// Versioned durability policy identity.
///
/// The policy names describe behavior rather than one provider product.
/// `ZonalLowLatency` is admitted only when the selected object leaf proves the
/// required zonal semantics. The remaining stable names are deliberate
/// reservations and return a typed unavailable refusal.
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DurabilityClass {
    RegionalObject,
    ZonalLowLatency,
    Relaxed,
    Ephemeral,
}

impl DurabilityClass {
    /// Returns the stable wire name.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::RegionalObject => "regional_object",
            Self::ZonalLowLatency => "zonal_low_latency",
            Self::Relaxed => "relaxed",
            Self::Ephemeral => "ephemeral",
        }
    }

    /// Refuses every durability class without an admitted execution protocol.
    pub const fn require_available(self) -> Result<(), DurabilityClassUnavailable> {
        match self {
            Self::RegionalObject | Self::ZonalLowLatency => Ok(()),
            Self::Relaxed | Self::Ephemeral => Err(DurabilityClassUnavailable { class: self }),
        }
    }

    /// Refuses classes without an admitted state-index acknowledgement model.
    ///
    /// P12.0 admits `ephemeral` at a whole-incarnation memory boundary, but
    /// does not make it reachable through the ordinary registered-WAL writer.
    pub const fn require_state_available(self) -> Result<(), DurabilityClassUnavailable> {
        match self {
            Self::RegionalObject | Self::ZonalLowLatency | Self::Ephemeral => Ok(()),
            Self::Relaxed => Err(DurabilityClassUnavailable { class: self }),
        }
    }
}

impl Display for DurabilityClass {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// A recognized durability policy that has no admitted execution backend.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DurabilityClassUnavailable {
    pub class: DurabilityClass,
}

impl Display for DurabilityClassUnavailable {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(
            formatter,
            "durability class '{}' is recognized but unavailable",
            self.class
        )
    }
}

impl Error for DurabilityClassUnavailable {}

/// Client response visibility independent of durability.
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum VisibilityMode {
    #[default]
    Applied,
    WaitFor,
}

/// The durable acknowledgement boundary actually achieved.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DurabilityBoundary {
    Registered,
    /// P12 state is visible only inside the current fenced RAM incarnation.
    IncarnationMemory,
}

/// Maximum failure loss promised by a successful acknowledgement.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum DurabilityLoss {
    Zero,
    WholeIncarnation,
}

/// Observable write-pipeline progress.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum WriteStage {
    Validated,
    Admitted,
    Sequenced,
    ObjectPut,
    Registered,
    AppliedToActiveMemtable,
    MicroCommitted,
    Sealed,
}

/// Per-index mutation mode.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum IndexMode {
    Mutable,
    AppendLog,
    /// Primary truth is a fenced, fully resident state table; Tantivy is an
    /// asynchronously derived mutable projection.
    State,
}

impl IndexMode {
    /// Whether indexed versions require identity replacement and tombstones.
    #[must_use]
    pub const fn has_mutation_visibility(self) -> bool {
        matches!(self, Self::Mutable | Self::State)
    }

    #[must_use]
    pub const fn is_state(self) -> bool {
        matches!(self, Self::State)
    }
}

/// Current version of the normalized semantic mapping model shared by the
/// catalog and Tantivy compiler. Format 8 adds the closed mapping-owned record
/// model used by native typed records, format 9 adds relationships, and format
/// 10 adds privacy-bound continuous-profile batches. Older mappings are
/// deliberately refused rather than reinterpreted.
pub const MAPPING_FORMAT_VERSION: u16 = 10;

/// Maximum fields admitted by one normalized mapping.
pub const MAX_MAPPING_FIELDS: usize = 16_384;

/// Version of the vector mapping extension embedded in the catalog mapping.
pub const VECTOR_MAPPING_FORMAT_VERSION: u16 = 2;

/// Version of the sparse-vector mapping extension embedded in the catalog.
pub const SPARSE_VECTOR_MAPPING_FORMAT_VERSION: u16 = 1;

/// Version of the vector-set mapping extension embedded in the catalog.
pub const VECTOR_SET_MAPPING_FORMAT_VERSION: u16 = 1;

/// Maximum vector fields admitted by one mapping generation.
pub const MAX_VECTOR_MAPPING_FIELDS: usize = 8;

/// Hard logical dimension ceiling shared by mappings, source values, queries,
/// codecs, and generated protocol projections.
pub const MAX_VECTOR_DIMENSIONS: u16 = 16_384;

/// Maximum encoded payload admitted for one dense value.
pub const MAX_VECTOR_BYTES_PER_FIELD: usize = 64 * 1_024;

/// Maximum encoded dense-vector payload admitted for one document before WAL.
pub const MAX_VECTOR_BYTES_PER_DOCUMENT: usize = 64 * 1_024;

/// Maximum encoded dense-vector payload admitted for one query branch.
pub const MAX_VECTOR_BYTES_PER_QUERY: usize = 64 * 1_024;

/// Maximum aggregate encoded dense-vector payload admitted in one request.
pub const MAX_VECTOR_BYTES_PER_REQUEST: usize = 16 * MAX_VECTOR_BYTES_PER_QUERY;

/// Maximum numeric sparse dimension admitted by mappings and values.
pub const MAX_SPARSE_VECTOR_DIMENSION: u32 = 16_777_215;

/// Maximum non-zero entries admitted for one sparse field or query.
pub const MAX_SPARSE_VECTOR_NONZERO: usize = 4_096;

/// Maximum encoded sparse payload admitted for one field or query. Each
/// entry owns one u32 dimension and one mapping-selected weight.
pub const MAX_SPARSE_VECTOR_BYTES: usize = 32 * 1_024;

/// Maximum token vectors admitted for one document vector-set field.
pub const MAX_VECTOR_SET_DOCUMENT_VECTORS: usize = 128;

/// Maximum token vectors admitted for one late-interaction query branch.
pub const MAX_VECTOR_SET_QUERY_VECTORS: usize = 32;

/// Maximum canonical element bytes admitted for one document vector set.
pub const MAX_VECTOR_SET_BYTES_PER_FIELD: usize = 64 * 1_024;

/// Maximum canonical element bytes admitted for one query vector set.
pub const MAX_VECTOR_SET_BYTES_PER_QUERY: usize = 64 * 1_024;

/// Maximum vector records retained by one mutable fresh-generation overlay.
pub const MAX_FRESH_VECTOR_RECORDS: usize = 16_384;

/// Maximum decoded vector bytes retained by one mutable fresh-generation
/// overlay. This is deliberately independent of one document's byte ceiling.
pub const MAX_FRESH_VECTOR_BYTES: usize = 128 * 1_024 * 1_024;

/// Absolute ceiling for one immutable-vector build's reconstructible scratch.
/// Operators may and normally do assign a smaller resource budget.
pub const MAX_VECTOR_BUILD_SCRATCH_BYTES: u64 = 16 * 1_024 * 1_024 * 1_024;

/// Absolute ceiling for one process's decoded immutable-vector cache. The
/// runtime resource specification remains the smaller operational authority.
pub const MAX_VECTOR_DECODED_CACHE_BYTES: u64 = 16 * 1_024 * 1_024 * 1_024;

/// Hard logical ceiling for vector bytes returned as explicitly requested
/// result features. Ordinary source bytes use the ordinary response budget.
pub const MAX_VECTOR_BYTES_PER_RESPONSE: usize = 64 * 1_024;

/// Logical dense element representation. Physical ANN strategy remains a
/// reconstructible implementation choice and is not encoded here.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DenseVectorElementType {
    F32,
    F16,
    I8,
}

/// Logical sparse weight representation. Dimension identifiers remain u32
/// and are never hashed or dictionary-resolved by the engine.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SparseVectorWeightType {
    F32,
    F16,
}

impl SparseVectorWeightType {
    #[must_use]
    pub const fn width_bytes(self) -> usize {
        match self {
            Self::F32 => 4,
            Self::F16 => 2,
        }
    }
}

/// One caller-supplied sparse coordinate. Canonical values are strictly
/// increasing by dimension and have a finite, non-zero weight.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SparseVectorEntry {
    pub dimension: u32,
    #[serde(deserialize_with = "deserialize_sparse_weight")]
    pub weight: f64,
}

fn deserialize_sparse_weight<'de, D>(deserializer: D) -> Result<f64, D::Error>
where
    D: Deserializer<'de>,
{
    serde_json::Number::deserialize(deserializer)?
        .to_string()
        .parse::<f64>()
        .map_err(de::Error::custom)
}

/// Native sparse value. Object form keeps this distinct from dense arrays and
/// leaves no string-feature hashing or dictionary semantics to infer.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SparseVectorValue {
    #[serde(deserialize_with = "deserialize_bounded_sparse_entries")]
    pub entries: Vec<SparseVectorEntry>,
}

/// Exact mapping semantics required to encode one sparse value.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SparseVectorDefinition {
    pub maximum_dimension: u32,
    pub weight_type: SparseVectorWeightType,
}

/// One validated sparse coordinate in the mapping-selected logical precision.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct DecodedSparseVectorEntry {
    pub dimension: u32,
    pub weight: f32,
}

impl SparseVectorDefinition {
    pub fn validate(self) -> Result<(), SparseVectorValueError> {
        if self.maximum_dimension > MAX_SPARSE_VECTOR_DIMENSION {
            return Err(SparseVectorValueError::InvalidMaximumDimension(self.maximum_dimension));
        }
        let maximum_bytes = MAX_SPARSE_VECTOR_NONZERO
            .checked_mul(4_usize.saturating_add(self.weight_type.width_bytes()))
            .ok_or(SparseVectorValueError::ByteLengthOverflow)?;
        if maximum_bytes > MAX_SPARSE_VECTOR_BYTES {
            return Err(SparseVectorValueError::FieldBytesExceeded(maximum_bytes));
        }
        Ok(())
    }
}

impl SparseVectorValue {
    pub fn decode(
        &self,
        definition: SparseVectorDefinition,
    ) -> Result<Vec<DecodedSparseVectorEntry>, SparseVectorValueError> {
        definition.validate()?;
        self.validate_bounded(definition, MAX_SPARSE_VECTOR_BYTES)?;
        self.entries
            .iter()
            .enumerate()
            .map(|(ordinal, entry)| {
                let weight = entry.weight as f32;
                if !weight.is_finite() {
                    return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
                }
                let weight = match definition.weight_type {
                    SparseVectorWeightType::F32 => canonical_vector_zero(weight),
                    SparseVectorWeightType::F16 => {
                        let bits = f32_to_f16_rne(weight).ok_or(SparseVectorValueError::F16Overflow(ordinal))?;
                        canonical_vector_zero(f16_to_f32(bits))
                    }
                };
                if weight == 0.0 {
                    return Err(SparseVectorValueError::ZeroWeight(ordinal));
                }
                Ok(DecodedSparseVectorEntry {
                    dimension: entry.dimension,
                    weight,
                })
            })
            .collect()
    }

    pub fn validate_bounded(
        &self,
        definition: SparseVectorDefinition,
        maximum_bytes: usize,
    ) -> Result<usize, SparseVectorValueError> {
        definition.validate()?;
        if self.entries.is_empty() {
            return Err(SparseVectorValueError::Empty);
        }
        if self.entries.len() > MAX_SPARSE_VECTOR_NONZERO {
            return Err(SparseVectorValueError::TooManyEntries(self.entries.len()));
        }
        let bytes = self
            .entries
            .len()
            .checked_mul(4_usize.saturating_add(definition.weight_type.width_bytes()))
            .ok_or(SparseVectorValueError::ByteLengthOverflow)?;
        if bytes > maximum_bytes.min(MAX_SPARSE_VECTOR_BYTES) {
            return Err(SparseVectorValueError::FieldBytesExceeded(bytes));
        }
        let mut previous = None;
        for (ordinal, entry) in self.entries.iter().enumerate() {
            if entry.dimension > definition.maximum_dimension {
                return Err(SparseVectorValueError::DimensionOutOfRange {
                    ordinal,
                    dimension: entry.dimension,
                    maximum: definition.maximum_dimension,
                });
            }
            if previous.is_some_and(|dimension| entry.dimension <= dimension) {
                return Err(SparseVectorValueError::DimensionsNotStrict(ordinal));
            }
            if !entry.weight.is_finite() {
                return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
            }
            if entry.weight == 0.0 {
                return Err(SparseVectorValueError::ZeroWeight(ordinal));
            }
            previous = Some(entry.dimension);
        }
        Ok(bytes)
    }

    pub fn canonical_native_bytes(
        &self,
        definition: SparseVectorDefinition,
    ) -> Result<Vec<u8>, SparseVectorValueError> {
        let decoded = self.decode(definition)?;
        let mut bytes = Vec::with_capacity(
            decoded
                .len()
                .saturating_mul(4_usize.saturating_add(definition.weight_type.width_bytes())),
        );
        for entry in decoded {
            bytes.extend_from_slice(&entry.dimension.to_le_bytes());
            match definition.weight_type {
                SparseVectorWeightType::F32 => bytes.extend_from_slice(&entry.weight.to_bits().to_le_bytes()),
                SparseVectorWeightType::F16 => {
                    let bits = f32_to_f16_rne(entry.weight).ok_or(SparseVectorValueError::F16Overflow(0))?;
                    bytes.extend_from_slice(&bits.to_le_bytes());
                }
            }
        }
        Ok(bytes)
    }
}

/// Decodes canonical interleaved `(u32 dimension, weight)` bytes. Immutable
/// bundle readers use this to reject malformed postings without inventing a
/// second sparse codec.
pub fn decode_sparse_native_bytes(
    definition: SparseVectorDefinition,
    bytes: &[u8],
) -> Result<Vec<DecodedSparseVectorEntry>, SparseVectorValueError> {
    definition.validate()?;
    let entry_bytes = 4_usize.saturating_add(definition.weight_type.width_bytes());
    if bytes.is_empty() || !bytes.len().is_multiple_of(entry_bytes) {
        return Err(SparseVectorValueError::FieldBytesExceeded(bytes.len()));
    }
    let count = bytes.len() / entry_bytes;
    if count > MAX_SPARSE_VECTOR_NONZERO || bytes.len() > MAX_SPARSE_VECTOR_BYTES {
        return Err(SparseVectorValueError::FieldBytesExceeded(bytes.len()));
    }
    let mut entries = Vec::with_capacity(count);
    let mut previous = None;
    for (ordinal, chunk) in bytes.chunks_exact(entry_bytes).enumerate() {
        let dimension = u32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
        if dimension > definition.maximum_dimension {
            return Err(SparseVectorValueError::DimensionOutOfRange {
                ordinal,
                dimension,
                maximum: definition.maximum_dimension,
            });
        }
        if previous.is_some_and(|value| dimension <= value) {
            return Err(SparseVectorValueError::DimensionsNotStrict(ordinal));
        }
        let weight = match definition.weight_type {
            SparseVectorWeightType::F32 => f32::from_le_bytes([chunk[4], chunk[5], chunk[6], chunk[7]]),
            SparseVectorWeightType::F16 => f16_to_f32(u16::from_le_bytes([chunk[4], chunk[5]])),
        };
        if !weight.is_finite() {
            return Err(SparseVectorValueError::NonFiniteWeight(ordinal));
        }
        let weight = canonical_vector_zero(weight);
        if weight == 0.0 {
            return Err(SparseVectorValueError::ZeroWeight(ordinal));
        }
        entries.push(DecodedSparseVectorEntry { dimension, weight });
        previous = Some(dimension);
    }
    Ok(entries)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SparseVectorValueError {
    Empty,
    InvalidMaximumDimension(u32),
    TooManyEntries(usize),
    ByteLengthOverflow,
    FieldBytesExceeded(usize),
    DimensionsNotStrict(usize),
    DimensionOutOfRange {
        ordinal: usize,
        dimension: u32,
        maximum: u32,
    },
    NonFiniteWeight(usize),
    ZeroWeight(usize),
    F16Overflow(usize),
}

impl Display for SparseVectorValueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => formatter.write_str("sparse vector value is empty"),
            Self::InvalidMaximumDimension(value) => {
                write!(formatter, "sparse maximum dimension {value} is invalid")
            }
            Self::TooManyEntries(value) => write!(formatter, "sparse vector has {value} non-zero entries"),
            Self::ByteLengthOverflow => formatter.write_str("sparse vector byte length overflowed"),
            Self::FieldBytesExceeded(value) => {
                write!(formatter, "sparse vector uses {value} bytes above its field ceiling")
            }
            Self::DimensionsNotStrict(ordinal) => {
                write!(formatter, "sparse vector dimension {ordinal} is duplicate or unsorted")
            }
            Self::DimensionOutOfRange {
                ordinal,
                dimension,
                maximum,
            } => write!(
                formatter,
                "sparse vector dimension {ordinal} value {dimension} exceeds {maximum}"
            ),
            Self::NonFiniteWeight(ordinal) => write!(formatter, "sparse vector weight {ordinal} is non-finite"),
            Self::ZeroWeight(ordinal) => write!(formatter, "sparse vector weight {ordinal} is zero"),
            Self::F16Overflow(ordinal) => write!(formatter, "sparse f16 weight {ordinal} overflows binary16"),
        }
    }
}

impl Error for SparseVectorValueError {}

fn deserialize_bounded_sparse_entries<'de, D>(deserializer: D) -> Result<Vec<SparseVectorEntry>, D::Error>
where
    D: Deserializer<'de>,
{
    struct SparseEntriesVisitor;

    impl<'de> serde::de::Visitor<'de> for SparseEntriesVisitor {
        type Value = Vec<SparseVectorEntry>;

        fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
            formatter.write_str("a bounded sparse entry sequence")
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: serde::de::SeqAccess<'de>,
        {
            if sequence
                .size_hint()
                .is_some_and(|length| length > MAX_SPARSE_VECTOR_NONZERO)
            {
                return Err(de::Error::custom("sparse vector exceeds the non-zero ceiling"));
            }
            let mut entries =
                Vec::with_capacity(sequence.size_hint().unwrap_or_default().min(MAX_SPARSE_VECTOR_NONZERO));
            while let Some(entry) = sequence.next_element::<SparseVectorEntry>()? {
                if entries.len() == MAX_SPARSE_VECTOR_NONZERO {
                    return Err(de::Error::custom("sparse vector exceeds the non-zero ceiling"));
                }
                entries.push(entry);
            }
            Ok(entries)
        }
    }

    deserializer.deserialize_seq(SparseEntriesVisitor)
}

impl DenseVectorElementType {
    #[must_use]
    pub const fn width_bytes(self) -> usize {
        match self {
            Self::F32 => 4,
            Self::F16 => 2,
            Self::I8 => 1,
        }
    }
}

/// Positive finite scale for an `i8` logical dense field. The canonical wire
/// form is the shortest Rust decimal spelling of the exact f32 value.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct VectorQuantizationScale(u32);

impl VectorQuantizationScale {
    pub fn new(value: f32) -> Result<Self, DenseVectorValueError> {
        if !value.is_finite() || value <= 0.0 {
            return Err(DenseVectorValueError::InvalidQuantizationScale);
        }
        Ok(Self(value.to_bits()))
    }

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

impl Serialize for VectorQuantizationScale {
    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 VectorQuantizationScale {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let text = String::deserialize(deserializer)?;
        let value = text.parse::<f32>().map_err(de::Error::custom)?;
        if value.to_string() != text {
            return Err(de::Error::custom("vector quantization scale is not canonical"));
        }
        Self::new(value).map_err(de::Error::custom)
    }
}

/// Only little-endian binary vector values are canonical.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DenseVectorByteOrder {
    LittleEndian,
}

/// Binary half of the native dense `VectorValue` union.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DenseVectorBinaryValue {
    pub element_type: DenseVectorElementType,
    pub byte_order: DenseVectorByteOrder,
    pub data_base64: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub quantization_scale: Option<VectorQuantizationScale>,
}

/// One native dense value. Document source also admits the numeric variant as
/// its compact bare-array spelling; exact `_source` bytes are never rewritten.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum DenseVectorValue {
    Numeric {
        #[serde(deserialize_with = "deserialize_bounded_dense_numbers")]
        values: Vec<f64>,
    },
    Binary {
        #[serde(flatten)]
        value: DenseVectorBinaryValue,
    },
}

impl DenseVectorValue {
    pub fn decode(&self, definition: DenseVectorDefinition) -> Result<Vec<f32>, DenseVectorValueError> {
        definition.validate()?;
        match self {
            Self::Numeric { values } => decode_dense_numeric_value(definition, values),
            Self::Binary { value } => decode_dense_binary_value(definition, value),
        }
    }

    /// Returns the canonical native element bytes without metric
    /// normalization. These bytes are reconstructible from exact source and
    /// let immutable accelerators preserve the mapping's declared element
    /// representation instead of narrowing normalized scoring coordinates.
    pub fn canonical_native_bytes(&self, definition: DenseVectorDefinition) -> Result<Vec<u8>, DenseVectorValueError> {
        definition.validate()?;
        match self {
            Self::Numeric { values } => encode_dense_numeric_native_bytes(definition, values),
            Self::Binary { value } => decode_dense_binary_native_bytes(definition, value),
        }
    }

    #[must_use]
    pub fn element_count(&self) -> Option<usize> {
        match self {
            Self::Numeric { values } => Some(values.len()),
            Self::Binary { .. } => None,
        }
    }

    pub fn validate_bounded(&self, maximum_bytes: usize) -> Result<usize, DenseVectorValueError> {
        match self {
            Self::Numeric { values } => {
                if values.is_empty() {
                    return Err(DenseVectorValueError::Empty);
                }
                let bytes = values
                    .len()
                    .checked_mul(DenseVectorElementType::F32.width_bytes())
                    .ok_or(DenseVectorValueError::ByteLengthOverflow)?;
                if values.len() > usize::from(MAX_VECTOR_DIMENSIONS) || bytes > maximum_bytes {
                    return Err(DenseVectorValueError::FieldBytesExceeded(bytes));
                }
                if let Some(ordinal) = values.iter().position(|value| !value.is_finite()) {
                    return Err(DenseVectorValueError::NonFiniteElement(ordinal));
                }
                Ok(bytes)
            }
            Self::Binary { value } => {
                match (value.element_type, value.quantization_scale) {
                    (DenseVectorElementType::I8, Some(_))
                    | (DenseVectorElementType::F32 | DenseVectorElementType::F16, None) => {}
                    (DenseVectorElementType::I8, None) => {
                        return Err(DenseVectorValueError::MissingQuantizationScale);
                    }
                    (DenseVectorElementType::F32 | DenseVectorElementType::F16, Some(_)) => {
                        return Err(DenseVectorValueError::UnexpectedQuantizationScale);
                    }
                }
                if value.data_base64.is_empty() || value.data_base64.len() > maximum_bytes.div_ceil(3) * 4 {
                    return Err(DenseVectorValueError::FieldBytesExceeded(value.data_base64.len()));
                }
                let bytes = BASE64_STANDARD
                    .decode(value.data_base64.as_bytes())
                    .map_err(|_| DenseVectorValueError::InvalidBase64)?;
                if bytes.is_empty()
                    || bytes.len() > maximum_bytes
                    || !bytes.len().is_multiple_of(value.element_type.width_bytes())
                    || bytes.len() / value.element_type.width_bytes() > usize::from(MAX_VECTOR_DIMENSIONS)
                {
                    return Err(DenseVectorValueError::FieldBytesExceeded(bytes.len()));
                }
                if BASE64_STANDARD.encode(&bytes) != value.data_base64 {
                    return Err(DenseVectorValueError::InvalidBase64);
                }
                Ok(bytes.len())
            }
        }
    }
}

/// Exact mapping semantics required to decode one dense value.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DenseVectorDefinition {
    pub dimensions: u16,
    pub element_type: DenseVectorElementType,
    pub metric: VectorMetric,
    pub quantization_scale: Option<VectorQuantizationScale>,
}

impl DenseVectorDefinition {
    pub fn validate(self) -> Result<(), DenseVectorValueError> {
        if self.dimensions == 0 || self.dimensions > MAX_VECTOR_DIMENSIONS {
            return Err(DenseVectorValueError::InvalidDimensions(self.dimensions));
        }
        let bytes = usize::from(self.dimensions)
            .checked_mul(self.element_type.width_bytes())
            .ok_or(DenseVectorValueError::ByteLengthOverflow)?;
        if bytes > MAX_VECTOR_BYTES_PER_FIELD {
            return Err(DenseVectorValueError::FieldBytesExceeded(bytes));
        }
        match (self.element_type, self.quantization_scale) {
            (DenseVectorElementType::I8, Some(_))
            | (DenseVectorElementType::F32 | DenseVectorElementType::F16, None) => Ok(()),
            (DenseVectorElementType::I8, None) => Err(DenseVectorValueError::MissingQuantizationScale),
            (DenseVectorElementType::F32 | DenseVectorElementType::F16, Some(_)) => {
                Err(DenseVectorValueError::UnexpectedQuantizationScale)
            }
        }
    }

    #[must_use]
    pub const fn encoded_bytes(self) -> usize {
        self.dimensions as usize * self.element_type.width_bytes()
    }
}

/// A vector-set value is an explicit object and can never be confused with one
/// dense numeric array. Each member uses the ordinary dense value codec.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct VectorSetValue {
    #[serde(deserialize_with = "deserialize_bounded_vector_set")]
    pub vectors: Vec<DenseVectorValue>,
}

/// Mapping-selected semantics for one bounded document vector set.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VectorSetDefinition {
    pub dense: DenseVectorDefinition,
    pub maximum_vectors: u16,
}

/// One decoded vector set and its per-vector canonical element bytes.
#[derive(Clone, Debug, PartialEq)]
pub struct DecodedVectorSet {
    pub vectors: Vec<Vec<f32>>,
    pub native_vectors: Vec<Vec<u8>>,
    pub native_bytes: usize,
}

impl VectorSetDefinition {
    pub fn validate(self) -> Result<(), VectorSetValueError> {
        self.dense.validate().map_err(VectorSetValueError::Dense)?;
        if self.maximum_vectors == 0 || usize::from(self.maximum_vectors) > MAX_VECTOR_SET_DOCUMENT_VECTORS {
            return Err(VectorSetValueError::InvalidMaximumVectors(self.maximum_vectors));
        }
        let maximum_bytes = usize::from(self.maximum_vectors)
            .checked_mul(self.dense.encoded_bytes())
            .ok_or(VectorSetValueError::ByteLengthOverflow)?;
        if maximum_bytes > MAX_VECTOR_SET_BYTES_PER_FIELD {
            return Err(VectorSetValueError::FieldBytesExceeded(maximum_bytes));
        }
        Ok(())
    }
}

impl VectorSetValue {
    pub fn decode(
        &self,
        definition: VectorSetDefinition,
        maximum_vectors: usize,
        maximum_bytes: usize,
    ) -> Result<DecodedVectorSet, VectorSetValueError> {
        definition.validate()?;
        let admitted_vectors = maximum_vectors
            .min(usize::from(definition.maximum_vectors))
            .min(MAX_VECTOR_SET_DOCUMENT_VECTORS);
        if self.vectors.is_empty() {
            return Err(VectorSetValueError::Empty);
        }
        if self.vectors.len() > admitted_vectors {
            return Err(VectorSetValueError::TooManyVectors(self.vectors.len()));
        }
        let native_bytes = self
            .vectors
            .len()
            .checked_mul(definition.dense.encoded_bytes())
            .ok_or(VectorSetValueError::ByteLengthOverflow)?;
        if native_bytes > maximum_bytes.min(MAX_VECTOR_SET_BYTES_PER_FIELD) {
            return Err(VectorSetValueError::FieldBytesExceeded(native_bytes));
        }
        let mut vectors = Vec::with_capacity(self.vectors.len());
        let mut native_vectors = Vec::with_capacity(self.vectors.len());
        for value in &self.vectors {
            vectors.push(value.decode(definition.dense).map_err(VectorSetValueError::Dense)?);
            native_vectors.push(
                value
                    .canonical_native_bytes(definition.dense)
                    .map_err(VectorSetValueError::Dense)?,
            );
        }
        Ok(DecodedVectorSet {
            vectors,
            native_vectors,
            native_bytes,
        })
    }
}

fn deserialize_bounded_vector_set<'de, D>(deserializer: D) -> Result<Vec<DenseVectorValue>, D::Error>
where
    D: Deserializer<'de>,
{
    struct VectorSetVisitor;

    impl<'de> serde::de::Visitor<'de> for VectorSetVisitor {
        type Value = Vec<DenseVectorValue>;

        fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
            formatter.write_str("a bounded sequence of dense vector values")
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: serde::de::SeqAccess<'de>,
        {
            if sequence
                .size_hint()
                .is_some_and(|length| length > MAX_VECTOR_SET_DOCUMENT_VECTORS)
            {
                return Err(de::Error::custom(
                    "vector set exceeds the document vector-count ceiling",
                ));
            }
            let mut vectors = Vec::with_capacity(
                sequence
                    .size_hint()
                    .unwrap_or_default()
                    .min(MAX_VECTOR_SET_DOCUMENT_VECTORS),
            );
            while let Some(vector) = sequence.next_element::<DenseVectorValue>()? {
                if vectors.len() == MAX_VECTOR_SET_DOCUMENT_VECTORS {
                    return Err(de::Error::custom(
                        "vector set exceeds the document vector-count ceiling",
                    ));
                }
                vectors.push(vector);
            }
            Ok(vectors)
        }
    }

    deserializer.deserialize_seq(VectorSetVisitor)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum VectorSetValueError {
    Empty,
    InvalidMaximumVectors(u16),
    TooManyVectors(usize),
    FieldBytesExceeded(usize),
    ByteLengthOverflow,
    Dense(DenseVectorValueError),
}

impl Display for VectorSetValueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => formatter.write_str("vector set cannot be empty"),
            Self::InvalidMaximumVectors(value) => {
                write!(formatter, "vector-set maximum vector count {value} is invalid")
            }
            Self::TooManyVectors(value) => write!(formatter, "vector set contains {value} vectors"),
            Self::FieldBytesExceeded(value) => write!(formatter, "vector set uses {value} element bytes"),
            Self::ByteLengthOverflow => formatter.write_str("vector-set byte length overflowed"),
            Self::Dense(error) => write!(formatter, "vector-set member is invalid: {error}"),
        }
    }
}

impl Error for VectorSetValueError {}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DenseVectorValueError {
    Empty,
    InvalidDimensions(u16),
    DimensionMismatch { expected: u16, observed: usize },
    InvalidQuantizationScale,
    MissingQuantizationScale,
    UnexpectedQuantizationScale,
    QuantizationScaleMismatch,
    ElementTypeMismatch,
    ByteLengthOverflow,
    FieldBytesExceeded(usize),
    NonFiniteElement(usize),
    F16Overflow(usize),
    I8ElementOutOfRange(usize),
    InvalidBase64Length { expected: usize, observed: usize },
    InvalidBase64,
    DecodedByteLength { expected: usize, observed: usize },
    ZeroNorm,
}

impl Display for DenseVectorValueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => formatter.write_str("dense vector value is empty"),
            Self::InvalidDimensions(value) => write!(formatter, "dense vector dimensions {value} are invalid"),
            Self::DimensionMismatch { expected, observed } => {
                write!(formatter, "dense vector has {observed} elements; expected {expected}")
            }
            Self::InvalidQuantizationScale => {
                formatter.write_str("dense i8 quantization scale must be positive and finite")
            }
            Self::MissingQuantizationScale => {
                formatter.write_str("dense i8 value requires quantization scale metadata")
            }
            Self::UnexpectedQuantizationScale => {
                formatter.write_str("dense f32/f16 value cannot carry quantization metadata")
            }
            Self::QuantizationScaleMismatch => {
                formatter.write_str("dense i8 value quantization scale differs from its mapping")
            }
            Self::ElementTypeMismatch => {
                formatter.write_str("dense binary value element type differs from its mapping")
            }
            Self::ByteLengthOverflow => formatter.write_str("dense vector byte length overflowed"),
            Self::FieldBytesExceeded(value) => {
                write!(formatter, "dense vector uses {value} bytes above its field ceiling")
            }
            Self::NonFiniteElement(ordinal) => write!(formatter, "dense vector element {ordinal} is non-finite"),
            Self::F16Overflow(ordinal) => write!(formatter, "dense f16 element {ordinal} overflows binary16"),
            Self::I8ElementOutOfRange(ordinal) => {
                write!(
                    formatter,
                    "dense i8 element {ordinal} is not an exact integer in -128..=127"
                )
            }
            Self::InvalidBase64Length { expected, observed } => {
                write!(
                    formatter,
                    "dense binary base64 has {observed} bytes; expected {expected}"
                )
            }
            Self::InvalidBase64 => formatter.write_str("dense binary value is not canonical base64"),
            Self::DecodedByteLength { expected, observed } => {
                write!(
                    formatter,
                    "dense binary value has {observed} decoded bytes; expected {expected}"
                )
            }
            Self::ZeroNorm => formatter.write_str("dense cosine vector has zero norm"),
        }
    }
}

impl Error for DenseVectorValueError {}

fn deserialize_bounded_dense_numbers<'de, D>(deserializer: D) -> Result<Vec<f64>, D::Error>
where
    D: Deserializer<'de>,
{
    struct DenseNumbersVisitor;

    impl<'de> Visitor<'de> for DenseNumbersVisitor {
        type Value = Vec<f64>;

        fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
            write!(formatter, "at most {MAX_VECTOR_DIMENSIONS} dense vector numbers")
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: SeqAccess<'de>,
        {
            if sequence
                .size_hint()
                .is_some_and(|length| length > usize::from(MAX_VECTOR_DIMENSIONS))
            {
                return Err(de::Error::custom("dense vector exceeds the element ceiling"));
            }
            let mut values = Vec::with_capacity(
                sequence
                    .size_hint()
                    .unwrap_or_default()
                    .min(usize::from(MAX_VECTOR_DIMENSIONS)),
            );
            while let Some(value) = sequence.next_element::<f64>()? {
                if values.len() == usize::from(MAX_VECTOR_DIMENSIONS) {
                    return Err(de::Error::custom("dense vector exceeds the element ceiling"));
                }
                values.push(value);
            }
            Ok(values)
        }
    }

    deserializer.deserialize_seq(DenseNumbersVisitor)
}

pub fn decode_dense_numeric_value(
    definition: DenseVectorDefinition,
    values: &[f64],
) -> Result<Vec<f32>, DenseVectorValueError> {
    let bytes = encode_dense_numeric_native_bytes(definition, values)?;
    decode_dense_native_bytes(definition, &bytes)
}

pub fn decode_dense_binary_value(
    definition: DenseVectorDefinition,
    value: &DenseVectorBinaryValue,
) -> Result<Vec<f32>, DenseVectorValueError> {
    let bytes = decode_dense_binary_native_bytes(definition, value)?;
    decode_dense_native_bytes(definition, &bytes)
}

fn decode_dense_binary_native_bytes(
    definition: DenseVectorDefinition,
    value: &DenseVectorBinaryValue,
) -> Result<Vec<u8>, DenseVectorValueError> {
    definition.validate()?;
    if value.element_type != definition.element_type {
        return Err(DenseVectorValueError::ElementTypeMismatch);
    }
    if value.quantization_scale != definition.quantization_scale {
        return Err(DenseVectorValueError::QuantizationScaleMismatch);
    }
    let expected_bytes = definition.encoded_bytes();
    let expected_base64 = expected_bytes.div_ceil(3) * 4;
    if value.data_base64.len() != expected_base64 {
        return Err(DenseVectorValueError::InvalidBase64Length {
            expected: expected_base64,
            observed: value.data_base64.len(),
        });
    }
    let bytes = BASE64_STANDARD
        .decode(value.data_base64.as_bytes())
        .map_err(|_| DenseVectorValueError::InvalidBase64)?;
    if bytes.len() != expected_bytes {
        return Err(DenseVectorValueError::DecodedByteLength {
            expected: expected_bytes,
            observed: bytes.len(),
        });
    }
    if BASE64_STANDARD.encode(&bytes) != value.data_base64 {
        return Err(DenseVectorValueError::InvalidBase64);
    }
    Ok(bytes)
}

fn encode_dense_numeric_native_bytes(
    definition: DenseVectorDefinition,
    values: &[f64],
) -> Result<Vec<u8>, DenseVectorValueError> {
    definition.validate()?;
    if values.len() != usize::from(definition.dimensions) {
        return Err(DenseVectorValueError::DimensionMismatch {
            expected: definition.dimensions,
            observed: values.len(),
        });
    }
    let mut bytes = Vec::with_capacity(definition.encoded_bytes());
    for (ordinal, value) in values.iter().copied().enumerate() {
        if !value.is_finite() {
            return Err(DenseVectorValueError::NonFiniteElement(ordinal));
        }
        match definition.element_type {
            DenseVectorElementType::F32 => {
                let value = canonical_vector_zero(value as f32);
                if !value.is_finite() {
                    return Err(DenseVectorValueError::NonFiniteElement(ordinal));
                }
                bytes.extend_from_slice(&value.to_bits().to_le_bytes());
            }
            DenseVectorElementType::F16 => {
                let value = value as f32;
                if !value.is_finite() {
                    return Err(DenseVectorValueError::F16Overflow(ordinal));
                }
                let bits = f32_to_f16_rne(value).ok_or(DenseVectorValueError::F16Overflow(ordinal))?;
                bytes.extend_from_slice(&bits.to_le_bytes());
            }
            DenseVectorElementType::I8 => {
                if value.fract() != 0.0 || !(-128.0..=127.0).contains(&value) {
                    return Err(DenseVectorValueError::I8ElementOutOfRange(ordinal));
                }
                bytes.push(value as i8 as u8);
            }
        }
    }
    Ok(bytes)
}

/// Decodes one exact native element sequence and applies the mapping's metric
/// normalization. Durable bundle readers use this instead of widening and
/// then re-encoding a caller value.
pub fn decode_dense_native_bytes(
    definition: DenseVectorDefinition,
    bytes: &[u8],
) -> Result<Vec<f32>, DenseVectorValueError> {
    definition.validate()?;
    let expected = definition.encoded_bytes();
    if bytes.len() != expected {
        return Err(DenseVectorValueError::DecodedByteLength {
            expected,
            observed: bytes.len(),
        });
    }
    let mut decoded = Vec::with_capacity(usize::from(definition.dimensions));
    match definition.element_type {
        DenseVectorElementType::F32 => {
            for (ordinal, chunk) in bytes.as_chunks::<4>().0.iter().enumerate() {
                let value = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
                if !value.is_finite() {
                    return Err(DenseVectorValueError::NonFiniteElement(ordinal));
                }
                decoded.push(canonical_vector_zero(value));
            }
        }
        DenseVectorElementType::F16 => {
            for (ordinal, chunk) in bytes.as_chunks::<2>().0.iter().enumerate() {
                let value = f16_to_f32(u16::from_le_bytes([chunk[0], chunk[1]]));
                if !value.is_finite() {
                    return Err(DenseVectorValueError::NonFiniteElement(ordinal));
                }
                decoded.push(canonical_vector_zero(value));
            }
        }
        DenseVectorElementType::I8 => {
            let scale = definition
                .quantization_scale
                .ok_or(DenseVectorValueError::MissingQuantizationScale)?
                .get();
            decoded.extend(
                bytes
                    .iter()
                    .map(|value| canonical_vector_zero((*value as i8 as f32) * scale)),
            );
        }
    }
    normalize_dense_for_metric(definition.metric, &mut decoded)?;
    Ok(decoded)
}

fn normalize_dense_for_metric(metric: VectorMetric, values: &mut [f32]) -> Result<(), DenseVectorValueError> {
    if metric != VectorMetric::Cosine {
        return Ok(());
    }
    let squared_norm = values.iter().try_fold(0.0_f64, |sum, value| {
        let next = sum + f64::from(*value) * f64::from(*value);
        next.is_finite().then_some(next)
    });
    let norm = squared_norm
        .map(f64::sqrt)
        .filter(|value| *value > 0.0 && value.is_finite())
        .ok_or(DenseVectorValueError::ZeroNorm)?;
    for value in values {
        *value = canonical_vector_zero((f64::from(*value) / norm) as f32);
        if !value.is_finite() {
            return Err(DenseVectorValueError::NonFiniteElement(0));
        }
    }
    Ok(())
}

const fn canonical_vector_zero(value: f32) -> f32 {
    if value == 0.0 { 0.0 } else { value }
}

fn f32_to_f16_rne(value: f32) -> Option<u16> {
    if !value.is_finite() {
        return None;
    }
    let bits = canonical_vector_zero(value).to_bits();
    let sign = ((bits >> 16) & 0x8000) as u16;
    let exponent = ((bits >> 23) & 0xff) as i32 - 127;
    let fraction = bits & 0x7f_ffff;
    if exponent > 15 {
        return None;
    }
    if exponent >= -14 {
        let mut half_exponent = (exponent + 15) as u16;
        let mut half_fraction = (fraction >> 13) as u16;
        let remainder = fraction & 0x1fff;
        if remainder > 0x1000 || remainder == 0x1000 && half_fraction & 1 == 1 {
            half_fraction += 1;
            if half_fraction == 0x400 {
                half_fraction = 0;
                half_exponent += 1;
                if half_exponent >= 0x1f {
                    return None;
                }
            }
        }
        return Some(sign | (half_exponent << 10) | half_fraction);
    }
    let significand = u64::from((1_u32 << 23) | fraction);
    let shift = u32::try_from(-exponent - 1).ok()?;
    let rounded = round_shift_right_ties_even(significand, shift);
    if rounded >= 0x400 {
        Some(sign | 0x0400)
    } else {
        Some(sign | rounded as u16)
    }
}

fn round_shift_right_ties_even(value: u64, shift: u32) -> u64 {
    if shift == 0 {
        return value;
    }
    if shift >= 64 {
        return 0;
    }
    let quotient = value >> shift;
    let remainder_mask = (1_u64 << shift) - 1;
    let remainder = value & remainder_mask;
    let halfway = 1_u64 << (shift - 1);
    quotient + u64::from(remainder > halfway || remainder == halfway && quotient & 1 == 1)
}

fn f16_to_f32(bits: u16) -> f32 {
    let sign = u32::from(bits & 0x8000) << 16;
    let exponent = (bits >> 10) & 0x1f;
    let fraction = bits & 0x03ff;
    match (exponent, fraction) {
        (0, 0) => f32::from_bits(sign),
        (0, _) => {
            let value = f32::from(fraction) * 2.0_f32.powi(-24);
            if sign == 0 { value } else { -value }
        }
        (0x1f, _) => f32::from_bits(sign | 0x7f80_0000 | (u32::from(fraction) << 13)),
        _ => {
            let f32_exponent = u32::from(exponent) + 112;
            f32::from_bits(sign | (f32_exponent << 23) | (u32::from(fraction) << 13))
        }
    }
}

/// Maximum UTF-8 bytes admitted by one mapped field path.
pub const MAX_MAPPING_FIELD_NAME_BYTES: usize = 1_024;

/// Normalized mapping semantics for one schema generation.
///
/// Durable catalog records own the canonical bytes for this exact type.
/// Execution crates compile it but do not define a second mapping model.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Mapping {
    pub format_version: u16,
    pub schema_generation: SchemaGeneration,
    pub mode: IndexMode,
    pub fields: BTreeMap<String, FieldMapping>,
    /// Semantics owned by the sole current mapping format.
    pub extensions: MappingExtensions,
}

impl Mapping {
    #[must_use]
    pub fn empty(mode: IndexMode) -> Self {
        Self {
            format_version: MAPPING_FORMAT_VERSION,
            schema_generation: SchemaGeneration::new(1),
            mode,
            fields: BTreeMap::new(),
            extensions: crate::MappingExtensions::default(),
        }
    }

    /// Completes format-4 metadata whose defaults are intrinsic to the native
    /// mapping model. Adapter-specific policy defaults are never applied here.
    #[must_use]
    pub fn normalized(mut self) -> Self {
        if self.format_version == MAPPING_FORMAT_VERSION {
            if self.extensions.analyzer_identities.is_empty() {
                self.extensions.analyzer_identities = self
                    .fields
                    .iter()
                    .filter(|(_, definition)| definition.field_type == FieldType::Text)
                    .map(|(field, _)| AnalyzerIdentity {
                        field: field.clone(),
                        index_analyzer: AnalyzerName::Standard,
                        search_analyzer: AnalyzerName::Standard,
                    })
                    .collect();
            }
            if self.extensions.keyword_limits.is_empty() {
                self.extensions.keyword_limits = self
                    .fields
                    .iter()
                    .filter(|(_, definition)| definition.field_type == FieldType::Keyword)
                    .map(|(field, _)| KeywordFieldPolicy {
                        field: field.clone(),
                        max_bytes: DEFAULT_KEYWORD_IGNORE_ABOVE,
                        normalizer: KeywordNormalizer::Keyword,
                    })
                    .collect();
            }
            if self.extensions.nested_paths.is_empty() {
                self.extensions.nested_paths = self
                    .fields
                    .iter()
                    .filter(|(_, definition)| definition.field_type == FieldType::Nested)
                    .map(|(field, _)| field.clone())
                    .collect();
            }
        }
        self
    }

    /// Decodes exact canonical catalog/public bytes after a minimal,
    /// duplicate-safe current-format probe.
    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, MappingCodecError> {
        if bytes.len() > MAX_MAPPING_BYTES {
            return Err(MappingCodecError::TooLarge(bytes.len()));
        }
        let mut probe = serde_json::Deserializer::from_slice(bytes);
        let version = MappingVersionProbe::deserialize(&mut probe).map_err(|error| {
            let message = error.to_string();
            if message.starts_with("duplicate mapping key '") {
                MappingCodecError::DuplicateKey(message)
            } else {
                MappingCodecError::Malformed(message)
            }
        })?;
        probe
            .end()
            .map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
        if version != MAPPING_FORMAT_VERSION {
            return Err(MappingCodecError::UnsupportedVersion(version));
        }
        let mapping: Self =
            serde_json::from_slice(bytes).map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
        mapping.validate().map_err(MappingCodecError::Invalid)?;
        let canonical =
            serde_json::to_vec(&mapping).map_err(|error| MappingCodecError::Malformed(error.to_string()))?;
        if canonical != bytes {
            return Err(MappingCodecError::NonCanonical);
        }
        Ok(mapping)
    }

    pub fn encode_canonical(&self) -> Result<Vec<u8>, MappingCodecError> {
        self.validate().map_err(MappingCodecError::Invalid)?;
        serde_json::to_vec(self).map_err(|error| MappingCodecError::Malformed(error.to_string()))
    }

    /// Validates the dependency-free semantic mapping contract.
    pub fn validate(&self) -> Result<(), MappingValidationError> {
        if self.format_version != MAPPING_FORMAT_VERSION {
            return Err(MappingValidationError::UnsupportedVersion(self.format_version));
        }
        let normalized = self.clone().normalized();
        let mapping = &normalized;
        if mapping.schema_generation.get() == 0 {
            return Err(MappingValidationError::ZeroSchemaGeneration);
        }
        if mapping.fields.len() > MAX_MAPPING_FIELDS {
            return Err(MappingValidationError::TooManyFields(mapping.fields.len()));
        }
        let mut vector_field_ids = std::collections::BTreeSet::new();
        let mut vector_fields = 0_usize;
        for (name, definition) in &mapping.fields {
            validate_mapping_field_name(name)?;
            for (separator, _) in name.match_indices('.') {
                let ancestor = &name[..separator];
                if mapping.fields.contains_key(ancestor)
                    && !mapping.extensions.nested_paths.iter().any(|path| path == ancestor)
                    && mapping
                        .extensions
                        .multi_field_projections
                        .iter()
                        .find(|projection| projection.child == *name)
                        .is_none_or(|projection| projection.parent != ancestor)
                {
                    return Err(MappingValidationError::ConflictingFieldPaths {
                        ancestor: ancestor.to_owned(),
                        descendant: name.clone(),
                    });
                }
            }
            match definition.field_type {
                FieldType::Vector {
                    format_version,
                    field_id,
                    dimensions,
                    metric,
                    element_type,
                    quantization_scale,
                } => {
                    vector_fields = vector_fields.saturating_add(1);
                    if format_version != VECTOR_MAPPING_FORMAT_VERSION {
                        return Err(MappingValidationError::UnsupportedVectorVersion(format_version));
                    }
                    if dimensions == 0 || dimensions > MAX_VECTOR_DIMENSIONS {
                        return Err(MappingValidationError::InvalidVectorDimensions {
                            field: name.clone(),
                            dimensions,
                        });
                    }
                    DenseVectorDefinition {
                        dimensions,
                        element_type,
                        metric,
                        quantization_scale,
                    }
                    .validate()
                    .map_err(|_| MappingValidationError::InvalidVectorElementContract(name.clone()))?;
                    if definition.stored {
                        return Err(MappingValidationError::StoredVectorField(name.clone()));
                    }
                    if !vector_field_ids.insert(field_id) {
                        return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
                    }
                }
                FieldType::SparseVector {
                    format_version,
                    field_id,
                    maximum_dimension,
                    weight_type,
                } => {
                    vector_fields = vector_fields.saturating_add(1);
                    if format_version != SPARSE_VECTOR_MAPPING_FORMAT_VERSION {
                        return Err(MappingValidationError::UnsupportedSparseVectorVersion(format_version));
                    }
                    SparseVectorDefinition {
                        maximum_dimension,
                        weight_type,
                    }
                    .validate()
                    .map_err(|_| MappingValidationError::InvalidSparseVectorContract(name.clone()))?;
                    if definition.stored {
                        return Err(MappingValidationError::StoredVectorField(name.clone()));
                    }
                    if !vector_field_ids.insert(field_id) {
                        return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
                    }
                }
                FieldType::VectorSet {
                    format_version,
                    field_id,
                    dimensions,
                    metric,
                    element_type,
                    quantization_scale,
                    maximum_vectors,
                } => {
                    vector_fields = vector_fields.saturating_add(1);
                    if format_version != VECTOR_SET_MAPPING_FORMAT_VERSION {
                        return Err(MappingValidationError::UnsupportedVectorSetVersion(format_version));
                    }
                    VectorSetDefinition {
                        dense: DenseVectorDefinition {
                            dimensions,
                            element_type,
                            metric,
                            quantization_scale,
                        },
                        maximum_vectors,
                    }
                    .validate()
                    .map_err(|_| MappingValidationError::InvalidVectorSetContract(name.clone()))?;
                    if definition.stored {
                        return Err(MappingValidationError::StoredVectorField(name.clone()));
                    }
                    if !vector_field_ids.insert(field_id) {
                        return Err(MappingValidationError::DuplicateVectorFieldId(field_id));
                    }
                }
                _ => {}
            }
            definition.validate(name)?;
        }
        if vector_fields > MAX_VECTOR_MAPPING_FIELDS {
            return Err(MappingValidationError::TooManyVectorFields(vector_fields));
        }
        mapping.extensions.validate(mapping)?;
        mapping
            .extensions
            .record_model
            .validate(mapping.mode)
            .map_err(|error| MappingValidationError::InvalidRecordModel(error.to_string()))?;
        Ok(())
    }

    /// Validates a mapping admitted for a new index or successor generation.
    pub fn validate_current(&self) -> Result<(), MappingValidationError> {
        self.validate()
    }

    /// Returns whether `self` is an additive successor of `predecessor`.
    ///
    /// This is the shared semantic check used before entering the catalog
    /// CAS. Existing field definitions and every format-4 extension identity
    /// are immutable; a successor may only append new definitions and their
    /// matching metadata.
    #[must_use]
    pub fn is_additive_successor_of(&self, predecessor: &Self) -> bool {
        let predecessor = predecessor.clone().normalized();
        let successor = self.clone().normalized();
        predecessor.format_version == MAPPING_FORMAT_VERSION
            && successor.format_version == MAPPING_FORMAT_VERSION
            && predecessor.mode == successor.mode
            && predecessor.extensions.record_model == successor.extensions.record_model
            && predecessor.extensions.dynamic_policy == successor.extensions.dynamic_policy
            && predecessor
                .fields
                .iter()
                .all(|(path, field)| successor.fields.get(path) == Some(field))
            && predecessor
                .extensions
                .multi_field_projections
                .iter()
                .all(|entry| successor.extensions.multi_field_projections.contains(entry))
            && predecessor
                .extensions
                .analyzer_identities
                .iter()
                .all(|entry| successor.extensions.analyzer_identities.contains(entry))
            && predecessor
                .extensions
                .keyword_limits
                .iter()
                .all(|entry| successor.extensions.keyword_limits.contains(entry))
            && predecessor
                .extensions
                .nested_paths
                .iter()
                .all(|entry| successor.extensions.nested_paths.contains(entry))
            && successor
                .extensions
                .nested_paths
                .iter()
                .filter(|path| !predecessor.extensions.nested_paths.contains(path))
                .all(|path| {
                    predecessor.fields.keys().all(|field| {
                        field != path
                            && !field
                                .strip_prefix(path.as_str())
                                .is_some_and(|suffix| suffix.starts_with('.'))
                    })
                })
    }
}

struct MappingVersionProbe;

impl MappingVersionProbe {
    fn deserialize<'de, D>(deserializer: D) -> Result<u16, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        struct VersionVisitor;

        impl<'de> Visitor<'de> for VersionVisitor {
            type Value = u16;

            fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                formatter.write_str("a mapping JSON object")
            }

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: MapAccess<'de>,
            {
                let mut keys = std::collections::BTreeSet::new();
                let mut version = None;
                while let Some(key) = map.next_key::<String>()? {
                    if !keys.insert(key.clone()) {
                        return Err(de::Error::custom(format!("duplicate mapping key '{key}'")));
                    }
                    let raw = map.next_value::<Box<serde_json::value::RawValue>>()?;
                    if key == "format_version" {
                        version = Some(
                            serde_json::from_str::<u16>(raw.get())
                                .map_err(|_| de::Error::custom("mapping format_version is invalid"))?,
                        );
                    }
                }
                version.ok_or_else(|| de::Error::custom("mapping format_version is missing"))
            }
        }

        deserializer.deserialize_map(VersionVisitor)
    }
}

impl Serialize for Mapping {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        use serde::ser::Error as _;

        if self.format_version != MAPPING_FORMAT_VERSION {
            return Err(S::Error::custom(format!(
                "unsupported mapping format version {}",
                self.format_version
            )));
        }
        let normalized = self.clone().normalized();
        MappingWireRef {
            format_version: normalized.format_version,
            generation: normalized.schema_generation,
            mode: normalized.mode,
            record_model: &normalized.extensions.record_model,
            dynamic_policy: &normalized.extensions.dynamic_policy,
            fields: &normalized.fields,
            multi_field_projections: &normalized.extensions.multi_field_projections,
            analyzer_identities: &normalized.extensions.analyzer_identities,
            keyword_limits: &normalized.extensions.keyword_limits,
            nested_paths: &normalized.extensions.nested_paths,
        }
        .serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for Mapping {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        use serde::de::Error as _;

        let value = serde_json::Value::deserialize(deserializer)?;
        let version = value
            .get("format_version")
            .and_then(serde_json::Value::as_u64)
            .and_then(|value| u16::try_from(value).ok())
            .ok_or_else(|| D::Error::custom("mapping format_version is missing or invalid"))?;
        if version != MAPPING_FORMAT_VERSION {
            return Err(D::Error::custom(format!(
                "unsupported mapping format version {version}"
            )));
        }
        let wire: MappingWire = serde_json::from_value(value).map_err(D::Error::custom)?;
        Ok(Self {
            format_version: wire.format_version,
            schema_generation: wire.generation,
            mode: wire.mode,
            fields: wire.fields,
            extensions: MappingExtensions {
                record_model: wire.record_model,
                dynamic_policy: wire.dynamic_policy,
                multi_field_projections: wire.multi_field_projections,
                analyzer_identities: wire.analyzer_identities,
                keyword_limits: wire.keyword_limits,
                nested_paths: wire.nested_paths,
            },
        })
    }
}

#[derive(Serialize)]
struct MappingWireRef<'a> {
    format_version: u16,
    generation: SchemaGeneration,
    mode: IndexMode,
    record_model: &'a IndexRecordModel,
    dynamic_policy: &'a DynamicMappingPolicy,
    fields: &'a BTreeMap<String, FieldMapping>,
    multi_field_projections: &'a Vec<MultiFieldProjection>,
    analyzer_identities: &'a Vec<AnalyzerIdentity>,
    keyword_limits: &'a Vec<KeywordFieldPolicy>,
    nested_paths: &'a Vec<String>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct MappingWire {
    format_version: u16,
    generation: SchemaGeneration,
    mode: IndexMode,
    record_model: IndexRecordModel,
    dynamic_policy: DynamicMappingPolicy,
    fields: BTreeMap<String, FieldMapping>,
    multi_field_projections: Vec<MultiFieldProjection>,
    analyzer_identities: Vec<AnalyzerIdentity>,
    keyword_limits: Vec<KeywordFieldPolicy>,
    nested_paths: Vec<String>,
}

/// Current mapping semantics kept separate from the base field definitions.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct MappingExtensions {
    pub record_model: IndexRecordModel,
    pub dynamic_policy: DynamicMappingPolicy,
    pub multi_field_projections: Vec<MultiFieldProjection>,
    pub analyzer_identities: Vec<AnalyzerIdentity>,
    pub keyword_limits: Vec<KeywordFieldPolicy>,
    pub nested_paths: Vec<String>,
}

impl MappingExtensions {
    fn validate(&self, mapping: &Mapping) -> Result<(), MappingValidationError> {
        if self.multi_field_projections.len() > MAX_MAPPING_FIELDS
            || self.analyzer_identities.len() > MAX_MAPPING_FIELDS
            || self.keyword_limits.len() > MAX_MAPPING_FIELDS
        {
            return Err(MappingValidationError::TooManyMappingMetadata);
        }
        if self.nested_paths.len() > MAX_NESTED_MAPPING_PATHS {
            return Err(MappingValidationError::TooManyNestedPaths(self.nested_paths.len()));
        }
        require_sorted_unique(
            self.multi_field_projections
                .iter()
                .map(|projection| projection.child.as_str()),
            "multi_field_projections",
        )?;
        require_sorted_unique(
            self.analyzer_identities.iter().map(|identity| identity.field.as_str()),
            "analyzer_identities",
        )?;
        require_sorted_unique(
            self.keyword_limits.iter().map(|policy| policy.field.as_str()),
            "keyword_limits",
        )?;
        require_sorted_unique(self.nested_paths.iter().map(String::as_str), "nested_paths")?;
        let nested_paths = self
            .nested_paths
            .iter()
            .map(String::as_str)
            .collect::<std::collections::BTreeSet<_>>();
        for child in &self.nested_paths {
            for (separator, _) in child.match_indices('.') {
                let parent = &child[..separator];
                if nested_paths.contains(parent) {
                    return Err(MappingValidationError::NestedPathOverlap {
                        parent: parent.to_owned(),
                        child: child.clone(),
                    });
                }
            }
        }
        for projection in &self.multi_field_projections {
            validate_mapping_field_name(&projection.child)?;
            validate_mapping_field_name(&projection.parent)?;
            let Some(child) = mapping.fields.get(&projection.child) else {
                return Err(MappingValidationError::MissingMultiField(projection.child.clone()));
            };
            let Some(parent) = mapping.fields.get(&projection.parent) else {
                return Err(MappingValidationError::MissingMultiFieldSource {
                    field: projection.child.clone(),
                    source: projection.parent.clone(),
                });
            };
            if !projection
                .child
                .strip_prefix(&projection.parent)
                .is_some_and(|suffix| suffix.starts_with('.') && suffix.len() > 1)
            {
                return Err(MappingValidationError::InvalidMultiFieldPath(projection.child.clone()));
            }
            if matches!(
                child.field_type,
                FieldType::Vector { .. }
                    | FieldType::SparseVector { .. }
                    | FieldType::VectorSet { .. }
                    | FieldType::Nested
            ) || matches!(
                parent.field_type,
                FieldType::Vector { .. }
                    | FieldType::SparseVector { .. }
                    | FieldType::VectorSet { .. }
                    | FieldType::Nested
            ) {
                return Err(MappingValidationError::InvalidMultiFieldEndpoint(
                    projection.child.clone(),
                ));
            }
            if !multi_field_types_are_compatible(parent.field_type, child.field_type) {
                return Err(MappingValidationError::InvalidMultiFieldEndpoint(
                    projection.child.clone(),
                ));
            }
        }
        for identity in &self.analyzer_identities {
            let Some(field) = mapping.fields.get(&identity.field) else {
                return Err(MappingValidationError::UnknownFieldMetadata(identity.field.clone()));
            };
            if field.field_type != FieldType::Text {
                return Err(MappingValidationError::AnalyzerOnNonTextField(identity.field.clone()));
            }
        }
        for policy in &self.keyword_limits {
            let Some(field) = mapping.fields.get(&policy.field) else {
                return Err(MappingValidationError::UnknownFieldMetadata(policy.field.clone()));
            };
            if field.field_type != FieldType::Keyword {
                return Err(MappingValidationError::KeywordPolicyOnNonKeywordField(
                    policy.field.clone(),
                ));
            }
            policy.validate()?;
            if let Some(MappingScalar::Keyword(value)) = &field.null_value
                && value.len() > policy.max_bytes as usize
            {
                return Err(MappingValidationError::InvalidKeywordNullValue(policy.field.clone()));
            }
        }
        for path in &self.nested_paths {
            if mapping
                .fields
                .get(path)
                .is_none_or(|field| field.field_type != FieldType::Nested)
            {
                return Err(MappingValidationError::InvalidNestedPath(path.clone()));
            }
            if path.split('.').count() > MAX_NESTED_MAPPING_DEPTH {
                return Err(MappingValidationError::NestedPathTooDeep(path.clone()));
            }
        }
        let text_fields = mapping
            .fields
            .iter()
            .filter_map(|(path, field)| (field.field_type == FieldType::Text).then_some(path.as_str()))
            .collect::<std::collections::BTreeSet<_>>();
        let analyzed_fields = self
            .analyzer_identities
            .iter()
            .map(|identity| identity.field.as_str())
            .collect::<std::collections::BTreeSet<_>>();
        if text_fields != analyzed_fields {
            return Err(MappingValidationError::IncompleteAnalyzerCoverage);
        }
        let keyword_fields = mapping
            .fields
            .iter()
            .filter_map(|(path, field)| (field.field_type == FieldType::Keyword).then_some(path.as_str()))
            .collect::<std::collections::BTreeSet<_>>();
        let limited_fields = self
            .keyword_limits
            .iter()
            .map(|policy| policy.field.as_str())
            .collect::<std::collections::BTreeSet<_>>();
        if keyword_fields != limited_fields {
            return Err(MappingValidationError::IncompleteKeywordCoverage);
        }
        let declared_nested = mapping
            .fields
            .iter()
            .filter_map(|(path, field)| (field.field_type == FieldType::Nested).then_some(path.as_str()))
            .collect::<std::collections::BTreeSet<_>>();
        let listed_nested = self
            .nested_paths
            .iter()
            .map(String::as_str)
            .collect::<std::collections::BTreeSet<_>>();
        if declared_nested != listed_nested {
            return Err(MappingValidationError::IncompleteNestedCoverage);
        }
        Ok(())
    }
}

const fn multi_field_types_are_compatible(parent: FieldType, child: FieldType) -> bool {
    matches!(
        (parent, child),
        (
            FieldType::Text | FieldType::Keyword,
            FieldType::Text | FieldType::Keyword
        ) | (
            FieldType::U64 | FieldType::Date,
            FieldType::U64 | FieldType::Date | FieldType::F64
        ) | (FieldType::I64, FieldType::I64 | FieldType::F64)
            | (FieldType::F64, FieldType::F64)
            | (FieldType::Bool, FieldType::Bool)
    )
}

fn require_sorted_unique<'a>(
    values: impl IntoIterator<Item = &'a str>,
    section: &'static str,
) -> Result<(), MappingValidationError> {
    let mut previous = None;
    for value in values {
        if previous.is_some_and(|prior| prior >= value) {
            return Err(MappingValidationError::MetadataNotSorted(section));
        }
        previous = Some(value);
    }
    Ok(())
}

pub const MAX_DYNAMIC_FIELDS_PER_REQUEST: usize = 128;
pub const MAX_NESTED_MAPPING_DEPTH: usize = 8;
pub const MAX_NESTED_MAPPING_PATHS: usize = 64;
pub const DEFAULT_KEYWORD_IGNORE_ABOVE: u32 = 32_766;
pub const MAX_SCALAR_ARRAY_DEPTH: usize = 32;
pub const MAX_SCALAR_ARRAY_NODES: usize = 4_096;
pub const MAX_SCALAR_ARRAY_VALUES: usize = 2_048;
pub const MAX_MAPPING_BYTES: usize = 1_048_576;
pub const DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION: u16 = 1;

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DynamicMappingPolicy {
    True,
    #[default]
    False,
    Strict,
}

/// Closed inferred type carried by the catalog-owned dynamic proposal.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum DynamicProposedTypeV1 {
    Boolean,
    I64,
    U64,
    F64,
    TextWithKeyword {
        index_analyzer: AnalyzerName,
        search_analyzer: AnalyzerName,
        keyword_child: String,
        keyword_normalizer: KeywordNormalizer,
        keyword_max_bytes: u32,
    },
}

impl DynamicProposedTypeV1 {
    pub fn from_first_candidate(
        field_path: &str,
        candidate: &MappingScalar,
    ) -> Result<Self, DynamicMappingProposalError> {
        validate_mapping_field_name(field_path).map_err(DynamicMappingProposalError::InvalidField)?;
        Ok(match candidate {
            MappingScalar::Boolean(_) => Self::Boolean,
            MappingScalar::I64(_) => Self::I64,
            MappingScalar::U64(_) => Self::U64,
            MappingScalar::F64Bits(bits) => {
                let value = f64::from_bits(*bits);
                if !value.is_finite() {
                    return Err(DynamicMappingProposalError::InvalidCandidate);
                }
                Self::F64
            }
            MappingScalar::Keyword(_) => Self::TextWithKeyword {
                index_analyzer: AnalyzerName::Standard,
                search_analyzer: AnalyzerName::Standard,
                keyword_child: format!("{field_path}.keyword"),
                keyword_normalizer: KeywordNormalizer::Keyword,
                keyword_max_bytes: DEFAULT_KEYWORD_IGNORE_ABOVE,
            },
        })
    }

    fn validate(&self, field_path: &str) -> Result<(), DynamicMappingProposalError> {
        if let Self::TextWithKeyword {
            keyword_child,
            keyword_max_bytes,
            ..
        } = self
            && (keyword_child != &format!("{field_path}.keyword")
                || *keyword_max_bytes == 0
                || *keyword_max_bytes > DEFAULT_KEYWORD_IGNORE_ABOVE)
        {
            return Err(DynamicMappingProposalError::InvalidProposedType);
        }
        Ok(())
    }
}

#[derive(Serialize)]
struct DynamicMappingProposalDigestInput<'a> {
    format_version: u16,
    collection_index_ref: CollectionIndexRef,
    base_generation: SchemaGeneration,
    field_path: &'a str,
    first_non_null_candidate: &'a MappingScalar,
    proposed_type: &'a DynamicProposedTypeV1,
}

/// Canonical proposal-v1 identity frozen by EC4. Attempt fencing is supplied
/// out of band by the coordinated command and is deliberately not serialized
/// here.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DynamicMappingProposalV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub base_generation: SchemaGeneration,
    pub field_path: String,
    pub first_non_null_candidate: MappingScalar,
    pub proposed_type: DynamicProposedTypeV1,
    pub proposal_digest: u64,
}

impl DynamicMappingProposalV1 {
    pub fn new(
        collection_index_ref: CollectionIndexRef,
        base_generation: SchemaGeneration,
        field_path: String,
        mut first_non_null_candidate: MappingScalar,
    ) -> Result<Self, DynamicMappingProposalError> {
        if let MappingScalar::F64Bits(bits) = &mut first_non_null_candidate
            && f64::from_bits(*bits) == 0.0
        {
            *bits = 0.0_f64.to_bits();
        }
        let proposed_type = DynamicProposedTypeV1::from_first_candidate(&field_path, &first_non_null_candidate)?;
        let mut proposal = Self {
            format_version: DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION,
            collection_index_ref,
            base_generation,
            field_path,
            first_non_null_candidate,
            proposed_type,
            proposal_digest: 0,
        };
        proposal.proposal_digest = proposal.exact_digest()?;
        proposal.validate()?;
        Ok(proposal)
    }

    pub fn exact_digest(&self) -> Result<u64, DynamicMappingProposalError> {
        let input = DynamicMappingProposalDigestInput {
            format_version: self.format_version,
            collection_index_ref: self.collection_index_ref,
            base_generation: self.base_generation,
            field_path: &self.field_path,
            first_non_null_candidate: &self.first_non_null_candidate,
            proposed_type: &self.proposed_type,
        };
        let bytes =
            serde_json::to_vec(&input).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
        if bytes.len() > MAX_MAPPING_BYTES {
            return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
        }
        Ok(bytes.iter().fold(0xcbf2_9ce4_8422_2325_u64, |digest, byte| {
            (digest ^ u64::from(*byte)).wrapping_mul(0x0000_0100_0000_01b3)
        }))
    }

    pub fn validate(&self) -> Result<(), DynamicMappingProposalError> {
        if self.format_version != DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION {
            return Err(DynamicMappingProposalError::UnsupportedVersion(self.format_version));
        }
        validate_mapping_field_name(&self.field_path).map_err(DynamicMappingProposalError::InvalidField)?;
        if matches!(
            self.first_non_null_candidate,
            MappingScalar::F64Bits(bits) if f64::from_bits(bits) == 0.0 && bits != 0.0_f64.to_bits()
        ) {
            return Err(DynamicMappingProposalError::InvalidCandidate);
        }
        self.proposed_type.validate(&self.field_path)?;
        if DynamicProposedTypeV1::from_first_candidate(&self.field_path, &self.first_non_null_candidate)?
            != self.proposed_type
            || self.exact_digest()? != self.proposal_digest
        {
            return Err(DynamicMappingProposalError::IdentityMismatch);
        }
        Ok(())
    }

    /// Verifies that `successor` is exactly the one-field (or text plus
    /// generated keyword child) additive delta described by this proposal.
    #[must_use]
    pub fn matches_successor(&self, predecessor: &Mapping, successor: &Mapping) -> bool {
        let Some(next_generation) = predecessor.schema_generation.checked_next() else {
            return false;
        };
        if self.validate().is_err()
            || self.base_generation != predecessor.schema_generation
            || successor.schema_generation != next_generation
        {
            return false;
        }
        let mut expected = predecessor.clone().normalized();
        expected.schema_generation = successor.schema_generation;
        let (field_type, text) = match &self.proposed_type {
            DynamicProposedTypeV1::Boolean => (FieldType::Bool, None),
            DynamicProposedTypeV1::I64 => (FieldType::I64, None),
            DynamicProposedTypeV1::U64 => (FieldType::U64, None),
            DynamicProposedTypeV1::F64 => (FieldType::F64, None),
            DynamicProposedTypeV1::TextWithKeyword {
                index_analyzer,
                search_analyzer,
                keyword_child,
                keyword_normalizer,
                keyword_max_bytes,
            } => (
                FieldType::Text,
                Some((
                    *index_analyzer,
                    *search_analyzer,
                    keyword_child,
                    *keyword_normalizer,
                    *keyword_max_bytes,
                )),
            ),
        };
        if expected
            .fields
            .insert(
                self.field_path.clone(),
                FieldMapping {
                    field_type,
                    stored: false,
                    null_value: None,
                },
            )
            .is_some()
        {
            return false;
        }
        if let Some((index_analyzer, search_analyzer, keyword_child, keyword_normalizer, keyword_max_bytes)) = text {
            if expected
                .fields
                .insert(
                    keyword_child.clone(),
                    FieldMapping {
                        field_type: FieldType::Keyword,
                        stored: false,
                        null_value: None,
                    },
                )
                .is_some()
            {
                return false;
            }
            expected.extensions.multi_field_projections.push(MultiFieldProjection {
                parent: self.field_path.clone(),
                child: keyword_child.clone(),
            });
            expected.extensions.analyzer_identities.push(AnalyzerIdentity {
                field: self.field_path.clone(),
                index_analyzer,
                search_analyzer,
            });
            expected.extensions.keyword_limits.push(KeywordFieldPolicy {
                field: keyword_child.clone(),
                max_bytes: keyword_max_bytes,
                normalizer: keyword_normalizer,
            });
            expected
                .extensions
                .multi_field_projections
                .sort_by(|left, right| left.child.cmp(&right.child));
            expected
                .extensions
                .analyzer_identities
                .sort_by(|left, right| left.field.cmp(&right.field));
            expected
                .extensions
                .keyword_limits
                .sort_by(|left, right| left.field.cmp(&right.field));
        }
        expected == successor.clone().normalized()
    }

    pub fn encode_canonical(&self) -> Result<Vec<u8>, DynamicMappingProposalError> {
        self.validate()?;
        let bytes = serde_json::to_vec(self).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
        if bytes.len() > MAX_MAPPING_BYTES {
            return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
        }
        Ok(bytes)
    }

    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, DynamicMappingProposalError> {
        if bytes.len() > MAX_MAPPING_BYTES {
            return Err(DynamicMappingProposalError::TooLarge(bytes.len()));
        }
        let mut probe = serde_json::Deserializer::from_slice(bytes);
        let version = MappingVersionProbe::deserialize(&mut probe)
            .map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
        probe
            .end()
            .map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
        if version != DYNAMIC_MAPPING_PROPOSAL_FORMAT_VERSION {
            return Err(DynamicMappingProposalError::UnsupportedVersion(version));
        }
        let proposal: Self =
            serde_json::from_slice(bytes).map_err(|error| DynamicMappingProposalError::Codec(error.to_string()))?;
        proposal.validate()?;
        if proposal.encode_canonical()? != bytes {
            return Err(DynamicMappingProposalError::NonCanonical);
        }
        Ok(proposal)
    }
}

#[derive(Debug)]
pub enum DynamicMappingProposalError {
    UnsupportedVersion(u16),
    InvalidField(MappingValidationError),
    InvalidCandidate,
    InvalidProposedType,
    IdentityMismatch,
    TooLarge(usize),
    NonCanonical,
    Codec(String),
}

impl Display for DynamicMappingProposalError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported dynamic mapping proposal version {version}")
            }
            Self::InvalidField(error) => write!(formatter, "invalid dynamic mapping field: {error}"),
            Self::InvalidCandidate => formatter.write_str("invalid dynamic mapping first candidate"),
            Self::InvalidProposedType => formatter.write_str("invalid dynamic mapping proposed type"),
            Self::IdentityMismatch => formatter.write_str("dynamic mapping proposal identity mismatch"),
            Self::TooLarge(bytes) => write!(formatter, "dynamic mapping proposal is too large: {bytes} bytes"),
            Self::NonCanonical => formatter.write_str("dynamic mapping proposal is not canonical"),
            Self::Codec(error) => write!(formatter, "dynamic mapping proposal codec error: {error}"),
        }
    }
}

impl Error for DynamicMappingProposalError {}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MultiFieldProjection {
    pub parent: String,
    pub child: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AnalyzerIdentity {
    pub field: String,
    pub index_analyzer: AnalyzerName,
    pub search_analyzer: AnalyzerName,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AnalyzerName {
    Keyword,
    Simple,
    Standard,
    Whitespace,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct KeywordFieldPolicy {
    pub field: String,
    pub max_bytes: u32,
    pub normalizer: KeywordNormalizer,
}

impl KeywordFieldPolicy {
    fn validate(&self) -> Result<(), MappingValidationError> {
        if self.max_bytes == 0 || self.max_bytes > DEFAULT_KEYWORD_IGNORE_ABOVE {
            return Err(MappingValidationError::InvalidKeywordLimit {
                field: self.field.clone(),
                limit: self.max_bytes,
            });
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum KeywordNormalizer {
    Keyword,
    Lowercase,
}

/// Explicit normalized field definition.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FieldMapping {
    #[serde(rename = "type")]
    pub field_type: FieldType,
    #[serde(default)]
    pub stored: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub null_value: Option<MappingScalar>,
}

impl FieldMapping {
    fn validate(&self, field: &str) -> Result<(), MappingValidationError> {
        let Some(value) = &self.null_value else {
            return Ok(());
        };
        let valid = match (self.field_type, value) {
            (FieldType::Text | FieldType::Keyword, MappingScalar::Keyword(_))
            | (FieldType::U64 | FieldType::Date, MappingScalar::U64(_))
            | (FieldType::I64, MappingScalar::I64(_))
            | (FieldType::Bool, MappingScalar::Boolean(_)) => true,
            (FieldType::U64 | FieldType::Date, MappingScalar::I64(value)) => *value >= 0,
            (FieldType::F64, MappingScalar::F64Bits(bits)) => f64::from_bits(*bits).is_finite(),
            (FieldType::F64, MappingScalar::I64(_) | MappingScalar::U64(_)) => true,
            _ => false,
        };
        if valid {
            Ok(())
        } else {
            Err(MappingValidationError::InvalidNullValue(field.to_owned()))
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum MappingScalar {
    Boolean(bool),
    I64(i64),
    U64(u64),
    F64Bits(u64),
    Keyword(String),
}

/// Mapping types admitted by the native mapping model.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldType {
    Text,
    Keyword,
    U64,
    /// Unsigned epoch milliseconds stored in the existing u64 fast column.
    Date,
    I64,
    F64,
    Bool,
    /// Object-correlation boundary reserved by format 4. EC6 owns physical
    /// nested rows and makes this field kind externally reachable.
    Nested,
    /// Immutable vector values live in a manifest-rooted segment sidecar, not
    /// in Tantivy. The catalog mapping remains their sole field authority.
    Vector {
        format_version: u16,
        field_id: VectorFieldId,
        dimensions: u16,
        metric: VectorMetric,
        element_type: DenseVectorElementType,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        quantization_scale: Option<VectorQuantizationScale>,
    },
    /// Sparse numeric dimensions are explicit and immutable. No string
    /// feature hashing or dictionary authority exists inside the engine.
    SparseVector {
        format_version: u16,
        field_id: VectorFieldId,
        maximum_dimension: u32,
        weight_type: SparseVectorWeightType,
    },
    /// A bounded set of same-shaped dense vectors scored by the versioned
    /// late-interaction MaxSim contract. The exact source object is authority.
    VectorSet {
        format_version: u16,
        field_id: VectorFieldId,
        dimensions: u16,
        metric: VectorMetric,
        element_type: DenseVectorElementType,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        quantization_scale: Option<VectorQuantizationScale>,
        maximum_vectors: u16,
    },
}

/// Frozen vector distance semantics. Greater returned score is always better.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum VectorMetric {
    Cosine,
    DotProduct,
    SquaredL2,
}

/// Returns whether a field path has one or more non-empty dot-delimited
/// components and contains no control bytes.
///
/// Dots are path separators, never literal key bytes. Callers that retain
/// arbitrary JSON therefore must reject object keys containing dots rather
/// than choosing a different interpretation at ingest and read time.
#[must_use]
pub fn is_canonical_field_path(path: &str) -> bool {
    !path.is_empty()
        && path.len() <= MAX_MAPPING_FIELD_NAME_BYTES
        && !path.bytes().any(|byte| byte <= 0x1f || byte == 0x7f)
        && path.split('.').all(|component| !component.is_empty())
}

/// Resolves one canonical dotted path through a caller-provided object member
/// lookup. Invalid paths resolve to `None`.
///
/// Keeping component splitting here gives ingestion, native execution, and
/// compatibility response materialization one path interpretation without
/// making foundational types depend on a JSON implementation.
pub fn resolve_field_path<'a, T>(
    root: &'a T,
    path: &str,
    member: impl FnMut(&'a T, &str) -> Option<&'a T>,
) -> Option<&'a T> {
    if !is_canonical_field_path(path) {
        return None;
    }
    path.split('.').try_fold(root, member)
}

/// Exact outcome of traversing a canonical field path.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FieldPathResolution<'a, T> {
    Missing,
    Found(&'a T),
    /// A present scalar/array was encountered where another object component
    /// was required.
    Blocked,
}

/// Resolves a canonical path while distinguishing absence from a present
/// non-container intermediate.
pub fn resolve_field_path_checked<'a, T>(
    root: &'a T,
    path: &str,
    mut member: impl FnMut(&'a T, &str) -> Option<&'a T>,
    is_container: impl Fn(&T) -> bool,
) -> FieldPathResolution<'a, T> {
    if !is_canonical_field_path(path) {
        return FieldPathResolution::Missing;
    }
    let mut current = root;
    for component in path.split('.') {
        if !is_container(current) {
            return FieldPathResolution::Blocked;
        }
        let Some(next) = member(current, component) else {
            return FieldPathResolution::Missing;
        };
        current = next;
    }
    FieldPathResolution::Found(current)
}

/// Validates one mapped field path without depending on an execution engine.
pub fn validate_mapping_field_name(name: &str) -> Result<(), MappingValidationError> {
    if !is_canonical_field_path(name) || name.starts_with('_') {
        return Err(MappingValidationError::InvalidFieldName(name.to_owned()));
    }
    Ok(())
}

/// Invalid normalized mapping semantics.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MappingValidationError {
    UnsupportedVersion(u16),
    ZeroSchemaGeneration,
    TooManyFields(usize),
    InvalidFieldName(String),
    ConflictingFieldPaths { ancestor: String, descendant: String },
    UnsupportedVectorVersion(u16),
    UnsupportedSparseVectorVersion(u16),
    UnsupportedVectorSetVersion(u16),
    TooManyVectorFields(usize),
    InvalidVectorDimensions { field: String, dimensions: u16 },
    InvalidVectorElementContract(String),
    InvalidSparseVectorContract(String),
    InvalidVectorSetContract(String),
    DuplicateVectorFieldId(VectorFieldId),
    StoredVectorField(String),
    MissingMultiField(String),
    MissingMultiFieldSource { field: String, source: String },
    InvalidMultiFieldPath(String),
    InvalidMultiFieldEndpoint(String),
    UnknownFieldMetadata(String),
    AnalyzerOnNonTextField(String),
    KeywordPolicyOnNonKeywordField(String),
    InvalidKeywordLimit { field: String, limit: u32 },
    InvalidKeywordNullValue(String),
    InvalidNullValue(String),
    InvalidNestedPath(String),
    InvalidRecordModel(String),
    NestedPathTooDeep(String),
    NestedPathOverlap { parent: String, child: String },
    IncompleteAnalyzerCoverage,
    IncompleteKeywordCoverage,
    IncompleteNestedCoverage,
    MetadataNotSorted(&'static str),
    TooManyMappingMetadata,
    TooManyNestedPaths(usize),
}

impl Display for MappingValidationError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported mapping format version {version}"),
            Self::ZeroSchemaGeneration => formatter.write_str("mapping schema generation must be nonzero"),
            Self::TooManyFields(count) => write!(formatter, "mapping has {count} fields"),
            Self::InvalidFieldName(name) => write!(formatter, "invalid mapped field name '{name}'"),
            Self::ConflictingFieldPaths { ancestor, descendant } => write!(
                formatter,
                "mapped scalar path '{ancestor}' conflicts with descendant '{descendant}'"
            ),
            Self::UnsupportedVectorVersion(version) => {
                write!(formatter, "unsupported vector mapping format version {version}")
            }
            Self::UnsupportedSparseVectorVersion(version) => {
                write!(formatter, "unsupported sparse-vector mapping format version {version}")
            }
            Self::UnsupportedVectorSetVersion(version) => {
                write!(formatter, "unsupported vector-set mapping format version {version}")
            }
            Self::TooManyVectorFields(count) => write!(formatter, "mapping has {count} vector fields"),
            Self::InvalidVectorDimensions { field, dimensions } => {
                write!(formatter, "vector field '{field}' has invalid dimensions {dimensions}")
            }
            Self::InvalidVectorElementContract(field) => {
                write!(
                    formatter,
                    "vector field '{field}' has invalid element or quantization semantics"
                )
            }
            Self::InvalidSparseVectorContract(field) => {
                write!(
                    formatter,
                    "sparse vector field '{field}' has invalid dimension or weight semantics"
                )
            }
            Self::InvalidVectorSetContract(field) => {
                write!(
                    formatter,
                    "vector-set field '{field}' has invalid element, dimension, or count semantics"
                )
            }
            Self::DuplicateVectorFieldId(field_id) => {
                write!(formatter, "vector field identity {field_id} is duplicated")
            }
            Self::StoredVectorField(field) => {
                write!(
                    formatter,
                    "vector field '{field}' cannot use Tantivy stored-field semantics"
                )
            }
            Self::MissingMultiField(field) => write!(formatter, "multi-field projection '{field}' has no field"),
            Self::MissingMultiFieldSource { field, source } => {
                write!(
                    formatter,
                    "multi-field projection '{field}' has missing source '{source}'"
                )
            }
            Self::InvalidMultiFieldPath(field) => {
                write!(
                    formatter,
                    "multi-field projection '{field}' is not a direct child of its source"
                )
            }
            Self::InvalidMultiFieldEndpoint(field) => {
                write!(
                    formatter,
                    "multi-field projection '{field}' has a nested or vector endpoint"
                )
            }
            Self::UnknownFieldMetadata(field) => {
                write!(formatter, "mapping metadata references unknown field '{field}'")
            }
            Self::AnalyzerOnNonTextField(field) => {
                write!(formatter, "analyzer identity is attached to non-text field '{field}'")
            }
            Self::KeywordPolicyOnNonKeywordField(field) => {
                write!(formatter, "keyword policy is attached to non-keyword field '{field}'")
            }
            Self::InvalidKeywordLimit { field, limit } => {
                write!(formatter, "keyword field '{field}' has invalid ignore_above {limit}")
            }
            Self::InvalidKeywordNullValue(field) => {
                write!(
                    formatter,
                    "keyword field '{field}' has a null_value beyond ignore_above"
                )
            }
            Self::InvalidNullValue(field) => write!(formatter, "field '{field}' has an invalid null_value"),
            Self::InvalidNestedPath(path) => write!(formatter, "nested path '{path}' has no nested field"),
            Self::InvalidRecordModel(reason) => write!(formatter, "invalid index record model: {reason}"),
            Self::NestedPathTooDeep(path) => write!(formatter, "nested path '{path}' exceeds maximum depth"),
            Self::NestedPathOverlap { parent, child } => {
                write!(formatter, "nested path '{child}' is below nested path '{parent}'")
            }
            Self::IncompleteAnalyzerCoverage => {
                formatter.write_str("every text field must have exactly one analyzer identity")
            }
            Self::IncompleteKeywordCoverage => {
                formatter.write_str("every keyword field must have exactly one keyword limit")
            }
            Self::IncompleteNestedCoverage => formatter.write_str("nested_paths must exactly cover nested fields"),
            Self::MetadataNotSorted(section) => {
                write!(formatter, "mapping section '{section}' is not sorted and unique")
            }
            Self::TooManyMappingMetadata => formatter.write_str("mapping metadata exceeds the field bound"),
            Self::TooManyNestedPaths(count) => write!(formatter, "mapping has {count} nested paths"),
        }
    }
}

impl Error for MappingValidationError {}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MappingCodecError {
    TooLarge(usize),
    DuplicateKey(String),
    Malformed(String),
    UnsupportedVersion(u16),
    Invalid(MappingValidationError),
    NonCanonical,
}

impl Display for MappingCodecError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooLarge(bytes) => write!(formatter, "mapping payload has {bytes} bytes"),
            Self::DuplicateKey(reason) | Self::Malformed(reason) => formatter.write_str(reason),
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported mapping format version {version}"),
            Self::Invalid(error) => write!(formatter, "invalid mapping: {error}"),
            Self::NonCanonical => formatter.write_str("mapping payload is not canonical JSON"),
        }
    }
}

impl Error for MappingCodecError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::Invalid(error) => Some(error),
            _ => None,
        }
    }
}

/// Object-storage residency assigned to one immutable segment file.
///
/// This enum reserves lifecycle policy identities. Only [`Self::Standard`] is
/// admitted until a provider-specific PUT plus metadata-readback probe passes.
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StorageClass {
    #[default]
    Standard,
    Cold,
}

/// Reconstructible local-residency preference for one index.
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheClass {
    Pinned,
    #[default]
    Normal,
    ColdTolerant,
}

impl CacheClass {
    /// Returns the stable catalog identity.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Pinned => "pinned",
            Self::Normal => "normal",
            Self::ColdTolerant => "cold_tolerant",
        }
    }
}

impl Display for CacheClass {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl StorageClass {
    /// Returns whether serde may omit this value from a versioned parent.
    #[must_use]
    pub const fn is_standard(&self) -> bool {
        matches!(self, Self::Standard)
    }

    /// Returns the stable manifest identity.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Standard => "standard",
            Self::Cold => "cold",
        }
    }
}

impl Display for StorageClass {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Reserved verb carried by a future versioned lifecycle-stage parent.
///
/// This enum does not execute lifecycle work. ADR-0016 reserves the wire names
/// before an index-config format depends on them.
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum LifecycleVerb {
    Transition,
    AggregateThenDelete,
}

/// A validated document identifier.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct DocumentId(String);

impl DocumentId {
    /// Constructs a non-empty document identifier no larger than 512 bytes.
    pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
        let value = value.into();
        if value.is_empty() {
            return Err(IdentifierError::EmptyDocumentId);
        }
        if value.len() > 512 {
            return Err(IdentifierError::DocumentIdTooLong(value.len()));
        }
        Ok(Self(value))
    }

    /// Returns the identifier text.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Consumes this identifier.
    #[must_use]
    pub fn into_inner(self) -> String {
        self.0
    }
}

impl<'de> Deserialize<'de> for DocumentId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::new(value).map_err(serde::de::Error::custom)
    }
}

impl Display for DocumentId {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// A bounded client token used to deduplicate one logical write.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct IdempotencyKey(String);

impl IdempotencyKey {
    /// Constructs a non-empty UTF-8 key no larger than 256 bytes.
    pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
        let value = value.into();
        if value.is_empty() {
            return Err(IdentifierError::EmptyIdempotencyKey);
        }
        if value.len() > 256 {
            return Err(IdentifierError::IdempotencyKeyTooLong(value.len()));
        }
        Ok(Self(value))
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl<'de> Deserialize<'de> for IdempotencyKey {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::new(value).map_err(serde::de::Error::custom)
    }
}

impl Display for IdempotencyKey {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// A bounded opaque point-in-time lease identifier.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct PitId(String);

impl PitId {
    /// Constructs a non-empty opaque PIT identifier no larger than 512 bytes.
    pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
        let value = value.into();
        if value.is_empty() {
            return Err(IdentifierError::EmptyPitId);
        }
        if value.len() > 512 {
            return Err(IdentifierError::PitIdTooLong(value.len()));
        }
        Ok(Self(value))
    }

    /// Returns the opaque identifier text.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl<'de> Deserialize<'de> for PitId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::new(value).map_err(serde::de::Error::custom)
    }
}

impl Display for PitId {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Durable format version for recovery-era-scoped PIT identifiers.
pub const SCOPED_PIT_ID_FORMAT_VERSION: u16 = 1;

/// A PIT identifier whose namespace is fenced by the coordinated recovery era.
///
/// The era is part of the identifier rather than adjacent command metadata so
/// a nonce may be reused only after the durable authority has advanced.
#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ScopedPitId {
    format_version: u16,
    recovery_era: RecoveryEra,
    nonce: PitId,
}

impl ScopedPitId {
    #[must_use]
    pub const fn new(recovery_era: RecoveryEra, nonce: PitId) -> Self {
        Self {
            format_version: SCOPED_PIT_ID_FORMAT_VERSION,
            recovery_era,
            nonce,
        }
    }

    #[must_use]
    pub const fn recovery_era(&self) -> RecoveryEra {
        self.recovery_era
    }

    #[must_use]
    pub const fn nonce(&self) -> &PitId {
        &self.nonce
    }
}

impl Ord for ScopedPitId {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.recovery_era
            .to_be_bytes()
            .cmp(&other.recovery_era.to_be_bytes())
            .then_with(|| self.nonce.cmp(&other.nonce))
    }
}

impl PartialOrd for ScopedPitId {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl<'de> Deserialize<'de> for ScopedPitId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct Wire {
            format_version: u16,
            recovery_era: RecoveryEra,
            nonce: PitId,
        }

        let wire = Wire::deserialize(deserializer)?;
        if wire.format_version != SCOPED_PIT_ID_FORMAT_VERSION {
            return Err(serde::de::Error::custom(
                IdentifierError::UnsupportedScopedPitIdVersion(wire.format_version),
            ));
        }
        Ok(Self::new(wire.recovery_era, wire.nonce))
    }
}

/// Stable BLAKE3 fingerprint of the complete client mutation intent.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct OperationFingerprint([u8; 32]);

impl OperationFingerprint {
    #[must_use]
    pub const fn new(bytes: [u8; 32]) -> Self {
        Self(bytes)
    }

    #[must_use]
    pub fn from_payload(bytes: &[u8]) -> Self {
        Self(*blake3::hash(bytes).as_bytes())
    }

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

/// A validated relative immutable-object key.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct ObjectKey(String);

impl ObjectKey {
    /// Constructs a safe relative object key.
    pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
        let value = value.into();
        if value.is_empty() {
            return Err(IdentifierError::EmptyObjectKey);
        }
        if value.starts_with('/') || value.contains('\\') {
            return Err(IdentifierError::InvalidObjectKey(value));
        }
        if value
            .split('/')
            .any(|component| component.is_empty() || component == "." || component == "..")
        {
            return Err(IdentifierError::InvalidObjectKey(value));
        }
        Ok(Self(value))
    }

    /// Returns the normalized relative key.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Consumes this object key.
    ///
    /// The allocation can exceed the encoded key length. Memory admissions
    /// retaining this key must use `allocated_capacity_bytes` instead.
    #[must_use]
    pub fn into_inner(self) -> String {
        self.0
    }

    /// Owned string allocation capacity, excluding the inline string header.
    #[must_use]
    pub fn allocated_capacity_bytes(&self) -> usize {
        self.0.capacity()
    }
}

impl<'de> Deserialize<'de> for ObjectKey {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::new(value).map_err(serde::de::Error::custom)
    }
}

impl Display for ObjectKey {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Current transitive root-set locator. The page count selects the current
/// single-page or linked-page physical encoding.
pub const ROOT_SET_REF_FORMAT_VERSION: u16 = 2;
pub const MAX_ROOT_SET_PAGES: u32 = 256;
pub const MAX_ROOT_SET_SCHEMA_GENERATIONS: usize = 256;
pub const MAX_ROOT_SET_OBJECT_KEY_BYTES: usize = 1_024;
pub const PINNED_MANIFEST_BINDING_FORMAT_VERSION: u16 = 1;

/// Never-reused physical identity of one immutable object.
#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PhysicalObjectId {
    pub backend_authority: BackendAuthorityId,
    pub object_incarnation: ObjectIncarnation,
    pub object_key: ObjectKey,
}

impl PhysicalObjectId {
    pub fn validate(&self) -> Result<(), RootSetRefError> {
        let length = self.object_key.as_str().len();
        if !(1..=MAX_ROOT_SET_OBJECT_KEY_BYTES).contains(&length) {
            return Err(RootSetRefError::ObjectKeyLength(length));
        }
        Ok(())
    }
}

/// Integrity-qualified physical object used inside immutable root-set pages.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ObjectIdentity {
    pub physical: PhysicalObjectId,
    pub content_checksum: [u8; 32],
}

impl ObjectIdentity {
    pub fn validate(&self) -> Result<(), RootSetRefError> {
        self.physical.validate()
    }
}

/// Bounded immutable root-set header copied only from an authoritative source.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RootSetRef {
    pub format_version: u16,
    pub root: PhysicalObjectId,
    pub content_digest: [u8; 32],
    pub item_count: u64,
    pub page_count: u32,
    pub schema_generations: Vec<SchemaGeneration>,
}

impl RootSetRef {
    pub fn validate(&self) -> Result<(), RootSetRefError> {
        if self.format_version != ROOT_SET_REF_FORMAT_VERSION {
            return Err(RootSetRefError::UnsupportedVersion(self.format_version));
        }
        self.root.validate()?;
        if self.item_count == 0
            || self.page_count == 0
            || self.page_count > MAX_ROOT_SET_PAGES
            || u64::from(self.page_count) > self.item_count
        {
            return Err(RootSetRefError::InvalidCounts {
                items: self.item_count,
                pages: self.page_count,
            });
        }
        if self.schema_generations.len() > MAX_ROOT_SET_SCHEMA_GENERATIONS
            || self.schema_generations.windows(2).any(|pair| pair[0] >= pair[1])
        {
            return Err(RootSetRefError::InvalidSchemaGenerations);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RootSetRefError {
    UnsupportedVersion(u16),
    ObjectKeyLength(usize),
    InvalidCounts { items: u64, pages: u32 },
    InvalidSchemaGenerations,
}

impl Display for RootSetRefError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported root-set reference version {version}"),
            Self::ObjectKeyLength(length) => {
                write!(formatter, "root-set object key is {length} bytes; expected 1..=1024")
            }
            Self::InvalidCounts { items, pages } => {
                write!(formatter, "root-set counts are invalid: items={items}, pages={pages}")
            }
            Self::InvalidSchemaGenerations => {
                formatter.write_str("root-set schema generations are not bounded strict ascending values")
            }
        }
    }
}

impl Error for RootSetRefError {}

/// Exact immutable-manifest codec selected by a durable PIT binding.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum PinnedManifestDecodeProvenanceV1 {
    ScopedManifestV3,
}

/// Complete immutable manifest binding copied into a durable PIT lease.
///
/// This is read authority only. It cannot publish or replace a manifest, and
/// retaining it beside a PIT is the immutable snapshot ownership transfer
/// required when the mutable current pointer later advances.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PinnedManifestBindingV1 {
    pub format_version: u16,
    pub shard_key: ShardKey,
    pub admission_generation: AdmissionGeneration,
    pub routing_revision: RoutingRevision,
    pub data_backend_authority: BackendAuthorityId,
    pub generation: ManifestGeneration,
    pub fence: FenceToken,
    pub primary_term: PrimaryTerm,
    pub manifest: ObjectIdentity,
    pub root_set: RootSetRef,
    pub committed_through: SeqNo,
    pub decode_provenance: PinnedManifestDecodeProvenanceV1,
}

impl PinnedManifestBindingV1 {
    pub fn validate(&self) -> Result<(), PinnedManifestBindingError> {
        if self.format_version != PINNED_MANIFEST_BINDING_FORMAT_VERSION {
            return Err(PinnedManifestBindingError::UnsupportedVersion(self.format_version));
        }
        if self.admission_generation.get() == 0
            || self.routing_revision.get() == 0
            || self.generation.get() == 0
            || self.fence.epoch().get() == 0
            || self.fence.epoch() != self.primary_term.epoch_component()
        {
            return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
        }
        self.manifest
            .validate()
            .map_err(|error| PinnedManifestBindingError::InvalidObject(error.to_string()))?;
        self.root_set
            .validate()
            .map_err(|error| PinnedManifestBindingError::InvalidRootSet(error.to_string()))?;
        if self.manifest.physical.backend_authority != self.data_backend_authority
            || self.root_set.root.backend_authority != self.data_backend_authority
        {
            return Err(PinnedManifestBindingError::BackendAuthorityMismatch);
        }
        Ok(())
    }
}

/// Stable identity of one exact committed vector-search cut.
///
/// The digest binds the logical index/mapping incarnation and every immutable
/// manifest/root visible to the vector branch. Callers must supply roots in
/// ascending shard order so the same cut has exactly one identity.
pub fn vector_snapshot_digest_v1(
    scope: CollectionIndexRef,
    schema_generation: SchemaGeneration,
    roots: &[PinnedManifestBindingV1],
) -> Result<[u8; 32], PinnedManifestBindingError> {
    if schema_generation.get() == 0
        || roots
            .iter()
            .any(|root| root.shard_key.collection_index_ref() != scope || root.validate().is_err())
        || roots
            .windows(2)
            .any(|pair| pair[0].shard_key.shard_id() >= pair[1].shard_key.shard_id())
    {
        return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
    }
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone-vector-snapshot-v1\0");
    hasher.update(&scope.collection_id().get().to_be_bytes());
    hasher.update(&scope.index_id().get().to_be_bytes());
    hasher.update(&schema_generation.get().to_be_bytes());
    let count = u32::try_from(roots.len()).map_err(|_| PinnedManifestBindingError::InvalidSemanticIdentity)?;
    hasher.update(&count.to_be_bytes());
    for root in roots {
        hasher.update(&root.shard_key.shard_id().get().to_be_bytes());
        hasher.update(&root.admission_generation.get().to_be_bytes());
        hasher.update(&root.routing_revision.get().to_be_bytes());
        hasher.update(&root.generation.get().to_be_bytes());
        hasher.update(&root.fence.epoch().get().to_be_bytes());
        hasher.update(&root.primary_term.get().to_be_bytes());
        hasher.update(&root.manifest.content_checksum);
        hasher.update(&root.root_set.content_digest);
        hasher.update(&root.committed_through.get().to_be_bytes());
    }
    Ok(*hasher.finalize().as_bytes())
}

/// Public semantic format of an exact committed read-cut projection.
pub const COMMITTED_READ_CUT_FORMAT_VERSION: u16 = 1;

/// Maximum number of exact index incarnations in one committed read cut.
pub const MAX_COMMITTED_READ_CUT_INDEXES: usize = 128;

/// One index entry in a provider-neutral committed read-cut projection.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommittedReadCutIndexV1 {
    pub index: IndexName,
    pub collection_index_ref: CollectionIndexRef,
    pub pinned_schema_generation: SchemaGeneration,
    pub cut_digest: [u8; 32],
}

/// Exact provider-neutral identity of one durable committed PIT cut.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommittedReadCutEvidenceV1 {
    pub format_version: u16,
    pub collection_id: CollectionId,
    pub recovery_era: RecoveryEra,
    pub catalog_revision: ClusterRevision,
    pub indexes: Vec<CommittedReadCutIndexV1>,
    pub aggregate_digest: [u8; 32],
}

impl CommittedReadCutEvidenceV1 {
    pub fn new(
        collection_id: CollectionId,
        recovery_era: RecoveryEra,
        catalog_revision: ClusterRevision,
        indexes: Vec<CommittedReadCutIndexV1>,
    ) -> Result<Self, PinnedManifestBindingError> {
        let aggregate_digest = committed_read_cut_digest_v1(collection_id, recovery_era, catalog_revision, &indexes)?;
        Ok(Self {
            format_version: COMMITTED_READ_CUT_FORMAT_VERSION,
            collection_id,
            recovery_era,
            catalog_revision,
            indexes,
            aggregate_digest,
        })
    }

    pub fn validate(&self) -> Result<(), PinnedManifestBindingError> {
        if self.format_version != COMMITTED_READ_CUT_FORMAT_VERSION
            || self.aggregate_digest
                != committed_read_cut_digest_v1(
                    self.collection_id,
                    self.recovery_era,
                    self.catalog_revision,
                    &self.indexes,
                )?
        {
            return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
        }
        Ok(())
    }
}

/// Stable opaque identity of one index's complete immutable committed root vector.
pub fn committed_index_cut_digest_v1(
    scope: CollectionIndexRef,
    pinned_schema_generation: SchemaGeneration,
    roots: &[PinnedManifestBindingV1],
) -> Result<[u8; 32], PinnedManifestBindingError> {
    if pinned_schema_generation.get() == 0
        || roots.is_empty()
        || roots
            .iter()
            .any(|root| root.shard_key.collection_index_ref() != scope || root.validate().is_err())
        || roots
            .windows(2)
            .any(|pair| pair[0].shard_key.shard_id() >= pair[1].shard_key.shard_id())
    {
        return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
    }
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone-committed-index-cut-v1\0");
    hasher.update(&scope.collection_id().get().to_be_bytes());
    hasher.update(&scope.index_id().get().to_be_bytes());
    hasher.update(&pinned_schema_generation.get().to_be_bytes());
    let count = u32::try_from(roots.len()).map_err(|_| PinnedManifestBindingError::InvalidSemanticIdentity)?;
    hasher.update(&count.to_be_bytes());
    for root in roots {
        hasher.update(&root.shard_key.shard_id().get().to_be_bytes());
        hasher.update(&root.admission_generation.get().to_be_bytes());
        hasher.update(&root.routing_revision.get().to_be_bytes());
        hasher.update(&root.generation.get().to_be_bytes());
        hasher.update(&root.fence.epoch().get().to_be_bytes());
        hasher.update(&root.primary_term.get().to_be_bytes());
        hasher.update(&root.manifest.content_checksum);
        hasher.update(&root.root_set.content_digest);
        hasher.update(&root.committed_through.get().to_be_bytes());
    }
    Ok(*hasher.finalize().as_bytes())
}

/// Stable aggregate identity of a canonical complete multi-index committed cut.
pub fn committed_read_cut_digest_v1(
    collection_id: CollectionId,
    recovery_era: RecoveryEra,
    catalog_revision: ClusterRevision,
    indexes: &[CommittedReadCutIndexV1],
) -> Result<[u8; 32], PinnedManifestBindingError> {
    if catalog_revision.get() == 0
        || indexes.is_empty()
        || indexes.len() > MAX_COMMITTED_READ_CUT_INDEXES
        || indexes.iter().any(|entry| {
            entry.collection_index_ref.collection_id() != collection_id
                || entry.pinned_schema_generation.get() == 0
                || entry.cut_digest == [0; 32]
        })
        || indexes
            .windows(2)
            .any(|pair| pair[0].collection_index_ref >= pair[1].collection_index_ref)
        || indexes.iter().map(|entry| &entry.index).collect::<BTreeSet<_>>().len() != indexes.len()
    {
        return Err(PinnedManifestBindingError::InvalidSemanticIdentity);
    }
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone-committed-read-cut-v1\0");
    hasher.update(&COMMITTED_READ_CUT_FORMAT_VERSION.to_be_bytes());
    hasher.update(&recovery_era.to_be_bytes());
    hasher.update(&collection_id.get().to_be_bytes());
    hasher.update(&catalog_revision.get().to_be_bytes());
    let count = u32::try_from(indexes.len()).map_err(|_| PinnedManifestBindingError::InvalidSemanticIdentity)?;
    hasher.update(&count.to_be_bytes());
    for entry in indexes {
        let name = entry.index.as_str().as_bytes();
        let name_length = u16::try_from(name.len()).map_err(|_| PinnedManifestBindingError::InvalidSemanticIdentity)?;
        hasher.update(&name_length.to_be_bytes());
        hasher.update(name);
        hasher.update(&entry.collection_index_ref.index_id().get().to_be_bytes());
        hasher.update(&entry.pinned_schema_generation.get().to_be_bytes());
        hasher.update(&entry.cut_digest);
    }
    Ok(*hasher.finalize().as_bytes())
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PinnedManifestBindingError {
    UnsupportedVersion(u16),
    InvalidSemanticIdentity,
    InvalidObject(String),
    InvalidRootSet(String),
    BackendAuthorityMismatch,
}

impl Display for PinnedManifestBindingError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported pinned manifest binding version {version}")
            }
            Self::InvalidSemanticIdentity => formatter.write_str("pinned manifest semantic identity is invalid"),
            Self::InvalidObject(reason) => write!(formatter, "pinned manifest object is invalid: {reason}"),
            Self::InvalidRootSet(reason) => write!(formatter, "pinned manifest root set is invalid: {reason}"),
            Self::BackendAuthorityMismatch => formatter.write_str("pinned manifest authorities do not match"),
        }
    }
}

impl Error for PinnedManifestBindingError {}

/// Identifier validation failure.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum IdentifierError {
    EmptyDocumentId,
    DocumentIdTooLong(usize),
    EmptyIdempotencyKey,
    IdempotencyKeyTooLong(usize),
    EmptyPitId,
    PitIdTooLong(usize),
    UnsupportedScopedPitIdVersion(u16),
    EmptyObjectKey,
    InvalidObjectKey(String),
}

impl Display for IdentifierError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyDocumentId => formatter.write_str("document id must not be empty"),
            Self::DocumentIdTooLong(length) => {
                write!(formatter, "document id is {length} bytes; maximum is 512")
            }
            Self::EmptyIdempotencyKey => formatter.write_str("idempotency key must not be empty"),
            Self::IdempotencyKeyTooLong(length) => {
                write!(formatter, "idempotency key is {length} bytes; maximum is 256")
            }
            Self::EmptyPitId => formatter.write_str("PIT id must not be empty"),
            Self::PitIdTooLong(length) => {
                write!(formatter, "PIT id is {length} bytes; maximum is 512")
            }
            Self::UnsupportedScopedPitIdVersion(version) => {
                write!(formatter, "unsupported scoped PIT identifier version {version}")
            }
            Self::EmptyObjectKey => formatter.write_str("object key must not be empty"),
            Self::InvalidObjectKey(key) => {
                write!(formatter, "invalid relative object key '{key}'")
            }
        }
    }
}

impl Error for IdentifierError {}

/// Stable P0 routing hash stored as `_rh`.
#[must_use]
pub const fn routing_hash_v1(bytes: &[u8]) -> RoutingHash {
    let mut hash = 0x811c_9dc5_u32;
    let mut offset = 0;
    while offset < bytes.len() {
        hash ^= bytes[offset] as u32;
        hash = hash.wrapping_mul(0x0100_0193);
        offset += 1;
    }
    RoutingHash::new(hash)
}

/// Stable P0 BLAKE3-derived mutable identity hash stored as two u64 fast fields.
#[must_use]
pub fn identity_hash_v1(bytes: &[u8]) -> IdHash {
    let digest = blake3::hash(bytes);
    let digest_bytes = digest.as_bytes();
    IdHash::new(u128::from_le_bytes([
        digest_bytes[0],
        digest_bytes[1],
        digest_bytes[2],
        digest_bytes[3],
        digest_bytes[4],
        digest_bytes[5],
        digest_bytes[6],
        digest_bytes[7],
        digest_bytes[8],
        digest_bytes[9],
        digest_bytes[10],
        digest_bytes[11],
        digest_bytes[12],
        digest_bytes[13],
        digest_bytes[14],
        digest_bytes[15],
    ]))
}

#[cfg(test)]
mod policy_axis_tests {
    use super::{LifecycleVerb, StorageClass};

    #[test]
    fn reserved_policy_axis_names_are_stable() {
        assert!(serde_json::to_string(&StorageClass::Standard).is_ok_and(|value| value == "\"standard\""));
        assert!(serde_json::to_string(&StorageClass::Cold).is_ok_and(|value| value == "\"cold\""));
        assert!(serde_json::to_string(&LifecycleVerb::Transition).is_ok_and(|value| value == "\"transition\""));
        assert!(
            serde_json::to_string(&LifecycleVerb::AggregateThenDelete)
                .is_ok_and(|value| value == "\"aggregate_then_delete\"")
        );
    }
}

/// Computes CRC32C (Castagnoli) without platform-specific unsafe code.
#[must_use]
pub fn crc32c(bytes: &[u8]) -> u32 {
    let mut crc = u32::MAX;
    for byte in bytes {
        crc ^= u32::from(*byte);
        for _ in 0..8 {
            let mask = (crc & 1).wrapping_neg();
            crc = (crc >> 1) ^ (0x82f6_3b78 & mask);
        }
    }
    !crc
}

/// A runtime role that can be composed into the Lodestone process.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum Role {
    Gateway,
    Indexer,
    Searcher,
    Compactor,
    Metastore,
}

impl Role {
    /// All roles in stable display and startup order.
    pub const ALL: [Self; 5] = [
        Self::Gateway,
        Self::Indexer,
        Self::Searcher,
        Self::Compactor,
        Self::Metastore,
    ];

    /// The stable CLI name of this role.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Gateway => "gateway",
            Self::Indexer => "indexer",
            Self::Searcher => "searcher",
            Self::Compactor => "compactor",
            Self::Metastore => "metastore",
        }
    }
}

impl Display for Role {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl FromStr for Role {
    type Err = RoleParseError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "gateway" => Ok(Self::Gateway),
            "indexer" => Ok(Self::Indexer),
            "searcher" => Ok(Self::Searcher),
            "compactor" => Ok(Self::Compactor),
            "metastore" => Ok(Self::Metastore),
            _ => Err(RoleParseError::Unknown(value.to_owned())),
        }
    }
}

/// A validated, non-empty set of runtime roles.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RoleSelection {
    roles: Vec<Role>,
}

impl RoleSelection {
    /// Selects every runtime role.
    #[must_use]
    pub fn all() -> Self {
        Self {
            roles: Role::ALL.to_vec(),
        }
    }

    /// Parses `all` or a comma-separated role list.
    pub fn parse(value: &str) -> Result<Self, RoleParseError> {
        if value == "all" {
            return Ok(Self::all());
        }

        let mut roles = Vec::new();
        for candidate in value.split(',') {
            if candidate.is_empty() {
                return Err(RoleParseError::Empty);
            }

            let role = candidate.parse::<Role>()?;
            if !roles.contains(&role) {
                roles.push(role);
            }
        }

        if roles.is_empty() {
            return Err(RoleParseError::Empty);
        }

        roles.sort_unstable();
        Ok(Self { roles })
    }

    /// Returns roles in stable order.
    #[must_use]
    pub fn roles(&self) -> &[Role] {
        &self.roles
    }

    /// Returns a stable comma-separated representation.
    #[must_use]
    pub fn display_list(&self) -> String {
        self.roles
            .iter()
            .map(|role| role.as_str())
            .collect::<Vec<_>>()
            .join(",")
    }
}

impl Default for RoleSelection {
    fn default() -> Self {
        Self::all()
    }
}

/// Failure to parse a runtime role selection.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RoleParseError {
    Empty,
    Unknown(String),
}

impl Display for RoleParseError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => formatter.write_str("role selection must not be empty"),
            Self::Unknown(role) => write!(
                formatter,
                "unknown role '{role}'; expected all, gateway, indexer, searcher, compactor, or metastore"
            ),
        }
    }
}

impl Error for RoleParseError {}

/// Arms a bounded external SIGKILL handshake at a named P1.5 durability
/// boundary.
///
/// The hook is inert unless all three `LOD_TEST_P15_FAULT_*` variables are
/// present. A live harness observes the ready file and kills the child; the
/// deadline prevents a misconfigured harness from wedging production code.
pub fn process_fault_point(point: &str) {
    process_fault_point_inner(point, None);
}

/// Arms the same bounded process-fault handshake while attaching one small,
/// printable fact from the durable transition that reached the boundary.
///
/// Evidence is test-only observability, never protocol authority. Invalid or
/// oversized evidence refuses to publish the handshake so a harness fails by
/// timeout instead of accepting a truncated or ambiguous fact.
pub fn process_fault_point_with_evidence(point: &str, evidence: &str) {
    const MAX_EVIDENCE_BYTES: usize = 512;
    if evidence.is_empty()
        || evidence.len() > MAX_EVIDENCE_BYTES
        || !evidence
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'=' | b'.'))
    {
        return;
    }
    process_fault_point_inner(point, Some(evidence));
}

/// Test-only path receiving one exact P5.2 GC protection observation.
///
/// Unlike the process-fault handshake, this side channel never waits for a
/// release file. The Ready executor publishes it before advancing beyond the
/// observed protection boundary, and later fault hooks remain independently
/// armed.
pub const GC_PROTECTION_OBSERVATION_PATH_ENV: &str = "LOD_TEST_P52_GC_PROTECTION_OBSERVATION_PATH";

/// Hard ceiling for one test-only GC protection observation.
pub const MAX_GC_PROTECTION_OBSERVATION_BYTES: usize = 8 * 1024 * 1024;

/// Failure to publish one bounded test-only GC protection observation.
#[derive(Debug)]
pub enum GcProtectionObservationError {
    InvalidPath,
    Empty,
    TooLarge {
        actual: usize,
        maximum: usize,
    },
    AlreadyPublished(PathBuf),
    Io {
        operation: &'static str,
        path: PathBuf,
        source: std::io::Error,
    },
}

impl Display for GcProtectionObservationError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidPath => formatter.write_str("GC protection observation path has no file name"),
            Self::Empty => formatter.write_str("GC protection observation is empty"),
            Self::TooLarge { actual, maximum } => write!(
                formatter,
                "GC protection observation is {actual} bytes; maximum is {maximum}"
            ),
            Self::AlreadyPublished(path) => {
                write!(
                    formatter,
                    "GC protection observation was already published at {}",
                    path.display()
                )
            }
            Self::Io {
                operation,
                path,
                source,
            } => write!(
                formatter,
                "GC protection observation {operation} failed for {}: {source}",
                path.display()
            ),
        }
    }
}

impl Error for GcProtectionObservationError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::Io { source, .. } => Some(source),
            _ => None,
        }
    }
}

/// Returns the optional test-only observation destination.
#[must_use]
pub fn gc_protection_observation_path() -> Option<PathBuf> {
    std::env::var_os(GC_PROTECTION_OBSERVATION_PATH_ENV)
        .filter(|value| !value.is_empty())
        .map(PathBuf::from)
}

/// Atomically publishes one bounded observation without replacing an older
/// one.
///
/// Callers must execute this blocking filesystem helper on a bounded blocking
/// pool. The completed temporary file is linked into place only after every
/// byte is written and synced, so readers never observe a partial record.
pub fn write_gc_protection_observation(path: &Path, encoded: &[u8]) -> Result<(), GcProtectionObservationError> {
    if encoded.is_empty() {
        return Err(GcProtectionObservationError::Empty);
    }
    if encoded.len() > MAX_GC_PROTECTION_OBSERVATION_BYTES {
        return Err(GcProtectionObservationError::TooLarge {
            actual: encoded.len(),
            maximum: MAX_GC_PROTECTION_OBSERVATION_BYTES,
        });
    }
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .filter(|name| !name.is_empty())
        .ok_or(GcProtectionObservationError::InvalidPath)?;
    let temporary = path.with_file_name(format!(".{file_name}.{}.tmp", std::process::id()));
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temporary)
        .map_err(|source| GcProtectionObservationError::Io {
            operation: "temporary-file creation",
            path: temporary.clone(),
            source,
        })?;
    let write_result = file.write_all(encoded).and_then(|()| file.sync_all());
    drop(file);
    if let Err(source) = write_result {
        let _cleanup_result = std::fs::remove_file(&temporary);
        return Err(GcProtectionObservationError::Io {
            operation: "temporary-file write",
            path: temporary,
            source,
        });
    }
    if let Err(source) = std::fs::hard_link(&temporary, path) {
        let _cleanup_result = std::fs::remove_file(&temporary);
        if source.kind() == std::io::ErrorKind::AlreadyExists {
            return Err(GcProtectionObservationError::AlreadyPublished(path.to_path_buf()));
        }
        return Err(GcProtectionObservationError::Io {
            operation: "atomic publication",
            path: path.to_path_buf(),
            source,
        });
    }
    std::fs::remove_file(&temporary).map_err(|source| GcProtectionObservationError::Io {
        operation: "temporary-file cleanup",
        path: temporary,
        source,
    })?;
    Ok(())
}

fn process_fault_point_inner(point: &str, evidence: Option<&str>) {
    struct FaultConfig {
        point: String,
        ready_path: PathBuf,
        release_path: PathBuf,
        timeout: Duration,
    }
    static CONFIG: OnceLock<Option<FaultConfig>> = OnceLock::new();
    let config = CONFIG.get_or_init(|| {
        let point = std::env::var("LOD_TEST_LISTENER_FAULT_POINT").ok()?;
        let ready_path = std::env::var_os("LOD_TEST_LISTENER_FAULT_READY").map(PathBuf::from)?;
        let release_path = std::env::var_os("LOD_TEST_LISTENER_FAULT_RELEASE").map(PathBuf::from)?;
        let timeout_millis = std::env::var("LOD_TEST_LISTENER_FAULT_TIMEOUT_MILLIS")
            .ok()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(30_000)
            .min(60_000);
        Some(FaultConfig {
            point,
            ready_path,
            release_path,
            timeout: Duration::from_millis(timeout_millis),
        })
    });
    let Some(config) = config.as_ref().filter(|config| config.point == point) else {
        return;
    };
    let payload = evidence.map_or_else(|| point.to_owned(), |fact| format!("{point}\n{fact}"));
    // Publish the handshake only after its complete payload is visible. A
    // direct write creates the ready path before writing its bytes, which can
    // let a loaded parent observe an empty or partial fault identity.
    let temporary = config
        .ready_path
        .with_extension(format!("fault-{}.tmp", std::process::id()));
    if std::fs::write(&temporary, payload.as_bytes()).is_err() {
        let _cleanup_result = std::fs::remove_file(&temporary);
        return;
    }
    if std::fs::rename(&temporary, &config.ready_path).is_err() {
        let _cleanup_result = std::fs::remove_file(&temporary);
        return;
    }
    let deadline = Instant::now() + config.timeout;
    while !config.release_path.exists() && Instant::now() < deadline {
        std::thread::yield_now();
    }
}

#[cfg(test)]
mod hash_backend_tests;

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::sync::atomic::{AtomicU64, Ordering};

    use super::{
        CollectionId, CollectionIndexRef, DocumentId, DurabilityClass, DynamicMappingProposalError,
        DynamicMappingProposalV1, Epoch, FieldMapping, FieldType, GcProtectionObservationError, IndexId, IndexMode,
        MAPPING_FORMAT_VERSION, MAX_GC_PROTECTION_OBSERVATION_BYTES, MAX_MAPPING_FIELD_NAME_BYTES, MAX_MAPPING_FIELDS,
        ManifestGeneration, Mapping, MappingExtensions, MappingScalar, MappingValidationError, ObjectKey, PitId, Role,
        RoleSelection, RoutingHash, RoutingRangeEnd, SchemaGeneration, ScopedPitId, SeqNo, ShardDescriptor, ShardId,
        TaskId, VisibilityMode, crc32c, validate_mapping_field_name, write_gc_protection_observation,
    };

    static OBSERVATION_TEST_SEQUENCE: AtomicU64 = AtomicU64::new(0);

    #[test]
    fn object_key_memory_charge_preserves_spare_string_capacity() -> Result<(), Box<dyn std::error::Error>> {
        let mut value = String::with_capacity(4096);
        value.push_str("a/b");
        let capacity = value.capacity();
        let key = ObjectKey::new(value)?;
        assert_eq!(key.as_str(), "a/b");
        assert_eq!(key.allocated_capacity_bytes(), capacity);
        assert!(key.allocated_capacity_bytes() > key.as_str().len());
        assert_eq!(serde_json::to_string(&key)?, "\"a/b\"");
        Ok(())
    }

    #[test]
    fn all_expands_to_every_role() {
        let selection = RoleSelection::parse("all");

        assert_eq!(selection.as_ref().map(|value| value.roles()), Ok(Role::ALL.as_slice()));
    }

    #[test]
    fn explicit_roles_are_deduplicated_and_stable() {
        let selection = RoleSelection::parse("searcher,gateway,searcher");

        assert_eq!(
            selection.as_ref().map(|value| value.roles()),
            Ok([Role::Gateway, Role::Searcher].as_slice())
        );
    }

    #[test]
    fn unknown_role_is_rejected() {
        let selection = RoleSelection::parse("unknown");

        assert!(selection.is_err());
    }

    #[test]
    fn semantic_numbers_advance_without_crossing_types() {
        assert_eq!(Epoch::new(7).checked_next(), Some(Epoch::new(8)));
        assert_eq!(SeqNo::new(41).checked_next(), Some(SeqNo::new(42)));
        assert_eq!(ManifestGeneration::new(u64::MAX).checked_next(), None);
    }

    #[test]
    fn sparse_values_are_canonical_bounded_and_mapping_typed() -> Result<(), Box<dyn std::error::Error>> {
        let definition = super::SparseVectorDefinition {
            maximum_dimension: 12,
            weight_type: super::SparseVectorWeightType::F16,
        };
        let value = super::SparseVectorValue {
            entries: vec![
                super::SparseVectorEntry {
                    dimension: 1,
                    weight: 1.5,
                },
                super::SparseVectorEntry {
                    dimension: 9,
                    weight: -2.0,
                },
            ],
        };
        let decoded = value.decode(definition)?;
        let native = value.canonical_native_bytes(definition)?;
        assert_eq!(super::decode_sparse_native_bytes(definition, &native)?, decoded);
        assert_eq!(native.len(), 12);
        let encoded = serde_json::to_vec(&value)?;
        assert_eq!(serde_json::from_slice::<super::SparseVectorValue>(&encoded)?, value);
        #[derive(Debug, serde::Deserialize, PartialEq, serde::Serialize)]
        #[serde(tag = "kind", rename_all = "snake_case")]
        enum TaggedSparse {
            Sparse {
                query_vector: super::SparseVectorValue,
                #[serde(deserialize_with = "super::deserialize_sparse_weight")]
                weight: f64,
            },
        }
        let tagged = TaggedSparse::Sparse {
            query_vector: value.clone(),
            weight: 0.5,
        };
        let encoded = serde_json::to_vec(&tagged)?;
        assert_eq!(serde_json::from_slice::<TaggedSparse>(&encoded)?, tagged);

        for invalid in [
            super::SparseVectorValue {
                entries: vec![
                    super::SparseVectorEntry {
                        dimension: 1,
                        weight: 1.0,
                    },
                    super::SparseVectorEntry {
                        dimension: 1,
                        weight: 2.0,
                    },
                ],
            },
            super::SparseVectorValue {
                entries: vec![super::SparseVectorEntry {
                    dimension: 2,
                    weight: 0.0,
                }],
            },
            super::SparseVectorValue {
                entries: vec![super::SparseVectorEntry {
                    dimension: 13,
                    weight: 1.0,
                }],
            },
            super::SparseVectorValue {
                entries: vec![super::SparseVectorEntry {
                    dimension: 2,
                    weight: f64::NAN,
                }],
            },
        ] {
            assert!(invalid.decode(definition).is_err());
        }
        Ok(())
    }

    #[test]
    fn vector_set_values_are_explicit_typed_and_bounded() -> Result<(), Box<dyn std::error::Error>> {
        let definition = super::VectorSetDefinition {
            dense: super::DenseVectorDefinition {
                dimensions: 2,
                element_type: super::DenseVectorElementType::F32,
                metric: super::VectorMetric::DotProduct,
                quantization_scale: None,
            },
            maximum_vectors: 2,
        };
        let value = super::VectorSetValue {
            vectors: vec![
                super::DenseVectorValue::Numeric { values: vec![1.0, 0.0] },
                super::DenseVectorValue::Numeric { values: vec![0.0, 1.0] },
            ],
        };
        let decoded = value.decode(
            definition,
            super::MAX_VECTOR_SET_DOCUMENT_VECTORS,
            super::MAX_VECTOR_SET_BYTES_PER_FIELD,
        )?;
        assert_eq!(decoded.vectors, vec![vec![1.0, 0.0], vec![0.0, 1.0]]);
        assert_eq!(decoded.native_bytes, 16);
        assert!(
            super::VectorSetValue { vectors: Vec::new() }
                .decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
                .is_err()
        );
        assert!(
            super::VectorSetValue {
                vectors: vec![super::DenseVectorValue::Numeric { values: vec![1.0, 0.0] }; 3],
            }
            .decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
            .is_err()
        );
        assert!(
            super::VectorSetValue {
                vectors: vec![super::DenseVectorValue::Numeric { values: vec![1.0] }],
            }
            .decode(definition, 2, super::MAX_VECTOR_SET_BYTES_PER_FIELD)
            .is_err()
        );
        let oversized_json = serde_json::json!({
            "vectors": (0..=super::MAX_VECTOR_SET_DOCUMENT_VECTORS)
                .map(|_| serde_json::json!({"kind": "numeric", "values": [1.0, 0.0]}))
                .collect::<Vec<_>>()
        });
        assert!(serde_json::from_value::<super::VectorSetValue>(oversized_json).is_err());
        Ok(())
    }

    #[test]
    fn task_ids_round_trip_the_complete_u128_domain_and_reject_numbers() {
        let task_id = TaskId::new(u128::MAX - 7);
        let encoded = serde_json::to_string(&task_id);
        let expected = format!("\"{}\"", u128::MAX - 7);
        assert_eq!(encoded.as_ref().ok().map(String::as_str), Some(expected.as_str()));
        assert_eq!(
            encoded
                .as_deref()
                .ok()
                .and_then(|encoded| serde_json::from_str(encoded).ok()),
            Some(task_id)
        );
        assert!(serde_json::from_str::<TaskId>("41").is_err());
    }

    #[test]
    fn gc_protection_observation_writer_is_bounded_atomic_and_single_assignment()
    -> Result<(), Box<dyn std::error::Error>> {
        let sequence = OBSERVATION_TEST_SEQUENCE.fetch_add(1, Ordering::Relaxed);
        let root = std::env::temp_dir().join(format!(
            "lodestone-p52-gc-protection-observation-{}-{sequence}",
            std::process::id()
        ));
        std::fs::create_dir(&root)?;
        let path = root.join("applied.json");
        let encoded = br#"{"format_version":1}"#;

        write_gc_protection_observation(&path, encoded)?;
        assert_eq!(std::fs::read(&path)?, encoded);
        assert!(matches!(
            write_gc_protection_observation(&path, encoded),
            Err(GcProtectionObservationError::AlreadyPublished(actual)) if actual == path
        ));
        assert!(matches!(
            write_gc_protection_observation(
                &root.join("oversized.json"),
                &vec![0; MAX_GC_PROTECTION_OBSERVATION_BYTES.saturating_add(1)]
            ),
            Err(GcProtectionObservationError::TooLarge { .. })
        ));
        assert!(!root.join("oversized.json").exists());

        std::fs::remove_dir_all(root)?;
        Ok(())
    }

    #[test]
    fn durable_identifiers_reject_ambiguous_values() {
        assert!(DocumentId::new("").is_err());
        assert!(PitId::new("").is_err());
        assert!(PitId::new("opaque-pit-lease").is_ok());
        assert!(ObjectKey::new("../escape").is_err());
        assert!(ObjectKey::new("valid/prefix/object").is_ok());
    }

    #[test]
    fn scoped_pit_id_round_trips_and_rejects_newer_formats() -> Result<(), Box<dyn std::error::Error>> {
        let pit_id = ScopedPitId::new(super::RecoveryEra::new(7)?, PitId::new("opaque-pit-lease")?);
        let encoded = serde_json::to_vec(&pit_id)?;
        assert_eq!(serde_json::from_slice::<ScopedPitId>(&encoded)?, pit_id);

        let mut unsupported: serde_json::Value = serde_json::from_slice(&encoded)?;
        let Some(object) = unsupported.as_object_mut() else {
            return Err("scoped PIT identifier was not an object".into());
        };
        object.insert("format_version".to_owned(), serde_json::json!(2));
        assert!(serde_json::from_value::<ScopedPitId>(unsupported).is_err());
        Ok(())
    }

    #[test]
    fn crc32c_matches_the_standard_check_value() {
        assert_eq!(crc32c(b"123456789"), 0xe306_9283);
    }

    #[test]
    fn durability_names_round_trip_and_admitted_classes_are_available() {
        let classes = [
            DurabilityClass::RegionalObject,
            DurabilityClass::ZonalLowLatency,
            DurabilityClass::Relaxed,
            DurabilityClass::Ephemeral,
        ];
        for class in classes {
            let encoded = serde_json::to_string(&class);
            assert_eq!(
                encoded.as_ref().ok().and_then(|bytes| serde_json::from_str(bytes).ok()),
                Some(class)
            );
        }
        assert!(DurabilityClass::RegionalObject.require_available().is_ok());
        assert!(DurabilityClass::ZonalLowLatency.require_available().is_ok());
        assert!(DurabilityClass::Relaxed.require_available().is_err());
        assert!(DurabilityClass::Ephemeral.require_available().is_err());
        assert!(DurabilityClass::RegionalObject.require_state_available().is_ok());
        assert!(DurabilityClass::ZonalLowLatency.require_state_available().is_ok());
        assert!(DurabilityClass::Ephemeral.require_state_available().is_ok());
        assert!(DurabilityClass::Relaxed.require_state_available().is_err());
        assert!(
            serde_json::to_string(&super::DurabilityBoundary::IncarnationMemory)
                .is_ok_and(|encoded| encoded == "\"incarnation_memory\"")
        );
        assert!(serde_json::from_str::<DurabilityClass>("\"future\"").is_err());
        assert!(matches!(
            serde_json::from_str::<VisibilityMode>("\"wait_for\""),
            Ok(VisibilityMode::WaitFor)
        ));
    }

    #[test]
    fn shard_descriptor_covers_exactly_its_hash_range() {
        let descriptor = ShardDescriptor::new(
            ShardId::new(7),
            Epoch::new(3),
            RoutingHash::new(10),
            RoutingRangeEnd::new(20).unwrap_or(RoutingRangeEnd::FULL),
        );
        assert!(descriptor.is_ok());
        let descriptor = descriptor.unwrap_or(ShardDescriptor {
            shard_id: ShardId::new(7),
            epoch: Epoch::new(3),
            hash_start: RoutingHash::new(10),
            hash_end: RoutingRangeEnd::FULL,
        });
        assert!(!descriptor.contains(RoutingHash::new(9)));
        assert!(descriptor.contains(RoutingHash::new(10)));
        assert!(descriptor.contains(RoutingHash::new(19)));
        assert!(!descriptor.contains(RoutingHash::new(20)));
    }

    #[test]
    fn shared_mapping_model_validates_semantics_and_refuses_retired_versions_and_unknown_fields()
    -> Result<(), Box<dyn std::error::Error>> {
        let mapping = Mapping {
            format_version: MAPPING_FORMAT_VERSION,
            schema_generation: SchemaGeneration::new(1),
            mode: IndexMode::Mutable,
            fields: BTreeMap::from([(
                "message".to_owned(),
                FieldMapping {
                    field_type: FieldType::Text,
                    stored: false,
                    null_value: None,
                },
            )]),
            extensions: crate::MappingExtensions::default(),
        };
        assert_eq!(mapping.validate(), Ok(()));
        assert_eq!(mapping.validate_current(), Ok(()));
        let canonical = serde_json::to_value(&mapping)?;
        for retired_version in 1..MAPPING_FORMAT_VERSION {
            let retired = Mapping {
                format_version: retired_version,
                ..mapping.clone()
            };
            assert_eq!(
                retired.validate(),
                Err(MappingValidationError::UnsupportedVersion(retired_version))
            );
            assert!(matches!(
                retired.encode_canonical(),
                Err(super::MappingCodecError::Invalid(
                    MappingValidationError::UnsupportedVersion(version)
                )) if version == retired_version
            ));

            let mut retired_bytes = canonical.clone();
            retired_bytes["format_version"] = serde_json::json!(retired_version);
            let retired_bytes = serde_json::to_vec(&retired_bytes)?;
            let original = retired_bytes.clone();
            assert!(matches!(
                Mapping::decode_canonical(&retired_bytes),
                Err(super::MappingCodecError::UnsupportedVersion(version)) if version == retired_version
            ));
            assert_eq!(retired_bytes, original);
            assert!(serde_json::from_slice::<Mapping>(&retired_bytes).is_err());
        }

        let future = Mapping {
            format_version: MAPPING_FORMAT_VERSION + 1,
            ..mapping.clone()
        };
        assert_eq!(
            future.validate(),
            Err(MappingValidationError::UnsupportedVersion(MAPPING_FORMAT_VERSION + 1))
        );

        let invalid = Mapping {
            schema_generation: SchemaGeneration::new(0),
            ..mapping.clone()
        };
        assert_eq!(invalid.validate(), Err(MappingValidationError::ZeroSchemaGeneration));
        assert!(
            serde_json::from_str::<Mapping>(
                r#"{"format_version":1,"schema_generation":1,"mode":"mutable","fields":{},"future":true}"#
            )
            .is_err()
        );
        for invalid in [".message", "message.", "object..message"] {
            let mut invalid_path = mapping.clone();
            invalid_path.fields = BTreeMap::from([(
                invalid.to_owned(),
                FieldMapping {
                    field_type: FieldType::Text,
                    stored: false,
                    null_value: None,
                },
            )]);
            assert_eq!(
                invalid_path.validate(),
                Err(MappingValidationError::InvalidFieldName(invalid.to_owned()))
            );
        }
        for (ancestor, descendant) in [("a", "a.b"), ("a.b", "a.b.c")] {
            let mut conflicting = mapping.clone();
            conflicting.fields = BTreeMap::from([
                (
                    ancestor.to_owned(),
                    FieldMapping {
                        field_type: FieldType::Keyword,
                        stored: false,
                        null_value: None,
                    },
                ),
                (
                    descendant.to_owned(),
                    FieldMapping {
                        field_type: FieldType::U64,
                        stored: false,
                        null_value: None,
                    },
                ),
            ]);
            assert_eq!(
                conflicting.validate(),
                Err(MappingValidationError::ConflictingFieldPaths {
                    ancestor: ancestor.to_owned(),
                    descendant: descendant.to_owned(),
                })
            );
        }
        let mut siblings = mapping;
        siblings.fields = BTreeMap::from([
            (
                "a.b".to_owned(),
                FieldMapping {
                    field_type: FieldType::Keyword,
                    stored: false,
                    null_value: None,
                },
            ),
            (
                "a.bc".to_owned(),
                FieldMapping {
                    field_type: FieldType::U64,
                    stored: false,
                    null_value: None,
                },
            ),
        ]);
        assert_eq!(siblings.validate(), Ok(()));
        Ok(())
    }

    #[test]
    fn ec6_mapping_refuses_nested_paths_below_nested_paths() {
        let nested = || FieldMapping {
            field_type: FieldType::Nested,
            stored: false,
            null_value: None,
        };
        let mapping = Mapping {
            format_version: MAPPING_FORMAT_VERSION,
            schema_generation: SchemaGeneration::new(1),
            mode: IndexMode::Mutable,
            fields: BTreeMap::from([
                ("items".to_owned(), nested()),
                ("items-options".to_owned(), nested()),
                ("items.options".to_owned(), nested()),
            ]),
            extensions: MappingExtensions {
                nested_paths: vec![
                    "items".to_owned(),
                    "items-options".to_owned(),
                    "items.options".to_owned(),
                ],
                ..MappingExtensions::default()
            },
        };
        assert_eq!(
            mapping.validate(),
            Err(MappingValidationError::NestedPathOverlap {
                parent: "items".to_owned(),
                child: "items.options".to_owned(),
            })
        );

        let mut predecessor = Mapping::empty(IndexMode::Mutable);
        predecessor.fields.insert(
            "items.color".to_owned(),
            FieldMapping {
                field_type: FieldType::Keyword,
                stored: false,
                null_value: None,
            },
        );
        let mut successor = predecessor.clone();
        successor.schema_generation = SchemaGeneration::new(2);
        successor.fields.insert("items".to_owned(), nested());
        successor.extensions.nested_paths = vec!["items".to_owned()];
        assert!(!successor.is_additive_successor_of(&predecessor));

        let mut safe_successor = predecessor.clone();
        safe_successor.schema_generation = SchemaGeneration::new(2);
        safe_successor.fields.insert("other".to_owned(), nested());
        safe_successor.fields.insert(
            "other.color".to_owned(),
            FieldMapping {
                field_type: FieldType::Keyword,
                stored: false,
                null_value: None,
            },
        );
        safe_successor.extensions.nested_paths = vec!["other".to_owned()];
        assert!(safe_successor.is_additive_successor_of(&predecessor));
    }

    #[test]
    fn ec5_dynamic_mapping_proposal_normalizes_signed_zero_and_refuses_noncanonical_decode()
    -> Result<(), Box<dyn std::error::Error>> {
        let proposal = DynamicMappingProposalV1::new(
            CollectionIndexRef::new(CollectionId::new(7)?, IndexId::new(3)?),
            SchemaGeneration::new(1),
            "temperature".to_owned(),
            MappingScalar::F64Bits((-0.0_f64).to_bits()),
        )?;
        assert_eq!(
            proposal.first_non_null_candidate,
            MappingScalar::F64Bits(0.0_f64.to_bits())
        );

        let mut noncanonical = proposal;
        noncanonical.first_non_null_candidate = MappingScalar::F64Bits((-0.0_f64).to_bits());
        noncanonical.proposal_digest = noncanonical.exact_digest()?;
        let encoded = serde_json::to_vec(&noncanonical)?;
        assert!(matches!(
            DynamicMappingProposalV1::decode_canonical(&encoded),
            Err(DynamicMappingProposalError::InvalidCandidate)
        ));
        Ok(())
    }

    #[test]
    fn ec5_deployment_mapping_fixtures_use_the_current_validated_format() -> Result<(), Box<dyn std::error::Error>> {
        for encoded in [
            include_str!("../../../fixtures/runtime/secure-strict-mapping.json"),
            include_str!("../../../fixtures/performance/search-mapping.json"),
            include_str!("../../../fixtures/reindex/mapping.json"),
            include_str!("../../../tests/performance/es-comparison-mapping-v1.json"),
        ] {
            let mapping: Mapping = serde_json::from_str(encoded)?;
            mapping.validate_current()?;
            assert_eq!(mapping.schema_generation, SchemaGeneration::new(1));
        }
        Ok(())
    }

    #[test]
    fn public_replayable_source_mapping_contract_matches_current_model() -> Result<(), serde_json::Error> {
        let contract: serde_json::Value = serde_json::from_str(include_str!(
            "../../../public/development/replayable-source-mapping-v1.json"
        ))?;
        assert_eq!(contract["mapping"]["format_version"], MAPPING_FORMAT_VERSION);
        assert_eq!(contract["mapping"]["maximum_fields"], MAX_MAPPING_FIELDS);
        assert_eq!(
            contract["mapping"]["maximum_field_name_bytes"],
            MAX_MAPPING_FIELD_NAME_BYTES
        );
        assert_eq!(
            contract["mapping"]["scalar_field_types"],
            serde_json::json!(["text", "keyword", "u64", "date", "i64", "f64", "bool"])
        );

        let mapping = Mapping {
            format_version: MAPPING_FORMAT_VERSION,
            schema_generation: SchemaGeneration::new(1),
            mode: IndexMode::AppendLog,
            fields: BTreeMap::from([
                (
                    "message".to_owned(),
                    FieldMapping {
                        field_type: FieldType::Text,
                        stored: false,
                        null_value: None,
                    },
                ),
                (
                    "record_id".to_owned(),
                    FieldMapping {
                        field_type: FieldType::Keyword,
                        stored: false,
                        null_value: None,
                    },
                ),
            ]),
            extensions: crate::MappingExtensions::default(),
        };
        assert_eq!(
            serde_json::Value::String(serde_json::to_string(&mapping)?),
            contract["example"]
        );
        for invalid in ["", ".message", "message.", "object..message", "_reserved", "a\u{7f}b"] {
            assert_eq!(
                validate_mapping_field_name(invalid),
                Err(MappingValidationError::InvalidFieldName(invalid.to_owned()))
            );
        }
        Ok(())
    }

    #[test]
    fn shared_field_path_resolver_uses_canonical_nested_components() {
        #[derive(Default)]
        struct Node {
            children: BTreeMap<&'static str, Node>,
            value: Option<u64>,
        }

        let root = Node {
            children: BTreeMap::from([(
                "metrics",
                Node {
                    children: BTreeMap::from([(
                        "count",
                        Node {
                            children: BTreeMap::new(),
                            value: Some(7),
                        },
                    )]),
                    value: None,
                },
            )]),
            value: None,
        };
        assert_eq!(
            super::resolve_field_path(&root, "metrics.count", |node, component| node.children.get(component))
                .and_then(|node| node.value),
            Some(7)
        );
        assert!(
            super::resolve_field_path(&root, "metrics..count", |node, component| node.children.get(component))
                .is_none()
        );
    }

    #[test]
    fn root_set_reference_is_exact_bounded_and_canonical() -> Result<(), Box<dyn std::error::Error>> {
        let root = super::RootSetRef {
            format_version: super::ROOT_SET_REF_FORMAT_VERSION,
            root: super::PhysicalObjectId {
                backend_authority: super::BackendAuthorityId::new("standard-data")?,
                object_incarnation: super::ObjectIncarnation::new(9)?,
                object_key: super::ObjectKey::new("root-sets/0009.json")?,
            },
            content_digest: [7; 32],
            item_count: 4,
            page_count: 2,
            schema_generations: vec![super::SchemaGeneration::new(1), super::SchemaGeneration::new(3)],
        };
        root.validate()?;
        assert_eq!(
            serde_json::from_slice::<super::RootSetRef>(&serde_json::to_vec(&root)?)?,
            root
        );

        let mut duplicate = root.clone();
        duplicate.schema_generations = vec![super::SchemaGeneration::new(1), super::SchemaGeneration::new(1)];
        assert_eq!(
            duplicate.validate(),
            Err(super::RootSetRefError::InvalidSchemaGenerations)
        );
        let mut empty = root;
        empty.item_count = 0;
        assert_eq!(
            empty.validate(),
            Err(super::RootSetRefError::InvalidCounts { items: 0, pages: 2 })
        );
        let mut single_page = empty;
        single_page.item_count = 1;
        single_page.page_count = 1;
        single_page.validate()?;
        let mut too_many_pages = single_page.clone();
        too_many_pages.page_count = 2;
        assert_eq!(
            too_many_pages.validate(),
            Err(super::RootSetRefError::InvalidCounts { items: 1, pages: 2 })
        );
        let mut predecessor = single_page;
        predecessor.format_version = 1;
        assert_eq!(
            predecessor.validate(),
            Err(super::RootSetRefError::UnsupportedVersion(1))
        );
        Ok(())
    }

    #[test]
    fn kubernetes_object_uid_digest_refuses_zero_on_construction_and_decode() -> Result<(), Box<dyn std::error::Error>>
    {
        assert!(super::KubernetesObjectUidDigest::new([0; 32]).is_err());
        assert!(
            serde_json::from_str::<super::KubernetesObjectUidDigest>(&format!("[{}]", "0,".repeat(31) + "0")).is_err()
        );
        let digest = super::KubernetesObjectUidDigest::new([7; 32])?;
        let decoded: super::KubernetesObjectUidDigest = serde_json::from_slice(&serde_json::to_vec(&digest)?)?;
        assert_eq!(decoded, digest);
        Ok(())
    }

    #[test]
    fn gateway_d1_semantic_identities_and_listener_profiles_are_exact() -> Result<(), Box<dyn std::error::Error>> {
        assert!(super::GatewayIncarnation::new([0; 32]).is_err());
        assert!(serde_json::from_str::<super::GatewayIncarnation>(&format!("[{}]", "0,".repeat(31) + "0")).is_err());
        let incarnation = super::GatewayIncarnation::new([7; 32])?;
        assert_eq!(
            serde_json::from_slice::<super::GatewayIncarnation>(&serde_json::to_vec(&incarnation)?)?,
            incarnation
        );
        assert_eq!(super::AuditRangeId::new(4).get(), 4);
        assert_eq!(super::QuotaGrantId::new(5).get(), 5);
        assert_eq!(super::QuotaWindow::new(6).get(), 6);
        assert_eq!(super::ProcessMonotonicMillis::new(7).get(), 7);
        assert_eq!(super::UnixMillis::new(8).get(), 8);
        for (listener, profile) in [
            (
                super::PublicGatewayListenerV1::LodestoneApi,
                super::PublicTlsProfileV1::Native,
            ),
            (
                super::PublicGatewayListenerV1::ElasticsearchHttp,
                super::PublicTlsProfileV1::Http,
            ),
            (
                super::PublicGatewayListenerV1::OtlpHttp,
                super::PublicTlsProfileV1::Http,
            ),
            (
                super::PublicGatewayListenerV1::JaegerHttp,
                super::PublicTlsProfileV1::Http,
            ),
            (
                super::PublicGatewayListenerV1::OtlpGrpc,
                super::PublicTlsProfileV1::Grpc,
            ),
            (
                super::PublicGatewayListenerV1::SubscriptionGrpc,
                super::PublicTlsProfileV1::Grpc,
            ),
            (
                super::PublicGatewayListenerV1::SubscriptionWebSocket,
                super::PublicTlsProfileV1::Http,
            ),
            (super::PublicGatewayListenerV1::Resp, super::PublicTlsProfileV1::Resp),
        ] {
            assert_eq!(listener.required_tls_profile(), profile);
        }
        Ok(())
    }

    #[test]
    fn ordinary_segment_allocation_ignores_reserved_ids_and_stops_at_boundary() {
        let reserved = super::SegmentId::new(super::RESERVED_SEGMENT_ID_NAMESPACE_START + 17);
        assert_eq!(
            super::next_ordinary_segment_id([super::SegmentId::new(7), reserved, super::SegmentId::new(9),]),
            Some(super::SegmentId::new(10))
        );
        assert_eq!(
            super::next_ordinary_segment_id([reserved]),
            Some(super::SegmentId::new(1))
        );
        assert_eq!(
            super::next_ordinary_segment_id([super::SegmentId::new(super::RESERVED_SEGMENT_ID_NAMESPACE_START - 1,)]),
            None
        );
    }

    #[test]
    fn dense_numeric_and_binary_values_decode_to_identical_semantics() -> Result<(), Box<dyn std::error::Error>> {
        use base64::Engine as _;

        for (element_type, scale, numeric, bytes) in [
            (
                super::DenseVectorElementType::F32,
                None,
                vec![1.5, -2.0],
                [1.5_f32.to_le_bytes(), (-2.0_f32).to_le_bytes()].concat(),
            ),
            (
                super::DenseVectorElementType::F16,
                None,
                vec![1.5, -2.0],
                [0x3e00_u16.to_le_bytes(), 0xc000_u16.to_le_bytes()].concat(),
            ),
            (
                super::DenseVectorElementType::I8,
                Some(super::VectorQuantizationScale::new(0.25)?),
                vec![6.0, -8.0],
                vec![6_u8, (-8_i8) as u8],
            ),
        ] {
            let definition = super::DenseVectorDefinition {
                dimensions: 2,
                element_type,
                metric: super::VectorMetric::DotProduct,
                quantization_scale: scale,
            };
            let numeric = super::decode_dense_numeric_value(definition, &numeric)?;
            let binary = super::decode_dense_binary_value(
                definition,
                &super::DenseVectorBinaryValue {
                    element_type,
                    byte_order: super::DenseVectorByteOrder::LittleEndian,
                    data_base64: super::BASE64_STANDARD.encode(bytes),
                    quantization_scale: scale,
                },
            )?;
            assert_eq!(numeric, binary);
        }
        Ok(())
    }

    #[test]
    fn dense_value_refuses_narrowing_and_noncanonical_binary_controls() -> Result<(), Box<dyn std::error::Error>> {
        use base64::Engine as _;

        let f16 = super::DenseVectorDefinition {
            dimensions: 1,
            element_type: super::DenseVectorElementType::F16,
            metric: super::VectorMetric::DotProduct,
            quantization_scale: None,
        };
        assert_eq!(super::decode_dense_numeric_value(f16, &[1.000_488_281_25])?, vec![1.0]);
        assert_eq!(
            super::decode_dense_numeric_value(f16, &[1.001_464_843_75])?,
            vec![1.001_953_1]
        );
        assert!(matches!(
            super::decode_dense_numeric_value(f16, &[70_000.0]),
            Err(super::DenseVectorValueError::F16Overflow(0))
        ));

        let i8_definition = super::DenseVectorDefinition {
            dimensions: 1,
            element_type: super::DenseVectorElementType::I8,
            metric: super::VectorMetric::DotProduct,
            quantization_scale: Some(super::VectorQuantizationScale::new(0.5)?),
        };
        assert!(matches!(
            super::decode_dense_numeric_value(i8_definition, &[1.5]),
            Err(super::DenseVectorValueError::I8ElementOutOfRange(0))
        ));
        let encoded = super::BASE64_STANDARD.encode([1_u8]);
        let mut binary = super::DenseVectorBinaryValue {
            element_type: super::DenseVectorElementType::I8,
            byte_order: super::DenseVectorByteOrder::LittleEndian,
            data_base64: encoded,
            quantization_scale: Some(super::VectorQuantizationScale::new(0.25)?),
        };
        assert_eq!(
            super::decode_dense_binary_value(i8_definition, &binary),
            Err(super::DenseVectorValueError::QuantizationScaleMismatch)
        );
        binary.quantization_scale = i8_definition.quantization_scale;
        binary.data_base64.push('=');
        assert!(matches!(
            super::decode_dense_binary_value(i8_definition, &binary),
            Err(super::DenseVectorValueError::InvalidBase64Length { .. })
        ));

        let wrong_order =
            br#"{"kind":"binary","element_type":"f32","byte_order":"big_endian","data_base64":"AAAAAA=="}"#;
        assert!(serde_json::from_slice::<super::DenseVectorValue>(wrong_order).is_err());
        let signed_zero = super::decode_dense_numeric_value(
            super::DenseVectorDefinition {
                dimensions: 2,
                element_type: super::DenseVectorElementType::F32,
                metric: super::VectorMetric::DotProduct,
                quantization_scale: None,
            },
            &[-0.0, 0.0],
        )?;
        assert!(signed_zero.iter().all(|value| value.to_bits() == 0));
        assert_eq!(
            super::decode_dense_numeric_value(
                super::DenseVectorDefinition {
                    dimensions: 2,
                    element_type: super::DenseVectorElementType::F32,
                    metric: super::VectorMetric::Cosine,
                    quantization_scale: None,
                },
                &[0.0, -0.0],
            ),
            Err(super::DenseVectorValueError::ZeroNorm)
        );
        Ok(())
    }

    #[test]
    fn dense_value_refuses_length_scale_and_numeric_boundary_controls() -> Result<(), Box<dyn std::error::Error>> {
        use base64::Engine as _;

        let f32_definition = super::DenseVectorDefinition {
            dimensions: 2,
            element_type: super::DenseVectorElementType::F32,
            metric: super::VectorMetric::DotProduct,
            quantization_scale: None,
        };
        for bytes in [vec![0_u8; 4], vec![0_u8; 12]] {
            let value = super::DenseVectorBinaryValue {
                element_type: super::DenseVectorElementType::F32,
                byte_order: super::DenseVectorByteOrder::LittleEndian,
                data_base64: super::BASE64_STANDARD.encode(bytes),
                quantization_scale: None,
            };
            assert!(matches!(
                super::decode_dense_binary_value(f32_definition, &value),
                Err(super::DenseVectorValueError::InvalidBase64Length { .. })
            ));
        }
        let wrong_type = super::DenseVectorBinaryValue {
            element_type: super::DenseVectorElementType::F16,
            byte_order: super::DenseVectorByteOrder::LittleEndian,
            data_base64: super::BASE64_STANDARD.encode([0_u8; 8]),
            quantization_scale: None,
        };
        assert_eq!(
            super::decode_dense_binary_value(f32_definition, &wrong_type),
            Err(super::DenseVectorValueError::ElementTypeMismatch)
        );
        assert!(matches!(
            super::decode_dense_numeric_value(f32_definition, &[f64::INFINITY, 0.0]),
            Err(super::DenseVectorValueError::NonFiniteElement(0))
        ));

        let f16_definition = super::DenseVectorDefinition {
            dimensions: 1,
            element_type: super::DenseVectorElementType::F16,
            metric: super::VectorMetric::DotProduct,
            quantization_scale: None,
        };
        assert_eq!(
            super::decode_dense_numeric_value(f16_definition, &[2.0_f64.powi(-25)])?[0].to_bits(),
            0
        );
        assert_eq!(
            super::decode_dense_numeric_value(f16_definition, &[2.0_f64.powi(-24)])?[0],
            2.0_f32.powi(-24)
        );
        assert_eq!(
            super::decode_dense_numeric_value(f16_definition, &[3.0 * 2.0_f64.powi(-25)])?[0],
            2.0_f32.powi(-23)
        );
        let infinite_half = super::DenseVectorBinaryValue {
            element_type: super::DenseVectorElementType::F16,
            byte_order: super::DenseVectorByteOrder::LittleEndian,
            data_base64: super::BASE64_STANDARD.encode(0x7c00_u16.to_le_bytes()),
            quantization_scale: None,
        };
        assert!(matches!(
            super::decode_dense_binary_value(f16_definition, &infinite_half),
            Err(super::DenseVectorValueError::NonFiniteElement(0))
        ));

        let i8_definition = super::DenseVectorDefinition {
            dimensions: 1,
            element_type: super::DenseVectorElementType::I8,
            metric: super::VectorMetric::DotProduct,
            quantization_scale: Some(super::VectorQuantizationScale::new(1.0)?),
        };
        for value in [-129.0, 128.0] {
            assert!(matches!(
                super::decode_dense_numeric_value(i8_definition, &[value]),
                Err(super::DenseVectorValueError::I8ElementOutOfRange(0))
            ));
        }
        for scale in [r#""0""#, r#""-1""#, r#""1.0""#] {
            assert!(serde_json::from_str::<super::VectorQuantizationScale>(scale).is_err());
        }
        Ok(())
    }
}