lodestone-proto 0.0.17

//! Canonical P6.2 remote-search application payloads (ADR-0077).
//!
//! The authenticated RPC prelude fixes who may invoke a method. These DTOs
//! additionally fix the exact route, node incarnation, manifest generation,
//! writer fence, and writer-first `(G,W,C,D)` cut used by one search. The
//! writer payload is carried in the shared `WriterRealtimeRead` method-level
//! envelope, distinct from ADR-0076's point-GET variant.

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

use lodestone_types::{
    FenceToken, IdHash, IndexMode, LogicalMillis, NodeToken, PinnedManifestBindingV1, RoutingHash, RoutingRangeEnd,
    ScopedPitId, SegmentId, SeqNo, ShardId, ShardKey,
};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use crate::{
    EngineApiCanonicalPlan, EngineApiContinuation, EngineApiResolvedRouteEntry, EngineApiResolvedRouteVector,
    EngineApiResolvedShardRoute, EngineApiRouteIdentity, EngineApiScopedSearchHit, EngineApiSearchResult, Query,
    RealtimeReadEnvelopeError, RealtimeReadKind, SCOPED_SEARCH_FORMAT_VERSION, SearchExecutionFreshness,
    SearchFreshness, decode_realtime_read_request, decode_realtime_read_response, encode_realtime_read_request,
    encode_realtime_read_response, validate_scoped_search_continuation, validate_scoped_search_plan,
};

