lodestone-proto 0.0.16

//! Current transport-neutral scoped data-plane contracts shared by authority adapters.
//!
//! These types carry authority-native data, routing, query, and PIT semantics.
//! They do not define a public listener, command framing, dispatcher, or fallback.

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

use lodestone_types::{
    AdmissionGeneration, AssignmentGeneration, ClusterRevision, CollectionIndexRef, DocumentId, DurabilityBoundary,
    DurabilityClass, IdempotencyKey, IndexId, IndexName, IndexRevision, LogicalMillis, Mapping, PrimaryTerm,
    RecoveryEra, RoutingHash, RoutingRangeEnd, RoutingRevision, SchemaGeneration, ScopedPitId, SeqNo, ShardId,
    ShardKey, VisibilityMode, WriteStage, is_canonical_field_path,
};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_json::value::RawValue;

use crate::{Aggregation, MutationResult, Query, RequestId, SearchExecutionFreshness, SearchFreshness, SortSpec};

pub const SCOPED_CATALOG_FORMAT_VERSION: u16 = 1;
pub const SCOPED_DOCUMENT_FORMAT_VERSION: u16 = 2;
pub const SCOPED_SEARCH_FORMAT_VERSION: u16 = 5;
pub const SCOPED_PIT_FORMAT_VERSION: u16 = 2;
pub const MAX_ENGINE_API_INDEXES: usize = 128;
pub const MAX_ENGINE_API_ROUTING_VALUES: usize = 1_024;
pub const MAX_ENGINE_API_INDEX_EXPRESSION_BYTES: usize = 4_096;
pub const MAX_ENGINE_API_SEARCH_WINDOW: usize = 10_000;
const MAX_ENGINE_API_ROUTING_VALUE_BYTES: usize = 4_096;
const MAX_ENGINE_API_SORT_VALUES: usize = 16;
const MAX_ENGINE_API_AGGREGATIONS: usize = 256;
#[derive(Debug)]
pub struct EngineApiCreateIndexRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub mapping: Mapping,
}

#[derive(Debug)]
pub struct EngineApiUpdateMappingRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: IndexId,
    pub expected_revision: IndexRevision,
    pub mapping: Mapping,
}

#[derive(Debug)]
pub struct EngineApiDeleteIndexRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: IndexId,
    pub expected_revision: IndexRevision,
}

#[derive(Debug)]
pub struct EngineApiPutRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: Option<IndexId>,
    pub id: DocumentId,
    pub routing: Option<String>,
    pub source: Box<RawValue>,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum EngineApiIndexSelector {
    Exact(IndexName),
    ExactScope { name: IndexName, index_id: IndexId },
    Expression(String),
}

#[derive(Deserialize, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
enum TaggedIndexSelector {
    Exact(IndexName),
    ExactScope { name: IndexName, index_id: IndexId },
    Expression(String),
}

impl Serialize for EngineApiIndexSelector {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Exact(index) => index.serialize(serializer),
            Self::ExactScope { name, index_id } => TaggedIndexSelector::ExactScope {
                name: name.clone(),
                index_id: *index_id,
            }
            .serialize(serializer),
            Self::Expression(expression) => TaggedIndexSelector::Expression(expression.clone()).serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for EngineApiIndexSelector {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum Repr {
            Scalar(IndexName),
            Tagged(TaggedIndexSelector),
        }
        Ok(match Repr::deserialize(deserializer)? {
            Repr::Scalar(index) | Repr::Tagged(TaggedIndexSelector::Exact(index)) => Self::Exact(index),
            Repr::Tagged(TaggedIndexSelector::ExactScope { name, index_id }) => Self::ExactScope { name, index_id },
            Repr::Tagged(TaggedIndexSelector::Expression(expression)) => Self::Expression(expression),
        })
    }
}