pub const SEARCH_WIRE_FORMAT_VERSION: u16 = 5;
pub const SEARCH_MAX_REQUEST_PAYLOAD_BYTES: usize = 2 * 1024 * 1024 - 64 * 1024;
/// `SearchShardExecute` carries the writer partial as data, not as control.
pub const SEARCH_MAX_SHARD_REQUEST_PAYLOAD_BYTES: usize = 15 * 1024 * 1024;
/// Leaves two MiB for the complete route/plan when embedded in a shard request
/// and one MiB for the method envelope/internal frame on the writer response.
pub const SEARCH_MAX_WRITER_RESPONSE_PAYLOAD_BYTES: usize = 13 * 1024 * 1024;
/// Leaves one MiB for the existing internal frame and future envelope fields.
pub const SEARCH_MAX_RESPONSE_PAYLOAD_BYTES: usize = 15 * 1024 * 1024;
pub const SEARCH_MAX_SHARDS: usize = 4_096;
pub const SEARCH_MAX_INHERITED_SOURCES: usize = 2;
/// Exact accumulated source lineages admitted by P6 range topology.
pub const SEARCH_MAX_RESULT_LINEAGES: usize = 32;
pub const SEARCH_MAX_HITS: usize = 100_000;
pub const SEARCH_MAX_TOMBSTONES: usize = 100_000;
pub const SEARCH_MAX_AGGREGATION_ENTRIES: usize = 1_000_000;
pub const SEARCH_MAX_AGGREGATION_DEPTH: usize = 16;
pub const SEARCH_MAX_FAILURE_REASON_BYTES: usize = 4_096;
pub const SEARCH_MAX_NESTED_GROUPS_PER_HIT: usize = 64;
pub const SEARCH_MAX_NESTED_HITS_PER_GROUP: usize = 100;
pub const SEARCH_MAX_NESTED_HITS: usize = 10_000;
pub const SEARCH_MAX_NESTED_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
pub const SEARCH_MAX_NESTED_NAME_BYTES: usize = 512;
pub const SEARCH_MAX_NESTED_PATH_BYTES: usize = 512;
pub const SEARCH_MAX_NESTED_SOURCE_BYTES: usize = 65_536;
pub const SEARCH_APPROXIMATE_SKETCH_FORMAT_VERSION: u16 = 1;
pub const SEARCH_MIN_HLL_PRECISION: u8 = 4;
pub const SEARCH_MAX_HLL_PRECISION: u8 = 16;
pub const SEARCH_MIN_PERCENTILE_PRECISION: u8 = 4;
pub const SEARCH_MAX_PERCENTILE_PRECISION: u8 = 12;
pub const SEARCH_MAX_NESTED_ORDINAL: u32 = 1_024;
pub const SEARCH_MAX_NESTED_SORT_VALUES: usize = 16;

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearcherTargetV1 {
    pub node_token: NodeToken,
    pub assignment_generation: lodestone_types::AssignmentGeneration,
    pub manifest_binding: PinnedManifestBindingV1,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WriterTargetV1 {
    pub node_token: NodeToken,
    pub fence: FenceToken,
}

/// Node-independent entitlement sent byte-for-byte to primary and hedge.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearchShardEntitlementV1 {
    pub index_mode: IndexMode,
    pub assignment_generation: lodestone_types::AssignmentGeneration,
    pub manifest_binding: PinnedManifestBindingV1,
    pub allowed_nodes: Vec<NodeToken>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ShardDispatchTargetsV1 {
    pub shard_key: ShardKey,
    pub index_mode: IndexMode,
    /// Exact result lineages admitted by this output. The output shard itself
    /// is always present; settled range artifacts retain the source shard that
    /// assigned their sequence number.
    pub result_lineages: Vec<ShardId>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub committed_segment_lineages: Vec<CommittedSegmentLineageV1>,
    pub primary: SearcherTargetV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hedge: Option<SearcherTargetV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub writer: Option<WriterTargetV1>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub inherited_sources: Vec<InheritedDispatchSourceV1>,
}

/// Reconstructible execution label for one immutable settlement segment.
///
/// The segment identity is allocated by the settlement state machine; this
/// projection never grants object access and cannot invent a durable lineage.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommittedSegmentLineageV1 {
    pub segment_id: SegmentId,
    pub result_lineage: ShardId,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct InheritedDispatchSourceV1 {
    pub source_shard: ShardKey,
    pub predicate: InheritedHashRangeV1,
    pub sequence_ceiling: SeqNo,
    pub result_lineage: ShardId,
    pub primary: SearcherTargetV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hedge: Option<SearcherTargetV1>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct InheritedHashRangeV1 {
    pub start: RoutingHash,
    pub end: RoutingRangeEnd,
}

/// Exact inherited WAL interval which an output writer may expose while a
/// P6.4 transition is live. Search targets are deliberately absent: the
/// current writer validates this projection against its fresh range
/// authority before returning any source-lineage hit.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WriterInheritedReplayV1 {
    pub source_shard: ShardKey,
    pub predicate: InheritedHashRangeV1,
    pub committed_through: SeqNo,
    pub sequence_ceiling: SeqNo,
    pub result_lineage: ShardId,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PitQueryGuardEntryV1 {
    pub collection_index_ref: lodestone_types::CollectionIndexRef,
    pub admission_generation: lodestone_types::AdmissionGeneration,
    pub schema_generation: lodestone_types::SchemaGeneration,
    pub routing_revision: lodestone_types::RoutingRevision,
    pub roots: Vec<PinnedManifestBindingV1>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PitQueryGuardV1 {
    pub format_version: u16,
    pub pit_id: ScopedPitId,
    pub entries: Vec<PitQueryGuardEntryV1>,
    pub grantor_expires_at: LogicalMillis,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearchCoordinateRequestV1 {
    pub format_version: u16,
    pub route: EngineApiResolvedRouteVector,
    pub plan: EngineApiCanonicalPlan,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub continuation: Option<EngineApiContinuation>,
    pub targets: Vec<ShardDispatchTargetsV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pit_guard: Option<PitQueryGuardV1>,
    pub remaining_worker_millis: u64,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearchCoordinateResponseV1 {
    pub format_version: u16,
    pub route: EngineApiResolvedRouteVector,
    /// Checked projection of every result lineage admitted by each output
    /// shard. This is required because inherited P6.4 hits retain the shard
    /// identity that assigned their sequence number.
    pub result_lineages: Vec<ShardResultLineagesV1>,
    pub result: EngineApiSearchResult,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ShardResultLineagesV1 {
    pub shard_key: ShardKey,
    pub result_lineages: Vec<ShardId>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WriterRealtimeRequestV1 {
    pub format_version: u16,
    pub route: EngineApiResolvedRouteVector,
    pub route_entry: EngineApiResolvedRouteEntry,
    pub shard: EngineApiResolvedShardRoute,
    pub plan: EngineApiCanonicalPlan,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub continuation: Option<EngineApiContinuation>,
    pub index_mode: IndexMode,
    pub target: WriterTargetV1,
    pub manifest_binding: PinnedManifestBindingV1,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub inherited_replays: Vec<WriterInheritedReplayV1>,
    pub candidate_limit: usize,
    pub remaining_worker_millis: u64,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WireTombstoneV1 {
    #[serde(with = "wire_identity_hash_decimal")]
    pub identity_hash: IdHash,
    pub kill_below_seq: SeqNo,
}

mod wire_identity_hash_decimal {
    use std::fmt::{self, Formatter};

    use lodestone_types::IdHash;
    use serde::de::{self, Visitor};

    use super::{Deserializer, Serializer};

    pub fn serialize<S>(value: &IdHash, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(&value.get())
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<IdHash, D::Error>
    where
        D: Deserializer<'de>,
    {
        struct CanonicalIdentityHashVisitor;

        impl Visitor<'_> for CanonicalIdentityHashVisitor {
            type Value = IdHash;

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

            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
            where
                E: de::Error,
            {
                if value.is_empty()
                    || value.len() > 1 && value.starts_with('0')
                    || !value.bytes().all(|byte| byte.is_ascii_digit())
                {
                    return Err(E::custom("identity hash is not a canonical decimal string"));
                }
                value
                    .parse::<u128>()
                    .map(IdHash::new)
                    .map_err(|_| E::custom("identity hash is outside the u128 domain"))
            }
        }

        deserializer.deserialize_str(CanonicalIdentityHashVisitor)
    }
}

#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum AggregationScalarV1 {
    Bool(bool),
    I64(i64),
    U64(u64),
    F64(f64),
    String(String),
    HistogramIndex(i64),
    Range { ordinal: u32, key: String },
    Tuple(Vec<AggregationScalarV1>),
}

impl<'de> Deserialize<'de> for AggregationScalarV1 {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct TaggedScalar {
            kind: String,
            value: Box<serde_json::value::RawValue>,
        }

        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct RangeValue {
            ordinal: u32,
            key: String,
        }

        fn decode<E, T>(raw: &serde_json::value::RawValue) -> Result<T, E>
        where
            E: serde::de::Error,
            T: for<'value> Deserialize<'value>,
        {
            serde_json::from_str(raw.get()).map_err(E::custom)
        }

        let tagged = TaggedScalar::deserialize(deserializer)?;
        match tagged.kind.as_str() {
            "bool" => decode(&tagged.value).map(Self::Bool),
            "i64" => decode(&tagged.value).map(Self::I64),
            "u64" => decode(&tagged.value).map(Self::U64),
            "f64" => decode(&tagged.value).map(Self::F64),
            "string" => decode(&tagged.value).map(Self::String),
            "histogram_index" => decode(&tagged.value).map(Self::HistogramIndex),
            "range" => decode::<D::Error, RangeValue>(&tagged.value).map(|value| Self::Range {
                ordinal: value.ordinal,
                key: value.key,
            }),
            "tuple" => decode(&tagged.value).map(Self::Tuple),
            value => Err(serde::de::Error::unknown_variant(
                value,
                &[
                    "bool",
                    "i64",
                    "u64",
                    "f64",
                    "string",
                    "histogram_index",
                    "range",
                    "tuple",
                ],
            )),
        }
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AggregationBucketPartialV1 {
    pub key: AggregationScalarV1,
    pub doc_count: u64,
    pub aggregations: BTreeMap<String, AggregationPartialV1>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum AggregationPartialV1 {
    Buckets {
        buckets: Vec<AggregationBucketPartialV1>,
    },
    Stats {
        count: u64,
        min: Option<f64>,
        max: Option<f64>,
        sum: f64,
    },
    Min {
        value: Option<f64>,
    },
    Max {
        value: Option<f64>,
    },
    Avg {
        count: u64,
        sum: f64,
    },
    Sum {
        sum: f64,
    },
    ValueCount {
        count: u64,
    },
    Cardinality {
        values: Vec<AggregationScalarV1>,
    },
    Percentiles {
        samples: Vec<f64>,
    },
    ApproximateCardinality {
        format_version: u16,
        precision: u8,
        registers: Vec<u8>,
    },
    ApproximatePercentiles {
        format_version: u16,
        precision: u8,
        count: u64,
        counts: Vec<u64>,
        minima: Vec<u64>,
        maxima: Vec<u64>,
    },
}

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

        impl<'de> serde::de::Visitor<'de> for AggregationPartialVisitor {
            type Value = AggregationPartialV1;

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

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: serde::de::MapAccess<'de>,
            {
                let mut kind = None;
                let mut buckets = None;
                let mut count = None;
                let mut min = None;
                let mut max = None;
                let mut sum = None;
                let mut value = None;
                let mut values = None;
                let mut samples = None;
                let mut format_version = None;
                let mut precision = None;
                let mut registers = None;
                let mut counts = None;
                let mut minima = None;
                let mut maxima = None;
                while let Some(field) = map.next_key::<String>()? {
                    match field.as_str() {
                        "kind" => {
                            if kind.replace(map.next_value::<String>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("kind"));
                            }
                        }
                        "buckets" => {
                            if buckets
                                .replace(map.next_value::<Vec<AggregationBucketPartialV1>>()?)
                                .is_some()
                            {
                                return Err(serde::de::Error::duplicate_field("buckets"));
                            }
                        }
                        "count" => {
                            if count.replace(map.next_value::<u64>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("count"));
                            }
                        }
                        "min" => {
                            if min.replace(map.next_value::<Option<f64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("min"));
                            }
                        }
                        "max" => {
                            if max.replace(map.next_value::<Option<f64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("max"));
                            }
                        }
                        "sum" => {
                            if sum.replace(map.next_value::<f64>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("sum"));
                            }
                        }
                        "value" => {
                            if value.replace(map.next_value::<Option<f64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("value"));
                            }
                        }
                        "values" => {
                            if values.replace(map.next_value::<Vec<AggregationScalarV1>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("values"));
                            }
                        }
                        "samples" => {
                            if samples.replace(map.next_value::<Vec<f64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("samples"));
                            }
                        }
                        "format_version" => {
                            if format_version.replace(map.next_value::<u16>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("format_version"));
                            }
                        }
                        "precision" => {
                            if precision.replace(map.next_value::<u8>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("precision"));
                            }
                        }
                        "registers" => {
                            if registers.replace(map.next_value::<Vec<u8>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("registers"));
                            }
                        }
                        "counts" => {
                            if counts.replace(map.next_value::<Vec<u64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("counts"));
                            }
                        }
                        "minima" => {
                            if minima.replace(map.next_value::<Vec<u64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("minima"));
                            }
                        }
                        "maxima" => {
                            if maxima.replace(map.next_value::<Vec<u64>>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("maxima"));
                            }
                        }
                        _ => {
                            return Err(serde::de::Error::unknown_field(
                                &field,
                                &[
                                    "kind",
                                    "buckets",
                                    "count",
                                    "min",
                                    "max",
                                    "sum",
                                    "value",
                                    "values",
                                    "samples",
                                    "format_version",
                                    "precision",
                                    "registers",
                                    "counts",
                                    "minima",
                                    "maxima",
                                ],
                            ));
                        }
                    }
                }
                let kind = kind.ok_or_else(|| serde::de::Error::missing_field("kind"))?;
                let fields_do_not_match =
                    || serde::de::Error::custom("aggregation partial fields do not match its kind");
                match kind.as_str() {
                    "buckets"
                        if count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        buckets
                            .map(|buckets| AggregationPartialV1::Buckets { buckets })
                            .ok_or_else(|| serde::de::Error::missing_field("buckets"))
                    }
                    "stats"
                        if buckets.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::Stats {
                            count: count.ok_or_else(|| serde::de::Error::missing_field("count"))?,
                            min: min.ok_or_else(|| serde::de::Error::missing_field("min"))?,
                            max: max.ok_or_else(|| serde::de::Error::missing_field("max"))?,
                            sum: sum.ok_or_else(|| serde::de::Error::missing_field("sum"))?,
                        })
                    }
                    "min"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        value
                            .map(|value| AggregationPartialV1::Min { value })
                            .ok_or_else(|| serde::de::Error::missing_field("value"))
                    }
                    "max"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        value
                            .map(|value| AggregationPartialV1::Max { value })
                            .ok_or_else(|| serde::de::Error::missing_field("value"))
                    }
                    "avg"
                        if buckets.is_none()
                            && min.is_none()
                            && max.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::Avg {
                            count: count.ok_or_else(|| serde::de::Error::missing_field("count"))?,
                            sum: sum.ok_or_else(|| serde::de::Error::missing_field("sum"))?,
                        })
                    }
                    "sum"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::Sum {
                            sum: sum.ok_or_else(|| serde::de::Error::missing_field("sum"))?,
                        })
                    }
                    "value_count"
                        if buckets.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::ValueCount {
                            count: count.ok_or_else(|| serde::de::Error::missing_field("count"))?,
                        })
                    }
                    "cardinality"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && samples.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::Cardinality {
                            values: values.ok_or_else(|| serde::de::Error::missing_field("values"))?,
                        })
                    }
                    "percentiles"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && values.is_none()
                            && format_version.is_none()
                            && precision.is_none()
                            && registers.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::Percentiles {
                            samples: samples.ok_or_else(|| serde::de::Error::missing_field("samples"))?,
                        })
                    }
                    "approximate_cardinality"
                        if buckets.is_none()
                            && count.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && counts.is_none()
                            && minima.is_none()
                            && maxima.is_none() =>
                    {
                        Ok(AggregationPartialV1::ApproximateCardinality {
                            format_version: format_version
                                .ok_or_else(|| serde::de::Error::missing_field("format_version"))?,
                            precision: precision.ok_or_else(|| serde::de::Error::missing_field("precision"))?,
                            registers: registers.ok_or_else(|| serde::de::Error::missing_field("registers"))?,
                        })
                    }
                    "approximate_percentiles"
                        if buckets.is_none()
                            && min.is_none()
                            && max.is_none()
                            && sum.is_none()
                            && value.is_none()
                            && values.is_none()
                            && samples.is_none()
                            && registers.is_none() =>
                    {
                        Ok(AggregationPartialV1::ApproximatePercentiles {
                            format_version: format_version
                                .ok_or_else(|| serde::de::Error::missing_field("format_version"))?,
                            precision: precision.ok_or_else(|| serde::de::Error::missing_field("precision"))?,
                            count: count.ok_or_else(|| serde::de::Error::missing_field("count"))?,
                            counts: counts.ok_or_else(|| serde::de::Error::missing_field("counts"))?,
                            minima: minima.ok_or_else(|| serde::de::Error::missing_field("minima"))?,
                            maxima: maxima.ok_or_else(|| serde::de::Error::missing_field("maxima"))?,
                        })
                    }
                    "buckets"
                    | "stats"
                    | "min"
                    | "max"
                    | "avg"
                    | "sum"
                    | "value_count"
                    | "cardinality"
                    | "percentiles"
                    | "approximate_cardinality"
                    | "approximate_percentiles" => Err(fields_do_not_match()),
                    other => Err(serde::de::Error::unknown_variant(
                        other,
                        &[
                            "buckets",
                            "stats",
                            "min",
                            "max",
                            "avg",
                            "sum",
                            "value_count",
                            "cardinality",
                            "percentiles",
                            "approximate_cardinality",
                            "approximate_percentiles",
                        ],
                    )),
                }
            }
        }

        // INVARIANT: as with the realtime disposition below, direct field
        // decoding avoids serde Content buffering of arbitrary-precision JSON
        // numbers while preserving the established tagged wire shape.
        deserializer.deserialize_map(AggregationPartialVisitor)
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ShardSearchPartialV1 {
    pub format_version: u16,
    pub shard_key: ShardKey,
    pub total: u64,
    pub hits: Vec<EngineApiScopedSearchHit>,
    pub aggregations: BTreeMap<String, AggregationPartialV1>,
    pub freshness: SearchExecutionFreshness,
    pub freshness_degraded: bool,
    pub has_more_hits: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub applied_continuation: Option<EngineApiContinuation>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WriterRealtimeResponseV1 {
    pub format_version: u16,
    pub target: WriterTargetV1,
    pub manifest_binding: PinnedManifestBindingV1,
    pub cut: SeqNo,
    pub tombstones: Vec<WireTombstoneV1>,
    pub fresh_partial: ShardSearchPartialV1,
}

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

#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum RealtimeDispositionV1 {
    Exact { writer: Box<WriterRealtimeResponseV1> },
    DegradedCommitted { reason: CommittedDegradationV1 },
}

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

        impl<'de> serde::de::Visitor<'de> for RealtimeDispositionVisitor {
            type Value = RealtimeDispositionV1;

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

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: serde::de::MapAccess<'de>,
            {
                let mut kind = None;
                let mut writer = None;
                let mut reason = None;
                while let Some(field) = map.next_key::<String>()? {
                    match field.as_str() {
                        "kind" => {
                            if kind.replace(map.next_value::<String>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("kind"));
                            }
                        }
                        "writer" => {
                            if writer
                                .replace(Box::new(map.next_value::<WriterRealtimeResponseV1>()?))
                                .is_some()
                            {
                                return Err(serde::de::Error::duplicate_field("writer"));
                            }
                        }
                        "reason" => {
                            if reason.replace(map.next_value::<CommittedDegradationV1>()?).is_some() {
                                return Err(serde::de::Error::duplicate_field("reason"));
                            }
                        }
                        _ => {
                            return Err(serde::de::Error::unknown_field(&field, &["kind", "writer", "reason"]));
                        }
                    }
                }
                match kind.as_deref() {
                    Some("exact") if reason.is_none() => writer
                        .map(|writer| RealtimeDispositionV1::Exact { writer })
                        .ok_or_else(|| serde::de::Error::missing_field("writer")),
                    Some("degraded_committed") if writer.is_none() => reason
                        .map(|reason| RealtimeDispositionV1::DegradedCommitted { reason })
                        .ok_or_else(|| serde::de::Error::missing_field("reason")),
                    Some("exact" | "degraded_committed") => Err(serde::de::Error::custom(
                        "realtime disposition fields do not match its kind",
                    )),
                    Some(value) => Err(serde::de::Error::unknown_variant(
                        value,
                        &["exact", "degraded_committed"],
                    )),
                    None => Err(serde::de::Error::missing_field("kind")),
                }
            }
        }

        // INVARIANT: derive's internally-tagged decoder buffers through
        // serde's Content representation. With serde_json arbitrary-precision
        // numbers that turns nested f32 scores into a reserved map before the
        // typed writer partial sees them. This streaming map visitor preserves
        // the existing wire shape while decoding every nested number directly.
        deserializer.deserialize_map(RealtimeDispositionVisitor)
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearchShardRequestV1 {
    pub format_version: u16,
    pub route: EngineApiResolvedRouteVector,
    pub route_entry: EngineApiResolvedRouteEntry,
    pub shard: EngineApiResolvedShardRoute,
    pub plan: EngineApiCanonicalPlan,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub continuation: Option<EngineApiContinuation>,
    pub entitlement: SearchShardEntitlementV1,
    pub result_lineages: Vec<ShardId>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub committed_segment_lineages: Vec<CommittedSegmentLineageV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub inherited_execution: Option<InheritedShardExecutionV1>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub writer_replays: Vec<WriterInheritedReplayV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub realtime: Option<RealtimeDispositionV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pit_guard: Option<PitQueryGuardV1>,
    pub candidate_limit: usize,
    pub remaining_worker_millis: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct InheritedShardExecutionV1 {
    pub source_shard: ShardKey,
    pub predicate: InheritedHashRangeV1,
    pub sequence_ceiling: SeqNo,
    pub result_lineage: ShardId,
    pub output_manifest_binding: PinnedManifestBindingV1,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SearchShardResponseV1 {
    pub format_version: u16,
    pub served_by: SearcherTargetV1,
    pub partial: ShardSearchPartialV1,
}

impl SearchCoordinateRequestV1 {
    pub fn result_lineages(&self) -> Result<Vec<ShardResultLineagesV1>, SearchWireError> {
        self.targets
            .iter()
            .map(|target| {
                if target.result_lineages.is_empty()
                    || target.result_lineages.len() > SEARCH_MAX_RESULT_LINEAGES
                    || target.result_lineages.windows(2).any(|pair| pair[0] >= pair[1])
                    || target
                        .result_lineages
                        .binary_search(&target.shard_key.shard_id())
                        .is_err()
                {
                    return Err(SearchWireError::InvalidTargetVector);
                }
                Ok(ShardResultLineagesV1 {
                    shard_key: target.shard_key,
                    result_lineages: target.result_lineages.clone(),
                })
            })
            .collect()
    }

    pub fn validate(&self) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        let route_shards = validate_route(&self.route, requires_complete_partition(&self.plan))?;
        validate_plan_for_route(&self.plan, self.continuation.as_ref(), &self.route)?;
        if self.targets.is_empty() || self.targets.len() > SEARCH_MAX_SHARDS || self.remaining_worker_millis == 0 {
            return Err(SearchWireError::InvalidCoordinate);
        }
        let target_shards = self
            .targets
            .iter()
            .map(|target| target.shard_key)
            .collect::<BTreeSet<_>>();
        if route_shards.len() != self.targets.len()
            || target_shards.len() != self.targets.len()
            || route_shards != target_shards
        {
            return Err(SearchWireError::InvalidTargetVector);
        }
        for target in &self.targets {
            validate_dispatch_target(target)?;
        }
        let _result_lineages = self.result_lineages()?;
        for entry in &self.route.entries {
            for shard in &entry.shards {
                let target = self
                    .targets
                    .iter()
                    .find(|target| target.shard_key == shard.shard_key)
                    .ok_or(SearchWireError::InvalidTargetVector)?;
                validate_searcher_target(&target.primary, *shard)?;
                if let Some(hedge) = &target.hedge {
                    validate_searcher_target(hedge, *shard)?;
                }
                if target.primary.manifest_binding.admission_generation != entry.admission_generation
                    || target.primary.manifest_binding.routing_revision != entry.routing_revision
                {
                    return Err(SearchWireError::InvalidTargetVector);
                }
                for inherited in &target.inherited_sources {
                    validate_inherited_source(inherited, *shard)?;
                    if inherited.primary.manifest_binding.admission_generation != entry.admission_generation {
                        return Err(SearchWireError::InvalidTargetVector);
                    }
                }
            }
        }
        match (&self.plan.pit_id, &self.pit_guard) {
            (Some(pit_id), Some(guard)) => guard.validate(pit_id, &self.route)?,
            (None, None) => {}
            _ => return Err(SearchWireError::InvalidPitGuard),
        }
        if self.plan.pit_id.is_some() {
            for target in &self.targets {
                if target.writer.is_some()
                    || target.primary.manifest_binding != root_for(&self.pit_guard, target.shard_key)?
                {
                    return Err(SearchWireError::InvalidPitGuard);
                }
            }
        }
        Ok(())
    }
}

impl SearchCoordinateResponseV1 {
    pub fn validate(&self) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        let route_shards =
            validate_route(&self.route, false).map_err(|_| SearchWireError::InvalidCoordinateResponse)?;
        let route_count = u32::try_from(route_shards.len()).map_err(|_| SearchWireError::InvalidCoordinateResponse)?;
        let accounted = self
            .result
            .shards
            .successful
            .checked_add(self.result.shards.skipped)
            .and_then(|count| count.checked_add(self.result.shards.failed));
        if self.route != self.result.route
            || self.route.format_version != SCOPED_SEARCH_FORMAT_VERSION
            || self.result.format_version != SCOPED_SEARCH_FORMAT_VERSION
            || self.result.shards.total != route_count
            || accounted != Some(route_count)
            || self.result.shards.skipped != 0
            || usize::try_from(self.result.shards.failed).ok() != Some(self.result.failures.len())
            || self.result.partial != (self.result.shards.failed != 0)
            || self.result.total.partial != self.result.partial
            || self.result.hits_partial != self.result.partial
            || self.result.aggregations_partial != self.result.partial
            || self.result.freshness == SearchExecutionFreshness::Realtime && self.result.freshness_degraded
            || self
                .result
                .failures
                .windows(2)
                .any(|pair| pair[0].shard_key >= pair[1].shard_key)
            || self
                .result
                .failures
                .iter()
                .any(|failure| !route_shards.contains(&failure.shard_key))
            || validate_result_lineages(&self.route, &self.result_lineages).is_err()
            || self.result.hits.iter().any(|hit| {
                self.route
                    .entries
                    .iter()
                    .find(|entry| entry.collection_index_ref == hit.collection_index_ref)
                    .is_none_or(|entry| {
                        entry.index != hit.index
                            || !entry.shards.iter().any(|shard| {
                                self.result_lineages
                                    .iter()
                                    .find(|lineages| lineages.shard_key == shard.shard_key)
                                    .is_some_and(|lineages| {
                                        lineages.result_lineages.binary_search(&hit.shard_id).is_ok()
                                    })
                            })
                    })
            })
        {
            return Err(SearchWireError::InvalidCoordinateResponse);
        }
        Ok(())
    }
}

impl WriterRealtimeRequestV1 {
    pub fn validate(&self) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        validate_route(&self.route, false)?;
        validate_route_pair(&self.route, &self.route_entry, &self.shard)?;
        validate_plan_for_route(&self.plan, self.continuation.as_ref(), &self.route)?;
        validate_writer_target(&self.target)?;
        self.manifest_binding
            .validate()
            .map_err(|_| SearchWireError::InvalidManifestBinding)?;
        validate_writer_replays(&self.inherited_replays, self.shard)?;
        if self.plan.pit_id.is_some()
            || !search_wants_realtime(self.plan.freshness, self.index_mode)
            || self.manifest_binding.shard_key != self.shard.shard_key
            || self.manifest_binding.admission_generation != self.route_entry.admission_generation
            || self.manifest_binding.routing_revision != self.route_entry.routing_revision
            || self.target.fence.recovery_era() != self.target.node_token.recovery_era
            || self.target.node_token.recovery_era != self.manifest_binding.fence.recovery_era()
            || self.target.fence.epoch() < self.manifest_binding.fence.epoch()
            || self.candidate_limit == 0
            || self.candidate_limit > SEARCH_MAX_HITS
            || self.remaining_worker_millis == 0
        {
            return Err(SearchWireError::InvalidWriterRequest);
        }
        Ok(())
    }
}

impl WriterRealtimeResponseV1 {
    pub fn validate(&self) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        validate_writer_target(&self.target)?;
        self.manifest_binding
            .validate()
            .map_err(|_| SearchWireError::InvalidManifestBinding)?;
        validate_partial_shape(&self.fresh_partial)?;
        if self.target.node_token.recovery_era != self.manifest_binding.fence.recovery_era()
            || self.target.fence.epoch() < self.manifest_binding.fence.epoch()
            || self.cut < self.manifest_binding.committed_through
            || self.fresh_partial.shard_key != self.manifest_binding.shard_key
            || self.fresh_partial.freshness != SearchExecutionFreshness::Realtime
            || self.fresh_partial.freshness_degraded
            || self.fresh_partial.hits.iter().any(|hit| hit.seq_no > self.cut)
            || self
                .tombstones
                .windows(2)
                .any(|pair| pair[0].identity_hash >= pair[1].identity_hash)
            || self
                .tombstones
                .iter()
                .any(|tombstone| tombstone.kill_below_seq > self.cut)
            || self.tombstones.len() > SEARCH_MAX_TOMBSTONES
        {
            return Err(SearchWireError::InvalidWriterResponse);
        }
        Ok(())
    }

    pub fn validate_for(&self, request: &WriterRealtimeRequestV1) -> Result<(), SearchWireError> {
        self.validate()?;
        validate_partial_for_limit(&self.fresh_partial, request.candidate_limit)?;
        let mut result_lineages = request
            .inherited_replays
            .iter()
            .map(|replay| replay.result_lineage)
            .collect::<Vec<_>>();
        result_lineages.push(request.shard.shard_key.shard_id());
        result_lineages.sort_unstable();
        result_lineages.dedup();
        validate_partial_for_route_lineages(
            &self.fresh_partial,
            &request.route_entry,
            request.shard,
            &request.plan,
            &result_lineages,
        )?;
        if self.target != request.target
            || self.manifest_binding != request.manifest_binding
            || self.fresh_partial.shard_key != request.shard.shard_key
            || self.fresh_partial.applied_continuation != request.continuation
            || self.fresh_partial.hits.iter().any(|hit| {
                if hit.shard_id == request.shard.shard_key.shard_id() {
                    hit.seq_no <= self.manifest_binding.committed_through || hit.seq_no > self.cut
                } else {
                    request
                        .inherited_replays
                        .iter()
                        .find(|replay| replay.result_lineage == hit.shard_id)
                        .is_none_or(|replay| {
                            hit.seq_no <= replay.committed_through
                                || hit.seq_no > replay.sequence_ceiling
                                || hit.seq_no > self.cut
                        })
                }
            })
        {
            return Err(SearchWireError::InvalidWriterResponse);
        }
        Ok(())
    }
}