#[derive(Debug)]
pub struct EngineApiIngestRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: Option<IndexId>,
    pub documents: Vec<EngineApiIngestDocument>,
    pub visibility: VisibilityMode,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiIngestDocument {
    pub id: DocumentId,
    #[serde(default)]
    pub routing: Option<String>,
    pub source: Box<RawValue>,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum EngineApiMutationCondition {
    #[default]
    Any,
    IfSeqNo {
        seq_no: SeqNo,
    },
    IfSeqNoAndPrimaryTerm {
        seq_no: SeqNo,
        primary_term: PrimaryTerm,
    },
    MustNotExist,
}

#[derive(Clone, Debug, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum EngineApiMutationOperation {
    Put { source: Box<RawValue> },
    Update { patch: Box<RawValue> },
    Delete,
}

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

        impl<'de> serde::de::Visitor<'de> for OperationVisitor {
            type Value = EngineApiMutationOperation;

            fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                formatter.write_str("a strict engine-api mutation operation object")
            }

            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 source = None;
                let mut patch = None;
                while let Some(field) = map.next_key::<String>()? {
                    match field.as_str() {
                        "kind" if kind.is_none() => kind = Some(map.next_value::<String>()?),
                        "source" if source.is_none() => source = Some(map.next_value::<Box<RawValue>>()?),
                        "patch" if patch.is_none() => patch = Some(map.next_value::<Box<RawValue>>()?),
                        "kind" => return Err(serde::de::Error::duplicate_field("kind")),
                        "source" => return Err(serde::de::Error::duplicate_field("source")),
                        "patch" => return Err(serde::de::Error::duplicate_field("patch")),
                        _ => return Err(serde::de::Error::unknown_field(&field, &["kind", "source", "patch"])),
                    }
                }
                match (kind.as_deref(), source, patch) {
                    (Some("put"), Some(source), None) => Ok(EngineApiMutationOperation::Put { source }),
                    (Some("update"), None, Some(patch)) => Ok(EngineApiMutationOperation::Update { patch }),
                    (Some("delete"), None, None) => Ok(EngineApiMutationOperation::Delete),
                    (None, _, _) => Err(serde::de::Error::missing_field("kind")),
                    (Some("put"), None, _) => Err(serde::de::Error::missing_field("source")),
                    (Some("update"), _, None) => Err(serde::de::Error::missing_field("patch")),
                    (Some("delete"), _, _) => Err(serde::de::Error::custom(
                        "delete operation cannot contain source or patch",
                    )),
                    (Some(other), _, _) => Err(serde::de::Error::unknown_variant(other, &["put", "update", "delete"])),
                }
            }
        }

        deserializer.deserialize_map(OperationVisitor)
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiMutationItem {
    pub id: DocumentId,
    #[serde(default)]
    pub routing: Option<String>,
    pub operation: EngineApiMutationOperation,
    #[serde(default)]
    pub condition: EngineApiMutationCondition,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub idempotency_key: Option<IdempotencyKey>,
}

#[derive(Clone, Debug)]
pub struct EngineApiMutateRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: Option<IndexId>,
    pub mutations: Vec<EngineApiMutationItem>,
    pub visibility: VisibilityMode,
}

#[derive(Debug)]
pub struct EngineApiGetRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub index: IndexName,
    pub expected_index_id: Option<IndexId>,
    pub id: DocumentId,
    pub routing: Option<String>,
}

#[derive(Debug)]
pub struct EngineApiSearchRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub indexes: Option<Vec<EngineApiIndexSelector>>,
    pub routing: Vec<String>,
    pub query: Query,
    pub retrieval: Option<crate::EngineRetrievalPlan>,
    pub highlight: Option<crate::HighlightSpec>,
    /// Reconstructible lexical query used only for fragment selection. Scalar
    /// search leaves this absent and uses `query`; retrieval derives it from
    /// its admitted lexical branch.
    pub highlight_query: Option<Query>,
    pub result_features: Option<crate::ResultFeatureRequest>,
    pub diversification: Option<crate::ResultDiversification>,
    pub aggregations: BTreeMap<String, Aggregation>,
    pub from: usize,
    pub limit: usize,
    pub sort: Vec<SortSpec>,
    pub freshness: SearchFreshness,
    pub read_after: Vec<EngineApiScopedWriteToken>,
    pub pit_id: Option<ScopedPitId>,
    /// Exact public-admission vector for a PIT query. Internal leaf adapters
    /// may omit it because their outer transport owns admission; the native
    /// wire supplies the vector carried by the opaque PIT authority whenever
    /// `pit_id` is present.
    pub pit_indexes: Option<Vec<EngineApiIndexSelector>>,
    pub allow_partial_search_results: bool,
    /// `None` requests an exact total. A value permits the coordinator to
    /// stop representing the exact count only after it proves the total is
    /// greater than this threshold.
    pub total_hits_threshold: Option<u64>,
    pub source_projection: crate::SearchSourceProjection,
    pub aggregation_only: bool,
    pub continuation: Option<EngineApiContinuation>,
    /// Physical vector work is returned only for an explicit diagnostic
    /// request. Ordinary search keeps this disabled.
    pub vector_diagnostics: bool,
}

#[derive(Debug)]
pub struct EngineApiOpenPitRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub indexes: Vec<EngineApiIndexSelector>,
    /// Exact resource vector admitted by the public edge. Compatibility
    /// syntax may resolve `indexes` into aliases or expressions while this
    /// vector preserves the pre-body authorization binding for PIT reuse.
    pub admission_indexes: Vec<EngineApiIndexSelector>,
    pub routing: Vec<String>,
    pub keep_alive_millis: u64,
}

#[derive(Debug)]
pub struct EngineApiRenewPitRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub pit_id: ScopedPitId,
    pub pit_indexes: Option<Vec<EngineApiIndexSelector>>,
    pub keep_alive_millis: u64,
}