impl SearchShardRequestV1 {
    pub fn validate(&self) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        validate_route(&self.route, false)?;
        validate_route_pair(&self.route, &self.route_entry, &self.shard)?;
        validate_plan_for_route(&self.plan, self.continuation.as_ref(), &self.route)?;
        let (execution_shard, assignment_generation) = self
            .inherited_execution
            .as_ref()
            .map_or((self.shard.shard_key, self.shard.assignment_generation), |inherited| {
                (inherited.source_shard, self.entitlement.assignment_generation)
            });
        validate_search_entitlement(&self.entitlement, execution_shard, assignment_generation)?;
        if self.result_lineages.is_empty()
            || self.result_lineages.len() > SEARCH_MAX_RESULT_LINEAGES
            || self.result_lineages.windows(2).any(|pair| pair[0] >= pair[1])
            || self
                .result_lineages
                .binary_search(&self.shard.shard_key.shard_id())
                .is_err()
            || self
                .committed_segment_lineages
                .windows(2)
                .any(|pair| pair[0] >= pair[1])
            || self
                .committed_segment_lineages
                .iter()
                .any(|segment| self.result_lineages.binary_search(&segment.result_lineage).is_err())
        {
            return Err(SearchWireError::InvalidShardRequest);
        }
        if let Some(inherited) = &self.inherited_execution {
            inherited
                .output_manifest_binding
                .validate()
                .map_err(|_| SearchWireError::InvalidManifestBinding)?;
            if inherited.source_shard.collection_index_ref() != self.shard.shard_key.collection_index_ref()
                || inherited.source_shard == self.shard.shard_key
                || inherited.result_lineage != inherited.source_shard.shard_id()
                || inherited.source_shard != self.entitlement.manifest_binding.shard_key
                || inherited.output_manifest_binding.shard_key != self.shard.shard_key
                || inherited.output_manifest_binding.admission_generation != self.route_entry.admission_generation
                || inherited.output_manifest_binding.routing_revision != self.route_entry.routing_revision
                || self.entitlement.manifest_binding.committed_through > inherited.sequence_ceiling
                || u64::from(inherited.predicate.start.get()) >= inherited.predicate.end.get()
                || inherited.predicate.start < self.shard.hash_start
                || inherited.predicate.end > self.shard.hash_end
            {
                return Err(SearchWireError::InvalidShardRequest);
            }
        }
        if self.candidate_limit == 0 || self.candidate_limit > SEARCH_MAX_HITS || self.remaining_worker_millis == 0 {
            return Err(SearchWireError::InvalidShardRequest);
        }
        match (&self.plan.pit_id, &self.pit_guard, &self.realtime) {
            (Some(pit_id), Some(guard), None) => {
                guard.validate(pit_id, &self.route)?;
                let expected_root = self.inherited_execution.as_ref().map_or_else(
                    || self.entitlement.manifest_binding.clone(),
                    |inherited| inherited.output_manifest_binding.clone(),
                );
                if root_for(&Some(guard.clone()), self.shard.shard_key)? != expected_root {
                    return Err(SearchWireError::InvalidPitGuard);
                }
            }
            (None, None, Some(RealtimeDispositionV1::Exact { writer })) => {
                let manifest_binding = self.inherited_execution.as_ref().map_or_else(
                    || self.entitlement.manifest_binding.clone(),
                    |inherited| inherited.output_manifest_binding.clone(),
                );
                let writer_request = WriterRealtimeRequestV1 {
                    format_version: SEARCH_WIRE_FORMAT_VERSION,
                    route: self.route.clone(),
                    route_entry: self.route_entry.clone(),
                    shard: self.shard,
                    plan: self.plan.clone(),
                    continuation: self.continuation.clone(),
                    index_mode: self.entitlement.index_mode,
                    target: writer.target,
                    manifest_binding,
                    inherited_replays: self.writer_replays.clone(),
                    candidate_limit: self.candidate_limit,
                    remaining_worker_millis: self.remaining_worker_millis,
                };
                writer_request.validate()?;
                writer.validate_for(&writer_request)?;
            }
            (None, None, Some(RealtimeDispositionV1::DegradedCommitted { .. }))
                if search_wants_realtime(self.plan.freshness, self.entitlement.index_mode)
                    && self.writer_replays.is_empty() => {}
            (None, None, None)
                if !search_wants_realtime(self.plan.freshness, self.entitlement.index_mode)
                    && self.writer_replays.is_empty() => {}
            _ => return Err(SearchWireError::InvalidShardRequest),
        }
        Ok(())
    }
}

impl SearchShardResponseV1 {
    pub fn validate_for(&self, request: &SearchShardRequestV1) -> Result<(), SearchWireError> {
        check_version(self.format_version)?;
        validate_partial_for_limit(&self.partial, request.candidate_limit)?;
        let result_lineages = request.inherited_execution.as_ref().map_or_else(
            || request.result_lineages.clone(),
            |inherited| vec![inherited.result_lineage],
        );
        validate_partial_for_route_lineages(
            &self.partial,
            &request.route_entry,
            request.shard,
            &request.plan,
            &result_lineages,
        )?;
        if self.served_by.assignment_generation != request.entitlement.assignment_generation
            || self.served_by.manifest_binding != request.entitlement.manifest_binding
            || !request.entitlement.allowed_nodes.contains(&self.served_by.node_token)
            || self.partial.shard_key != request.shard.shard_key
            || self.partial.applied_continuation != request.continuation
            || matches!(request.realtime, Some(RealtimeDispositionV1::Exact { .. }))
                && (self.partial.freshness != SearchExecutionFreshness::Realtime || self.partial.freshness_degraded)
            || matches!(request.realtime, Some(RealtimeDispositionV1::DegradedCommitted { .. }))
                && (self.partial.freshness != SearchExecutionFreshness::Committed || !self.partial.freshness_degraded)
            || request.plan.pit_id.is_some() && self.partial.freshness != SearchExecutionFreshness::Committed
        {
            return Err(SearchWireError::InvalidShardResponse);
        }
        Ok(())
    }
}

impl PitQueryGuardV1 {
    fn validate(&self, pit_id: &ScopedPitId, route: &EngineApiResolvedRouteVector) -> Result<(), SearchWireError> {
        if self.format_version != SEARCH_WIRE_FORMAT_VERSION
            || &self.pit_id != pit_id
            || self.grantor_expires_at.get() == 0
            || self.entries.len() != route.entries.len()
        {
            return Err(SearchWireError::InvalidPitGuard);
        }
        for (guard_entry, route_entry) in self.entries.iter().zip(&route.entries) {
            self.validate_entry(pit_id, guard_entry, route_entry)?;
            if guard_entry.roots.len() != route_entry.shards.len()
                || guard_entry
                    .roots
                    .iter()
                    .map(|root| root.shard_key)
                    .ne(route_entry.shards.iter().map(|shard| shard.shard_key))
            {
                return Err(SearchWireError::InvalidPitGuard);
            }
        }
        Ok(())
    }

    fn validate_entry(
        &self,
        pit_id: &ScopedPitId,
        guard_entry: &PitQueryGuardEntryV1,
        route_entry: &EngineApiResolvedRouteEntry,
    ) -> Result<(), SearchWireError> {
        if guard_entry.collection_index_ref != route_entry.collection_index_ref
            || guard_entry.admission_generation != route_entry.admission_generation
            || guard_entry.schema_generation != route_entry.schema_generation
            || guard_entry.routing_revision != route_entry.routing_revision
            || guard_entry.roots.is_empty()
        {
            return Err(SearchWireError::InvalidPitGuard);
        }
        let mut previous = None;
        for root in &guard_entry.roots {
            root.validate().map_err(|_| SearchWireError::InvalidPitGuard)?;
            if previous.is_some_and(|key| key >= root.shard_key)
                || root.fence.recovery_era() != pit_id.recovery_era()
                || root.shard_key.collection_index_ref() != guard_entry.collection_index_ref
                || root.admission_generation != guard_entry.admission_generation
                || root.routing_revision != guard_entry.routing_revision
            {
                return Err(SearchWireError::InvalidPitGuard);
            }
            previous = Some(root.shard_key);
        }
        Ok(())
    }
}

fn root_for(guard: &Option<PitQueryGuardV1>, shard: ShardKey) -> Result<PinnedManifestBindingV1, SearchWireError> {
    guard
        .as_ref()
        .and_then(|guard| {
            guard
                .entries
                .iter()
                .flat_map(|entry| &entry.roots)
                .find(|root| root.shard_key == shard)
        })
        .cloned()
        .ok_or(SearchWireError::InvalidPitGuard)
}

fn validate_route_pair(
    route: &EngineApiResolvedRouteVector,
    entry: &EngineApiResolvedRouteEntry,
    shard: &EngineApiResolvedShardRoute,
) -> Result<(), SearchWireError> {
    if shard.shard_key.collection_index_ref() != entry.collection_index_ref
        || shard.assignment_generation.get() == 0
        || !route.entries.iter().any(|candidate| candidate == entry)
        || !entry.shards.iter().any(|candidate| candidate == shard)
    {
        return Err(SearchWireError::InvalidRoute);
    }
    Ok(())
}

fn validate_route(
    route: &EngineApiResolvedRouteVector,
    require_complete_partition: bool,
) -> Result<BTreeSet<ShardKey>, SearchWireError> {
    if route.format_version != SCOPED_SEARCH_FORMAT_VERSION
        || route.entries.is_empty()
        || route.entries.len() > crate::MAX_ENGINE_API_INDEXES
        || route
            .entries
            .windows(2)
            .any(|pair| pair[0].collection_index_ref >= pair[1].collection_index_ref)
    {
        return Err(SearchWireError::InvalidRoute);
    }
    let mut keys = BTreeSet::new();
    let mut total_shards = 0_usize;
    for entry in &route.entries {
        total_shards = total_shards
            .checked_add(entry.shards.len())
            .ok_or(SearchWireError::InvalidRoute)?;
        if total_shards > SEARCH_MAX_SHARDS
            || entry.admission_generation.get() == 0
            || entry.schema_generation.get() == 0
            || entry.routing_revision.get() == 0
            || entry.shards.is_empty()
            || require_complete_partition
                && (entry.shards.first().is_none_or(|shard| shard.hash_start.get() != 0)
                    || entry
                        .shards
                        .last()
                        .is_none_or(|shard| shard.hash_end != lodestone_types::RoutingRangeEnd::FULL))
            || entry.shards.windows(2).any(|pair| {
                pair[0].hash_start >= pair[1].hash_start
                    || pair[0].hash_end.get() > u64::from(pair[1].hash_start.get())
                    || require_complete_partition && pair[0].hash_end.get() != u64::from(pair[1].hash_start.get())
            })
        {
            return Err(SearchWireError::InvalidRoute);
        }
        for shard in &entry.shards {
            if shard.shard_key.collection_index_ref() != entry.collection_index_ref
                || u64::from(shard.hash_start.get()) >= shard.hash_end.get()
                || shard.assignment_generation.get() == 0
                || !keys.insert(shard.shard_key)
            {
                return Err(SearchWireError::InvalidRoute);
            }
        }
    }
    Ok(keys)
}

const fn requires_complete_partition(plan: &EngineApiCanonicalPlan) -> bool {
    plan.pit_id.is_none() && plan.routing.is_empty()
}

fn validate_plan_for_route(
    plan: &EngineApiCanonicalPlan,
    continuation: Option<&EngineApiContinuation>,
    route: &EngineApiResolvedRouteVector,
) -> Result<(), SearchWireError> {
    validate_scoped_search_plan(plan).map_err(|_| SearchWireError::InvalidRoute)?;
    if plan
        .pit_id
        .as_ref()
        .is_some_and(|pit_id| pit_id.recovery_era() != route.cluster_recovery_era)
    {
        return Err(SearchWireError::InvalidRoute);
    }
    if let Some(continuation) = continuation {
        validate_scoped_search_continuation(continuation).map_err(|_| SearchWireError::InvalidRoute)?;
        if continuation.canonical_plan_identity != *plan
            || continuation.route_identity != route_identity(route)
            || continuation.cluster_recovery_era != route.cluster_recovery_era
            || continuation.pit_id != plan.pit_id
        {
            return Err(SearchWireError::InvalidRoute);
        }
    }
    Ok(())
}

fn route_identity(route: &EngineApiResolvedRouteVector) -> Vec<EngineApiRouteIdentity> {
    route
        .entries
        .iter()
        .map(|entry| EngineApiRouteIdentity {
            collection_index_ref: entry.collection_index_ref,
            admission_generation: entry.admission_generation,
            schema_generation: entry.schema_generation,
            routing_revision: entry.routing_revision,
        })
        .collect()
}