#[derive(Debug)]
pub struct EngineApiClosePitRequest {
    pub format_version: u16,
    pub request_id: RequestId,
    pub pit_id: ScopedPitId,
    pub pit_indexes: Option<Vec<EngineApiIndexSelector>>,
}
fn validate_selectors(selectors: &[EngineApiIndexSelector]) -> Result<(), ScopedSearchContractError> {
    if selectors.is_empty() || selectors.len() > MAX_ENGINE_API_INDEXES {
        return Err(ScopedSearchContractError::InvalidIndexCount(selectors.len()));
    }
    let unique = selectors.iter().collect::<BTreeSet<_>>();
    if unique.len() != selectors.len() {
        return Err(ScopedSearchContractError::DuplicateIndex);
    }
    for selector in selectors {
        if let EngineApiIndexSelector::Expression(expression) = selector {
            validate_expression(expression)?;
        }
    }
    Ok(())
}

fn validate_expression(expression: &str) -> Result<(), ScopedSearchContractError> {
    if expression.is_empty() || expression.len() > MAX_ENGINE_API_INDEX_EXPRESSION_BYTES {
        return Err(ScopedSearchContractError::InvalidExpressionLength(expression.len()));
    }
    let mut escaped = false;
    for character in expression.chars() {
        if escaped {
            if !matches!(character, '*' | '?' | '\\') {
                return Err(ScopedSearchContractError::InvalidExpressionEscape);
            }
            escaped = false;
        } else if character == '\\' {
            escaped = true;
        }
    }
    if escaped {
        return Err(ScopedSearchContractError::InvalidExpressionEscape);
    }
    Ok(())
}

fn validate_routing_values<'a>(values: impl Iterator<Item = &'a String>) -> Result<(), ScopedSearchContractError> {
    let values = values.collect::<Vec<_>>();
    if values.len() > MAX_ENGINE_API_ROUTING_VALUES {
        return Err(ScopedSearchContractError::TooManyRoutingValues(values.len()));
    }
    if values
        .iter()
        .any(|value| value.len() > MAX_ENGINE_API_ROUTING_VALUE_BYTES)
    {
        return Err(ScopedSearchContractError::RoutingValueTooLong);
    }
    Ok(())
}

fn validate_search_vectors(
    aggregations: &BTreeMap<String, Aggregation>,
    sort: &[SortSpec],
    read_after: &[EngineApiScopedWriteToken],
) -> Result<(), ScopedSearchContractError> {
    if read_after.len() > MAX_ENGINE_API_ROUTING_VALUES {
        return Err(ScopedSearchContractError::TooManyReadAfterTokens(read_after.len()));
    }
    if aggregations.len() > MAX_ENGINE_API_AGGREGATIONS {
        return Err(ScopedSearchContractError::TooManyAggregations(aggregations.len()));
    }
    if sort.len() > MAX_ENGINE_API_SORT_VALUES {
        return Err(ScopedSearchContractError::TooManySortValues(sort.len()));
    }
    let mut identities = BTreeSet::new();
    for token in read_after {
        if token.first_seq_no > token.last_seq_no
            || token.primary_term.get() == 0
            || !identities.insert((token.collection_index_ref, token.shard_id))
        {
            return Err(ScopedSearchContractError::InvalidReadAfterToken);
        }
    }
    Ok(())
}

fn validate_result_controls(plan: &EngineApiCanonicalPlan) -> Result<(), ScopedSearchContractError> {
    if let Some(highlight) = &plan.highlight {
        let fields = highlight.fields.iter().collect::<BTreeSet<_>>();
        if highlight.fields.is_empty()
            || highlight.fields.len() > crate::MAX_HIGHLIGHT_FIELDS
            || fields.len() != highlight.fields.len()
            || highlight.fields.iter().any(|field| !is_canonical_field_path(field))
            || highlight.fragment_size == 0
            || highlight.fragment_size > crate::MAX_HIGHLIGHT_FRAGMENT_BYTES
            || highlight.max_fragments == 0
            || highlight.max_fragments > crate::MAX_HIGHLIGHT_FRAGMENTS_PER_FIELD
            || plan.retrieval.is_some() && plan.highlight_query.is_none()
        {
            return Err(ScopedSearchContractError::InvalidResultControls);
        }
    } else if plan.highlight_query.is_some() {
        return Err(ScopedSearchContractError::InvalidResultControls);
    }
    if let Some(features) = &plan.result_features {
        let names = features
            .numeric
            .iter()
            .map(|feature| &feature.name)
            .collect::<BTreeSet<_>>();
        if features.top_n == 0
            || features.top_n > crate::MAX_RESULT_FEATURE_HITS
            || features.include_branch_components && plan.retrieval.is_none()
            || !features.include_branch_components && features.numeric.is_empty()
            || features.numeric.len() > crate::MAX_RESULT_FEATURES
            || names.len() != features.numeric.len()
            || features.numeric.iter().any(|feature| {
                feature.name.is_empty()
                    || feature.name.len() > crate::MAX_RESULT_FEATURE_NAME_BYTES
                    || !is_canonical_field_path(&feature.field)
                    || match feature.transform {
                        crate::NumericFeatureTransform::Identity | crate::NumericFeatureTransform::Log1p => false,
                        crate::NumericFeatureTransform::Clamp { minimum, maximum } => {
                            !minimum.is_finite() || !maximum.is_finite() || minimum > maximum
                        }
                    }
            })
        {
            return Err(ScopedSearchContractError::InvalidResultControls);
        }
    }
    if let Some(diversification) = &plan.diversification {
        let fields = diversification.fields.iter().collect::<BTreeSet<_>>();
        if diversification.fields.is_empty()
            || diversification.fields.len() > crate::MAX_DIVERSIFICATION_FIELDS
            || fields.len() != diversification.fields.len()
            || diversification
                .fields
                .iter()
                .any(|field| !is_canonical_field_path(field))
            || diversification.max_per_group == 0
            || diversification.max_per_group > crate::MAX_DIVERSIFICATION_PER_GROUP
            || plan.aggregation_only
        {
            return Err(ScopedSearchContractError::InvalidResultControls);
        }
    }
    Ok(())
}

pub fn validate_scoped_search_plan(plan: &EngineApiCanonicalPlan) -> Result<(), ScopedSearchContractError> {
    if plan.format_version != SCOPED_SEARCH_FORMAT_VERSION {
        return Err(ScopedSearchContractError::UnsupportedVersion(plan.format_version));
    }
    match (&plan.indexes, &plan.pit_id) {
        (Some(indexes), None) => validate_selectors(indexes)?,
        (None, Some(_)) if plan.routing.is_empty() => {}
        _ => return Err(ScopedSearchContractError::InvalidSearchScope),
    }
    validate_routing_values(plan.routing.iter())?;
    if plan.limit == 0 || plan.limit > MAX_ENGINE_API_SEARCH_WINDOW {
        return Err(ScopedSearchContractError::InvalidLimit(plan.limit));
    }
    if plan.from > MAX_ENGINE_API_SEARCH_WINDOW
        || plan
            .from
            .checked_add(plan.limit)
            .is_none_or(|window| window > MAX_ENGINE_API_SEARCH_WINDOW)
    {
        return Err(ScopedSearchContractError::InvalidFrom(plan.from));
    }
    if plan.pit_id.is_some() && (plan.freshness == SearchFreshness::Realtime || !plan.read_after.is_empty()) {
        return Err(ScopedSearchContractError::InvalidSearchScope);
    }
    if plan.aggregation_only && (plan.aggregations.is_empty() || plan.from != 0) {
        return Err(ScopedSearchContractError::InvalidSearchScope);
    }
    if let Some(retrieval) = &plan.retrieval {
        retrieval
            .validate()
            .map_err(|_| ScopedSearchContractError::InvalidRetrievalPlan)?;
        if plan.query != Query::MatchAll {
            return Err(ScopedSearchContractError::AmbiguousRetrieval);
        }
        if plan.from != 0
            || plan.limit > retrieval.result_window
            || !plan.sort.is_empty()
            || !plan.read_after.is_empty()
        {
            return Err(ScopedSearchContractError::InvalidRetrievalShape);
        }
    }
    validate_search_vectors(&plan.aggregations, &plan.sort, &plan.read_after)?;
    validate_result_controls(plan)
}