fn validate_dispatch_target(target: &ShardDispatchTargetsV1) -> Result<(), SearchWireError> {
    validate_searcher_target_key(&target.primary, target.shard_key)?;
    if let Some(hedge) = &target.hedge {
        validate_searcher_target_key(hedge, target.shard_key)?;
        if hedge.node_token == target.primary.node_token || hedge.manifest_binding != target.primary.manifest_binding {
            return Err(SearchWireError::InvalidTargetVector);
        }
    }
    if let Some(writer) = &target.writer {
        validate_writer_target(writer)?;
        if writer.node_token.recovery_era != target.primary.node_token.recovery_era
            || writer.fence.epoch() < target.primary.manifest_binding.fence.epoch()
        {
            return Err(SearchWireError::InvalidTargetVector);
        }
    }
    if target.inherited_sources.len() > SEARCH_MAX_INHERITED_SOURCES
        || target
            .inherited_sources
            .windows(2)
            .any(|pair| pair[0].source_shard >= pair[1].source_shard)
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    if target.result_lineages.is_empty()
        || target.result_lineages.len() > SEARCH_MAX_RESULT_LINEAGES
        || target.result_lineages.windows(2).any(|pair| pair[0] >= pair[1])
        || target
            .result_lineages
            .binary_search(&target.shard_key.shard_id())
            .is_err()
        || target
            .committed_segment_lineages
            .windows(2)
            .any(|pair| pair[0] >= pair[1])
        || target
            .committed_segment_lineages
            .iter()
            .any(|segment| target.result_lineages.binary_search(&segment.result_lineage).is_err())
        || target
            .inherited_sources
            .iter()
            .any(|source| target.result_lineages.binary_search(&source.result_lineage).is_err())
        || !target.inherited_sources.is_empty() && !target.committed_segment_lineages.is_empty()
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn validate_inherited_source(
    source: &InheritedDispatchSourceV1,
    output: EngineApiResolvedShardRoute,
) -> Result<(), SearchWireError> {
    validate_searcher_target_key(&source.primary, source.source_shard)?;
    if let Some(hedge) = &source.hedge {
        validate_searcher_target_key(hedge, source.source_shard)?;
        if hedge.node_token == source.primary.node_token
            || hedge.manifest_binding != source.primary.manifest_binding
            || hedge.assignment_generation != source.primary.assignment_generation
        {
            return Err(SearchWireError::InvalidTargetVector);
        }
    }
    if source.source_shard.collection_index_ref() != output.shard_key.collection_index_ref()
        || source.source_shard == output.shard_key
        || source.result_lineage != source.source_shard.shard_id()
        || u64::from(source.predicate.start.get()) >= source.predicate.end.get()
        || source.predicate.start < output.hash_start
        || source.predicate.end > output.hash_end
        || source.primary.manifest_binding.committed_through > source.sequence_ceiling
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn validate_writer_replays(
    replays: &[WriterInheritedReplayV1],
    output: EngineApiResolvedShardRoute,
) -> Result<(), SearchWireError> {
    if replays.len() > SEARCH_MAX_INHERITED_SOURCES
        || replays
            .windows(2)
            .any(|pair| pair[0].source_shard >= pair[1].source_shard)
        || replays.iter().any(|replay| {
            replay.source_shard.collection_index_ref() != output.shard_key.collection_index_ref()
                || replay.source_shard == output.shard_key
                || replay.result_lineage != replay.source_shard.shard_id()
                || replay.committed_through > replay.sequence_ceiling
                || u64::from(replay.predicate.start.get()) >= replay.predicate.end.get()
                || replay.predicate.start < output.hash_start
                || replay.predicate.end > output.hash_end
        })
    {
        return Err(SearchWireError::InvalidWriterRequest);
    }
    Ok(())
}

fn validate_result_lineages(
    route: &EngineApiResolvedRouteVector,
    lineages: &[ShardResultLineagesV1],
) -> Result<(), SearchWireError> {
    let route_shards = route
        .entries
        .iter()
        .flat_map(|entry| entry.shards.iter().map(|shard| shard.shard_key))
        .collect::<BTreeSet<_>>();
    if lineages.len() != route_shards.len()
        || lineages.windows(2).any(|pair| pair[0].shard_key >= pair[1].shard_key)
        || lineages
            .iter()
            .map(|lineage| lineage.shard_key)
            .ne(route_shards.iter().copied())
    {
        return Err(SearchWireError::InvalidCoordinateResponse);
    }
    for lineage in lineages {
        let output = lineage.shard_key;
        let admitted = &lineage.result_lineages;
        if admitted.is_empty()
            || admitted.len() > SEARCH_MAX_RESULT_LINEAGES
            || admitted.windows(2).any(|pair| pair[0] >= pair[1])
            || admitted.binary_search(&output.shard_id()).is_err()
        {
            return Err(SearchWireError::InvalidCoordinateResponse);
        }
    }
    Ok(())
}

fn validate_searcher_target(
    target: &SearcherTargetV1,
    shard: EngineApiResolvedShardRoute,
) -> Result<(), SearchWireError> {
    validate_searcher_target_key(target, shard.shard_key)?;
    if target.assignment_generation != shard.assignment_generation {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn validate_search_entitlement(
    entitlement: &SearchShardEntitlementV1,
    shard_key: ShardKey,
    assignment_generation: lodestone_types::AssignmentGeneration,
) -> Result<(), SearchWireError> {
    entitlement
        .manifest_binding
        .validate()
        .map_err(|_| SearchWireError::InvalidManifestBinding)?;
    if entitlement.assignment_generation != assignment_generation
        || entitlement.manifest_binding.shard_key != shard_key
        || entitlement.allowed_nodes.is_empty()
        || entitlement.allowed_nodes.len() > 2
        || entitlement
            .allowed_nodes
            .windows(2)
            .any(|pair| compare_nodes(&pair[0], &pair[1]).is_ge())
        || entitlement.allowed_nodes.iter().any(|node| {
            node.node_id.get() == 0
                || node.node_incarnation.get() == 0
                || node.recovery_era != entitlement.manifest_binding.fence.recovery_era()
        })
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn compare_nodes(left: &NodeToken, right: &NodeToken) -> std::cmp::Ordering {
    left.recovery_era
        .to_be_bytes()
        .cmp(&right.recovery_era.to_be_bytes())
        .then_with(|| left.node_id.get().cmp(&right.node_id.get()))
        .then_with(|| left.node_incarnation.get().cmp(&right.node_incarnation.get()))
}

#[must_use]
pub const fn search_wants_realtime(freshness: SearchFreshness, index_mode: IndexMode) -> bool {
    matches!(freshness, SearchFreshness::Realtime)
        || matches!(freshness, SearchFreshness::Auto) && index_mode.has_mutation_visibility()
}

fn validate_searcher_target_key(target: &SearcherTargetV1, shard: ShardKey) -> Result<(), SearchWireError> {
    target
        .manifest_binding
        .validate()
        .map_err(|_| SearchWireError::InvalidManifestBinding)?;
    if target.node_token.node_id.get() == 0
        || target.node_token.node_incarnation.get() == 0
        || target.assignment_generation.get() == 0
        || target.manifest_binding.shard_key != shard
        || target.manifest_binding.fence.recovery_era() != target.node_token.recovery_era
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn validate_writer_target(target: &WriterTargetV1) -> Result<(), SearchWireError> {
    if target.node_token.node_id.get() == 0
        || target.node_token.node_incarnation.get() == 0
        || target.fence.epoch().get() == 0
        || target.fence.recovery_era() != target.node_token.recovery_era
    {
        return Err(SearchWireError::InvalidTargetVector);
    }
    Ok(())
}

fn validate_partial_shape(partial: &ShardSearchPartialV1) -> Result<(), SearchWireError> {
    check_version(partial.format_version)?;
    if partial.hits.len() > SEARCH_MAX_HITS
        || u64::try_from(partial.hits.len()).map_or(true, |hits| hits > partial.total)
    {
        return Err(SearchWireError::InvalidPartial);
    }
    let mut nested_hits = 0_usize;
    let mut nested_bytes = 0_usize;
    for hit in &partial.hits {
        validate_nested_inner_hits(hit, &mut nested_hits, &mut nested_bytes)?;
    }
    let mut entries = 0_usize;
    for value in partial.aggregations.values() {
        validate_aggregation(value, 0, &mut entries)?;
    }
    Ok(())
}

fn validate_nested_inner_hits(
    hit: &EngineApiScopedSearchHit,
    nested_hits: &mut usize,
    nested_bytes: &mut usize,
) -> Result<(), SearchWireError> {
    if hit.inner_hits.len() > SEARCH_MAX_NESTED_GROUPS_PER_HIT {
        return Err(SearchWireError::InvalidPartial);
    }
    for (name, result) in &hit.inner_hits {
        if name.is_empty()
            || name.len() > SEARCH_MAX_NESTED_NAME_BYTES
            || result.hits.len() > SEARCH_MAX_NESTED_HITS_PER_GROUP
            || result.total > u64::from(SEARCH_MAX_NESTED_ORDINAL)
            || u64::try_from(result.hits.len()).map_or(true, |returned| returned > result.total)
        {
            return Err(SearchWireError::InvalidPartial);
        }
        *nested_bytes = nested_bytes
            .checked_add(name.len().saturating_add(64))
            .ok_or(SearchWireError::InvalidPartial)?;
        let mut path = None;
        let mut identities = BTreeSet::new();
        for nested in &result.hits {
            if nested.path.is_empty()
                || nested.path.len() > SEARCH_MAX_NESTED_PATH_BYTES
                || nested.ordinal >= SEARCH_MAX_NESTED_ORDINAL
                || !nested.score.is_finite()
                || nested.score < 0.0
                || nested.score == 0.0 && nested.score.is_sign_negative()
                || nested.sort.len() > SEARCH_MAX_NESTED_SORT_VALUES
                || nested.sort.iter().any(|value| {
                    !matches!(
                        value,
                        serde_json::Value::Null
                            | serde_json::Value::Bool(_)
                            | serde_json::Value::Number(_)
                            | serde_json::Value::String(_)
                    )
                })
            {
                return Err(SearchWireError::InvalidPartial);
            }
            if path.is_some_and(|path| path != nested.path)
                || !identities.insert((nested.path.as_str(), nested.ordinal))
            {
                return Err(SearchWireError::InvalidPartial);
            }
            path = Some(nested.path.as_str());
            let exact = nested.exact_source.as_deref().ok_or(SearchWireError::InvalidPartial)?;
            if exact.len() > SEARCH_MAX_NESTED_SOURCE_BYTES {
                return Err(SearchWireError::InvalidPartial);
            }
            let exact_value: serde_json::Value =
                serde_json::from_str(exact).map_err(|_| SearchWireError::InvalidPartial)?;
            if !exact_value.is_object() || exact_value != nested.source {
                return Err(SearchWireError::InvalidPartial);
            }
            *nested_hits = nested_hits.checked_add(1).ok_or(SearchWireError::InvalidPartial)?;
            if *nested_hits > SEARCH_MAX_NESTED_HITS {
                return Err(SearchWireError::InvalidPartial);
            }
            let sort_bytes = serde_json::to_vec(&nested.sort).map_err(|_| SearchWireError::InvalidPartial)?;
            *nested_bytes = nested_bytes
                .checked_add(exact.len())
                .and_then(|bytes| bytes.checked_add(nested.path.len()))
                .and_then(|bytes| bytes.checked_add(sort_bytes.len()))
                .and_then(|bytes| bytes.checked_add(96))
                .ok_or(SearchWireError::InvalidPartial)?;
        }
    }
    if *nested_bytes > SEARCH_MAX_NESTED_RESPONSE_BYTES {
        return Err(SearchWireError::InvalidPartial);
    }
    Ok(())
}

fn validate_partial_for_limit(partial: &ShardSearchPartialV1, candidate_limit: usize) -> Result<(), SearchWireError> {
    validate_partial_shape(partial)?;
    if candidate_limit == 0
        || candidate_limit > SEARCH_MAX_HITS
        || partial.hits.len() > candidate_limit
        || partial.has_more_hits && partial.hits.len() != candidate_limit
    {
        return Err(SearchWireError::InvalidPartial);
    }
    Ok(())
}

fn validate_partial_for_route_lineages(
    partial: &ShardSearchPartialV1,
    entry: &EngineApiResolvedRouteEntry,
    shard: EngineApiResolvedShardRoute,
    plan: &EngineApiCanonicalPlan,
    result_lineages: &[ShardId],
) -> Result<(), SearchWireError> {
    if partial.shard_key != shard.shard_key
        || partial.aggregations.len() != plan.aggregations.len()
        || partial.aggregations.keys().ne(plan.aggregations.keys())
        || result_lineages.is_empty()
        || result_lineages.windows(2).any(|pair| pair[0] >= pair[1])
        || partial.hits.iter().any(|hit| {
            hit.collection_index_ref != entry.collection_index_ref
                || hit.index != entry.index
                || result_lineages.binary_search(&hit.shard_id).is_err()
                || !hit.score.is_finite()
                || hit.score < 0.0
                || hit.score == 0.0 && hit.score.is_sign_negative()
                || hit.sort.len() != plan.sort.len()
        })
    {
        return Err(SearchWireError::InvalidPartial);
    }
    validate_inner_hits_for_plan(&partial.hits, &plan.query)?;
    Ok(())
}

fn validate_inner_hits_for_plan(hits: &[EngineApiScopedSearchHit], query: &Query) -> Result<(), SearchWireError> {
    let mut inner_hit_specs = BTreeMap::new();
    collect_inner_hit_specs(query, &mut inner_hit_specs)?;
    for hit in hits {
        for (name, result) in &hit.inner_hits {
            let Some((path, sort_values)) = inner_hit_specs.get(name) else {
                return Err(SearchWireError::InvalidPartial);
            };
            if result
                .hits
                .iter()
                .any(|nested| nested.path != *path || nested.sort.len() != *sort_values)
            {
                return Err(SearchWireError::InvalidPartial);
            }
        }
    }
    Ok(())
}

fn collect_inner_hit_specs(
    query: &Query,
    output: &mut BTreeMap<String, (String, usize)>,
) -> Result<(), SearchWireError> {
    match query {
        Query::Nested {
            path,
            query,
            inner_hits,
            ..
        } => {
            if let Some(inner_hits) = inner_hits {
                let name = inner_hits.name.clone().unwrap_or_else(|| path.clone());
                if output.insert(name, (path.clone(), inner_hits.sort.len())).is_some() {
                    return Err(SearchWireError::InvalidPartial);
                }
            }
            collect_inner_hit_specs(query, output)
        }
        Query::Boost { query, .. } => collect_inner_hit_specs(query, output),
        Query::Bool {
            must,
            should,
            filter,
            must_not,
            ..
        } => {
            for child in must.iter().chain(should).chain(filter).chain(must_not) {
                collect_inner_hit_specs(child, output)?;
            }
            Ok(())
        }
        _ => Ok(()),
    }
}

fn validate_aggregation(
    partial: &AggregationPartialV1,
    depth: usize,
    entries: &mut usize,
) -> Result<(), SearchWireError> {
    if depth >= SEARCH_MAX_AGGREGATION_DEPTH {
        return Err(SearchWireError::AggregationBudget);
    }
    *entries = entries.checked_add(1).ok_or(SearchWireError::AggregationBudget)?;
    if *entries > SEARCH_MAX_AGGREGATION_ENTRIES {
        return Err(SearchWireError::AggregationBudget);
    }
    match partial {
        AggregationPartialV1::Buckets { buckets } => {
            let mut previous: Option<&AggregationScalarV1> = None;
            for bucket in buckets {
                validate_scalar(&bucket.key)?;
                if previous.is_some_and(|key| compare_scalars(key, &bucket.key).is_ge()) {
                    return Err(SearchWireError::NonCanonicalAggregation);
                }
                previous = Some(&bucket.key);
                for child in bucket.aggregations.values() {
                    validate_aggregation(child, depth + 1, entries)?;
                }
            }
        }
        AggregationPartialV1::Stats { min, max, sum, .. } => {
            validate_optional_number(*min)?;
            validate_optional_number(*max)?;
            validate_number(*sum)?;
        }
        AggregationPartialV1::Min { value } | AggregationPartialV1::Max { value } => {
            validate_optional_number(*value)?;
        }
        AggregationPartialV1::Avg { sum, .. } | AggregationPartialV1::Sum { sum } => validate_number(*sum)?,
        AggregationPartialV1::ValueCount { .. } => {}
        AggregationPartialV1::Cardinality { values } => {
            for value in values {
                validate_scalar(value)?;
            }
            if values
                .windows(2)
                .any(|pair| compare_scalars(&pair[0], &pair[1]).is_ge())
            {
                return Err(SearchWireError::NonCanonicalAggregation);
            }
        }
        AggregationPartialV1::Percentiles { samples } => {
            for sample in samples {
                validate_number(*sample)?;
            }
        }
        AggregationPartialV1::ApproximateCardinality {
            format_version,
            precision,
            registers,
        } => {
            if *format_version != SEARCH_APPROXIMATE_SKETCH_FORMAT_VERSION
                || !(SEARCH_MIN_HLL_PRECISION..=SEARCH_MAX_HLL_PRECISION).contains(precision)
                || registers.len() != 1_usize << *precision
                || registers
                    .iter()
                    .any(|register| *register > 64_u8.saturating_sub(*precision).saturating_add(1))
            {
                return Err(SearchWireError::InvalidPartial);
            }
        }
        AggregationPartialV1::ApproximatePercentiles {
            format_version,
            precision,
            count,
            counts,
            minima,
            maxima,
        } => {
            let expected = 1_usize << *precision;
            if *format_version != SEARCH_APPROXIMATE_SKETCH_FORMAT_VERSION
                || !(SEARCH_MIN_PERCENTILE_PRECISION..=SEARCH_MAX_PERCENTILE_PRECISION).contains(precision)
                || counts.len() != expected
                || minima.len() != expected
                || maxima.len() != expected
            {
                return Err(SearchWireError::InvalidPartial);
            }
            let observed = counts.iter().try_fold(0_u64, |total, value| total.checked_add(*value));
            if observed != Some(*count)
                || counts.iter().enumerate().any(|(index, bucket_count)| {
                    if *bucket_count == 0 {
                        minima[index] != u64::MAX || maxima[index] != 0
                    } else {
                        minima[index] > maxima[index]
                            || (minima[index] >> (64 - *precision)) as usize != index
                            || (maxima[index] >> (64 - *precision)) as usize != index
                    }
                })
            {
                return Err(SearchWireError::InvalidPartial);
            }
        }
    }
    Ok(())
}

fn scalar_rank(value: &AggregationScalarV1) -> u8 {
    match value {
        AggregationScalarV1::Bool(_) => 0,
        AggregationScalarV1::I64(_) => 1,
        AggregationScalarV1::U64(_) => 2,
        AggregationScalarV1::F64(_) => 3,
        AggregationScalarV1::String(_) => 4,
        AggregationScalarV1::HistogramIndex(_) => 5,
        AggregationScalarV1::Range { .. } => 6,
        AggregationScalarV1::Tuple(_) => 7,
    }
}

fn compare_scalars(left: &AggregationScalarV1, right: &AggregationScalarV1) -> std::cmp::Ordering {
    scalar_rank(left)
        .cmp(&scalar_rank(right))
        .then_with(|| match (left, right) {
            (AggregationScalarV1::Bool(left), AggregationScalarV1::Bool(right)) => left.cmp(right),
            (AggregationScalarV1::I64(left), AggregationScalarV1::I64(right)) => left.cmp(right),
            (AggregationScalarV1::U64(left), AggregationScalarV1::U64(right)) => left.cmp(right),
            (AggregationScalarV1::F64(left), AggregationScalarV1::F64(right)) => left.total_cmp(right),
            (AggregationScalarV1::String(left), AggregationScalarV1::String(right)) => left.cmp(right),
            (AggregationScalarV1::HistogramIndex(left), AggregationScalarV1::HistogramIndex(right)) => left.cmp(right),
            (
                AggregationScalarV1::Range {
                    ordinal: left_ordinal,
                    key: left_key,
                },
                AggregationScalarV1::Range {
                    ordinal: right_ordinal,
                    key: right_key,
                },
            ) => left_ordinal.cmp(right_ordinal).then_with(|| left_key.cmp(right_key)),
            (AggregationScalarV1::Tuple(left), AggregationScalarV1::Tuple(right)) => compare_scalar_slices(left, right),
            _ => std::cmp::Ordering::Equal,
        })
}

fn compare_scalar_slices(left: &[AggregationScalarV1], right: &[AggregationScalarV1]) -> std::cmp::Ordering {
    left.iter()
        .zip(right)
        .find_map(|(left, right)| {
            let ordering = compare_scalars(left, right);
            (ordering != std::cmp::Ordering::Equal).then_some(ordering)
        })
        .unwrap_or_else(|| left.len().cmp(&right.len()))
}

fn validate_scalar(value: &AggregationScalarV1) -> Result<(), SearchWireError> {
    match value {
        AggregationScalarV1::F64(number) => validate_number(*number)?,
        AggregationScalarV1::Tuple(values) => {
            if values.is_empty() || values.len() > crate::MAX_COMPOSITE_SOURCES {
                return Err(SearchWireError::AggregationBudget);
            }
            for value in values {
                if matches!(value, AggregationScalarV1::Tuple(_)) {
                    return Err(SearchWireError::InvalidPartial);
                }
                validate_scalar(value)?;
            }
        }
        _ => {}
    }
    Ok(())
}

fn validate_optional_number(value: Option<f64>) -> Result<(), SearchWireError> {
    value.map_or(Ok(()), validate_number)
}

fn validate_number(value: f64) -> Result<(), SearchWireError> {
    if !value.is_finite() || value == 0.0 && value.is_sign_negative() {
        return Err(SearchWireError::NonCanonicalNumber);
    }
    Ok(())
}

fn check_version(version: u16) -> Result<(), SearchWireError> {
    if version != SEARCH_WIRE_FORMAT_VERSION {
        return Err(SearchWireError::UnsupportedVersion(version));
    }
    Ok(())
}

macro_rules! codec {
    ($encode:ident, $decode:ident, $ty:ty, $limit:expr, $validate:expr) => {
        pub fn $encode(value: &$ty) -> Result<Vec<u8>, SearchWireError> {
            ($validate)(value)?;
            encode_bounded(value, $limit)
        }

        pub fn $decode(bytes: &[u8]) -> Result<$ty, SearchWireError> {
            let value: $ty = decode_bounded(bytes, $limit)?;
            ($validate)(&value)?;
            Ok(value)
        }
    };
}

codec!(
    encode_search_coordinate_request,
    decode_search_coordinate_request,
    SearchCoordinateRequestV1,
    SEARCH_MAX_REQUEST_PAYLOAD_BYTES,
    SearchCoordinateRequestV1::validate
);
codec!(
    encode_search_coordinate_response,
    decode_search_coordinate_response,
    SearchCoordinateResponseV1,
    SEARCH_MAX_RESPONSE_PAYLOAD_BYTES,
    SearchCoordinateResponseV1::validate
);
pub fn encode_writer_realtime_request(value: &WriterRealtimeRequestV1) -> Result<Vec<u8>, SearchWireError> {
    value.validate()?;
    let inner = encode_bounded(value, SEARCH_MAX_REQUEST_PAYLOAD_BYTES)?;
    encode_realtime_read_request(RealtimeReadKind::SearchSlice, &inner).map_err(map_realtime_request_envelope_error)
}

pub fn decode_writer_realtime_request(bytes: &[u8]) -> Result<WriterRealtimeRequestV1, SearchWireError> {
    let inner = decode_realtime_read_request(RealtimeReadKind::SearchSlice, bytes)
        .map_err(map_realtime_request_envelope_error)?;
    let value: WriterRealtimeRequestV1 = decode_bounded(&inner, SEARCH_MAX_REQUEST_PAYLOAD_BYTES)?;
    value.validate()?;
    Ok(value)
}

pub fn encode_writer_realtime_response(value: &WriterRealtimeResponseV1) -> Result<Vec<u8>, SearchWireError> {
    value.validate()?;
    let inner = encode_bounded(value, SEARCH_MAX_WRITER_RESPONSE_PAYLOAD_BYTES)?;
    encode_realtime_read_response(RealtimeReadKind::SearchSlice, &inner).map_err(map_realtime_response_envelope_error)
}

pub fn decode_writer_realtime_response(bytes: &[u8]) -> Result<WriterRealtimeResponseV1, SearchWireError> {
    let inner = decode_realtime_read_response(RealtimeReadKind::SearchSlice, bytes)
        .map_err(map_realtime_response_envelope_error)?;
    let value: WriterRealtimeResponseV1 = decode_bounded(&inner, SEARCH_MAX_WRITER_RESPONSE_PAYLOAD_BYTES)?;
    value.validate()?;
    Ok(value)
}
codec!(
    encode_search_shard_request,
    decode_search_shard_request,
    SearchShardRequestV1,
    SEARCH_MAX_SHARD_REQUEST_PAYLOAD_BYTES,
    SearchShardRequestV1::validate
);
pub fn encode_search_shard_response(value: &SearchShardResponseV1) -> Result<Vec<u8>, SearchWireError> {
    check_version(value.format_version)?;
    validate_partial_shape(&value.partial)?;
    encode_bounded(value, SEARCH_MAX_RESPONSE_PAYLOAD_BYTES)
}

pub fn decode_search_shard_response(bytes: &[u8]) -> Result<SearchShardResponseV1, SearchWireError> {
    // This authenticated internal response is validated structurally below;
    // requiring its byte-for-byte re-encoding would duplicate every hit body
    // at the coordinator without strengthening the semantic fence.
    let value: SearchShardResponseV1 = decode_bounded_semantic(bytes, SEARCH_MAX_RESPONSE_PAYLOAD_BYTES)?;
    check_version(value.format_version)?;
    validate_partial_shape(&value.partial)?;
    Ok(value)
}

fn encode_bounded<T: Serialize>(value: &T, limit: usize) -> Result<Vec<u8>, SearchWireError> {
    let bytes = serde_json::to_vec(value).map_err(|error| SearchWireError::Codec(error.to_string()))?;
    if bytes.len() > limit {
        return Err(SearchWireError::MessageTooLarge {
            actual: bytes.len(),
            maximum: limit,
        });
    }
    Ok(bytes)
}

fn decode_bounded<T>(bytes: &[u8], limit: usize) -> Result<T, SearchWireError>
where
    T: for<'de> Deserialize<'de> + Serialize,
{
    if bytes.len() > limit {
        return Err(SearchWireError::MessageTooLarge {
            actual: bytes.len(),
            maximum: limit,
        });
    }
    let value = serde_json::from_slice(bytes).map_err(|error| SearchWireError::Codec(error.to_string()))?;
    let canonical = serde_json::to_vec(&value).map_err(|error| SearchWireError::Codec(error.to_string()))?;
    if canonical != bytes {
        return Err(SearchWireError::NonCanonical);
    }
    Ok(value)
}

fn decode_bounded_semantic<T>(bytes: &[u8], limit: usize) -> Result<T, SearchWireError>
where
    T: for<'de> Deserialize<'de>,
{
    if bytes.len() > limit {
        return Err(SearchWireError::MessageTooLarge {
            actual: bytes.len(),
            maximum: limit,
        });
    }
    serde_json::from_slice(bytes).map_err(|error| SearchWireError::Codec(error.to_string()))
}

fn map_realtime_request_envelope_error(error: RealtimeReadEnvelopeError) -> SearchWireError {
    map_realtime_envelope_error(error, SearchWireError::InvalidWriterRequest)
}

fn map_realtime_response_envelope_error(error: RealtimeReadEnvelopeError) -> SearchWireError {
    map_realtime_envelope_error(error, SearchWireError::InvalidWriterResponse)
}

fn map_realtime_envelope_error(error: RealtimeReadEnvelopeError, invalid_shape: SearchWireError) -> SearchWireError {
    match error {
        RealtimeReadEnvelopeError::MessageTooLarge { actual, maximum } => {
            SearchWireError::MessageTooLarge { actual, maximum }
        }
        RealtimeReadEnvelopeError::NonCanonical => SearchWireError::NonCanonical,
        RealtimeReadEnvelopeError::UnsupportedVersion(version) => SearchWireError::UnsupportedVersion(version),
        RealtimeReadEnvelopeError::Malformed | RealtimeReadEnvelopeError::UnexpectedKind { .. } => invalid_shape,
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SearchWireError {
    AggregationBudget,
    Codec(String),
    InvalidCoordinate,
    InvalidCoordinateResponse,
    InvalidManifestBinding,
    InvalidPartial,
    InvalidPitGuard,
    InvalidRoute,
    InvalidShardRequest,
    InvalidShardResponse,
    InvalidTargetVector,
    InvalidWriterRequest,
    InvalidWriterResponse,
    MessageTooLarge { actual: usize, maximum: usize },
    NonCanonical,
    NonCanonicalAggregation,
    NonCanonicalNumber,
    UnsupportedVersion(u16),
}

impl Display for SearchWireError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{self:?}")
    }
}

impl Error for SearchWireError {}

#[cfg(test)]
mod tests {
    use std::error::Error;
    use std::io;

    use lodestone_types::{
        AdmissionGeneration, BackendAuthorityId, CollectionId, DocumentId, Epoch, IndexId, IndexName,
        ManifestGeneration, NodeId, NodeIncarnation, ObjectIdentity, ObjectIncarnation, ObjectKey,
        PINNED_MANIFEST_BINDING_FORMAT_VERSION, PhysicalObjectId, PinnedManifestDecodeProvenanceV1, PrimaryTerm,
        ROOT_SET_REF_FORMAT_VERSION, RecoveryEra, RecoveryGeneration, RootSetRef, RoutingRevision, ShardId,
    };

    use super::*;

    fn writer_response(identity_hash: IdHash) -> Result<WriterRealtimeResponseV1, Box<dyn Error>> {
        let recovery_era = RecoveryEra::new(1)?;
        let epoch = Epoch::new(1);
        let fence = FenceToken::new(recovery_era, epoch)?;
        let shard_key = ShardKey::new(
            lodestone_types::CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(1)?),
            ShardId::new(0),
        );
        let backend_authority = BackendAuthorityId::new("p62b-wire-test")?;
        let manifest_binding = PinnedManifestBindingV1 {
            format_version: PINNED_MANIFEST_BINDING_FORMAT_VERSION,
            shard_key,
            admission_generation: AdmissionGeneration::new(1),
            routing_revision: RoutingRevision::new(1),
            data_backend_authority: backend_authority.clone(),
            generation: ManifestGeneration::new(1),
            fence,
            primary_term: PrimaryTerm::from_parts(RecoveryGeneration::genesis(), epoch)?,
            manifest: ObjectIdentity {
                physical: PhysicalObjectId {
                    backend_authority: backend_authority.clone(),
                    object_incarnation: ObjectIncarnation::new(1)?,
                    object_key: ObjectKey::new("p62b/wire/manifest")?,
                },
                content_checksum: [1; 32],
            },
            root_set: RootSetRef {
                format_version: ROOT_SET_REF_FORMAT_VERSION,
                root: PhysicalObjectId {
                    backend_authority,
                    object_incarnation: ObjectIncarnation::new(2)?,
                    object_key: ObjectKey::new("p62b/wire/root")?,
                },
                content_digest: [2; 32],
                item_count: 1,
                page_count: 1,
                schema_generations: Vec::new(),
            },
            committed_through: SeqNo::new(1),
            decode_provenance: PinnedManifestDecodeProvenanceV1::ScopedManifestV3,
        };
        Ok(WriterRealtimeResponseV1 {
            format_version: SEARCH_WIRE_FORMAT_VERSION,
            target: WriterTargetV1 {
                node_token: NodeToken::new(recovery_era, NodeId::new(1), NodeIncarnation::new(1)?)?,
                fence,
            },
            manifest_binding,
            cut: SeqNo::new(2),
            tombstones: vec![WireTombstoneV1 {
                identity_hash,
                kill_below_seq: SeqNo::new(2),
            }],
            fresh_partial: ShardSearchPartialV1 {
                format_version: SEARCH_WIRE_FORMAT_VERSION,
                shard_key,
                total: 0,
                hits: Vec::new(),
                aggregations: BTreeMap::new(),
                freshness: SearchExecutionFreshness::Realtime,
                freshness_degraded: false,
                has_more_hits: false,
                applied_continuation: None,
            },
        })
    }

    fn replace_once(bytes: &[u8], from: &str, to: &str) -> Result<Vec<u8>, Box<dyn Error>> {
        let encoded = std::str::from_utf8(bytes)?;
        if encoded.matches(from).count() != 1 {
            return Err(io::Error::other("writer-response mutation target was not unique").into());
        }
        Ok(encoded.replacen(from, to, 1).into_bytes())
    }

    #[test]
    fn writer_response_identity_hash_is_full_domain_canonical_decimal() -> Result<(), Box<dyn Error>> {
        let identity_hash = IdHash::new(u128::MAX);
        let response = writer_response(identity_hash)?;
        let encoded = encode_writer_realtime_response(&response)?;
        let canonical_field = format!("\"identity_hash\":\"{}\"", identity_hash.get());
        assert!(std::str::from_utf8(&encoded)?.contains(&canonical_field));
        assert_eq!(decode_writer_realtime_response(&encoded)?, response);

        let zero_response = writer_response(IdHash::new(0))?;
        let zero_encoded = encode_writer_realtime_response(&zero_response)?;
        assert!(std::str::from_utf8(&zero_encoded)?.contains("\"identity_hash\":\"0\""));
        assert_eq!(decode_writer_realtime_response(&zero_encoded)?, zero_response);

        let numeric_field = format!("\"identity_hash\":{}", identity_hash.get());
        let numeric = replace_once(&encoded, &canonical_field, &numeric_field)?;
        assert!(matches!(
            decode_writer_realtime_response(&numeric),
            Err(SearchWireError::Codec(_))
        ));

        let leading_zero_field = format!("\"identity_hash\":\"0{}\"", identity_hash.get());
        let leading_zero = replace_once(&encoded, &canonical_field, &leading_zero_field)?;
        assert!(matches!(
            decode_writer_realtime_response(&leading_zero),
            Err(SearchWireError::Codec(_))
        ));

        let overflow_field = "\"identity_hash\":\"340282366920938463463374607431768211456\"";
        let overflow = replace_once(&encoded, &canonical_field, overflow_field)?;
        assert!(matches!(
            decode_writer_realtime_response(&overflow),
            Err(SearchWireError::Codec(_))
        ));

        let unknown = replace_once(
            &encoded,
            "\"kill_below_seq\":2}",
            "\"kill_below_seq\":2,\"unknown\":true}",
        )?;
        assert!(matches!(
            decode_writer_realtime_response(&unknown),
            Err(SearchWireError::Codec(_))
        ));

        let substituted = replace_once(&encoded, "\"identity_hash\"", "\"document_hash\"")?;
        assert!(matches!(
            decode_writer_realtime_response(&substituted),
            Err(SearchWireError::Codec(_))
        ));
        Ok(())
    }

    #[test]
    fn ec6_nested_hit_wire_uses_exact_json_text_and_current_version() -> Result<(), Box<dyn Error>> {
        let mut response = writer_response(IdHash::new(7))?;
        response.fresh_partial.total = 1;
        response.fresh_partial.hits.push(EngineApiScopedSearchHit {
            collection_index_ref: response.fresh_partial.shard_key.collection_index_ref(),
            index: IndexName::new("events")?,
            id: DocumentId::new("doc-1")?,
            shard_id: response.fresh_partial.shard_key.shard_id(),
            seq_no: SeqNo::new(2),
            primary_term: response.manifest_binding.primary_term,
            routing: None,
            score: 1.0,
            source: serde_json::json!({"state": "new", "value": 2}),
            exact_source: None,
            sort: Vec::new(),
            highlight: BTreeMap::new(),
            fragments: Vec::new(),
            features: Vec::new(),
            group_values: Vec::new(),
            inner_hits: BTreeMap::from([(
                "matching".to_owned(),
                crate::EngineApiNestedInnerHitsResult {
                    total: 1,
                    hits: vec![crate::EngineApiNestedSearchHit {
                        path: "items".to_owned(),
                        ordinal: 0,
                        score: 0.5,
                        source: serde_json::json!({"state":"nested"}),
                        exact_source: Some(r#"{"state":"nested"}"#.to_owned()),
                        sort: vec![serde_json::json!(1)],
                    }],
                },
            )]),
        });
        let planned_query = Query::Nested {
            path: "items".to_owned(),
            query: Box::new(Query::MatchAll),
            score_mode: crate::NestedScoreMode::Avg,
            ignore_unmapped: false,
            inner_hits: Some(crate::NestedInnerHits {
                name: Some("matching".to_owned()),
                from: 0,
                size: 3,
                sort: vec![crate::SortSpec::Score {
                    order: crate::SortOrder::Desc,
                }],
            }),
        };
        assert!(validate_inner_hits_for_plan(&response.fresh_partial.hits, &planned_query).is_ok());
        for mutation in ["name", "path", "sort"] {
            let mut malformed = response.fresh_partial.hits.clone();
            let hit = malformed.first_mut().ok_or("nested wire hit disappeared")?;
            match mutation {
                "name" => {
                    let result = hit.inner_hits.remove("matching").ok_or("nested result disappeared")?;
                    hit.inner_hits.insert("unrequested".to_owned(), result);
                }
                "path" => {
                    hit.inner_hits
                        .get_mut("matching")
                        .and_then(|result| result.hits.first_mut())
                        .ok_or("nested result disappeared")?
                        .path = "other".to_owned();
                }
                "sort" => {
                    hit.inner_hits
                        .get_mut("matching")
                        .and_then(|result| result.hits.first_mut())
                        .ok_or("nested result disappeared")?
                        .sort
                        .clear();
                }
                _ => unreachable!("fixed test mutation"),
            }
            assert!(matches!(
                validate_inner_hits_for_plan(&malformed, &planned_query),
                Err(SearchWireError::InvalidPartial)
            ));
        }
        let encoded = encode_writer_realtime_response(&response)?;
        assert!(std::str::from_utf8(&encoded)?.contains("\"score\":1.0"));
        assert!(std::str::from_utf8(&encoded)?.contains(r#""exact_source":"{\"state\":\"nested\"}""#));
        assert_eq!(decode_writer_realtime_response(&encoded)?, response);

        let mut older_response = response.clone();
        older_response.format_version = 3;
        let older =
            encode_realtime_read_response(RealtimeReadKind::SearchSlice, &serde_json::to_vec(&older_response)?)?;
        assert!(matches!(
            decode_writer_realtime_response(&older),
            Err(SearchWireError::UnsupportedVersion(3)) | Err(SearchWireError::InvalidWriterResponse)
        ));

        let mut missing: serde_json::Value = serde_json::from_slice(&encoded)?;
        missing
            .get_mut("payload")
            .and_then(|payload| payload.get_mut("fresh_partial"))
            .and_then(|partial| partial.get_mut("hits"))
            .and_then(serde_json::Value::as_array_mut)
            .and_then(|hits| hits.first_mut())
            .and_then(serde_json::Value::as_object_mut)
            .ok_or("encoded hit shape changed")?
            .remove("inner_hits");
        assert!(matches!(
            decode_writer_realtime_response(&serde_json::to_vec(&missing)?),
            Err(SearchWireError::Codec(_))
        ));

        let malformed = replace_once(
            &encoded,
            "\"score\":1.0",
            "\"score\":{\"$serde_json::private::Number\":\"1.0\"}",
        )?;
        assert!(matches!(
            decode_writer_realtime_response(&malformed),
            Err(SearchWireError::Codec(_))
        ));
        Ok(())
    }

    #[test]
    fn ec6_nested_hit_near_semantic_cap_fits_current_wire() -> Result<(), Box<dyn Error>> {
        let mut response = writer_response(IdHash::new(7))?;
        response.fresh_partial.total = 1;
        let payload = "x".repeat(880);
        let exact_source = format!(r#"{{"payload":"{payload}"}}"#);
        let source: serde_json::Value = serde_json::from_str(&exact_source)?;
        let inner_hits = (0..64)
            .map(|group| {
                let hits = (0..64)
                    .map(|ordinal| crate::EngineApiNestedSearchHit {
                        path: "items".to_owned(),
                        ordinal,
                        score: 0.5,
                        source: source.clone(),
                        exact_source: Some(exact_source.clone()),
                        sort: Vec::new(),
                    })
                    .collect::<Vec<_>>();
                (
                    format!("group-{group}"),
                    crate::EngineApiNestedInnerHitsResult { total: 64, hits },
                )
            })
            .collect();
        response.fresh_partial.hits.push(EngineApiScopedSearchHit {
            collection_index_ref: response.fresh_partial.shard_key.collection_index_ref(),
            index: IndexName::new("events")?,
            id: DocumentId::new("doc-near-cap")?,
            shard_id: response.fresh_partial.shard_key.shard_id(),
            seq_no: SeqNo::new(2),
            primary_term: response.manifest_binding.primary_term,
            routing: None,
            score: 1.0,
            source: serde_json::json!({}),
            exact_source: None,
            sort: Vec::new(),
            highlight: BTreeMap::new(),
            fragments: Vec::new(),
            features: Vec::new(),
            group_values: Vec::new(),
            inner_hits,
        });
        let encoded = encode_writer_realtime_response(&response)?;
        assert!(encoded.len() < SEARCH_MAX_WRITER_RESPONSE_PAYLOAD_BYTES);
        assert_eq!(decode_writer_realtime_response(&encoded)?, response);
        Ok(())
    }

    #[test]
    fn writer_response_codec_refuses_duplicate_tombstone_identities() -> Result<(), Box<dyn Error>> {
        let response = writer_response(IdHash::new(7))?;
        let encoded = encode_writer_realtime_response(&response)?;
        let duplicate = replace_once(
            &encoded,
            "\"tombstones\":[{\"identity_hash\":\"7\",\"kill_below_seq\":2}]",
            "\"tombstones\":[{\"identity_hash\":\"7\",\"kill_below_seq\":1},{\"identity_hash\":\"7\",\"kill_below_seq\":2}]",
        )?;
        assert!(matches!(
            decode_writer_realtime_response(&duplicate),
            Err(SearchWireError::InvalidWriterResponse)
        ));
        Ok(())
    }

    #[test]
    fn approximate_partial_wire_round_trips_and_refuses_newer_or_malformed_state() -> Result<(), Box<dyn Error>> {
        let mut response = writer_response(IdHash::new(7))?;
        response.fresh_partial.aggregations = BTreeMap::from([
            (
                "distinct".to_owned(),
                AggregationPartialV1::ApproximateCardinality {
                    format_version: SEARCH_APPROXIMATE_SKETCH_FORMAT_VERSION,
                    precision: 4,
                    registers: vec![0; 16],
                },
            ),
            (
                "latency".to_owned(),
                AggregationPartialV1::ApproximatePercentiles {
                    format_version: SEARCH_APPROXIMATE_SKETCH_FORMAT_VERSION,
                    precision: 4,
                    count: 0,
                    counts: vec![0; 16],
                    minima: vec![u64::MAX; 16],
                    maxima: vec![0; 16],
                },
            ),
        ]);
        let encoded = encode_writer_realtime_response(&response)?;
        assert_eq!(decode_writer_realtime_response(&encoded)?, response);

        let mut newer = response.clone();
        let Some(AggregationPartialV1::ApproximateCardinality { format_version, .. }) =
            newer.fresh_partial.aggregations.get_mut("distinct")
        else {
            return Err("approximate cardinality version disappeared".into());
        };
        *format_version = 2;
        let newer = encode_realtime_read_response(RealtimeReadKind::SearchSlice, &serde_json::to_vec(&newer)?)?;
        assert!(matches!(
            decode_writer_realtime_response(&newer),
            Err(SearchWireError::InvalidPartial)
        ));

        let mut malformed = response.clone();
        let Some(AggregationPartialV1::ApproximatePercentiles { counts, .. }) =
            malformed.fresh_partial.aggregations.get_mut("latency")
        else {
            return Err("approximate percentile counts disappeared".into());
        };
        counts.pop();
        let malformed = encode_realtime_read_response(RealtimeReadKind::SearchSlice, &serde_json::to_vec(&malformed)?)?;
        assert!(matches!(
            decode_writer_realtime_response(&malformed),
            Err(SearchWireError::InvalidPartial)
        ));
        Ok(())
    }

    #[test]
    fn composite_tuple_partial_round_trips_and_refuses_noncurrent_or_nested_keys() -> Result<(), Box<dyn Error>> {
        let mut response = writer_response(IdHash::new(9))?;
        response.fresh_partial.aggregations.insert(
            "page".to_owned(),
            AggregationPartialV1::Buckets {
                buckets: vec![AggregationBucketPartialV1 {
                    key: AggregationScalarV1::Tuple(vec![
                        AggregationScalarV1::String("ap-south".to_owned()),
                        AggregationScalarV1::I64(1_700_000_000_000),
                    ]),
                    doc_count: 2,
                    aggregations: BTreeMap::new(),
                }],
            },
        );
        let encoded = encode_writer_realtime_response(&response)?;
        assert_eq!(decode_writer_realtime_response(&encoded)?, response);

        let mut predecessor = response.clone();
        predecessor.format_version = SEARCH_WIRE_FORMAT_VERSION.saturating_sub(1);
        let predecessor =
            encode_realtime_read_response(RealtimeReadKind::SearchSlice, &serde_json::to_vec(&predecessor)?)?;
        assert!(matches!(
            decode_writer_realtime_response(&predecessor),
            Err(SearchWireError::UnsupportedVersion(version)) if version == SEARCH_WIRE_FORMAT_VERSION - 1
        ));

        for invalid_key in [
            AggregationScalarV1::Tuple(Vec::new()),
            AggregationScalarV1::Tuple(vec![AggregationScalarV1::Tuple(vec![AggregationScalarV1::I64(1)])]),
            AggregationScalarV1::Tuple(
                (0..=crate::MAX_COMPOSITE_SOURCES)
                    .map(|value| AggregationScalarV1::U64(value as u64))
                    .collect(),
            ),
        ] {
            let mut malformed = response.clone();
            let Some(AggregationPartialV1::Buckets { buckets }) = malformed.fresh_partial.aggregations.get_mut("page")
            else {
                return Err("composite partial disappeared".into());
            };
            buckets[0].key = invalid_key;
            let malformed =
                encode_realtime_read_response(RealtimeReadKind::SearchSlice, &serde_json::to_vec(&malformed)?)?;
            assert!(matches!(
                decode_writer_realtime_response(&malformed),
                Err(SearchWireError::AggregationBudget | SearchWireError::InvalidPartial)
            ));
        }
        Ok(())
    }
}