pub fn validate_scoped_search_continuation(
    continuation: &EngineApiContinuation,
) -> Result<(), ScopedSearchContractError> {
    if continuation.format_version != SCOPED_SEARCH_FORMAT_VERSION
        || continuation.catalog_revision.get() == 0
        || continuation.route_identity.is_empty()
        || continuation.route_identity.len() > MAX_ENGINE_API_INDEXES
        || continuation
            .route_identity
            .windows(2)
            .any(|pair| pair[0].collection_index_ref >= pair[1].collection_index_ref)
        || continuation.route_identity.iter().any(|identity| {
            identity.admission_generation.get() == 0
                || identity.schema_generation.get() == 0
                || identity.routing_revision.get() == 0
        })
        || continuation.sort_values.len() > MAX_ENGINE_API_SORT_VALUES
    {
        return Err(ScopedSearchContractError::InvalidContinuation);
    }
    validate_scoped_search_plan(&continuation.canonical_plan_identity)
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiCanonicalPlan {
    pub format_version: u16,
    pub indexes: Option<Vec<EngineApiIndexSelector>>,
    pub routing: Vec<String>,
    pub query: Query,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retrieval: Option<crate::EngineRetrievalPlan>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub highlight: Option<crate::HighlightSpec>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub highlight_query: Option<Query>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_features: Option<crate::ResultFeatureRequest>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub diversification: Option<crate::ResultDiversification>,
    #[serde(default)]
    pub aggregations: BTreeMap<String, Aggregation>,
    #[serde(default)]
    pub from: usize,
    pub limit: usize,
    #[serde(default)]
    pub sort: Vec<SortSpec>,
    #[serde(default)]
    pub freshness: SearchFreshness,
    #[serde(default)]
    pub read_after: Vec<EngineApiScopedWriteToken>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pit_id: Option<ScopedPitId>,
    pub allow_partial_search_results: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub total_hits_threshold: Option<u64>,
    #[serde(default)]
    pub source_projection: crate::SearchSourceProjection,
    /// Exact hitless execution intent. Aggregation traversal still returns the
    /// complete total; no candidate may be retained only for response hits.
    pub aggregation_only: bool,
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiRouteIdentity {
    pub collection_index_ref: CollectionIndexRef,
    pub admission_generation: AdmissionGeneration,
    pub schema_generation: SchemaGeneration,
    pub routing_revision: RoutingRevision,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiCompositeContinuation {
    pub format_version: u16,
    pub cluster_recovery_era: RecoveryEra,
    pub catalog_revision: ClusterRevision,
    pub canonical_plan_digest: String,
    pub route_identity: Vec<EngineApiRouteIdentity>,
    pub aggregation_path: Vec<String>,
    pub key: Vec<crate::CompositeScalar>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pit_id: Option<ScopedPitId>,
}

pub fn validate_scoped_composite_continuation(
    continuation: &EngineApiCompositeContinuation,
) -> Result<(), ScopedSearchContractError> {
    if continuation.format_version != SCOPED_SEARCH_FORMAT_VERSION
        || continuation.catalog_revision.get() == 0
        || continuation.canonical_plan_digest.len() != 64
        || !continuation
            .canonical_plan_digest
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        || continuation.route_identity.is_empty()
        || continuation.route_identity.len() > MAX_ENGINE_API_INDEXES
        || continuation
            .route_identity
            .windows(2)
            .any(|pair| pair[0].collection_index_ref >= pair[1].collection_index_ref)
        || continuation.route_identity.iter().any(|identity| {
            identity.admission_generation.get() == 0
                || identity.schema_generation.get() == 0
                || identity.routing_revision.get() == 0
        })
        || continuation.aggregation_path.is_empty()
        || continuation.aggregation_path.len() > crate::MAX_COMPOSITE_AGGREGATION_PATH_DEPTH
        || continuation
            .aggregation_path
            .iter()
            .any(|name| name.is_empty() || name.len() > crate::MAX_COMPOSITE_NAME_BYTES)
        || continuation.key.is_empty()
        || continuation.key.len() > crate::MAX_COMPOSITE_SOURCES
        || continuation.key.iter().any(|value| !value.is_valid())
    {
        return Err(ScopedSearchContractError::InvalidContinuation);
    }
    Ok(())
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiScopedTiebreak {
    pub collection_index_ref: CollectionIndexRef,
    pub shard_id: ShardId,
    pub seq_no: SeqNo,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiContinuation {
    pub format_version: u16,
    pub cluster_recovery_era: RecoveryEra,
    pub catalog_revision: ClusterRevision,
    pub canonical_plan_identity: EngineApiCanonicalPlan,
    pub route_identity: Vec<EngineApiRouteIdentity>,
    pub sort_values: Vec<serde_json::Value>,
    pub tiebreak: EngineApiScopedTiebreak,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pit_id: Option<ScopedPitId>,
}

/// Borrowed wire view used when one continuation identity is rebound to a
/// visible hit's exact sort position.
///
/// The serialized object is byte-for-byte shape-compatible with
/// [`EngineApiContinuation`]. Keeping the immutable plan and route identity
/// borrowed prevents response adapters from deep-cloning the complete plan
/// once per hit merely to replace `sort_values` and `tiebreak`.
#[derive(Serialize)]
pub struct EngineApiContinuationEncoding<'a> {
    format_version: u16,
    cluster_recovery_era: RecoveryEra,
    catalog_revision: ClusterRevision,
    canonical_plan_identity: &'a EngineApiCanonicalPlan,
    route_identity: &'a [EngineApiRouteIdentity],
    sort_values: &'a [serde_json::Value],
    tiebreak: EngineApiScopedTiebreak,
    #[serde(skip_serializing_if = "Option::is_none")]
    pit_id: Option<&'a ScopedPitId>,
}

impl EngineApiContinuation {
    #[must_use]
    pub fn encoding_at<'a>(
        &'a self,
        sort_values: &'a [serde_json::Value],
        tiebreak: EngineApiScopedTiebreak,
    ) -> EngineApiContinuationEncoding<'a> {
        EngineApiContinuationEncoding {
            format_version: self.format_version,
            cluster_recovery_era: self.cluster_recovery_era,
            catalog_revision: self.catalog_revision,
            canonical_plan_identity: &self.canonical_plan_identity,
            route_identity: &self.route_identity,
            sort_values,
            tiebreak,
            pit_id: self.pit_id.as_ref(),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiScopedWriteToken {
    pub collection_index_ref: CollectionIndexRef,
    pub routing_revision: RoutingRevision,
    pub shard_id: ShardId,
    pub first_seq_no: SeqNo,
    pub last_seq_no: SeqNo,
    pub primary_term: PrimaryTerm,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiCreateIndexResult {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub revision: IndexRevision,
    pub schema_generation: SchemaGeneration,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiUpdateMappingResult {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub revision: IndexRevision,
    pub schema_generation: SchemaGeneration,
}

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

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiDeleteIndexResult {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub revision: IndexRevision,
    pub state: EngineApiDeleteIndexState,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiPutResult {
    pub format_version: u16,
    pub token: EngineApiScopedWriteToken,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiWriteOutcome {
    pub collection_index_ref: CollectionIndexRef,
    pub routing_revision: RoutingRevision,
    pub shard_id: ShardId,
    pub id: DocumentId,
    pub seq_no: SeqNo,
    pub primary_term: PrimaryTerm,
    pub schema_generation: SchemaGeneration,
    pub mapping_adopted: bool,
    pub result: MutationResult,
    pub deduplicated: bool,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiBatchWriteResult {
    pub format_version: u16,
    pub durability_class: DurabilityClass,
    pub durability_boundary: DurabilityBoundary,
    pub visibility_mode: VisibilityMode,
    pub response_stage: WriteStage,
    pub tokens: Vec<EngineApiScopedWriteToken>,
    pub outcomes: Vec<EngineApiWriteOutcome>,
    pub deduplicated: u64,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiGetDocument {
    pub collection_index_ref: CollectionIndexRef,
    pub routing_revision: RoutingRevision,
    pub shard_id: ShardId,
    pub id: DocumentId,
    pub seq_no: SeqNo,
    pub primary_term: PrimaryTerm,
    pub source: serde_json::Value,
    /// Exact stored `_source` bytes carried by the current internal P6 wire
    /// for adapters that must not reserialize the semantic JSON projection.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exact_source: Option<Vec<u8>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub routing: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiGetResult {
    pub format_version: u16,
    pub found: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub document: Option<EngineApiGetDocument>,
}

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

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiPitLeaseResult {
    pub format_version: u16,
    pub pit_id: ScopedPitId,
    pub expires_at: LogicalMillis,
    pub entries: Vec<EngineApiPitRouteEntry>,
    pub shard_count: usize,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiClosePitResult {
    pub format_version: u16,
    pub removed: bool,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiResolvedShardRoute {
    pub shard_key: ShardKey,
    pub hash_start: RoutingHash,
    pub hash_end: RoutingRangeEnd,
    pub assignment_generation: AssignmentGeneration,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiResolvedRouteEntry {
    pub index: IndexName,
    pub collection_index_ref: CollectionIndexRef,
    pub admission_generation: AdmissionGeneration,
    pub schema_generation: SchemaGeneration,
    pub routing_revision: RoutingRevision,
    pub shards: Vec<EngineApiResolvedShardRoute>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiResolvedRouteVector {
    pub format_version: u16,
    pub cluster_recovery_era: RecoveryEra,
    pub catalog_revision: ClusterRevision,
    pub entries: Vec<EngineApiResolvedRouteEntry>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiScopedSearchHit {
    pub collection_index_ref: CollectionIndexRef,
    pub index: IndexName,
    pub id: DocumentId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub routing: Option<String>,
    pub shard_id: ShardId,
    pub seq_no: SeqNo,
    pub primary_term: PrimaryTerm,
    pub score: f32,
    pub source: serde_json::Value,
    /// Exact stored `_source` bytes carried by the current internal P6 search
    /// wire for adapters that must not reserialize the semantic projection.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exact_source: Option<Vec<u8>>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub sort: Vec<serde_json::Value>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub highlight: BTreeMap<String, Vec<String>>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub fragments: Vec<crate::SearchFragment>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub features: Vec<crate::ResultFeature>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub group_values: Vec<serde_json::Value>,
    pub inner_hits: BTreeMap<String, EngineApiNestedInnerHitsResult>,
}

/// P6-internal nested hit group. Unlike the public semantic JSON projection,
/// this transport type carries byte-exact child JSON text for downstream
/// leaves without JSON's decimal-array expansion for raw bytes.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiNestedInnerHitsResult {
    pub total: u64,
    pub hits: Vec<EngineApiNestedSearchHit>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiNestedSearchHit {
    pub path: String,
    pub ordinal: u32,
    pub score: f32,
    pub source: serde_json::Value,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exact_source: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub sort: Vec<serde_json::Value>,
}

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

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EngineApiSearchFailureReason {
    Cancelled,
    Overloaded,
    ResourceLimit,
    DeadlineExceeded,
    Unavailable,
    InvalidResponse,
    /// Internal signal that the caller must refresh the route projection
    /// before exposing a public unavailable failure.
    StaleRoute,
}

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

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiSearchFailure {
    pub shard_key: ShardKey,
    pub stage: EngineApiFailureStage,
    pub reason: EngineApiSearchFailureReason,
    pub recovery: EngineApiFailureRecovery,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiShardCounts {
    pub total: u32,
    pub successful: u32,
    pub skipped: u32,
    pub failed: u32,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EngineApiTotalRelation {
    Exact,
    LowerBound,
    /// Successful-shard count is known, but failed shards prevent an exact
    /// total and the configured threshold has not itself been exceeded.
    Incomplete,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiTotal {
    pub value: u64,
    pub relation: EngineApiTotalRelation,
    pub partial: bool,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EngineVectorAchievedModeV1 {
    Exact,
    Ann,
    Mixed,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineVectorBranchDiagnosticsV1 {
    pub name: String,
    pub requested_mode: crate::EngineDenseRetrievalMode,
    pub achieved_mode: EngineVectorAchievedModeV1,
    pub result_count: u64,
    pub exact_segments: u64,
    pub ann_segments: u64,
    pub exact_fallback_segments: u64,
    pub distance_evaluations: u64,
    pub peak_scratch_bytes: u64,
    pub fresh_vectors_scanned: u64,
    pub realtime_overlay_queries: u64,
    pub realtime_overlay_candidates: u64,
    pub bundle_roots: u64,
    pub object_read_requests: u64,
    pub object_bytes_read: u64,
    pub cache_hits: u64,
    pub cache_misses: u64,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub snapshot_digests: Vec<[u8; 32]>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EngineApiSearchResult {
    pub format_version: u16,
    pub route: EngineApiResolvedRouteVector,
    pub freshness: SearchExecutionFreshness,
    pub freshness_degraded: bool,
    pub timed_out: bool,
    pub partial: bool,
    pub shards: EngineApiShardCounts,
    pub failures: Vec<EngineApiSearchFailure>,
    pub total: EngineApiTotal,
    pub hits_partial: bool,
    pub aggregations_partial: bool,
    pub hits: Vec<EngineApiScopedSearchHit>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub aggregations: BTreeMap<String, serde_json::Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub continuation: Option<EngineApiContinuation>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub vector_diagnostics: Vec<EngineVectorBranchDiagnosticsV1>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ScopedSearchContractError {
    UnsupportedVersion(u16),
    InvalidIndexCount(usize),
    DuplicateIndex,
    InvalidExpressionLength(usize),
    InvalidExpressionEscape,
    TooManyRoutingValues(usize),
    RoutingValueTooLong,
    InvalidLimit(usize),
    InvalidFrom(usize),
    TooManyReadAfterTokens(usize),
    TooManyAggregations(usize),
    TooManySortValues(usize),
    InvalidReadAfterToken,
    InvalidContinuation,
    InvalidSearchScope,
    InvalidRetrievalPlan,
    AmbiguousRetrieval,
    InvalidRetrievalShape,
    InvalidResultControls,
}

impl Display for ScopedSearchContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported scoped-search format version {version}")
            }
            Self::InvalidIndexCount(count) => {
                write!(formatter, "scoped-search index count {count} is outside its bound")
            }
            Self::DuplicateIndex => formatter.write_str("scoped-search index selectors must be unique"),
            Self::InvalidExpressionLength(length) => write!(
                formatter,
                "scoped-search index expression length {length} is outside its bound"
            ),
            Self::InvalidExpressionEscape => {
                formatter.write_str("scoped-search index expression contains an invalid escape")
            }
            Self::TooManyRoutingValues(count) => {
                write!(formatter, "scoped-search routing value count {count} exceeds its bound")
            }
            Self::RoutingValueTooLong => formatter.write_str("scoped-search routing value exceeds its byte bound"),
            Self::InvalidLimit(limit) => write!(formatter, "scoped-search limit {limit} is outside 1..=10000"),
            Self::InvalidFrom(from) => write!(formatter, "scoped-search from {from} exceeds its bound"),
            Self::TooManyReadAfterTokens(count) => write!(
                formatter,
                "scoped-search read-after token count {count} exceeds its bound"
            ),
            Self::TooManyAggregations(count) => {
                write!(formatter, "scoped-search aggregation count {count} exceeds its bound")
            }
            Self::TooManySortValues(count) => {
                write!(formatter, "scoped-search sort value count {count} exceeds its bound")
            }
            Self::InvalidReadAfterToken => formatter.write_str("scoped-search read-after token is noncanonical"),
            Self::InvalidContinuation => formatter.write_str("scoped-search continuation is noncanonical"),
            Self::InvalidSearchScope => {
                formatter.write_str("scoped search requires selectors or one PIT, but never both")
            }
            Self::InvalidRetrievalPlan => formatter.write_str("scoped search retrieval plan is invalid"),
            Self::AmbiguousRetrieval => {
                formatter.write_str("scoped search cannot combine scalar query and retrieval plan")
            }
            Self::InvalidRetrievalShape => formatter.write_str(
                "scoped search retrieval requires from zero, score order, no read-after tokens, and a page within its result window",
            ),
            Self::InvalidResultControls => {
                formatter.write_str("scoped search result fragments, features, or diversification are invalid")
            }
        }
    }
}
impl Error for ScopedSearchContractError {}

#[cfg(test)]
mod tests {
    use super::{
        EngineApiCanonicalPlan, EngineApiIndexSelector, SCOPED_SEARCH_FORMAT_VERSION, ScopedSearchContractError,
        validate_scoped_search_plan,
    };
    use crate::{
        FragmentBoundary, FragmentOffsetEncoding, FragmentRanker, HighlightSpec, NumericFeatureTransform,
        NumericResultFeatureSpec, Query, ResultDiversification, ResultFeatureRequest, SearchFreshness,
    };
    use lodestone_types::IndexName;
    use std::collections::BTreeMap;

    #[test]
    fn scoped_search_plan_refuses_duplicate_selectors() -> Result<(), Box<dyn std::error::Error>> {
        let index = EngineApiIndexSelector::Exact(IndexName::new("logs")?);
        let plan = EngineApiCanonicalPlan {
            format_version: SCOPED_SEARCH_FORMAT_VERSION,
            indexes: Some(vec![index.clone(), index]),
            routing: Vec::new(),
            query: Query::MatchAll,
            retrieval: None,
            highlight: None,
            highlight_query: None,
            result_features: None,
            diversification: None,
            aggregations: BTreeMap::new(),
            from: 0,
            limit: 10,
            sort: Vec::new(),
            freshness: SearchFreshness::Committed,
            read_after: Vec::new(),
            pit_id: None,
            total_hits_threshold: None,
            allow_partial_search_results: false,
            source_projection: crate::SearchSourceProjection::ParsedAndExact,
            aggregation_only: false,
        };
        assert_eq!(
            validate_scoped_search_plan(&plan),
            Err(ScopedSearchContractError::DuplicateIndex)
        );
        Ok(())
    }

    #[test]
    fn scoped_search_plan_refuses_unbounded_or_ambiguous_result_controls() -> Result<(), Box<dyn std::error::Error>> {
        let mut plan = EngineApiCanonicalPlan {
            format_version: SCOPED_SEARCH_FORMAT_VERSION,
            indexes: Some(vec![EngineApiIndexSelector::Exact(IndexName::new("logs")?)]),
            routing: Vec::new(),
            query: Query::MatchAll,
            retrieval: None,
            highlight: Some(HighlightSpec {
                fields: vec!["message".to_owned()],
                fragment_size: 128,
                max_fragments: 2,
                boundary: FragmentBoundary::Sentence,
                offset_encoding: FragmentOffsetEncoding::Utf16CodeUnits,
                ranker: FragmentRanker::LexicalMatchCountV1,
            }),
            highlight_query: None,
            result_features: Some(ResultFeatureRequest {
                top_n: 10,
                include_branch_components: false,
                numeric: vec![NumericResultFeatureSpec {
                    name: "quality".to_owned(),
                    field: "quality".to_owned(),
                    transform: NumericFeatureTransform::Clamp {
                        minimum: 0.0,
                        maximum: 1.0,
                    },
                }],
            }),
            diversification: Some(ResultDiversification {
                fields: vec!["tenant".to_owned()],
                max_per_group: 2,
            }),
            aggregations: BTreeMap::new(),
            from: 0,
            limit: 10,
            sort: Vec::new(),
            freshness: SearchFreshness::Committed,
            read_after: Vec::new(),
            pit_id: None,
            total_hits_threshold: None,
            allow_partial_search_results: false,
            source_projection: crate::SearchSourceProjection::ParsedAndExact,
            aggregation_only: false,
        };
        assert_eq!(validate_scoped_search_plan(&plan), Ok(()));

        plan.highlight
            .as_mut()
            .ok_or("highlight")?
            .fields
            .push("message".to_owned());
        assert_eq!(
            validate_scoped_search_plan(&plan),
            Err(ScopedSearchContractError::InvalidResultControls)
        );
        plan.highlight.as_mut().ok_or("highlight")?.fields.pop();

        plan.result_features.as_mut().ok_or("features")?.numeric[0].transform = NumericFeatureTransform::Clamp {
            minimum: 2.0,
            maximum: 1.0,
        };
        assert_eq!(
            validate_scoped_search_plan(&plan),
            Err(ScopedSearchContractError::InvalidResultControls)
        );
        plan.result_features.as_mut().ok_or("features")?.numeric[0].transform = NumericFeatureTransform::Identity;

        plan.diversification
            .as_mut()
            .ok_or("diversification")?
            .fields
            .push("tenant".to_owned());
        assert_eq!(
            validate_scoped_search_plan(&plan),
            Err(ScopedSearchContractError::InvalidResultControls)
        );
        plan.diversification.as_mut().ok_or("diversification")?.fields.pop();

        plan.result_features = Some(ResultFeatureRequest {
            top_n: 1,
            include_branch_components: true,
            numeric: Vec::new(),
        });
        assert_eq!(
            validate_scoped_search_plan(&plan),
            Err(ScopedSearchContractError::InvalidResultControls)
        );
        Ok(())
    }
}