engine.rs

use std::cell::OnceCell;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::ops::{Bound, Range};
use std::path::Path;
use std::sync::Arc;
use std::time::Instant;

use sha2::{Digest, Sha256};
use tantivy_fst::Automaton;

use tantivy::collector::sort_key::{NaturalComparator, ReverseComparator};
use tantivy::collector::{Collector, SegmentCollector, TopDocs, TopNComputer};
use tantivy::columnar::StrColumn;
use tantivy::directory::error::OpenReadError;
use tantivy::directory::Directory;
use tantivy::index::SegmentId;
use tantivy::query::{
	BooleanQuery, BoostQuery, ConstScoreQuery, DfaWrapper, DisjunctionMaxQuery, EmptyQuery, ExistsQuery,
	FuzzyTermQuery, Occur, PhraseQuery, Query, RangeQuery, TermQuery, TermSetQuery,
};
use tantivy::schema::{
	Field, IndexRecordOption, Schema, TantivyDocument, TextFieldIndexing, TextOptions, FAST, INDEXED,
};
use tantivy::tokenizer::{
	AsciiFoldingFilter, Language, LowerCaser, RemoveLongFilter, Stemmer, StopWordFilter, TextAnalyzer, Token,
	TokenFilter, TokenStream, Tokenizer, MAX_TOKEN_LEN,
};
use tantivy::{
	DocAddress, DocId, Index, IndexReader, IndexSettings, IndexWriter, Opstamp, ReloadPolicy, Score, Searcher,
	SegmentReader, Term,
};
use unicode_normalization::char::{canonical_combining_class, compose, decompose_compatible};
use unicode_normalization::is_nfkc;

use crate::catalog::{self, AggregationResult};
use crate::error::{index_error, FulltextError, Result};
use crate::protocol::{
	validate_record_id, EngineConfig, EngineIdentityConfig, FilterComparator, FilterExpression, FilterFieldType,
	FilterValue, MutationBatch, QueryOptions, SearchClause, SearchExpression, SearchMode, SearchRequest, SynonymRule,
	TraceRecord, MAX_FUZZY_TERMS, MAX_PREFIX_EXPANSIONS, MAX_QUERY_CLAUSES, MAX_SEARCH_RESPONSE_BYTES, MAX_TRACE_SPANS,
};

const ID_FIELD_NAME: &str = "__fulltext_id";
const VERSION_FIELD_NAME: &str = "__fulltext_version";
pub(crate) const IDENTITY_PATH: &str = ".harper-fulltext-identity";
const META_PATH: &str = "meta.json";
const ANALYZER_NAME: &str = "english@2";
const SURFACE_ANALYZER_NAME: &str = "english_surface@2";
pub const MAX_COMMIT_PAYLOAD_BYTES: usize = 64 * 1024;
const MAX_TRACE_TOKENS_PER_VALUE: usize = 262_144;
const MAX_TRACE_EXPRESSION_TOKENS: usize = 524_288;
const MAX_TRACE_EXPRESSION_VALUES: usize = 4_096;
const MAX_TRACE_WORK_TOKENS: usize = 8_388_608;
const MAX_TOKEN_CHARACTERS: usize = 40;
const MAX_NONSTARTERS: usize = 30;
const DELETE_MERGE_RATIO: f32 = 0.5;
const COMBINING_GRAPHEME_JOINER: char = '\u{034f}';
type SynonymMap = Arc<HashMap<String, Vec<String>>>;
type BuiltSchema = (Schema, Field, Field, Vec<EngineField>, Vec<EngineFilterField>);

struct CanonicalIdentity {
	config: EngineIdentityConfig,
	analyzer: TextAnalyzer,
	analyzed_synonyms: SynonymMap,
	surface_synonyms: SynonymMap,
}

#[derive(Clone)]
pub struct Engine {
	index: Index,
	id_field: Field,
	version_field: Field,
	fields: Vec<EngineField>,
	field_lookup: HashMap<String, usize>,
	filter_fields: Vec<EngineFilterField>,
	filter_field_lookup: HashMap<String, usize>,
	analyzer: TextAnalyzer,
	surface_analyzer: TextAnalyzer,
	index_analyzer: TextAnalyzer,
	surface_index_analyzer: TextAnalyzer,
	positions: bool,
	max_query_terms: usize,
	max_query_clauses: usize,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InspectionResult {
	Missing,
	Cursorless,
	Payload(String),
}

#[derive(Clone)]
struct EngineField {
	name: String,
	field: Field,
	surface_field: Option<Field>,
	weight: f32,
}

#[derive(Clone)]
struct EngineFilterField {
	name: String,
	field: Field,
	field_type: FilterFieldType,
}

struct BuiltQuery {
	query: Box<dyn Query>,
	clauses: usize,
	stable_scores: bool,
}

const MAX_PHRASE_CONSTRUCTION_TERMS: usize = 64 * 256;
const MAX_FUZZY_CONSTRUCTION_WORK_UNITS: usize = 256;

#[derive(Default)]
struct QueryConstructionBudget {
	phrase_terms: usize,
	fuzzy_work_units: usize,
}

impl QueryConstructionBudget {
	fn phrase(&mut self, terms: usize, fields: usize) -> Result<()> {
		self.phrase_terms = self.phrase_terms.saturating_add(terms.saturating_mul(fields));
		if self.phrase_terms > MAX_PHRASE_CONSTRUCTION_TERMS {
			return Err(FulltextError::invalid(format!(
				"phrase construction exceeds {MAX_PHRASE_CONSTRUCTION_TERMS} term-field positions"
			)));
		}
		Ok(())
	}

	fn fuzzy(&mut self, terms: usize, fields: usize, distance: u8) -> Result<()> {
		if distance == 0 {
			return Ok(());
		}
		let weight = if distance == 2 { 4 } else { 1 };
		self.fuzzy_work_units = self
			.fuzzy_work_units
			.saturating_add(terms.saturating_mul(fields).saturating_mul(weight));
		if self.fuzzy_work_units > MAX_FUZZY_CONSTRUCTION_WORK_UNITS {
			return Err(FulltextError::invalid(format!(
				"fuzzy construction exceeds {MAX_FUZZY_CONSTRUCTION_WORK_UNITS} construction work units"
			)));
		}
		Ok(())
	}
}

type StableScoreTieHit = ((Score, String), DocAddress);

struct StableScoreTieCollector {
	doc_range: Range<usize>,
}

impl StableScoreTieCollector {
	fn new(doc_range: Range<usize>) -> Self {
		Self { doc_range }
	}
}

struct StableScoreTieSegmentCollector {
	top_docs: TopNComputer<(Score, u64), DocId, (NaturalComparator, ReverseComparator)>,
	id_column: StrColumn,
	segment_ord: u32,
	match_count: u64,
	missing_id: bool,
}

impl SegmentCollector for StableScoreTieSegmentCollector {
	type Fruit = Result<(u64, Vec<StableScoreTieHit>)>;

	fn collect(&mut self, doc: DocId, score: Score) {
		self.match_count += 1;
		let Some(id_ord) = self.id_column.ords().first(doc) else {
			self.missing_id = true;
			return;
		};
		// A segment's term ordinals preserve the dictionary's byte order.
		self.top_docs.push((score, id_ord), doc);
	}

	fn harvest(self) -> Self::Fruit {
		if self.missing_id {
			return Err(FulltextError::new("E_NATIVE_FAILURE", "search hit has no ID ordinal"));
		}
		let top_docs = self.top_docs.into_vec();
		let ordinals = top_docs
			.iter()
			.enumerate()
			.map(|(index, hit)| (hit.sort_key.1, index))
			.collect::<Vec<_>>();
		let ids = resolve_string_ordinals(
			&self.id_column,
			ordinals,
			top_docs.len(),
			"search hit ID ordinal is missing",
		)?;
		let hits = top_docs
			.into_iter()
			.zip(ids)
			.map(|(hit, id)| {
				let id =
					id.ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search hit ID ordinal is missing"))?;
				let id = String::from_utf8(id)
					.map_err(|_| FulltextError::new("E_NATIVE_FAILURE", "search hit ID is not UTF-8"))?;
				Ok(((hit.sort_key.0, id), DocAddress::new(self.segment_ord, hit.doc)))
			})
			.collect::<Result<Vec<_>>>()?;
		Ok((self.match_count, hits))
	}
}

impl Collector for StableScoreTieCollector {
	type Fruit = Result<(usize, u64, Vec<StableScoreTieHit>)>;
	type Child = StableScoreTieSegmentCollector;

	fn for_segment(&self, segment_ord: u32, segment: &SegmentReader) -> tantivy::Result<Self::Child> {
		let id_column = segment
			.fast_fields()
			.str(ID_FIELD_NAME)?
			.ok_or_else(|| tantivy::TantivyError::InternalError("search segment has no ID fast field".to_owned()))?;
		Ok(StableScoreTieSegmentCollector {
			top_docs: TopNComputer::new_with_comparator(self.doc_range.end, (NaturalComparator, ReverseComparator)),
			id_column,
			segment_ord,
			match_count: 0,
			missing_id: false,
		})
	}

	fn requires_scoring(&self) -> bool {
		true
	}

	fn merge_fruits(&self, segment_fruits: Vec<Result<(u64, Vec<StableScoreTieHit>)>>) -> tantivy::Result<Self::Fruit> {
		let mut top_docs = TopNComputer::new_with_comparator(
			self.doc_range.end,
			((NaturalComparator, ReverseComparator), ReverseComparator),
		);
		let mut match_count = 0u64;
		for segment_hits in segment_fruits {
			let (segment_match_count, segment_hits) = match segment_hits {
				Ok(segment_hits) => segment_hits,
				Err(error) => return Ok(Err(error)),
			};
			match_count += segment_match_count;
			for (sort_key, address) in segment_hits {
				top_docs.push((sort_key, address), address);
			}
		}
		let top_docs = top_docs.into_sorted_vec();
		let retained = top_docs.len();
		Ok(Ok((
			retained,
			match_count,
			top_docs
				.into_iter()
				.skip(self.doc_range.start)
				.map(|hit| (hit.sort_key.0, hit.doc))
				.collect(),
		)))
	}
}

struct TraversalScoreTieCollector {
	inner: StableScoreTieCollector,
	after: Option<crate::protocol::SearchBoundary>,
	deadline: Option<Instant>,
}

struct TraversalScoreTieSegmentCollector {
	inner: StableScoreTieSegmentCollector,
	after: Option<(Score, Option<u64>)>,
}

impl SegmentCollector for TraversalScoreTieSegmentCollector {
	type Fruit = Result<(u64, Vec<StableScoreTieHit>)>;

	fn collect(&mut self, doc: DocId, score: Score) {
		self.inner.match_count += 1;
		let Some(id_ord) = self.inner.id_column.ords().first(doc) else {
			self.inner.missing_id = true;
			return;
		};
		if let Some((boundary_score, first_after_id)) = self.after {
			match score.total_cmp(&boundary_score) {
				std::cmp::Ordering::Greater => return,
				std::cmp::Ordering::Equal if first_after_id.is_none_or(|first| id_ord < first) => return,
				_ => {}
			}
		}
		self.inner.top_docs.push((score, id_ord), doc);
	}

	fn harvest(self) -> Self::Fruit {
		self.inner.harvest()
	}
}

impl Collector for TraversalScoreTieCollector {
	type Fruit = Result<(usize, u64, Vec<StableScoreTieHit>)>;
	type Child = TraversalScoreTieSegmentCollector;

	fn for_segment(&self, segment_ord: u32, segment: &SegmentReader) -> tantivy::Result<Self::Child> {
		let inner = self.inner.for_segment(segment_ord, segment)?;
		let after = self
			.after
			.as_ref()
			.map(|boundary| {
				let mut stream = inner
					.id_column
					.dictionary()
					.range()
					.gt(boundary.id.as_bytes())
					.into_stream()?;
				let first_after_id = stream.advance().then(|| stream.term_ord());
				Ok::<_, std::io::Error>((boundary.score, first_after_id))
			})
			.transpose()?;
		Ok(TraversalScoreTieSegmentCollector { inner, after })
	}

	fn requires_scoring(&self) -> bool {
		true
	}

	fn merge_fruits(&self, fruits: Vec<<Self::Child as SegmentCollector>::Fruit>) -> tantivy::Result<Self::Fruit> {
		self.inner.merge_fruits(fruits)
	}

	fn collect_segment(
		&self,
		weight: &dyn tantivy::query::Weight,
		segment_ord: u32,
		reader: &SegmentReader,
	) -> tantivy::Result<<Self::Child as SegmentCollector>::Fruit> {
		let mut collector = self.for_segment(segment_ord, reader)?;
		let mut scorer = weight.scorer(reader, 1.0)?;
		let alive = reader.alive_bitset();
		let mut visited = 0_u64;
		let mut doc = scorer.doc();
		while doc != tantivy::TERMINATED {
			if visited.is_multiple_of(1024) {
				if let Err(error) = check_deadline(self.deadline) {
					return Ok(Err(error));
				}
			}
			if alive.is_none_or(|alive| alive.is_alive(doc)) {
				collector.collect(doc, scorer.score());
			}
			visited += 1;
			doc = scorer.advance();
		}
		Ok(collector.harvest())
	}
}

pub struct Writer {
	inner: IndexWriter,
	checkpoint_required: bool,
	worker_ownership_unproven: bool,
	id_field: Field,
	version_field: Field,
	fields: Vec<EngineField>,
	field_lookup: HashMap<String, usize>,
	filter_fields: Vec<EngineFilterField>,
	filter_field_lookup: HashMap<String, usize>,
}

pub(crate) struct PreparedBatch {
	mutation_count: u64,
	deletes: Vec<String>,
	documents: Vec<(String, TantivyDocument)>,
}

#[derive(Clone, Debug, PartialEq)]
pub struct SearchHit {
	pub id: String,
	pub score: f32,
	pub version: Option<String>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TotalRelation {
	Exact,
	LowerBound,
}

#[derive(Clone, Debug, PartialEq)]
pub struct SearchResult {
	pub aggregations: Vec<AggregationResult>,
	pub(crate) encoded_aggregations: Option<Vec<u8>>,
	pub total: u64,
	pub total_relation: TotalRelation,
	pub hits: Vec<SearchHit>,
	pub traversal: Option<(String, bool)>,
}

#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
pub struct TraceSpan {
	pub start: u32,
	pub end: u32,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TraceValueMatch {
	pub field: String,
	pub value_index: u32,
	pub spans: Vec<TraceSpan>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TraceRecordMatch {
	pub id: String,
	pub values: Vec<TraceValueMatch>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TraceResult {
	pub complete: bool,
	pub records: Vec<TraceRecordMatch>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TraceRecordState {
	pub membership: Option<bool>,
	pub analysis_complete: bool,
	pub spans_complete: bool,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TracePageResult {
	pub trace: TraceResult,
	pub next_record: usize,
	pub states: Vec<TraceRecordState>,
}

struct TraceProgress {
	recovery: bool,
	states: Vec<TraceRecordState>,
}

impl TraceProgress {
	fn new(recovery: bool) -> Self {
		Self {
			recovery,
			states: Vec::new(),
		}
	}

	fn defer(&self, exhausted: bool) -> bool {
		self.recovery && exhausted && !self.states.is_empty()
	}

	fn consume(&mut self, state: TraceRecordState) {
		if self.recovery {
			self.states.push(state);
		}
	}

	fn finish(self, trace: TraceResult) -> TracePageResult {
		TracePageResult {
			trace,
			next_record: self.states.len(),
			states: self.states,
		}
	}
}

struct TraceSource<'a> {
	field: &'a str,
	value_index: u32,
	analyzed: Vec<SourceToken>,
	surface: Vec<SourceToken>,
	utf16: Option<Vec<u32>>,
	analyzed_truncated: bool,
	surface_truncated: bool,
}

struct TraceLeaf<'a> {
	plan: TracePlan,
	fields: HashSet<&'a str>,
	analyzed_fuzzy: bool,
}

impl TraceLeaf<'_> {
	fn needs_surface(&self) -> bool {
		self.plan.needs_surface(self.analyzed_fuzzy)
	}
}

enum TraceExpression {
	Leaf(usize),
	And(Vec<TraceExpression>),
	Or(Vec<TraceExpression>),
	Not(Box<TraceExpression>),
	Optional(Box<TraceExpression>, Vec<TraceExpression>),
}

struct TraceDecision {
	matched: Option<bool>,
	leaves: Vec<usize>,
}

impl TraceExpression {
	fn evaluate(&self, membership: &[Option<bool>]) -> TraceDecision {
		match self {
			Self::Leaf(index) => TraceDecision {
				matched: membership[*index],
				leaves: if membership[*index] == Some(true) {
					vec![*index]
				} else {
					Vec::new()
				},
			},
			Self::Not(child) => TraceDecision {
				matched: child.evaluate(membership).matched.map(|value| !value),
				leaves: Vec::new(),
			},
			Self::And(children) | Self::Or(children) => {
				let and = matches!(self, Self::And(_));
				let decisions: Vec<_> = children.iter().map(|child| child.evaluate(membership)).collect();
				let matched = if decisions.iter().any(|child| child.matched == Some(!and)) {
					Some(!and)
				} else if decisions.iter().any(|child| child.matched.is_none()) {
					None
				} else {
					Some(and)
				};
				let leaves = if matched == Some(true) {
					decisions
						.into_iter()
						.filter(|child| child.matched == Some(true))
						.flat_map(|child| child.leaves)
						.collect()
				} else {
					Vec::new()
				};
				TraceDecision { matched, leaves }
			}
			Self::Optional(required, optional) => {
				let mut result = required.evaluate(membership);
				if result.matched == Some(true) {
					for child in optional {
						let child = child.evaluate(membership);
						if child.matched == Some(true) {
							result.leaves.extend(child.leaves);
						}
					}
				}
				result
			}
		}
	}
}

#[derive(Clone)]
struct SourceToken {
	text: String,
	start: usize,
	end: usize,
	position: usize,
}

enum TraceTerms {
	Small(Vec<String>),
	Large(HashSet<String>),
}

impl FromIterator<String> for TraceTerms {
	fn from_iter<T: IntoIterator<Item = String>>(iter: T) -> Self {
		let terms: Vec<_> = iter.into_iter().collect();
		if terms.len() <= 8 {
			Self::Small(terms)
		} else {
			Self::Large(terms.into_iter().collect())
		}
	}
}

impl TraceTerms {
	fn len(&self) -> usize {
		match self {
			Self::Small(terms) => terms.len(),
			Self::Large(terms) => terms.len(),
		}
	}

	fn contains(&self, term: &str) -> bool {
		match self {
			Self::Small(terms) => terms.iter().any(|candidate| candidate == term),
			Self::Large(terms) => terms.contains(term),
		}
	}
}

enum TracePlan {
	Any(TraceTerms),
	All(TraceTerms),
	Phrase(Vec<(usize, String)>),
	Prefix {
		completed: TraceTerms,
		prefix: String,
		matcher: Option<FuzzyMatcher>,
	},
	Fuzzy {
		exact: TraceTerms,
		fuzzy: Vec<FuzzyTraceTerm>,
	},
}

struct FuzzyMatcher {
	text: String,
	field: Field,
	options: crate::protocol::QueryOptions,
	prefix: bool,
	automaton: OnceCell<Result<DfaWrapper>>,
}

impl FuzzyMatcher {
	fn new(text: &str, field: Field, options: crate::protocol::QueryOptions, prefix: bool) -> Self {
		Self {
			text: text.to_owned(),
			field,
			options,
			prefix,
			automaton: OnceCell::new(),
		}
	}

	fn matches(&self, candidate: &str) -> Result<bool> {
		if candidate == self.text || (self.prefix && candidate.starts_with(&self.text)) {
			return Ok(true);
		}
		if self.options.fuzzy_distance == 0 {
			return Ok(false);
		}

		let automaton = self
			.automaton
			.get_or_init(|| {
				let term = Term::from_field_text(self.field, &self.text);
				let query = if self.prefix {
					FuzzyTermQuery::new_prefix(term, self.options.fuzzy_distance, self.options.fuzzy_transpositions)
				} else {
					FuzzyTermQuery::new(term, self.options.fuzzy_distance, self.options.fuzzy_transpositions)
				};
				query.automaton().map_err(index_error)
			})
			.as_ref()
			.map_err(Clone::clone)?;
		let mut state = automaton.start();
		for byte in candidate.bytes() {
			state = automaton.accept(&state, byte);
			if !automaton.can_match(&state) {
				return Ok(false);
			}
		}
		Ok(automaton.is_match(&state))
	}
}

struct FuzzyTraceTerm {
	analyzed: String,
	surface: String,
	matchers: Option<NativeFuzzyTraceMatchers>,
}

struct NativeFuzzyTraceMatchers {
	analyzed: FuzzyMatcher,
	surface: Option<FuzzyMatcher>,
}

impl NativeFuzzyTraceMatchers {
	fn surface_matcher(&self) -> &FuzzyMatcher {
		self.surface.as_ref().unwrap_or(&self.analyzed)
	}
}

impl FuzzyTraceTerm {
	fn new(analyzed: String, surface: String, field: Field, options: QueryOptions) -> Self {
		let matchers =
			(options.fuzzy_distance != 1 || !options.fuzzy_transpositions).then(|| NativeFuzzyTraceMatchers {
				analyzed: FuzzyMatcher::new(&analyzed, field, options, false),
				surface: (analyzed != surface).then(|| FuzzyMatcher::new(&surface, field, options, false)),
			});
		Self {
			analyzed,
			surface,
			matchers,
		}
	}

	fn matches_analyzed(&self, analyzed: Option<&[&str]>) -> Result<bool> {
		if let Some(terms) = analyzed {
			for term in terms {
				if match &self.matchers {
					Some(matchers) => matchers.analyzed.matches(term)?,
					None => within_one_edit(&self.analyzed, term),
				} {
					return Ok(true);
				}
			}
		}
		Ok(false)
	}

	fn matches(&self, analyzed: Option<&[&str]>, surface: &str) -> Result<bool> {
		if let Some(matchers) = &self.matchers {
			if let Some(terms) = analyzed {
				for term in terms {
					if matchers.analyzed.matches(term)? {
						return Ok(true);
					}
				}
			}
			return matchers.surface_matcher().matches(surface);
		}
		Ok(
			analyzed.is_some_and(|terms| terms.iter().any(|term| within_one_edit(&self.analyzed, term)))
				|| within_one_edit(&self.surface, surface),
		)
	}
}

impl TracePlan {
	fn needs_surface(&self, analyzed_fuzzy: bool) -> bool {
		matches!(self, Self::Prefix { .. }) || (!analyzed_fuzzy && matches!(self, Self::Fuzzy { .. }))
	}

	fn record_matches(&self, found: &HashSet<String>) -> bool {
		// Each trace branch inserts only its own distinct terms or its private match marker.
		match self {
			Self::Any(_) | Self::Fuzzy { .. } => !found.is_empty(),
			Self::All(terms) => terms.len() != 0 && found.len() == terms.len(),
			Self::Phrase(terms) => !terms.is_empty() && found.contains("__phrase"),
			Self::Prefix { completed, prefix, .. } => {
				!prefix.is_empty() && found.contains("__prefix") && found.len() == completed.len() + 1
			}
		}
	}
}

impl Engine {
	pub(crate) fn validate(config: &EngineConfig) -> Result<()> {
		canonical_identity(&config.identity).map(|_| ())
	}

	pub fn inspect<D: Directory + Clone>(directory: D, config: &EngineIdentityConfig) -> Result<InspectionResult> {
		let identity = canonical_identity(config)?.config;
		let (expected_schema, _, _, _, _) = build_schema(&identity)?;
		let expected_identity = identity_bytes(&identity);
		let sidecar_exists = directory.exists(Path::new(IDENTITY_PATH)).map_err(storage_error)?;
		let meta_exists = directory.exists(Path::new(META_PATH)).map_err(storage_error)?;

		if !sidecar_exists && !meta_exists {
			return Ok(InspectionResult::Missing);
		}
		if !sidecar_exists {
			return Err(FulltextError::new(
				"E_INCOMPLETE_CREATE",
				"meta.json exists without a fulltext identity sidecar",
			));
		}
		let actual_identity = directory.atomic_read(Path::new(IDENTITY_PATH)).map_err(storage_error)?;
		if actual_identity != expected_identity {
			return Err(FulltextError::new(
				"E_IDENTITY_MISMATCH",
				"the persisted index identity does not match the requested configuration",
			));
		}
		if !meta_exists {
			return Ok(InspectionResult::Cursorless);
		}
		let index = Index::open(directory).map_err(recovery_index_error)?;
		if index.schema() != expected_schema {
			return Err(FulltextError::new(
				"E_SCHEMA_MISMATCH",
				"the persisted Tantivy schema does not match the requested configuration",
			));
		}
		Ok(match validated_committed_payload(&index)? {
			Some(payload) => InspectionResult::Payload(payload),
			None => InspectionResult::Cursorless,
		})
	}

	pub fn open<D: Directory + Clone>(directory: D, config: &EngineConfig) -> Result<Self> {
		crate::protocol::validate_query_limits(config.limits.max_query_terms, config.limits.max_query_clauses)?;
		let canonical = canonical_identity(&config.identity)?;
		let identity = canonical.config;
		let analyzer = canonical.analyzer;
		let (schema, id_field, version_field, fields, filter_fields) = build_schema(&identity)?;
		let expected_identity = identity_bytes(&identity);
		let sidecar_exists = directory.exists(Path::new(IDENTITY_PATH)).map_err(storage_error)?;
		let meta_exists = directory.exists(Path::new(META_PATH)).map_err(storage_error)?;

		if sidecar_exists {
			let actual = directory.atomic_read(Path::new(IDENTITY_PATH)).map_err(storage_error)?;
			if actual != expected_identity {
				return Err(FulltextError::new(
					"E_IDENTITY_MISMATCH",
					"the persisted index identity does not match the requested configuration",
				));
			}
		} else if meta_exists {
			return Err(FulltextError::new(
				"E_INCOMPLETE_CREATE",
				"meta.json exists without a fulltext identity sidecar",
			));
		} else {
			directory
				.atomic_write(Path::new(IDENTITY_PATH), &expected_identity)
				.map_err(storage_error)?;
			directory.sync_directory().map_err(storage_error)?;
		}

		let index = if meta_exists {
			Index::open(directory).map_err(recovery_index_error)?
		} else {
			Index::create(directory, schema.clone(), IndexSettings::default()).map_err(index_error)?
		};
		if index.schema() != schema {
			return Err(FulltextError::new(
				"E_SCHEMA_MISMATCH",
				"the persisted Tantivy schema does not match the requested configuration",
			));
		}
		let surface_analyzer = build_surface_analyzer(None);
		let index_analyzer = build_analyzer(identity.stop_words, Some(canonical.analyzed_synonyms))?;
		let surface_index_analyzer = build_surface_analyzer(Some(canonical.surface_synonyms));
		index.tokenizers().register(ANALYZER_NAME, index_analyzer.clone());
		index
			.tokenizers()
			.register(SURFACE_ANALYZER_NAME, surface_index_analyzer.clone());
		let field_lookup = fields
			.iter()
			.enumerate()
			.map(|(index, field)| (field.name.clone(), index))
			.collect();
		let filter_field_lookup = filter_fields
			.iter()
			.enumerate()
			.map(|(index, field)| (field.name.clone(), index))
			.collect();
		Ok(Self {
			index,
			id_field,
			version_field,
			fields,
			field_lookup,
			filter_fields,
			filter_field_lookup,
			analyzer,
			surface_analyzer,
			index_analyzer,
			surface_index_analyzer,
			positions: identity.positions,
			max_query_terms: config.limits.max_query_terms,
			max_query_clauses: config.limits.max_query_clauses,
		})
	}

	pub fn writer(&self, config: &EngineConfig) -> Result<Writer> {
		self.writer_with_payload_using(config, index_error)
			.map(|(writer, _)| writer)
	}

	pub(crate) fn writer_with_payload(&self, config: &EngineConfig) -> Result<(Writer, Option<String>)> {
		self.writer_with_payload_using(config, recovery_index_error)
	}

	fn writer_with_payload_using(
		&self,
		config: &EngineConfig,
		map_error: fn(tantivy::TantivyError) -> FulltextError,
	) -> Result<(Writer, Option<String>)> {
		let inner = self
			.index
			.writer_with_num_threads(config.limits.indexing_threads, config.limits.writer_memory_bytes)
			.map_err(map_error)?;
		let mut merge_policy = tantivy::merge_policy::LogMergePolicy::default();
		merge_policy.set_del_docs_ratio_before_merge(DELETE_MERGE_RATIO);
		inner.set_merge_policy(Box::new(merge_policy));
		// A competing process can publish until we acquire the writer lock.
		let payload = self.committed_payload()?;
		Ok((
			Writer {
				inner,
				checkpoint_required: payload.is_some(),
				worker_ownership_unproven: false,
				id_field: self.id_field,
				version_field: self.version_field,
				fields: self.fields.clone(),
				field_lookup: self.field_lookup.clone(),
				filter_fields: self.filter_fields.clone(),
				filter_field_lookup: self.filter_field_lookup.clone(),
			},
			payload,
		))
	}

	pub fn reader(&self) -> Result<IndexReader> {
		self.reader_using(index_error)
	}

	pub(crate) fn reader_for_open(&self) -> Result<IndexReader> {
		self.reader_using(recovery_index_error)
	}

	fn reader_using(&self, map_error: fn(tantivy::TantivyError) -> FulltextError) -> Result<IndexReader> {
		self.index
			.reader_builder()
			.reload_policy(ReloadPolicy::Manual)
			.try_into()
			.map_err(map_error)
	}

	pub fn committed_payload(&self) -> Result<Option<String>> {
		validated_committed_payload(&self.index)
	}

	pub fn committed_payload_for_searcher(&self, searcher: &Searcher) -> Result<Option<String>> {
		let metas = self.index.load_metas().map_err(recovery_index_error)?;
		let segments = metas
			.segments
			.iter()
			.map(|segment| (segment.id(), segment.delete_opstamp()))
			.collect::<std::collections::BTreeMap<_, _>>();
		if searcher.generation().segments() != &segments {
			return Err(FulltextError::new(
				"E_RELOAD_FAILED",
				"native index changed during reload; retry the reload",
			));
		}
		validated_payload(metas.payload)
	}

	pub fn search(&self, searcher: &Searcher, request: &SearchRequest) -> Result<SearchResult> {
		self.search_inner(searcher, request, None)
	}

	pub fn search_with_deadline(
		&self,
		searcher: &Searcher,
		request: &SearchRequest,
		deadline: Instant,
	) -> Result<SearchResult> {
		self.search_inner(searcher, request, Some(deadline))
	}

	fn search_inner(
		&self,
		searcher: &Searcher,
		request: &SearchRequest,
		deadline: Option<Instant>,
	) -> Result<SearchResult> {
		crate::request_validation::validate_search_admission(request)?;

		let has_catalog = request.sort.is_some() || !request.aggregations.is_empty();
		if request.traversal && has_catalog {
			return Err(FulltextError::invalid(
				"sort and aggregations are not supported by searchPage",
			));
		}
		let catalog_fields = has_catalog.then(|| {
			self.filter_fields
				.iter()
				.map(|field| crate::protocol::FilterFieldConfig {
					name: field.name.clone(),
					field_type: field.field_type,
				})
				.collect::<Vec<_>>()
		});
		let catalog = catalog_fields
			.as_deref()
			.map(|fields| catalog::validate_requests(request.sort.as_deref(), &request.aggregations, fields))
			.transpose()?;
		if catalog.is_some() {
			catalog::check_aggregation_shape(&request.aggregations, searcher.segment_readers().len())?;
		}
		check_deadline(deadline)?;
		let snapshot = if request.traversal {
			let snapshot = search_snapshot(searcher.generation().segments());
			if request.after.as_ref().is_some_and(|after| after.snapshot != snapshot) {
				return Err(FulltextError::new(
					"E_CURSOR_EXPIRED",
					"search snapshot changed; restart traversal on a dedicated reader",
				));
			}
			Some(snapshot)
		} else {
			None
		};
		let query = self.query(searcher, request)?;
		check_deadline(deadline)?;
		self.collect_search(searcher, request, deadline, catalog, snapshot, query)
	}
	fn collect_search(
		&self,
		searcher: &Searcher,
		request: &SearchRequest,
		deadline: Option<Instant>,
		catalog: Option<catalog::ValidatedCatalog<'_>>,
		snapshot: Option<String>,
		query: BuiltQuery,
	) -> Result<SearchResult> {
		if let Some(snapshot) = snapshot {
			let collector = TraversalScoreTieCollector {
				inner: StableScoreTieCollector::new(0..request.limit.saturating_add(1)),
				after: request.after.clone(),
				deadline,
			};
			let (_, total, mut scored_docs) = searcher
				.search(query.query.as_ref(), &collector)
				.map_err(index_error)??;
			check_deadline(deadline)?;
			let more = scored_docs.len() > request.limit;
			scored_docs.truncate(request.limit);
			let hits = search_hits(searcher, scored_docs)?;
			let bytes = search_response_bytes(&hits, 17 + 4 + snapshot.len() + 1)?;
			if bytes > MAX_SEARCH_RESPONSE_BYTES {
				return Err(FulltextError::new(
					"E_RESULT_TOO_LARGE",
					"search traversal response exceeds its byte limit",
				));
			}
			check_deadline(deadline)?;
			return Ok(SearchResult {
				aggregations: Vec::new(),
				encoded_aggregations: None,
				total,
				total_relation: TotalRelation::Exact,
				hits,
				traversal: Some((snapshot, more)),
			});
		}

		if request.sort.is_some() {
			let catalog =
				catalog.ok_or_else(|| FulltextError::native("Validated catalog missing for sorted search"))?;
			let catalog::SortedCatalogResult {
				hits: scored_docs,
				total,
				aggregations,
			} = catalog.sorted_catalog(
				searcher,
				query.query.as_ref(),
				request.offset,
				request.limit,
				self.index.tokenizers().clone(),
				deadline,
			)?;
			let hits = search_hits(searcher, scored_docs)?;
			let encoded_aggregations = if aggregations.is_empty() {
				None
			} else {
				Some(catalog::encode_results(&aggregations)?)
			};
			let aggregate_bytes = encoded_aggregations.as_ref().map_or(4, Vec::len);
			if search_response_bytes(&hits, 13)?.saturating_add(aggregate_bytes) > MAX_SEARCH_RESPONSE_BYTES {
				return Err(FulltextError::new(
					"E_RESULT_TOO_LARGE",
					"search response exceeds its byte limit",
				));
			}
			check_deadline(deadline)?;
			return Ok(SearchResult {
				total,
				total_relation: TotalRelation::Exact,
				hits,
				traversal: None,
				aggregations,
				encoded_aggregations,
			});
		}
		let window_end = request.offset + request.limit;
		let stable_scores = request.candidate_ids.is_none() && request.filter.is_none() && query.stable_scores;
		let (hits, bounded_total, bounded_total_relation, collected_total) = if stable_scores {
			let (retained, match_count, mut scored_docs) = searcher
				.search(
					query.query.as_ref(),
					&catalog::DeadlineCollector::new(
						StableScoreTieCollector::new(request.offset..window_end.saturating_add(1)),
						deadline,
					),
				)
				.map_err(collection_error)??;
			check_deadline(deadline)?;
			scored_docs.truncate(request.limit);
			let hits = search_hits(searcher, scored_docs)?;
			let (total, relation) = if retained < window_end.saturating_add(1) {
				(retained as u64, TotalRelation::Exact)
			} else {
				(window_end as u64, TotalRelation::LowerBound)
			};
			(hits, total, relation, Some(match_count))
		} else {
			let scored_docs = searcher
				.search(
					query.query.as_ref(),
					&TopDocs::with_limit(window_end.saturating_add(1)).order_by_score(),
				)
				.map_err(index_error)?;
			check_deadline(deadline)?;
			let boundary_tie =
				scored_docs.len() > window_end && scored_docs[window_end - 1].0 == scored_docs[window_end].0;
			let (hits, collected_total) = if boundary_tie {
				let (_, match_count, scored_docs) = searcher
					.search(
						query.query.as_ref(),
						&catalog::DeadlineCollector::new(
							StableScoreTieCollector::new(request.offset..window_end),
							deadline,
						),
					)
					.map_err(collection_error)??;
				let hits = search_hits(searcher, scored_docs)?;
				(hits, Some(match_count))
			} else {
				let ids = search_hit_ids(searcher, &scored_docs)?;
				let mut ranked = scored_docs
					.iter()
					.zip(ids)
					.map(|((score, address), id)| ((*score, id), *address))
					.collect::<Vec<_>>();
				ranked.sort_unstable_by(|((left_score, left_id), left), ((right_score, right_id), right)| {
					right_score
						.total_cmp(left_score)
						.then_with(|| left_id.as_bytes().cmp(right_id.as_bytes()))
						.then_with(|| left.cmp(right))
				});
				let page = ranked
					.into_iter()
					.skip(request.offset)
					.take(request.limit)
					.collect::<Vec<_>>();
				let hits = search_hits(searcher, page)?;
				(hits, None)
			};
			let (total, relation) = if scored_docs.len() <= window_end {
				(scored_docs.len() as u64, TotalRelation::Exact)
			} else {
				(window_end as u64, TotalRelation::LowerBound)
			};
			(hits, total, relation, collected_total)
		};
		let (aggregation_total, aggregations) = if let Some(catalog) = catalog {
			if request.exact_total && collected_total.is_none() && bounded_total_relation != TotalRelation::Exact {
				let (total, results) = catalog.aggregate_count(
					searcher,
					query.query.as_ref(),
					self.index.tokenizers().clone(),
					deadline,
				)?;
				(Some(total), results)
			} else {
				(
					None,
					catalog.aggregate(
						searcher,
						query.query.as_ref(),
						self.index.tokenizers().clone(),
						deadline,
					)?,
				)
			}
		} else {
			(None, Vec::new())
		};
		let collected_total = collected_total.or(aggregation_total);
		let (total, total_relation) = if request.exact_total {
			match collected_total {
				Some(total) => {
					check_deadline(deadline)?;
					(total, TotalRelation::Exact)
				}
				None if bounded_total_relation == TotalRelation::Exact => (bounded_total, TotalRelation::Exact),
				None => {
					check_deadline(deadline)?;
					(
						catalog::count_matches(searcher, query.query.as_ref(), deadline).map_err(collection_error)?,
						TotalRelation::Exact,
					)
				}
			}
		} else {
			(bounded_total, bounded_total_relation)
		};
		check_deadline(deadline)?;

		let encoded_aggregations = if aggregations.is_empty() {
			None
		} else {
			Some(catalog::encode_results(&aggregations)?)
		};
		let response_bytes =
			search_response_bytes(&hits, 13)?.saturating_add(encoded_aggregations.as_ref().map_or(4, Vec::len));
		check_deadline(deadline)?;
		if response_bytes > MAX_SEARCH_RESPONSE_BYTES {
			return Err(FulltextError::new(
				"E_RESULT_TOO_LARGE",
				format!("search response exceeds {MAX_SEARCH_RESPONSE_BYTES} bytes"),
			));
		}
		Ok(SearchResult {
			aggregations,
			encoded_aggregations,
			total,
			total_relation,
			hits,
			traversal: None,
		})
	}

	pub fn trace_matches(
		&self,
		request: &SearchRequest,
		records: &[TraceRecord],
		deadline: Option<Instant>,
	) -> Result<TraceResult> {
		Ok(self.trace_matches_inner(request, records, deadline, false)?.trace)
	}

	pub fn trace_matches_page(
		&self,
		request: &SearchRequest,
		records: &[TraceRecord],
		deadline: Option<Instant>,
	) -> Result<TracePageResult> {
		self.trace_matches_inner(request, records, deadline, true)
	}

	fn trace_matches_inner(
		&self,
		request: &SearchRequest,
		records: &[TraceRecord],
		deadline: Option<Instant>,
		recovery: bool,
	) -> Result<TracePageResult> {
		check_deadline(deadline)?;
		crate::request_validation::validate_trace_admission(request, records)?;
		let SearchExpression::Clause(clause) = &request.expression else {
			return self.trace_expression(request, records, deadline, recovery);
		};
		let selected = self.selected_fields(&clause.fields)?;
		self.require_surface_fields(&selected)?;
		let selected_names = selected.iter().map(|field| field.name.as_str()).collect::<HashSet<_>>();
		let candidates = request
			.candidate_ids
			.as_ref()
			.map(|ids| ids.iter().map(String::as_str).collect::<HashSet<_>>());
		if !recovery && candidates.as_ref().is_some_and(HashSet::is_empty) {
			return Ok(TraceProgress::new(false).finish(TraceResult {
				complete: true,
				records: Vec::new(),
			}));
		}
		let plan = self.trace_plan(clause)?;
		let mut complete = true;
		let mut remaining_spans = MAX_TRACE_SPANS;
		let mut response_bytes = 3usize + if recovery { 2 + records.len() } else { 0 };
		let mut progress = TraceProgress::new(recovery);
		let mut matched_records = Vec::new();
		let mut analyzed_source_analyzer = self.index_analyzer.clone();
		let mut surface_source_analyzer = self.surface_index_analyzer.clone();
		for record in records {
			check_deadline(deadline)?;
			if candidates
				.as_ref()
				.is_some_and(|candidates| !candidates.contains(record.id.as_str()))
			{
				progress.consume(TraceRecordState {
					membership: Some(false),
					analysis_complete: true,
					spans_complete: true,
				});
				continue;
			}
			let mut shared_exhausted = false;
			let mut pending_values = Vec::new();
			let mut found = HashSet::new();
			let mut record_span_budget = remaining_spans;
			let mut record_truncated = false;
			let mut record_analysis_truncated = false;
			for (field, source_values) in &record.fields {
				if !self.field_lookup.contains_key(field) {
					return Err(FulltextError::invalid(format!("unknown trace field {field}")));
				}
				if !selected_names.contains(field.as_str()) {
					continue;
				}
				for (value_index, value) in source_values.iter().enumerate() {
					check_deadline(deadline)?;
					let (spans, terms, spans_truncated, analysis_truncated) = self.trace_value(
						&plan,
						value,
						record_span_budget,
						deadline,
						&mut analyzed_source_analyzer,
						&mut surface_source_analyzer,
					)?;
					record_truncated |= spans_truncated || analysis_truncated;
					record_analysis_truncated |= analysis_truncated;
					shared_exhausted |= spans_truncated;
					found.extend(terms);
					if spans.is_empty() {
						continue;
					}
					record_span_budget = record_span_budget.saturating_sub(spans.len());
					pending_values.push((field.as_str(), value_index as u32, spans));
				}
			}
			complete &= !record_analysis_truncated;
			let matched = plan.record_matches(&found);
			if matched {
				complete &= !record_truncated;
				if !recovery && pending_values.is_empty() {
					continue;
				}
				let record_header_bytes = 6usize.saturating_add(record.id.len());
				let mut record_bytes = 0usize;
				let mut values = Vec::new();
				for (field, value_index, spans) in pending_values {
					let value_header_bytes = 10usize.saturating_add(field.len());
					let available = MAX_SEARCH_RESPONSE_BYTES
						.saturating_sub(response_bytes)
						.saturating_sub(record_header_bytes)
						.saturating_sub(record_bytes);
					let byte_limited_spans = available.saturating_sub(value_header_bytes) / 8;
					let keep = spans.len().min(remaining_spans).min(byte_limited_spans);
					if keep < spans.len() {
						shared_exhausted = true;
						complete = false;
					}
					if keep == 0 {
						continue;
					}
					values.push(TraceValueMatch {
						field: field.to_owned(),
						value_index,
						spans: spans.into_iter().take(keep).collect(),
					});
					remaining_spans -= keep;
					record_bytes = record_bytes.saturating_add(value_header_bytes + keep * 8);
				}
				if !recovery && values.is_empty() {
					continue;
				}
				shared_exhausted |=
					record_header_bytes + record_bytes > MAX_SEARCH_RESPONSE_BYTES.saturating_sub(response_bytes);
				if progress.defer(shared_exhausted) {
					complete = false;
					break;
				}
				response_bytes = response_bytes.saturating_add(record_header_bytes + record_bytes);
				matched_records.push(TraceRecordMatch {
					id: record.id.clone(),
					values,
				});
			}
			let state = TraceRecordState {
				membership: if matched {
					Some(true)
				} else if record_analysis_truncated {
					None
				} else {
					Some(false)
				},
				analysis_complete: !record_analysis_truncated,
				spans_complete: !record_analysis_truncated && (!matched || !shared_exhausted),
			};
			progress.consume(state);
			if recovery && shared_exhausted && matched {
				break;
			}
		}
		if recovery {
			complete &= progress.states.len() == records.len();
		}
		Ok(progress.finish(TraceResult {
			complete,
			records: matched_records,
		}))
	}

	fn prepare_trace_expression<'a>(
		&'a self,
		expression: &SearchExpression,
		leaves: &mut Vec<TraceLeaf<'a>>,
		negative: bool,
		budget: &mut QueryConstructionBudget,
	) -> Result<TraceExpression> {
		Ok(match expression {
			SearchExpression::Clause(clause) => {
				let fields = self.selected_fields(&clause.fields)?;
				self.require_surface_fields(&fields)?;
				let plan = self.trace_plan(clause)?;
				match &plan {
					TracePlan::Phrase(terms) => budget.phrase(terms.len(), fields.len())?,
					TracePlan::Fuzzy { fuzzy, .. } => {
						budget.fuzzy(fuzzy.len(), fields.len(), clause.options.fuzzy_distance)?
					}
					TracePlan::Prefix { matcher, .. } => budget.fuzzy(
						usize::from(matcher.is_some()),
						fields.len(),
						clause.options.fuzzy_distance,
					)?,
					_ => {}
				}
				let index = leaves.len();
				leaves.push(TraceLeaf {
					plan,
					fields: fields.iter().map(|field| field.name.as_str()).collect(),
					analyzed_fuzzy: negative,
				});
				TraceExpression::Leaf(index)
			}
			SearchExpression::And(children) => TraceExpression::And(
				children
					.iter()
					.map(|child| self.prepare_trace_expression(child, leaves, negative, budget))
					.collect::<Result<_>>()?,
			),
			SearchExpression::Or(children) => TraceExpression::Or(
				children
					.iter()
					.map(|child| self.prepare_trace_expression(child, leaves, negative, budget))
					.collect::<Result<_>>()?,
			),
			SearchExpression::Not(child) => {
				TraceExpression::Not(Box::new(self.prepare_trace_expression(child, leaves, true, budget)?))
			}
			SearchExpression::Boost { clause, .. } => {
				self.prepare_trace_expression(clause, leaves, negative, budget)?
			}
			SearchExpression::Optional { required, optional } => TraceExpression::Optional(
				Box::new(self.prepare_trace_expression(required, leaves, negative, budget)?),
				optional
					.iter()
					.map(|child| self.prepare_trace_expression(child, leaves, negative, budget))
					.collect::<Result<_>>()?,
			),
		})
	}

	fn trace_expression(
		&self,
		request: &SearchRequest,
		records: &[TraceRecord],
		deadline: Option<Instant>,
		recovery: bool,
	) -> Result<TracePageResult> {
		crate::protocol::validate_search_structure(&request.expression, self.max_query_clauses, self.fields.len())?;
		let mut leaves = Vec::new();
		let expression = self.prepare_trace_expression(
			&request.expression,
			&mut leaves,
			false,
			&mut QueryConstructionBudget::default(),
		)?;
		let selected: HashSet<&str> = leaves.iter().flat_map(|leaf| leaf.fields.iter().copied()).collect();
		let candidates = request
			.candidate_ids
			.as_ref()
			.map(|ids| ids.iter().map(String::as_str).collect::<HashSet<_>>());
		let mut result = TraceResult {
			complete: true,
			records: Vec::new(),
		};
		let mut remaining_spans = MAX_TRACE_SPANS;
		let mut remaining_bytes = MAX_SEARCH_RESPONSE_BYTES - 3 - if recovery { 2 + records.len() } else { 0 };
		let mut progress = TraceProgress::new(recovery);
		let mut work = MAX_TRACE_WORK_TOKENS;
		let surface_fields: HashSet<&str> = leaves
			.iter()
			.filter(|leaf| leaf.needs_surface())
			.flat_map(|leaf| leaf.fields.iter().copied())
			.collect();
		let mut analyzed_analyzer = self.index_analyzer.clone();
		let mut surface_analyzer = self.surface_index_analyzer.clone();
		'records: for record in records {
			check_deadline(deadline)?;
			if candidates.as_ref().is_some_and(|ids| !ids.contains(record.id.as_str())) {
				progress.consume(TraceRecordState {
					membership: Some(false),
					analysis_complete: true,
					spans_complete: true,
				});
				continue;
			}
			let mut shared_exhausted = false;
			let mut analysis_complete = true;
			let mut spans_complete = true;
			let mut sources = Vec::new();
			let mut remaining_tokens = MAX_TRACE_EXPRESSION_TOKENS;
			let mut missing_fields = HashSet::new();
			for (field, values) in &record.fields {
				if !self.field_lookup.contains_key(field) {
					return Err(FulltextError::invalid(format!("unknown trace field {field}")));
				}
				if !selected.contains(field.as_str()) {
					continue;
				}
				for (value_index, value) in values.iter().enumerate() {
					if sources.len() == MAX_TRACE_EXPRESSION_VALUES || (remaining_tokens == 0 && !value.is_empty()) {
						missing_fields.insert(field.as_str());
						continue;
					}
					let (mut analyzed, mut analyzed_truncated) =
						Self::source_tokens(&mut analyzed_analyzer, value, deadline)?;
					if analyzed.len() > remaining_tokens {
						analyzed.truncate(remaining_tokens);
						analyzed_truncated = true;
					}
					remaining_tokens -= analyzed.len();
					let (mut surface, mut surface_truncated) = if !surface_fields.contains(field.as_str()) {
						(Vec::new(), false)
					} else if remaining_tokens == 0 {
						(Vec::new(), !value.is_empty())
					} else {
						Self::source_tokens(&mut surface_analyzer, value, deadline)?
					};
					if surface.len() > remaining_tokens {
						surface.truncate(remaining_tokens);
						surface_truncated = true;
					}
					remaining_tokens -= surface.len();
					if analyzed.is_empty() && surface.is_empty() && !analyzed_truncated && !surface_truncated {
						continue;
					}
					sources.push(TraceSource {
						field: field.as_str(),
						value_index: value_index as u32,
						analyzed,
						surface,
						utf16: utf16_offsets(value),
						analyzed_truncated,
						surface_truncated,
					});
				}
			}
			let mut membership = Vec::with_capacity(leaves.len());
			for leaf in &leaves {
				let mut found = HashSet::new();
				let mut unknown = leaf.fields.iter().any(|field| missing_fields.contains(*field));
				for source in &sources {
					if !leaf.fields.contains(source.field) {
						continue;
					}
					let cost = source.analyzed.len() + if leaf.needs_surface() { source.surface.len() } else { 0 };
					if cost > work {
						if progress.defer(true) {
							result.complete = false;
							break 'records;
						}
						shared_exhausted = true;
						unknown = true;
						continue;
					}
					work -= cost;
					let (_, terms, _, truncated) =
						self.trace_tokens(&leaf.plan, source, 0, deadline, leaf.analyzed_fuzzy)?;
					found.extend(terms);
					unknown |= truncated;
				}
				membership.push(if leaf.plan.record_matches(&found) {
					Some(true)
				} else if unknown {
					None
				} else {
					Some(false)
				});
				result.complete &= !unknown;
				analysis_complete &= !unknown;
			}
			let decision = expression.evaluate(&membership);
			if progress.defer(shared_exhausted) {
				result.complete = false;
				break;
			}
			if decision.matched != Some(true) {
				progress.consume(TraceRecordState {
					membership: decision.matched,
					analysis_complete,
					spans_complete: analysis_complete,
				});
				if recovery && shared_exhausted {
					break;
				}
				continue;
			}
			let mut values = BTreeMap::<(&str, u32), BTreeSet<(u32, u32)>>::new();
			let mut pending_spans = remaining_spans;
			for index in decision.leaves {
				let leaf = &leaves[index];
				for source in &sources {
					if !leaf.fields.contains(source.field) {
						continue;
					}
					let cost = source.analyzed.len() + if leaf.needs_surface() { source.surface.len() } else { 0 };
					if cost > work {
						if progress.defer(true) {
							result.complete = false;
							break 'records;
						}
						shared_exhausted = true;
						spans_complete = false;
						result.complete = false;
						continue;
					}
					work -= cost;
					let entry = values.entry((source.field, source.value_index)).or_default();
					let candidate_spans = pending_spans + entry.len();
					let (spans, _, truncated, _) =
						self.trace_tokens(&leaf.plan, source, candidate_spans, deadline, false)?;
					if truncated && progress.defer(true) {
						result.complete = false;
						break 'records;
					}
					result.complete &= !truncated;
					shared_exhausted |= truncated;
					spans_complete &= !truncated;
					for span in spans {
						if entry.contains(&(span.start, span.end)) {
							continue;
						}
						if pending_spans == 0 {
							if progress.defer(true) {
								result.complete = false;
								break 'records;
							}
							shared_exhausted = true;
							spans_complete = false;
							result.complete = false;
							break;
						}
						entry.insert((span.start, span.end));
						pending_spans -= 1;
					}
				}
			}
			let mut record_bytes = 6 + record.id.len();
			let mut matched_values = Vec::new();
			for ((field, value_index), spans) in values {
				let header = 10 + field.len();
				let keep = spans
					.len()
					.min(remaining_bytes.saturating_sub(record_bytes + header) / 8);
				if keep < spans.len() {
					shared_exhausted = true;
					spans_complete = false;
					result.complete = false;
				}
				if keep == 0 {
					continue;
				}
				let spans = spans
					.into_iter()
					.take(keep)
					.map(|(start, end)| TraceSpan { start, end })
					.collect();
				record_bytes += header + keep * 8;
				remaining_spans -= keep;
				matched_values.push(TraceValueMatch {
					field: field.to_owned(),
					value_index,
					spans,
				});
			}
			if record_bytes > remaining_bytes {
				result.complete = false;
				if !recovery {
					continue;
				}
				shared_exhausted = true;
				spans_complete = false;
			}
			if progress.defer(shared_exhausted) {
				result.complete = false;
				break;
			}
			remaining_bytes = remaining_bytes.saturating_sub(record_bytes);
			result.records.push(TraceRecordMatch {
				id: record.id.clone(),
				values: matched_values,
			});
			progress.consume(TraceRecordState {
				membership: Some(true),
				analysis_complete,
				spans_complete: spans_complete && analysis_complete,
			});
			if recovery && shared_exhausted {
				break;
			}
		}
		if recovery {
			result.complete &= progress.states.len() == records.len();
		}
		Ok(progress.finish(result))
	}

	fn trace_plan(&self, request: &SearchClause) -> Result<TracePlan> {
		match request.mode {
			SearchMode::Any => Ok(TracePlan::Any(
				self.analyze(&request.text, false, true)?.into_iter().collect(),
			)),
			SearchMode::All => Ok(TracePlan::All(
				self.analyze(&request.text, false, true)?.into_iter().collect(),
			)),
			SearchMode::Phrase => {
				if !self.positions {
					return Err(FulltextError::invalid("phrase search requires positions to be enabled"));
				}
				Ok(TracePlan::Phrase(self.analyze_positioned(&request.text)?))
			}
			SearchMode::Prefix | SearchMode::FuzzyPrefix => {
				if request.text.chars().last().is_some_and(char::is_whitespace) {
					return Ok(TracePlan::All(
						self.analyze(&request.text, false, true)?.into_iter().collect(),
					));
				}
				let (completed, prefix) = self.final_surface_term(&request.text)?;
				let prefix = prefix.unwrap_or_default();
				let minimum = if request.mode == SearchMode::FuzzyPrefix { 4 } else { 3 };
				if !prefix.is_empty() && prefix.chars().count() < minimum {
					return Err(FulltextError::invalid(format!(
						"{} prefix must contain at least {minimum} Unicode characters",
						if request.mode == SearchMode::FuzzyPrefix {
							"fuzzy"
						} else {
							"exact"
						}
					)));
				}
				let matcher = (request.mode == SearchMode::FuzzyPrefix && fuzzy_eligible(&prefix))
					.then(|| FuzzyMatcher::new(&prefix, self.id_field, request.options, true));
				Ok(TracePlan::Prefix {
					completed: completed.into_iter().collect(),
					prefix,
					matcher,
				})
			}
			SearchMode::Fuzzy => {
				let mut terms = Vec::new();
				for surface in self.analyze(&request.text, true, true)? {
					if let [analyzed] = self.analyze(&surface, false, false)?.as_slice() {
						terms.push((analyzed.clone(), surface));
					}
				}
				if terms.iter().filter(|(_, surface)| fuzzy_eligible(surface)).count() > MAX_FUZZY_TERMS {
					return Err(FulltextError::invalid(format!(
						"fuzzy search contains more than {MAX_FUZZY_TERMS} eligible terms"
					)));
				}
				Ok(TracePlan::Fuzzy {
					exact: terms.iter().map(|(term, _)| term.clone()).collect(),
					fuzzy: terms
						.into_iter()
						.filter(|(_, surface)| fuzzy_eligible(surface))
						.map(|(analyzed, surface)| {
							FuzzyTraceTerm::new(analyzed, surface, self.id_field, request.options)
						})
						.collect(),
				})
			}
		}
	}

	fn trace_value(
		&self,
		plan: &TracePlan,
		value: &str,
		max_spans: usize,
		deadline: Option<Instant>,
		analyzed_source_analyzer: &mut TextAnalyzer,
		surface_source_analyzer: &mut TextAnalyzer,
	) -> Result<(Vec<TraceSpan>, HashSet<String>, bool, bool)> {
		let (analyzed, analyzed_truncated) = Self::source_tokens(analyzed_source_analyzer, value, deadline)?;
		let needs_surface = plan.needs_surface(false);
		let (surface, surface_truncated) = if needs_surface {
			Self::source_tokens(surface_source_analyzer, value, deadline)?
		} else {
			(Vec::new(), false)
		};
		let source = TraceSource {
			field: "",
			value_index: 0,
			analyzed,
			surface,
			utf16: utf16_offsets(value),
			analyzed_truncated,
			surface_truncated,
		};
		self.trace_tokens(plan, &source, max_spans, deadline, false)
	}

	fn trace_tokens(
		&self,
		plan: &TracePlan,
		source: &TraceSource<'_>,
		max_spans: usize,
		deadline: Option<Instant>,
		analyzed_fuzzy: bool,
	) -> Result<(Vec<TraceSpan>, HashSet<String>, bool, bool)> {
		let analyzed = &source.analyzed;
		let surface = &source.surface;
		let utf16_offsets = &source.utf16;
		let needs_surface = plan.needs_surface(analyzed_fuzzy);
		let analysis_truncated = source.analyzed_truncated || (needs_surface && source.surface_truncated);
		let mut spans = Vec::new();
		let mut seen_spans = HashSet::new();
		let mut found = HashSet::new();
		let mut spans_truncated = false;
		match plan {
			TracePlan::Any(terms) | TracePlan::All(terms) => {
				for (index, token) in analyzed.iter().enumerate() {
					if index % 256 == 0 {
						check_deadline(deadline)?;
					}
					if terms.contains(&token.text) {
						found.insert(token.text.clone());
						spans_truncated |= push_trace_span(
							&mut spans,
							&mut seen_spans,
							source_span(utf16_offsets, token.start, token.end),
							max_spans,
						);
						if spans_truncated && plan.record_matches(&found) {
							break;
						}
					}
				}
			}
			TracePlan::Phrase(terms) => {
				if !terms.is_empty() {
					for (anchor_index, anchor) in analyzed.iter().enumerate() {
						check_deadline(deadline)?;
						if anchor.text != terms[0].1 {
							continue;
						}
						let first_source_position = anchor.position;
						let first_query_position = terms[0].0;
						let mut source_index = anchor_index + 1;
						let mut final_token = anchor;
						let mut matches = true;
						for (position, term) in terms.iter().skip(1) {
							let expected_position = first_source_position + (position - first_query_position);
							while source_index < analyzed.len() && analyzed[source_index].position < expected_position {
								source_index += 1;
							}
							let position_start = source_index;
							while source_index < analyzed.len() && analyzed[source_index].position == expected_position
							{
								source_index += 1;
							}
							let Some(token) = analyzed[position_start..source_index]
								.iter()
								.find(|token| token.text == *term)
							else {
								matches = false;
								break;
							};
							final_token = token;
						}
						if matches {
							found.insert("__phrase".to_owned());
							spans_truncated |= push_trace_span(
								&mut spans,
								&mut seen_spans,
								source_span(utf16_offsets, anchor.start, final_token.end),
								max_spans,
							);
							if spans_truncated && plan.record_matches(&found) {
								break;
							}
						}
					}
				}
			}
			TracePlan::Prefix {
				completed,
				prefix,
				matcher,
			} => {
				for (index, token) in analyzed.iter().enumerate() {
					if index % 256 == 0 {
						check_deadline(deadline)?;
					}
					if completed.contains(&token.text) {
						found.insert(token.text.clone());
						spans_truncated |= push_trace_span(
							&mut spans,
							&mut seen_spans,
							source_span(utf16_offsets, token.start, token.end),
							max_spans,
						);
						if spans_truncated && plan.record_matches(&found) {
							break;
						}
					}
				}
				for (index, token) in surface.iter().enumerate() {
					if index % 256 == 0 {
						check_deadline(deadline)?;
					}
					if token.text.starts_with(prefix)
						|| match matcher {
							Some(matcher) => matcher.matches(&token.text)?,
							None => false,
						} {
						found.insert("__prefix".to_owned());
						spans_truncated |= push_trace_span(
							&mut spans,
							&mut seen_spans,
							source_span(utf16_offsets, token.start, token.end),
							max_spans,
						);
						if spans_truncated && plan.record_matches(&found) {
							break;
						}
					}
				}
			}
			TracePlan::Fuzzy { exact, fuzzy } => {
				let mut analyzed_by_position = HashMap::<usize, Vec<&str>>::new();
				if !analyzed_fuzzy {
					for token in analyzed {
						analyzed_by_position
							.entry(token.position)
							.or_default()
							.push(token.text.as_str());
					}
				}
				let tokens = if analyzed_fuzzy { analyzed } else { surface };
				for (index, token) in tokens.iter().enumerate() {
					if index % 256 == 0 {
						check_deadline(deadline)?;
					}
					let analyzed_term = [token.text.as_str()];
					let source_terms = if analyzed_fuzzy {
						Some(analyzed_term.as_slice())
					} else {
						analyzed_by_position.get(&token.position).map(Vec::as_slice)
					};
					let exact_match =
						source_terms.and_then(|tokens| tokens.iter().find(|term| exact.contains(term)).copied());
					let mut fuzzy_match = None;
					if exact_match.is_none() {
						for alternative in fuzzy {
							if if analyzed_fuzzy {
								alternative.matches_analyzed(source_terms)?
							} else {
								alternative.matches(source_terms, &token.text)?
							} {
								fuzzy_match = Some(alternative.analyzed.as_str());
								break;
							}
						}
					}
					if let Some(term) = exact_match.or(fuzzy_match) {
						found.insert(term.to_owned());
						spans_truncated |= push_trace_span(
							&mut spans,
							&mut seen_spans,
							source_span(utf16_offsets, token.start, token.end),
							max_spans,
						);
					}
					if spans_truncated && plan.record_matches(&found) {
						break;
					}
				}
			}
		}
		spans.sort_by_key(|span| (span.start, span.end));
		spans.dedup();
		Ok((spans, found, spans_truncated, analysis_truncated))
	}

	fn source_tokens(
		analyzer: &mut TextAnalyzer,
		text: &str,
		deadline: Option<Instant>,
	) -> Result<(Vec<SourceToken>, bool)> {
		let mut stream = analyzer.token_stream(text);
		let mut tokens = Vec::new();
		while stream.advance() {
			if tokens.len() % 256 == 0 {
				check_deadline(deadline)?;
			}
			if tokens.len() == MAX_TRACE_TOKENS_PER_VALUE {
				return Ok((tokens, true));
			}
			let token = stream.token();
			tokens.push(SourceToken {
				text: token.text.clone(),
				start: token.offset_from,
				end: token.offset_to,
				position: token.position,
			});
		}
		Ok((tokens, false))
	}

	fn selected_fields(&self, requested: &[String]) -> Result<Vec<&EngineField>> {
		if requested.is_empty() {
			return Ok(self.fields.iter().collect());
		}
		let mut seen = HashSet::with_capacity(requested.len());
		let mut fields = Vec::with_capacity(requested.len());
		for name in requested {
			if !seen.insert(name) {
				return Err(FulltextError::invalid(format!("duplicate search field {name}")));
			}
			let index = self
				.field_lookup
				.get(name)
				.ok_or_else(|| FulltextError::invalid(format!("unknown search field {name}")))?;
			fields.push(&self.fields[*index]);
		}
		Ok(fields)
	}

	fn query(&self, searcher: &Searcher, request: &SearchRequest) -> Result<BuiltQuery> {
		let BuiltQuery {
			query,
			clauses,
			stable_scores,
		} = self.expression_query(searcher, &request.expression)?;
		let query = self.with_candidates(query, request.candidate_ids.as_deref())?;
		Ok(BuiltQuery {
			query: self.with_filter(query, request.filter.as_ref())?,
			clauses,
			stable_scores,
		})
	}

	fn expression_query(&self, searcher: &Searcher, expression: &SearchExpression) -> Result<BuiltQuery> {
		crate::protocol::validate_search_structure(expression, self.max_query_clauses, self.fields.len())?;
		if let SearchExpression::Boost { boost, clause } = expression {
			if let SearchExpression::Clause(leaf) = clause.as_ref() {
				if matches!(leaf.mode, SearchMode::Any | SearchMode::All) {
					// Only the root has no inherited boost whose multiplication order must be preserved.
					let fields: Vec<_> = self
						.selected_fields(&leaf.fields)?
						.into_iter()
						.cloned()
						.map(|mut field| {
							field.weight *= *boost;
							field
						})
						.collect();
					let fields: Vec<_> = fields.iter().collect();
					let occur = if leaf.mode == SearchMode::Any {
						Occur::Should
					} else {
						Occur::Must
					};
					let (query, clauses, stable_scores) = self.term_query(&leaf.text, &fields, occur)?;
					self.check_total_clause_count(clauses, 1)?;
					return Ok(BuiltQuery {
						query,
						clauses: clauses + 1,
						stable_scores,
					});
				}
			}
		}
		self.expression_query_inner(searcher, expression, &mut QueryConstructionBudget::default())
	}

	fn expression_query_inner(
		&self,
		searcher: &Searcher,
		expression: &SearchExpression,
		budget: &mut QueryConstructionBudget,
	) -> Result<BuiltQuery> {
		match expression {
			SearchExpression::Clause(clause) => self.clause_query(searcher, clause, budget),
			SearchExpression::And(children) => self.boolean_query(searcher, children, Occur::Must, budget),
			SearchExpression::Or(children) => self.boolean_query(searcher, children, Occur::Should, budget),
			SearchExpression::Boost { boost, clause } => {
				let mut child = self.expression_query_inner(searcher, clause, budget)?;
				self.check_total_clause_count(child.clauses, 1)?;
				child.query = boosted(child.query, *boost);
				child.stable_scores &= *boost == 1.0;
				child.clauses += 1;
				Ok(child)
			}
			SearchExpression::Optional { required, optional } => {
				let mut clauses = optional.len() + 1;
				let required = self.expression_query_inner(searcher, required, budget)?;
				self.check_total_clause_count(clauses, required.clauses)?;
				clauses += required.clauses;
				let mut queries = Vec::with_capacity(optional.len() + 1);
				queries.push((Occur::Must, required.query));
				for child in optional {
					let child = self.expression_query_inner(searcher, child, budget)?;
					self.check_total_clause_count(clauses, child.clauses)?;
					clauses += child.clauses;
					queries.push((Occur::Should, child.query));
				}
				Ok(BuiltQuery {
					query: Box::new(BooleanQuery::new(queries)),
					clauses,
					stable_scores: false,
				})
			}
			SearchExpression::Not(child) => {
				let child = self.expression_query_inner(searcher, child, budget)?;
				self.check_total_clause_count(child.clauses, 2)?;
				Ok(BuiltQuery {
					query: Box::new(ConstScoreQuery::new(
						Box::new(BooleanQuery::new(vec![
							(Occur::Must, Box::new(tantivy::query::AllQuery)),
							(Occur::MustNot, child.query),
						])),
						0.0,
					)),
					clauses: child.clauses + 2,
					stable_scores: false,
				})
			}
		}
	}

	fn boolean_query(
		&self,
		searcher: &Searcher,
		children: &[SearchExpression],
		occur: Occur,
		budget: &mut QueryConstructionBudget,
	) -> Result<BuiltQuery> {
		let mut clauses = children.len();
		let mut queries = Vec::with_capacity(children.len());
		for child in children {
			let child = self.expression_query_inner(searcher, child, budget)?;
			self.check_total_clause_count(clauses, child.clauses)?;
			clauses += child.clauses;
			queries.push((occur, child.query));
		}
		Ok(BuiltQuery {
			query: Box::new(BooleanQuery::new(queries)),
			clauses,
			stable_scores: children.len() > if occur == Occur::Must { 2 } else { 1 },
		})
	}

	fn clause_query(
		&self,
		searcher: &Searcher,
		clause: &SearchClause,
		budget: &mut QueryConstructionBudget,
	) -> Result<BuiltQuery> {
		let fields = self.selected_fields(&clause.fields)?;
		let (query, clauses, stable_scores) = match clause.mode {
			SearchMode::Any => self.term_query(&clause.text, &fields, Occur::Should)?,
			SearchMode::All => self.term_query(&clause.text, &fields, Occur::Must)?,
			SearchMode::Phrase => self.phrase_query(&clause.text, &fields, budget)?,
			SearchMode::Prefix => self.prefix_query(searcher, &clause.text, &fields, false, clause.options, budget)?,
			SearchMode::Fuzzy => self.fuzzy_query(&clause.text, &fields, clause.options, budget)?,
			SearchMode::FuzzyPrefix => {
				self.prefix_query(searcher, &clause.text, &fields, true, clause.options, budget)?
			}
		};
		Ok(BuiltQuery {
			query,
			clauses,
			stable_scores,
		})
	}

	fn analyze(&self, text: &str, surface: bool, deduplicate: bool) -> Result<Vec<String>> {
		let mut analyzer = if surface {
			self.surface_analyzer.clone()
		} else {
			self.analyzer.clone()
		};
		let mut stream = analyzer.token_stream(text);
		let mut terms = Vec::new();
		let mut seen = HashSet::new();
		while stream.advance() {
			let token = stream.token();
			if !deduplicate || seen.insert(token.text.clone()) {
				if terms.len() == self.max_query_terms {
					return Err(FulltextError::invalid(format!(
						"search text produces more than {} terms",
						self.max_query_terms
					)));
				}
				terms.push(token.text.clone());
			}
		}
		Ok(terms)
	}

	fn analyze_positioned(&self, text: &str) -> Result<Vec<(usize, String)>> {
		let mut analyzer = self.analyzer.clone();
		let mut stream = analyzer.token_stream(text);
		let mut terms = Vec::new();
		while stream.advance() {
			let token = stream.token();
			terms.push((token.position, token.text.clone()));
			if terms.len() > self.max_query_terms {
				return Err(FulltextError::invalid(format!(
					"search text produces more than {} terms",
					self.max_query_terms
				)));
			}
		}
		Ok(terms)
	}

	fn term_query(&self, text: &str, fields: &[&EngineField], occur: Occur) -> Result<(Box<dyn Query>, usize, bool)> {
		let terms = self.analyze(text, false, true)?;
		if terms.is_empty() {
			return Ok((Box::new(EmptyQuery), 0, false));
		}
		self.check_clause_count(terms.len(), fields.len())?;
		let clauses = terms.len().saturating_mul(fields.len());
		if occur == Occur::Should {
			let mut alternatives = Vec::with_capacity(clauses);
			for term in terms {
				for field in fields {
					let query: Box<dyn Query> = Box::new(TermQuery::new(
						Term::from_field_text(field.field, &term),
						IndexRecordOption::WithFreqs,
					));
					alternatives.push((Occur::Should, boosted(query, field.weight)));
				}
			}
			return Ok((Box::new(BooleanQuery::new(alternatives)), clauses, clauses > 1));
		}
		let stable_scores = terms.len() > 2 && clauses > 0;
		Ok((
			Box::new(BooleanQuery::new(
				terms
					.into_iter()
					.map(|term| (occur, self.term_group(&term, fields)))
					.collect(),
			)),
			clauses,
			stable_scores,
		))
	}

	fn phrase_query(
		&self,
		text: &str,
		fields: &[&EngineField],
		budget: &mut QueryConstructionBudget,
	) -> Result<(Box<dyn Query>, usize, bool)> {
		if !self.positions {
			return Err(FulltextError::invalid("phrase search requires positions to be enabled"));
		}
		let terms = self.analyze_positioned(text)?;
		if terms.is_empty() {
			return Ok((Box::new(EmptyQuery), 0, false));
		}
		self.check_clause_count(1, fields.len())?;
		budget.phrase(terms.len(), fields.len())?;
		if terms.len() == 1 {
			return Ok((self.term_group(&terms[0].1, fields), fields.len(), fields.len() > 1));
		}
		let alternatives = fields
			.iter()
			.map(|field| {
				let first_position = terms[0].0;
				let query: Box<dyn Query> = Box::new(PhraseQuery::new_with_offset(
					terms
						.iter()
						.map(|(position, term)| (position - first_position, Term::from_field_text(field.field, term)))
						.collect(),
				));
				(Occur::Should, boosted(query, field.weight))
			})
			.collect();
		Ok((Box::new(BooleanQuery::new(alternatives)), fields.len(), false))
	}

	fn fuzzy_query(
		&self,
		text: &str,
		fields: &[&EngineField],
		options: QueryOptions,
		budget: &mut QueryConstructionBudget,
	) -> Result<(Box<dyn Query>, usize, bool)> {
		let surface = self.analyze(text, true, true)?;
		let mut pairs = Vec::new();
		for surface_term in surface {
			let analyzed = self.analyze(&surface_term, false, false)?;
			if let [analyzed_term] = analyzed.as_slice() {
				pairs.push((analyzed_term.clone(), surface_term));
			}
		}
		if pairs.is_empty() {
			return Ok((Box::new(EmptyQuery), 0, false));
		}
		let fuzzy_terms = pairs.iter().filter(|(_, term)| fuzzy_eligible(term)).count();
		if fuzzy_terms > MAX_FUZZY_TERMS {
			return Err(FulltextError::invalid(format!(
				"fuzzy search contains more than {MAX_FUZZY_TERMS} eligible terms"
			)));
		}
		self.check_clause_count(pairs.len(), fields.len().saturating_mul(3))?;
		budget.fuzzy(fuzzy_terms, fields.len(), options.fuzzy_distance)?;
		let stable_scores = pairs.len() - fuzzy_terms >= 2;
		let exact_bonus = fields.iter().map(|field| field.weight).fold(0.0f32, f32::max) * 0.25 + 1.0;
		let clause_count = pairs.len().saturating_mul(fields.len().saturating_mul(3));
		let clauses = pairs
			.into_iter()
			.map(|(analyzed, surface)| {
				let exact = self.term_group(&analyzed, fields);
				if !fuzzy_eligible(&surface) {
					return Ok((Occur::Should, exact));
				}
				let exact_branch: Box<dyn Query> = Box::new(BooleanQuery::new(vec![
					(Occur::Must, exact.box_clone()),
					(Occur::Must, Box::new(ConstScoreQuery::new(exact, exact_bonus))),
				]));
				if options.fuzzy_distance == 0 {
					return Ok((Occur::Should, exact_branch));
				}
				let fuzzy = self.fuzzy_group(&analyzed, fields, false, options)?;
				Ok((
					Occur::Should,
					Box::new(DisjunctionMaxQuery::new(vec![exact_branch, fuzzy])) as Box<dyn Query>,
				))
			})
			.collect::<Result<Vec<_>>>()?;
		Ok((Box::new(BooleanQuery::new(clauses)), clause_count, stable_scores))
	}

	fn prefix_query(
		&self,
		searcher: &Searcher,
		text: &str,
		fields: &[&EngineField],
		fuzzy: bool,
		options: QueryOptions,
		budget: &mut QueryConstructionBudget,
	) -> Result<(Box<dyn Query>, usize, bool)> {
		self.require_surface_fields(fields)?;
		if text.chars().last().is_some_and(char::is_whitespace) {
			return self.term_query(text, fields, Occur::Must);
		}
		let (completed, surface_prefix) = self.final_surface_term(text)?;
		let Some(surface_prefix) = surface_prefix else {
			return Ok((Box::new(EmptyQuery), 0, false));
		};
		let minimum = if fuzzy { 4 } else { 3 };
		if surface_prefix.chars().count() < minimum {
			return Err(FulltextError::invalid(format!(
				"{} prefix must contain at least {minimum} Unicode characters",
				if fuzzy { "fuzzy" } else { "exact" }
			)));
		}
		let completed_clause_count = completed.len().saturating_mul(fields.len());
		let fuzzy_clause_count = usize::from(fuzzy && fuzzy_eligible(&surface_prefix)) * fields.len();
		let known_clause_count = completed_clause_count.saturating_add(fuzzy_clause_count);
		if known_clause_count > self.max_query_clauses {
			return Err(FulltextError::new(
				"E_PREFIX_TOO_BROAD",
				format!("prefix query exceeds {} clauses", self.max_query_clauses),
			));
		}
		let stable_scores = completed.len() + 1 > 2 && !(fuzzy && fuzzy_eligible(&surface_prefix));
		let mut clauses = Vec::with_capacity(completed.len() + 1);
		for term in completed {
			clauses.push((Occur::Must, self.term_group(&term, fields)));
		}
		let (exact_prefix, prefix_clause_count) = self.expanded_prefix_group(
			searcher,
			&surface_prefix,
			fields,
			self.max_query_clauses - known_clause_count,
		)?;
		if completed_clause_count
			.saturating_add(prefix_clause_count)
			.saturating_add(fuzzy_clause_count)
			> self.max_query_clauses
		{
			return Err(FulltextError::new(
				"E_PREFIX_TOO_BROAD",
				format!("prefix query exceeds {} clauses", self.max_query_clauses),
			));
		}
		budget.fuzzy(
			usize::from(fuzzy && fuzzy_eligible(&surface_prefix)),
			fields.len(),
			options.fuzzy_distance,
		)?;
		let final_group = if fuzzy && fuzzy_eligible(&surface_prefix) {
			let exact_bonus = fields.iter().map(|field| field.weight).fold(0.0f32, f32::max) * 0.25 + 1.0;
			let exact_for_bonus = exact_prefix.box_clone();
			let exact_branch: Box<dyn Query> = Box::new(BooleanQuery::new(vec![
				(Occur::Must, exact_prefix),
				(
					Occur::Must,
					Box::new(ConstScoreQuery::new(exact_for_bonus, exact_bonus)),
				),
			]));
			if options.fuzzy_distance == 0 {
				exact_branch
			} else {
				Box::new(DisjunctionMaxQuery::new(vec![
					exact_branch,
					self.fuzzy_group(&surface_prefix, fields, true, options)?,
				])) as Box<dyn Query>
			}
		} else {
			exact_prefix
		};
		clauses.push((Occur::Must, final_group));
		Ok((
			Box::new(BooleanQuery::new(clauses)),
			completed_clause_count + prefix_clause_count + fuzzy_clause_count,
			stable_scores,
		))
	}

	fn final_surface_term(&self, text: &str) -> Result<(Vec<String>, Option<String>)> {
		let mut analyzer = TextAnalyzer::builder(NfkcTokenizer::default())
			.filter_dynamic(EnglishPossessiveFilter)
			.filter_dynamic(LowerCaser)
			.filter_dynamic(AsciiFoldingFilter)
			.build();
		let mut stream = analyzer.token_stream(text);
		let mut final_term = None;
		let mut final_position = None;
		while stream.advance() {
			let token = stream.token();
			final_term = Some(token.text.clone());
			final_position = Some(token.position);
		}
		if final_term.as_ref().is_some_and(|term| term.len() >= 40) {
			return Err(FulltextError::invalid(
				"the final prefix token must be shorter than 40 UTF-8 bytes",
			));
		}
		let mut seen = HashSet::new();
		let completed = match final_position {
			Some(final_position) => self
				.analyze_positioned(text)?
				.into_iter()
				.filter_map(|(position, term)| (position < final_position && seen.insert(term.clone())).then_some(term))
				.collect(),
			None => Vec::new(),
		};
		Ok((completed, final_term))
	}

	fn term_group(&self, term: &str, fields: &[&EngineField]) -> Box<dyn Query> {
		Box::new(BooleanQuery::new(
			fields
				.iter()
				.map(|field| {
					let query: Box<dyn Query> = Box::new(TermQuery::new(
						Term::from_field_text(field.field, term),
						IndexRecordOption::WithFreqs,
					));
					(Occur::Should, boosted(query, field.weight))
				})
				.collect(),
		))
	}

	fn fuzzy_group(
		&self,
		term: &str,
		fields: &[&EngineField],
		prefix: bool,
		options: QueryOptions,
	) -> Result<Box<dyn Query>> {
		let alternatives = fields
			.iter()
			.map(|field| {
				let query_field = if prefix {
					field
						.surface_field
						.ok_or_else(|| FulltextError::invalid("prefix search requires surfaceTerms to be enabled"))?
				} else {
					field.field
				};
				let term = Term::from_field_text(query_field, term);
				let query: Box<dyn Query> = if prefix {
					Box::new(FuzzyTermQuery::new_prefix(
						term,
						options.fuzzy_distance,
						options.fuzzy_transpositions,
					))
				} else {
					Box::new(FuzzyTermQuery::new(
						term,
						options.fuzzy_distance,
						options.fuzzy_transpositions,
					))
				};
				Ok(Box::new(BoostQuery::new(
					Box::new(ConstScoreQuery::new(query, 0.25)),
					field.weight,
				)) as Box<dyn Query>)
			})
			.collect::<Result<Vec<_>>>()?;
		Ok(Box::new(DisjunctionMaxQuery::new(alternatives)))
	}

	fn expanded_prefix_group(
		&self,
		searcher: &Searcher,
		prefix: &str,
		fields: &[&EngineField],
		remaining_clauses: usize,
	) -> Result<(Box<dyn Query>, usize)> {
		let mut alternatives = Vec::new();
		for field in fields {
			let surface_field = field
				.surface_field
				.ok_or_else(|| FulltextError::invalid("prefix search requires surfaceTerms to be enabled"))?;
			let mut terms = BTreeSet::new();
			for segment in searcher.segment_readers() {
				let inverted = segment.inverted_index(surface_field).map_err(index_error)?;
				let dictionary = inverted.terms();
				let mut range = dictionary.range().ge(prefix.as_bytes());
				let end = prefix_end(prefix.as_bytes());
				if let Some(end) = end.as_deref() {
					range = range.lt(end);
				}
				let mut stream = range.into_stream().map_err(storage_error)?;
				while stream.advance() {
					terms.insert(stream.key().to_vec());
					if terms.len() > MAX_PREFIX_EXPANSIONS {
						return Err(FulltextError::new(
							"E_PREFIX_TOO_BROAD",
							format!("prefix expands to more than {MAX_PREFIX_EXPANSIONS} terms"),
						));
					}
					if alternatives.len().saturating_add(terms.len()) > remaining_clauses {
						return Err(FulltextError::new(
							"E_PREFIX_TOO_BROAD",
							format!("prefix query exceeds {} clauses", self.max_query_clauses),
						));
					}
				}
			}
			for term in terms {
				let query: Box<dyn Query> = Box::new(TermQuery::new(
					Term::from_field_bytes(surface_field, &term),
					IndexRecordOption::WithFreqs,
				));
				alternatives.push(boosted(query, field.weight));
			}
		}
		let clause_count = alternatives.len();
		if alternatives.is_empty() {
			Ok((Box::new(EmptyQuery), 0))
		} else {
			Ok((Box::new(DisjunctionMaxQuery::new(alternatives)), clause_count))
		}
	}

	fn with_candidates(&self, query: Box<dyn Query>, candidate_ids: Option<&[String]>) -> Result<Box<dyn Query>> {
		let Some(candidate_ids) = candidate_ids else {
			return Ok(query);
		};
		if candidate_ids.is_empty() {
			return Ok(Box::new(EmptyQuery));
		}
		let terms = candidate_ids
			.iter()
			.collect::<HashSet<_>>()
			.into_iter()
			.map(|id| Term::from_field_text(self.id_field, id));
		let filter: Box<dyn Query> = Box::new(ConstScoreQuery::new(Box::new(TermSetQuery::new(terms)), 0.0));
		Ok(Box::new(BooleanQuery::new(vec![
			(Occur::Must, query),
			(Occur::Must, filter),
		])))
	}

	fn with_filter(&self, query: Box<dyn Query>, filter: Option<&FilterExpression>) -> Result<Box<dyn Query>> {
		let Some(filter) = filter else {
			return Ok(query);
		};
		let filter = Box::new(ConstScoreQuery::new(self.filter_query(filter)?, 0.0));
		Ok(Box::new(BooleanQuery::new(vec![
			(Occur::Must, query),
			(Occur::Must, filter),
		])))
	}

	fn filter_query(&self, expression: &FilterExpression) -> Result<Box<dyn Query>> {
		self.filter_query_inner(expression, 0, &mut 0)
	}

	fn filter_query_inner(
		&self,
		expression: &FilterExpression,
		depth: usize,
		leaves: &mut usize,
	) -> Result<Box<dyn Query>> {
		if depth > crate::protocol::MAX_EXPRESSION_DEPTH {
			return Err(FulltextError::invalid("filter expression nesting exceeds 8 levels"));
		}
		match expression {
			FilterExpression::Clause {
				field,
				comparator,
				values,
			} => {
				*leaves += 1;
				if *leaves > MAX_QUERY_CLAUSES {
					return Err(FulltextError::invalid("filter expression exceeds 256 clauses"));
				}
				self.filter_clause(field, *comparator, values)
			}
			FilterExpression::Not(child) => Ok(complement_query(self.filter_query_inner(child, depth + 1, leaves)?)),
			FilterExpression::And(children) | FilterExpression::Or(children) => {
				if children.is_empty() || children.len() > MAX_QUERY_CLAUSES {
					return Err(FulltextError::invalid(format!(
						"filter boolean expressions require 1 to {MAX_QUERY_CLAUSES} clauses"
					)));
				}
				let occur = if matches!(expression, FilterExpression::And(_)) {
					Occur::Must
				} else {
					Occur::Should
				};
				let mut queries = Vec::with_capacity(children.len());
				for child in children {
					queries.push((occur, self.filter_query_inner(child, depth + 1, leaves)?));
				}
				Ok(Box::new(BooleanQuery::new(queries)))
			}
		}
	}

	fn filter_clause(
		&self,
		name: &str,
		comparator: FilterComparator,
		values: &[FilterValue],
	) -> Result<Box<dyn Query>> {
		let expected_values = match comparator {
			FilterComparator::In => None,
			FilterComparator::Between => Some(2),
			FilterComparator::Exists | FilterComparator::Missing => Some(0),
			_ => Some(1),
		};
		if (values.is_empty() && expected_values != Some(0))
			|| values.len() > MAX_QUERY_CLAUSES
			|| expected_values.is_some_and(|count| count != values.len())
		{
			return Err(FulltextError::invalid("invalid filter value count"));
		}
		let index = self
			.filter_field_lookup
			.get(name)
			.ok_or_else(|| FulltextError::invalid(format!("unknown filter field {name}")))?;
		let field = &self.filter_fields[*index];
		if matches!(comparator, FilterComparator::Exists | FilterComparator::Missing) {
			let exists: Box<dyn Query> = if field.field_type == FilterFieldType::Boolean {
				Box::new(BooleanQuery::new(vec![
					(
						Occur::Should,
						Box::new(TermQuery::new(
							Term::from_field_bool(field.field, true),
							IndexRecordOption::Basic,
						)),
					),
					(
						Occur::Should,
						Box::new(TermQuery::new(
							Term::from_field_bool(field.field, false),
							IndexRecordOption::Basic,
						)),
					),
				]))
			} else {
				Box::new(ExistsQuery::new(
					self.index.schema().get_field_name(field.field).to_owned(),
					false,
				))
			};
			return Ok(if comparator == FilterComparator::Missing {
				complement_query(exists)
			} else {
				exists
			});
		}
		let terms = values
			.iter()
			.map(|value| filter_term(field, value))
			.collect::<Result<Vec<_>>>()?;
		match comparator {
			FilterComparator::Exists | FilterComparator::Missing => unreachable!(),
			FilterComparator::Equals => Ok(Box::new(TermQuery::new(terms[0].clone(), IndexRecordOption::Basic))),
			FilterComparator::In if terms.len() == 1 => {
				Ok(Box::new(TermQuery::new(terms[0].clone(), IndexRecordOption::Basic)))
			}
			FilterComparator::In => Ok(Box::new(TermSetQuery::new(terms))),
			FilterComparator::Lt
			| FilterComparator::Le
			| FilterComparator::Gt
			| FilterComparator::Ge
			| FilterComparator::Between => {
				if field.field_type != FilterFieldType::Number {
					return Err(FulltextError::invalid("range filters require a number field"));
				}
				let (lower, upper) = match comparator {
					FilterComparator::Lt => (Bound::Unbounded, Bound::Excluded(terms[0].clone())),
					FilterComparator::Le => (Bound::Unbounded, Bound::Included(terms[0].clone())),
					FilterComparator::Gt => (Bound::Excluded(terms[0].clone()), Bound::Unbounded),
					FilterComparator::Ge => (Bound::Included(terms[0].clone()), Bound::Unbounded),
					FilterComparator::Between => (Bound::Included(terms[0].clone()), Bound::Included(terms[1].clone())),
					_ => unreachable!(),
				};
				Ok(Box::new(RangeQuery::new(lower, upper)))
			}
		}
	}

	fn require_surface_fields(&self, fields: &[&EngineField]) -> Result<()> {
		if fields.iter().any(|field| field.surface_field.is_none()) {
			return Err(FulltextError::invalid(
				"prefix search and match tracing require surfaceTerms to be enabled",
			));
		}
		Ok(())
	}

	fn check_clause_count(&self, groups: usize, alternatives: usize) -> Result<()> {
		if groups.saturating_mul(alternatives) > self.max_query_clauses {
			return Err(FulltextError::invalid(format!(
				"search query exceeds {} clauses",
				self.max_query_clauses
			)));
		}
		Ok(())
	}

	fn check_total_clause_count(&self, clauses: usize, additional: usize) -> Result<()> {
		if clauses.saturating_add(additional) > self.max_query_clauses {
			return Err(FulltextError::invalid(format!(
				"search query exceeds {} clauses",
				self.max_query_clauses
			)));
		}
		Ok(())
	}
}

fn complement_query(query: Box<dyn Query>) -> Box<dyn Query> {
	Box::new(BooleanQuery::new(vec![
		(Occur::Must, Box::new(tantivy::query::AllQuery)),
		(Occur::MustNot, query),
	]))
}

fn filter_term(field: &EngineFilterField, value: &FilterValue) -> Result<Term> {
	match (field.field_type, value) {
		(FilterFieldType::String, FilterValue::String(value)) => {
			if value.len() > MAX_TOKEN_LEN {
				return Err(FulltextError::invalid(format!(
					"filter field {} exceeds {MAX_TOKEN_LEN} UTF-8 bytes",
					field.name
				)));
			}
			Ok(Term::from_field_text(field.field, value))
		}
		(FilterFieldType::Number, FilterValue::Number(value)) => {
			crate::request_validation::validate_filter_number(*value)?;
			Ok(Term::from_field_f64(field.field, canonical_number(*value)))
		}
		(FilterFieldType::Boolean, FilterValue::Boolean(value)) => Ok(Term::from_field_bool(field.field, *value)),
		_ => Err(FulltextError::invalid(format!(
			"filter field {} received the wrong value type",
			field.name
		))),
	}
}

fn check_deadline(deadline: Option<Instant>) -> Result<()> {
	if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
		Err(FulltextError::new("E_TIMEOUT", crate::error::SEARCH_TIMEOUT_MESSAGE))
	} else {
		Ok(())
	}
}

fn search_snapshot(segments: &BTreeMap<SegmentId, Option<Opstamp>>) -> String {
	let mut digest = Sha256::new();
	digest.update(b"fulltext-snapshot-v1\0");
	digest.update((segments.len() as u64).to_le_bytes());
	for (segment, deleted_at) in segments {
		digest.update(segment.uuid_string().as_bytes());
		digest.update([u8::from(deleted_at.is_some())]);
		if let Some(opstamp) = deleted_at {
			digest.update(opstamp.to_le_bytes());
		}
	}
	format!("{:x}", digest.finalize())
}

fn search_response_bytes(hits: &[SearchHit], header_bytes: usize) -> Result<usize> {
	hits.iter().try_fold(header_bytes, |bytes, hit| {
		if !hit.score.is_finite() {
			return Err(FulltextError::invalid(
				"search score overflow; reopen with lower field weights",
			));
		}
		bytes
			.checked_add(9 + hit.id.len() + hit.version.as_ref().map_or(0, |version| 4 + version.len()))
			.ok_or_else(|| FulltextError::new("E_RESULT_TOO_LARGE", "search response size overflow"))
	})
}

fn search_hits(searcher: &Searcher, scored_docs: Vec<StableScoreTieHit>) -> Result<Vec<SearchHit>> {
	let versions = search_hit_versions(searcher, scored_docs.iter().map(|(_, address)| *address))?;
	Ok(scored_docs
		.into_iter()
		.zip(versions)
		.map(|(((score, id), _), version)| SearchHit { id, score, version })
		.collect())
}

fn search_hit_versions(
	searcher: &Searcher,
	addresses: impl ExactSizeIterator<Item = DocAddress>,
) -> Result<Vec<Option<String>>> {
	let count = addresses.len();
	let mut hits_by_segment = HashMap::new();
	for (index, address) in addresses.enumerate() {
		hits_by_segment
			.entry(address.segment_ord)
			.or_insert_with(Vec::new)
			.push((index, address.doc_id));
	}
	let mut versions = vec![None; count];
	for (segment_ord, segment_hits) in hits_by_segment {
		let segment = &searcher.segment_readers()[segment_ord as usize];
		let Some(column) = segment.fast_fields().str(VERSION_FIELD_NAME).map_err(index_error)? else {
			continue;
		};
		let ordinals = segment_hits
			.iter()
			.enumerate()
			.filter_map(|(result_index, (_, doc_id))| {
				column.term_ords(*doc_id).next().map(|ordinal| (ordinal, result_index))
			})
			.collect();
		let values = resolve_string_ordinals(
			&column,
			ordinals,
			segment_hits.len(),
			"search hit version ordinal is missing",
		)?;
		for ((index, _), value) in segment_hits.into_iter().zip(values) {
			let Some(value) = value else {
				continue;
			};
			versions[index] = Some(
				String::from_utf8(value)
					.map_err(|_| FulltextError::new("E_NATIVE_FAILURE", "search hit version is not UTF-8"))?,
			);
		}
	}
	Ok(versions)
}

fn search_hit_ids(searcher: &Searcher, scored_docs: &[(f32, DocAddress)]) -> Result<Vec<String>> {
	let mut ids = vec![String::new(); scored_docs.len()];
	let mut hits_by_segment = HashMap::new();
	for (index, (_, address)) in scored_docs.iter().enumerate() {
		hits_by_segment
			.entry(address.segment_ord)
			.or_insert_with(Vec::new)
			.push((index, address.doc_id));
	}
	for (segment_ord, segment_hits) in hits_by_segment {
		let segment = &searcher.segment_readers()[segment_ord as usize];
		let id_column = segment
			.fast_fields()
			.str(ID_FIELD_NAME)
			.map_err(index_error)?
			.ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search segment has no ID fast field"))?;
		let id_ordinals = segment_hits
			.iter()
			.enumerate()
			.map(|(result_index, (_, doc_id))| {
				let ordinal = id_column
					.term_ords(*doc_id)
					.next()
					.ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search hit has no ID ordinal"))?;
				Ok((ordinal, result_index))
			})
			.collect::<Result<Vec<_>>>()?;
		let segment_ids = resolve_string_ordinals(
			&id_column,
			id_ordinals,
			segment_hits.len(),
			"search hit ID ordinal is missing",
		)?;
		for ((index, _), id) in segment_hits.iter().zip(segment_ids) {
			let id = id.ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search hit ID ordinal is missing"))?;
			ids[*index] = String::from_utf8(id)
				.map_err(|_| FulltextError::new("E_NATIVE_FAILURE", "search hit ID is not UTF-8"))?;
		}
	}
	Ok(ids)
}

fn resolve_string_ordinals(
	column: &StrColumn,
	mut ordinals: Vec<(u64, usize)>,
	value_count: usize,
	missing_message: &'static str,
) -> Result<Vec<Option<Vec<u8>>>> {
	ordinals.sort_unstable();
	let mut values = vec![None; value_count];
	let mut next_ordinal = 0;
	let mut extra_terms = false;
	let sorted_ordinals = ordinals.iter().map(|(ordinal, _)| *ordinal).collect::<Vec<_>>();
	let found_all = column
		.dictionary()
		.sorted_ords_to_term_cb(&sorted_ordinals, |value| {
			if let Some((_, result_index)) = ordinals.get(next_ordinal) {
				values[*result_index] = Some(value.to_vec());
			} else {
				extra_terms = true;
			}
			next_ordinal += 1;
		})
		.map_err(storage_error)?;
	if extra_terms {
		return Err(storage_error("string dictionary returned too many terms"));
	}
	if !found_all || next_ordinal != ordinals.len() {
		return Err(FulltextError::new("E_NATIVE_FAILURE", missing_message));
	}
	Ok(values)
}

fn boosted(query: Box<dyn Query>, weight: f32) -> Box<dyn Query> {
	if weight == 1.0 {
		query
	} else {
		Box::new(BoostQuery::new(query, weight))
	}
}

fn fuzzy_eligible(term: &str) -> bool {
	term.chars().count() >= 4
		&& term.chars().any(char::is_alphabetic)
		&& !term
			.chars()
			.any(|character| character.is_numeric() || matches!(character, '-' | '_' | '/'))
}

fn utf16_offsets(source: &str) -> Option<Vec<u32>> {
	if source.is_ascii() {
		return None;
	}
	let mut offsets = vec![0; source.len() + 1];
	let mut utf16_offset = 0u32;
	for (byte_offset, character) in source.char_indices() {
		offsets[byte_offset] = utf16_offset;
		utf16_offset += character.len_utf16() as u32;
		offsets[byte_offset + character.len_utf8()] = utf16_offset;
	}
	Some(offsets)
}

fn source_span(utf16_offsets: &Option<Vec<u32>>, start: usize, end: usize) -> Option<TraceSpan> {
	if start > end {
		return None;
	}
	let (start, end) = match utf16_offsets {
		Some(offsets) => (*offsets.get(start)?, *offsets.get(end)?),
		None => (u32::try_from(start).ok()?, u32::try_from(end).ok()?),
	};
	(start <= end).then_some(TraceSpan { start, end })
}

fn push_trace_span(
	spans: &mut Vec<TraceSpan>,
	seen: &mut HashSet<TraceSpan>,
	span: Option<TraceSpan>,
	max_spans: usize,
) -> bool {
	let Some(span) = span else {
		return false;
	};
	if seen.contains(&span) {
		return false;
	}
	if spans.len() < max_spans {
		seen.insert(span);
		spans.push(span);
		false
	} else {
		true
	}
}

fn within_one_edit(left: &str, right: &str) -> bool {
	let left_length = left.chars().count();
	let right_length = right.chars().count();
	if left_length.abs_diff(right_length) > 1 {
		return false;
	}
	if left == right {
		return true;
	}
	if left_length == right_length {
		let mut differences = [(0usize, '\0', '\0'); 2];
		let mut count = 0;
		for (index, (left, right)) in left.chars().zip(right.chars()).enumerate() {
			if left == right {
				continue;
			}
			if count == differences.len() {
				return false;
			}
			differences[count] = (index, left, right);
			count += 1;
		}
		return count == 1
			|| (count == 2
				&& differences[1].0 == differences[0].0 + 1
				&& differences[0].1 == differences[1].2
				&& differences[1].1 == differences[0].2);
	}
	let (shorter, longer) = if left_length < right_length {
		(left, right)
	} else {
		(right, left)
	};
	let mut shorter = shorter.chars().peekable();
	let mut longer = longer.chars().peekable();
	let mut edits = 0;
	while let (Some(short), Some(long)) = (shorter.peek(), longer.peek()) {
		if short == long {
			shorter.next();
		} else {
			edits += 1;
			if edits > 1 {
				return false;
			}
		}
		longer.next();
	}
	true
}

fn prefix_end(prefix: &[u8]) -> Option<Vec<u8>> {
	let mut end = prefix.to_vec();
	while let Some(last) = end.last_mut() {
		if *last != u8::MAX {
			*last += 1;
			return Some(end);
		}
		end.pop();
	}
	None
}

impl Writer {
	pub fn apply(&self, batch: MutationBatch) -> Result<u64> {
		let prepared = self.prepare(batch)?;
		self.apply_prepared(prepared)
	}

	pub(crate) fn prepare(&self, batch: MutationBatch) -> Result<PreparedBatch> {
		let mutation_count = batch.upserts.len() + batch.deletes.len();
		for id in &batch.deletes {
			validate_record_id(id)?;
		}
		let mut documents = Vec::with_capacity(batch.upserts.len());
		for upsert in batch.upserts {
			validate_record_id(&upsert.id)?;
			let mut seen = HashSet::with_capacity(upsert.fields.len());
			let mut document = TantivyDocument::default();
			document.add_text(self.id_field, &upsert.id);
			if let Some(version) = &upsert.version {
				document.add_text(self.version_field, version);
			}
			for (name, values) in upsert.fields {
				if !seen.insert(name.clone()) {
					return Err(FulltextError::invalid(format!("duplicate mutation field {name}")));
				}
				let index = self
					.field_lookup
					.get(&name)
					.ok_or_else(|| FulltextError::invalid(format!("unknown mutation field {name}")))?;
				for value in values {
					let field = &self.fields[*index];
					document.add_text(field.field, &value);
					if let Some(surface_field) = field.surface_field {
						document.add_text(surface_field, &value);
					}
				}
			}
			let mut seen_filters = HashSet::with_capacity(upsert.filters.len());
			for (name, values) in upsert.filters {
				if !seen_filters.insert(name.clone()) {
					return Err(FulltextError::invalid(format!("duplicate mutation filter {name}")));
				}
				let index = self.filter_field_lookup.get(&name).ok_or_else(|| {
					FulltextError::new("E_SCHEMA_MISMATCH", format!("unknown mutation filter {name}"))
				})?;
				let filter = &self.filter_fields[*index];
				let deduplicate = values.len() > 1;
				let mut seen_values = HashSet::new();
				for value in values {
					match (filter.field_type, value) {
						(FilterFieldType::String, FilterValue::String(value)) => {
							if value.len() > MAX_TOKEN_LEN {
								return Err(FulltextError::invalid(format!(
									"mutation filter {name} exceeds {MAX_TOKEN_LEN} UTF-8 bytes"
								)));
							}
							if deduplicate && seen_values.contains(value.as_str()) {
								continue;
							}
							document.add_text(filter.field, &value);
							if deduplicate {
								seen_values.insert(value);
							}
						}
						(FilterFieldType::Number, FilterValue::Number(value)) => {
							document.add_f64(filter.field, canonical_number(value));
						}
						(FilterFieldType::Boolean, FilterValue::Boolean(value)) => {
							document.add_bool(filter.field, value);
						}
						_ => {
							return Err(FulltextError::invalid(format!(
								"mutation filter {name} has the wrong value type"
							)));
						}
					}
				}
			}
			documents.push((upsert.id, document));
		}
		Ok(PreparedBatch {
			mutation_count: mutation_count as u64,
			deletes: batch.deletes,
			documents,
		})
	}

	pub(crate) fn apply_prepared(&self, prepared: PreparedBatch) -> Result<u64> {
		for id in prepared.deletes {
			self.inner.delete_term(Term::from_field_text(self.id_field, &id));
		}
		for (id, document) in prepared.documents {
			self.inner.delete_term(Term::from_field_text(self.id_field, &id));
			self.inner.add_document(document).map_err(index_error)?;
		}
		Ok(prepared.mutation_count)
	}

	pub fn commit(&mut self) -> Result<u64> {
		self.commit_with_payload(None)
	}

	pub fn commit_with_payload(&mut self, payload: Option<&str>) -> Result<u64> {
		if self.checkpoint_required && payload.is_none() {
			return Err(FulltextError::new(
				"E_CHECKPOINT_REQUIRED",
				"index has a checkpoint; use publish with a payload",
			));
		}
		if payload.is_some_and(|payload| payload.len() > MAX_COMMIT_PAYLOAD_BYTES) {
			return Err(FulltextError::invalid(format!(
				"commit payload exceeds {MAX_COMMIT_PAYLOAD_BYTES} UTF-8 bytes"
			)));
		}
		// Tantivy can detach unjoined workers when prepare_commit fails.
		let previously_unproven = self.worker_ownership_unproven;
		self.worker_ownership_unproven = true;
		let mut commit = self
			.inner
			.prepare_commit()
			.map_err(|error| commit_error("prepare_commit", error))?;
		self.worker_ownership_unproven = previously_unproven;
		if let Some(payload) = payload {
			commit.set_payload(payload);
		}
		self.checkpoint_required |= payload.is_some();
		let opstamp = commit.commit().map_err(|error| commit_error("commit", error))?;
		Ok(opstamp)
	}

	/// Discard pending mutations while keeping this writer usable. Tantivy replaces the writer
	/// without joining prior merges. For terminal shutdown, call `close` directly instead;
	/// closing a replacement cannot join the earlier writer's merges.
	pub fn rollback(&mut self) -> Result<u64> {
		self.inner.rollback().map_err(index_error)
	}

	pub fn close(self) -> Result<()> {
		let result = self.inner.wait_merging_threads().map_err(index_error);
		if self.worker_ownership_unproven {
			let mut message =
				"Tantivy commit preparation left worker ownership unproven; restart before reusing this generation"
					.to_owned();
			if let Err(error) = result {
				message.push_str(&format!("; writer join also reported: {error}"));
			}
			return Err(FulltextError::new("E_QUIESCENCE_FAILED", message));
		}
		result
	}
}

fn build_schema(config: &EngineIdentityConfig) -> Result<BuiltSchema> {
	build_schema_with_surface_record(config, IndexRecordOption::WithFreqs)
}

fn build_schema_with_surface_record(
	config: &EngineIdentityConfig,
	surface_record: IndexRecordOption,
) -> Result<BuiltSchema> {
	let mut builder = Schema::builder();
	let id_indexing = TextFieldIndexing::default()
		.set_tokenizer("raw")
		.set_index_option(IndexRecordOption::Basic)
		.set_fieldnorms(false);
	let id_options = TextOptions::default()
		.set_indexing_options(id_indexing)
		.set_fast(tantivy::tokenizer::RAW_TOKENIZER_NAME);
	let id_field = builder.add_text_field(ID_FIELD_NAME, id_options);
	let version_field = builder.add_text_field(
		VERSION_FIELD_NAME,
		TextOptions::default().set_fast(tantivy::tokenizer::RAW_TOKENIZER_NAME),
	);
	let record = if config.positions {
		IndexRecordOption::WithFreqsAndPositions
	} else {
		IndexRecordOption::WithFreqs
	};
	let mut fields = Vec::with_capacity(config.fields.len());
	for field in &config.fields {
		let indexing = TextFieldIndexing::default()
			.set_tokenizer(ANALYZER_NAME)
			.set_index_option(record);
		let options = TextOptions::default().set_indexing_options(indexing);
		let schema_field = builder.add_text_field(&field.name, options);
		let surface_field = if config.surface_terms {
			let surface_name = format!("__fulltext_surface_{}", fields.len());
			let surface_indexing = TextFieldIndexing::default()
				.set_tokenizer(SURFACE_ANALYZER_NAME)
				.set_index_option(surface_record);
			Some(builder.add_text_field(
				&surface_name,
				TextOptions::default().set_indexing_options(surface_indexing),
			))
		} else {
			None
		};
		fields.push(EngineField {
			name: field.name.clone(),
			field: schema_field,
			surface_field,
			weight: field.weight,
		});
	}
	let mut filter_fields = Vec::with_capacity(config.filter_fields.len());
	for (index, filter) in config.filter_fields.iter().enumerate() {
		let schema_name = format!("__fulltext_filter_{index}");
		let field = match filter.field_type {
			FilterFieldType::String => {
				let indexing = TextFieldIndexing::default()
					.set_tokenizer("raw")
					.set_index_option(IndexRecordOption::Basic)
					.set_fieldnorms(false);
				builder.add_text_field(
					&schema_name,
					TextOptions::default().set_indexing_options(indexing).set_fast("raw"),
				)
			}
			FilterFieldType::Number => builder.add_f64_field(&schema_name, INDEXED | FAST),
			FilterFieldType::Boolean => builder.add_bool_field(&schema_name, INDEXED),
		};
		filter_fields.push(EngineFilterField {
			name: filter.name.clone(),
			field,
			field_type: filter.field_type,
		});
	}
	Ok((builder.build(), id_field, version_field, fields, filter_fields))
}

fn canonical_number(value: f64) -> f64 {
	if value == 0.0 {
		0.0
	} else {
		value
	}
}

#[derive(Clone, Default)]
struct NfkcTokenizer {
	token: Token,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct SourceSpan {
	start: usize,
	end: usize,
}

impl SourceSpan {
	fn merge(&mut self, other: Self) {
		self.start = self.start.min(other.start);
		self.end = self.end.max(other.end);
	}
}

#[derive(Clone, Copy)]
struct MappedCharacter {
	character: char,
	span: SourceSpan,
}

enum MappedCharacters<'a> {
	Original(std::str::CharIndices<'a>),
	Normalized(MappedNfkc<'a>),
}

impl MappedCharacters<'_> {
	fn next(&mut self) -> Option<MappedCharacter> {
		match self {
			Self::Original(characters) => characters.next().map(|(start, character)| MappedCharacter {
				character,
				span: SourceSpan {
					start,
					end: start + character.len_utf8(),
				},
			}),
			Self::Normalized(characters) => characters.next(),
		}
	}
}

struct MappedNfkc<'a> {
	source: std::str::CharIndices<'a>,
	decomposed: Vec<char>,
	decomposition_pending: Vec<(u8, MappedCharacter)>,
	nonstarter_count: usize,
	composee: Option<MappedCharacter>,
	composition_pending: Vec<MappedCharacter>,
	last_combining_class: Option<u8>,
	output: Vec<MappedCharacter>,
	output_offset: usize,
	finished: bool,
}

impl<'a> MappedNfkc<'a> {
	fn new(source: &'a str) -> Self {
		Self {
			source: source.char_indices(),
			decomposed: Vec::new(),
			decomposition_pending: Vec::new(),
			nonstarter_count: 0,
			composee: None,
			composition_pending: Vec::new(),
			last_combining_class: None,
			output: Vec::new(),
			output_offset: 0,
			finished: false,
		}
	}

	fn push_decomposed(&mut self, character: MappedCharacter) {
		let combining_class = canonical_combining_class(character.character);
		if combining_class == 0 {
			self.flush_decomposition();
			self.push_recomposition(character);
		} else {
			self.decomposition_pending.push((combining_class, character));
		}
	}

	fn flush_decomposition(&mut self) {
		let mut pending = std::mem::take(&mut self.decomposition_pending);
		pending.sort_by_key(|(combining_class, _)| *combining_class);
		for (_, character) in pending.drain(..) {
			self.push_recomposition(character);
		}
		self.decomposition_pending = pending;
	}

	fn push_recomposition(&mut self, character: MappedCharacter) {
		let combining_class = canonical_combining_class(character.character);
		let Some(mut composee) = self.composee.take() else {
			if combining_class == 0 {
				self.composee = Some(character);
			} else {
				self.output.push(character);
			}
			return;
		};
		let composition = match self.last_combining_class {
			None => compose(composee.character, character.character),
			Some(last) if last < combining_class => compose(composee.character, character.character),
			Some(_) => None,
		};
		if let Some(composed) = composition {
			composee.character = composed;
			composee.span.merge(character.span);
			self.composee = Some(composee);
			return;
		}
		if combining_class == 0 {
			self.output.push(composee);
			self.output.append(&mut self.composition_pending);
			self.composee = Some(character);
			self.last_combining_class = None;
		} else {
			self.composee = Some(composee);
			self.composition_pending.push(character);
			self.last_combining_class = Some(combining_class);
		}
	}

	fn finish(&mut self) {
		self.flush_decomposition();
		if let Some(composee) = self.composee.take() {
			self.output.push(composee);
		}
		self.output.append(&mut self.composition_pending);
		self.finished = true;
	}

	fn next(&mut self) -> Option<MappedCharacter> {
		loop {
			if let Some(character) = self.output.get(self.output_offset).copied() {
				self.output_offset += 1;
				return Some(character);
			}
			self.output.clear();
			self.output_offset = 0;
			if self.finished {
				return None;
			}
			if let Some((start, source_character)) = self.source.next() {
				let span = SourceSpan {
					start,
					end: start + source_character.len_utf8(),
				};
				let mut decomposed = std::mem::take(&mut self.decomposed);
				decomposed.clear();
				decompose_compatible(source_character, |character| decomposed.push(character));
				let leading_nonstarters = decomposed
					.iter()
					.take_while(|character| canonical_combining_class(**character) != 0)
					.count();
				if self.nonstarter_count + leading_nonstarters > MAX_NONSTARTERS {
					self.push_decomposed(MappedCharacter {
						character: COMBINING_GRAPHEME_JOINER,
						span,
					});
					self.nonstarter_count = 0;
				}
				if leading_nonstarters == decomposed.len() {
					self.nonstarter_count += decomposed.len();
				} else {
					self.nonstarter_count = decomposed
						.iter()
						.rev()
						.take_while(|character| canonical_combining_class(**character) != 0)
						.count();
				}
				for character in decomposed.drain(..) {
					self.push_decomposed(MappedCharacter { character, span });
				}
				self.decomposed = decomposed;
			} else {
				self.finish();
			}
		}
	}
}

fn is_stream_safe_nfkc(text: &str) -> bool {
	if !is_nfkc(text) {
		return false;
	}
	let mut nonstarter_count = 0;
	for source_character in text.chars() {
		let mut decomposition_length = 0;
		let mut leading_nonstarters = 0;
		let mut trailing_nonstarters = 0;
		let mut saw_starter = false;
		decompose_compatible(source_character, |character| {
			decomposition_length += 1;
			if canonical_combining_class(character) == 0 {
				saw_starter = true;
				trailing_nonstarters = 0;
			} else {
				if !saw_starter {
					leading_nonstarters += 1;
				}
				trailing_nonstarters += 1;
			}
		});
		if nonstarter_count + leading_nonstarters > MAX_NONSTARTERS {
			return false;
		}
		nonstarter_count = if leading_nonstarters == decomposition_length {
			nonstarter_count + decomposition_length
		} else {
			trailing_nonstarters
		};
	}
	true
}

struct NfkcTokenStream<'a> {
	characters: MappedCharacters<'a>,
	token: &'a mut Token,
}

impl Tokenizer for NfkcTokenizer {
	type TokenStream<'a> = NfkcTokenStream<'a>;

	fn token_stream<'a>(&'a mut self, text: &'a str) -> Self::TokenStream<'a> {
		self.token.reset();
		let characters = if text.is_ascii() || is_stream_safe_nfkc(text) {
			MappedCharacters::Original(text.char_indices())
		} else {
			MappedCharacters::Normalized(MappedNfkc::new(text))
		};
		NfkcTokenStream {
			characters,
			token: &mut self.token,
		}
	}
}

impl TokenStream for NfkcTokenStream<'_> {
	fn advance(&mut self) -> bool {
		self.token.text.clear();
		self.token.position = self.token.position.wrapping_add(1);
		while let Some(character) = self.characters.next() {
			if !character.character.is_alphanumeric() {
				continue;
			}
			let mut source_span = character.span;
			let mut token_characters = 1;
			self.token.text.push(character.character);
			while let Some(character) = self.characters.next() {
				if !character.character.is_alphanumeric() {
					break;
				}
				source_span.merge(character.span);
				if token_characters < MAX_TOKEN_CHARACTERS {
					self.token.text.push(character.character);
					token_characters += 1;
				}
			}
			self.token.offset_from = source_span.start;
			self.token.offset_to = source_span.end;
			return true;
		}
		false
	}

	fn token(&self) -> &Token {
		self.token
	}

	fn token_mut(&mut self) -> &mut Token {
		self.token
	}
}

#[derive(Clone)]
struct EnglishPossessiveFilter;

impl TokenFilter for EnglishPossessiveFilter {
	type Tokenizer<T: Tokenizer> = EnglishPossessiveFilterWrapper<T>;

	fn transform<T: Tokenizer>(self, tokenizer: T) -> Self::Tokenizer<T> {
		EnglishPossessiveFilterWrapper { tokenizer }
	}
}

#[derive(Clone)]
struct EnglishPossessiveFilterWrapper<T> {
	tokenizer: T,
}

impl<T: Tokenizer> Tokenizer for EnglishPossessiveFilterWrapper<T> {
	type TokenStream<'a> = EnglishPossessiveTokenStream<'a, T::TokenStream<'a>>;

	fn token_stream<'a>(&'a mut self, text: &'a str) -> Self::TokenStream<'a> {
		EnglishPossessiveTokenStream {
			source: text,
			tail: self.tokenizer.token_stream(text),
		}
	}
}

struct EnglishPossessiveTokenStream<'a, T> {
	source: &'a str,
	tail: T,
}

impl<T: TokenStream> TokenStream for EnglishPossessiveTokenStream<'_, T> {
	fn advance(&mut self) -> bool {
		while self.tail.advance() {
			let (position, possessive) = {
				let token = self.tail.token();
				(
					token.position,
					token.text.eq_ignore_ascii_case("s") && is_possessive_suffix(self.source, token.offset_from),
				)
			};
			if possessive {
				self.tail.token_mut().position = position.wrapping_sub(1);
			} else {
				return true;
			}
		}
		false
	}

	fn token(&self) -> &Token {
		self.tail.token()
	}

	fn token_mut(&mut self) -> &mut Token {
		self.tail.token_mut()
	}
}

fn is_possessive_suffix(source: &str, offset: usize) -> bool {
	let Some(prefix) = source.get(..offset) else {
		return false;
	};
	let stem = prefix
		.strip_suffix('\'')
		.or_else(|| prefix.strip_suffix('’'))
		.or_else(|| prefix.strip_suffix('＇'));
	stem.and_then(|stem| stem.chars().next_back())
		.is_some_and(char::is_alphanumeric)
}

#[derive(Clone)]
struct SynonymFilter {
	rules: SynonymMap,
}

impl TokenFilter for SynonymFilter {
	type Tokenizer<T: Tokenizer> = SynonymFilterWrapper<T>;

	fn transform<T: Tokenizer>(self, tokenizer: T) -> Self::Tokenizer<T> {
		SynonymFilterWrapper {
			tokenizer,
			rules: self.rules,
			pending: Vec::new(),
		}
	}
}

#[derive(Clone)]
struct SynonymFilterWrapper<T> {
	tokenizer: T,
	rules: SynonymMap,
	pending: Vec<Token>,
}

impl<T: Tokenizer> Tokenizer for SynonymFilterWrapper<T> {
	type TokenStream<'a> = SynonymTokenStream<'a, T::TokenStream<'a>>;

	fn token_stream<'a>(&'a mut self, text: &'a str) -> Self::TokenStream<'a> {
		self.pending.clear();
		SynonymTokenStream {
			tail: self.tokenizer.token_stream(text),
			rules: &self.rules,
			pending: &mut self.pending,
		}
	}
}

struct SynonymTokenStream<'a, T> {
	tail: T,
	rules: &'a HashMap<String, Vec<String>>,
	pending: &'a mut Vec<Token>,
}

impl<T: TokenStream> TokenStream for SynonymTokenStream<'_, T> {
	fn advance(&mut self) -> bool {
		self.pending.pop();
		if !self.pending.is_empty() {
			return true;
		}
		if !self.tail.advance() {
			return false;
		}
		let token = self.tail.token();
		if let Some(replacements) = self.rules.get(&token.text) {
			for replacement in replacements.iter().rev() {
				let mut expanded = token.clone();
				expanded.text = replacement.clone();
				self.pending.push(expanded);
			}
			self.pending.push(token.clone());
		}
		true
	}

	fn token(&self) -> &Token {
		self.pending.last().unwrap_or_else(|| self.tail.token())
	}

	fn token_mut(&mut self) -> &mut Token {
		self.pending.last_mut().unwrap_or_else(|| self.tail.token_mut())
	}
}

fn build_analyzer(stop_words: bool, synonyms: Option<SynonymMap>) -> Result<TextAnalyzer> {
	let mut builder = TextAnalyzer::builder(NfkcTokenizer::default())
		.filter_dynamic(EnglishPossessiveFilter)
		.filter_dynamic(LowerCaser)
		.filter_dynamic(AsciiFoldingFilter)
		.filter_dynamic(RemoveLongFilter::limit(40));
	if stop_words {
		let stop_filter = StopWordFilter::new(Language::English)
			.ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "English stop words are unavailable"))?;
		builder = builder.filter_dynamic(stop_filter);
	}
	builder = builder.filter_dynamic(Stemmer::new(Language::English));
	if let Some(rules) = synonyms.filter(|rules| !rules.is_empty()) {
		builder = builder.filter_dynamic(SynonymFilter { rules });
	}
	Ok(builder.build())
}

fn build_surface_analyzer(synonyms: Option<SynonymMap>) -> TextAnalyzer {
	let mut builder = TextAnalyzer::builder(NfkcTokenizer::default())
		.filter_dynamic(EnglishPossessiveFilter)
		.filter_dynamic(LowerCaser)
		.filter_dynamic(AsciiFoldingFilter)
		.filter_dynamic(RemoveLongFilter::limit(40));
	if let Some(rules) = synonyms.filter(|rules| !rules.is_empty()) {
		builder = builder.filter_dynamic(SynonymFilter { rules });
	}
	builder.build()
}

fn canonical_identity(config: &EngineIdentityConfig) -> Result<CanonicalIdentity> {
	let mut analyzer = build_analyzer(config.stop_words, None)?;
	let mut normalizer = build_surface_analyzer(None);
	let mut canonical = config.clone();
	let mut analyzed_sources = HashSet::with_capacity(config.synonyms.len());
	let mut surface_sources = HashSet::with_capacity(config.synonyms.len());
	let mut rules = Vec::with_capacity(config.synonyms.len());
	let mut analyzed_lookup = HashMap::with_capacity(config.synonyms.len());
	let mut surface_lookup = HashMap::with_capacity(config.synonyms.len());
	for rule in &config.synonyms {
		let analyzed_source = canonical_synonym_term(&mut analyzer, &rule.source, "source")?;
		let surface_source = canonical_synonym_term(&mut normalizer, &rule.source, "source")?;
		if !analyzed_sources.insert(analyzed_source.clone()) || !surface_sources.insert(surface_source.clone()) {
			return Err(FulltextError::invalid(format!(
				"synonym source {surface_source:?} is declared more than once after analysis"
			)));
		}
		let mut analyzed_replacements = Vec::with_capacity(rule.replacements.len());
		let mut surface_replacements = Vec::with_capacity(rule.replacements.len());
		for replacement in &rule.replacements {
			let analyzed_replacement = canonical_synonym_term(&mut analyzer, replacement, "replacement")?;
			let surface_replacement = canonical_synonym_term(&mut normalizer, replacement, "replacement")?;
			if analyzed_replacement == analyzed_source || surface_replacement == surface_source {
				return Err(FulltextError::invalid(
					"a synonym replacement must differ from its source after analysis",
				));
			}
			analyzed_replacements.push(analyzed_replacement);
			surface_replacements.push(surface_replacement);
		}
		analyzed_replacements.sort();
		analyzed_replacements.dedup();
		surface_replacements.sort();
		surface_replacements.dedup();
		if analyzed_replacements.len() != rule.replacements.len()
			|| surface_replacements.len() != rule.replacements.len()
		{
			return Err(FulltextError::invalid(
				"synonym replacements must be unique after analysis",
			));
		}
		analyzed_lookup.insert(analyzed_source, analyzed_replacements);
		surface_lookup.insert(surface_source.clone(), surface_replacements.clone());
		rules.push(SynonymRule {
			source: surface_source,
			replacements: surface_replacements,
		});
	}
	rules.sort_by(|left, right| left.source.cmp(&right.source));
	canonical.synonyms = rules;
	Ok(CanonicalIdentity {
		config: canonical,
		analyzer,
		analyzed_synonyms: Arc::new(analyzed_lookup),
		surface_synonyms: Arc::new(surface_lookup),
	})
}

fn canonical_synonym_term(analyzer: &mut TextAnalyzer, text: &str, label: &str) -> Result<String> {
	let mut stream = analyzer.token_stream(text);
	if !stream.advance() {
		return Err(FulltextError::invalid(format!(
			"synonym {label} must produce exactly one analyzed term"
		)));
	}
	let term = stream.token().text.clone();
	if stream.advance() {
		return Err(FulltextError::invalid(format!(
			"synonym {label} must produce exactly one analyzed term"
		)));
	}
	Ok(term)
}

fn identity_bytes(config: &EngineIdentityConfig) -> Vec<u8> {
	let mut bytes = if config.filter_fields.is_empty() {
		b"HTFI\x06\x00".to_vec()
	} else if config
		.filter_fields
		.iter()
		.any(|field| field.field_type == FilterFieldType::String)
	{
		b"HTFI\x08\x00".to_vec()
	} else {
		b"HTFI\x07\x00".to_vec()
	};
	push_string(&mut bytes, &config.index_id);
	push_string(&mut bytes, &config.generation);
	push_string(&mut bytes, &config.analyzer);
	bytes.extend_from_slice(&[
		config.stop_words as u8,
		config.positions as u8,
		config.surface_terms as u8,
	]);
	bytes.extend_from_slice(&(config.synonyms.len() as u16).to_le_bytes());
	for rule in &config.synonyms {
		push_string(&mut bytes, &rule.source);
		bytes.extend_from_slice(&(rule.replacements.len() as u16).to_le_bytes());
		for replacement in &rule.replacements {
			push_string(&mut bytes, replacement);
		}
	}
	bytes.extend_from_slice(&(config.fields.len() as u16).to_le_bytes());
	for field in &config.fields {
		push_string(&mut bytes, &field.name);
	}
	if !config.filter_fields.is_empty() {
		bytes.extend_from_slice(&(config.filter_fields.len() as u16).to_le_bytes());
		for field in &config.filter_fields {
			push_string(&mut bytes, &field.name);
			bytes.push(match field.field_type {
				FilterFieldType::String => 0,
				FilterFieldType::Number => 1,
				FilterFieldType::Boolean => 2,
			});
		}
	}
	bytes
}

pub(crate) fn persisted_index_id(bytes: &[u8]) -> Option<&str> {
	let version = bytes.get(4..6)?;
	if bytes.get(..4)? != b"HTFI"
		|| !matches!(
			version,
			b"\x01\x00"
				| b"\x02\x00"
				| b"\x03\x00"
				| b"\x04\x00"
				| b"\x05\x00"
				| b"\x06\x00"
				| b"\x07\x00"
				| b"\x08\x00"
		) {
		return None;
	}
	let mut offset = 6;
	let index_id = take_string(bytes, &mut offset)?;
	take_string(bytes, &mut offset)?;
	take_string(bytes, &mut offset)?;
	if bytes
		.get(offset..offset.checked_add(3)?)?
		.iter()
		.any(|value| *value > 1)
	{
		return None;
	}
	offset += 3;
	if matches!(
		version,
		b"\x03\x00" | b"\x04\x00" | b"\x05\x00" | b"\x06\x00" | b"\x07\x00" | b"\x08\x00"
	) {
		let synonym_count = u16::from_le_bytes(bytes.get(offset..offset.checked_add(2)?)?.try_into().ok()?) as usize;
		offset += 2;
		for _ in 0..synonym_count {
			take_string(bytes, &mut offset)?;
			let replacement_count =
				u16::from_le_bytes(bytes.get(offset..offset.checked_add(2)?)?.try_into().ok()?) as usize;
			offset += 2;
			for _ in 0..replacement_count {
				take_string(bytes, &mut offset)?;
			}
		}
	}
	let field_count = u16::from_le_bytes(bytes.get(offset..offset.checked_add(2)?)?.try_into().ok()?) as usize;
	offset += 2;
	for _ in 0..field_count {
		take_string(bytes, &mut offset)?;
	}
	if matches!(version, b"\x05\x00" | b"\x07\x00" | b"\x08\x00") {
		let filter_count = u16::from_le_bytes(bytes.get(offset..offset.checked_add(2)?)?.try_into().ok()?) as usize;
		offset += 2;
		for _ in 0..filter_count {
			take_string(bytes, &mut offset)?;
			if *bytes.get(offset)? > 2 {
				return None;
			}
			offset += 1;
		}
	}
	(offset == bytes.len()).then_some(index_id)
}

fn take_string<'a>(bytes: &'a [u8], offset: &mut usize) -> Option<&'a str> {
	let length = u32::from_le_bytes(bytes.get(*offset..(*offset).checked_add(4)?)?.try_into().ok()?) as usize;
	*offset += 4;
	let end = (*offset).checked_add(length)?;
	let value = std::str::from_utf8(bytes.get(*offset..end)?).ok()?;
	*offset = end;
	Some(value)
}

fn push_string(bytes: &mut Vec<u8>, value: &str) {
	bytes.extend_from_slice(&(value.len() as u32).to_le_bytes());
	bytes.extend_from_slice(value.as_bytes());
}

fn storage_error(error: impl std::fmt::Display) -> FulltextError {
	FulltextError::new("E_STORAGE", error.to_string())
}

fn collection_error(error: tantivy::TantivyError) -> FulltextError {
	match &error {
		tantivy::TantivyError::InvalidArgument(message) if message == crate::error::SEARCH_TIMEOUT_MESSAGE => {
			FulltextError::new("E_TIMEOUT", crate::error::SEARCH_TIMEOUT_MESSAGE)
		}
		_ => index_error(error),
	}
}

fn commit_error(operation: &str, error: tantivy::TantivyError) -> FulltextError {
	let mut error = index_error(error);
	error.message = format!("Tantivy {operation}: {}", error.message);
	error
}

pub(crate) fn recovery_index_error(error: tantivy::TantivyError) -> FulltextError {
	match error {
		tantivy::TantivyError::DataCorruption(_) => FulltextError::new("E_INDEX_CORRUPT", error.to_string()),
		tantivy::TantivyError::IncompatibleIndex(_)
		| tantivy::TantivyError::OpenReadError(OpenReadError::IncompatibleIndex(_)) => {
			FulltextError::new("E_INDEX_FORMAT_INCOMPATIBLE", error.to_string())
		}
		tantivy::TantivyError::OpenReadError(OpenReadError::FileDoesNotExist(_)) => {
			FulltextError::new("E_INDEX_CORRUPT", error.to_string())
		}
		tantivy::TantivyError::OpenReadError(OpenReadError::IoError { ref io_error, .. })
			if matches!(
				io_error.kind(),
				std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof
			) =>
		{
			FulltextError::new("E_INDEX_CORRUPT", error.to_string())
		}
		tantivy::TantivyError::IoError(ref io_error)
			if matches!(
				io_error.kind(),
				std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof
			) =>
		{
			FulltextError::new("E_INDEX_CORRUPT", error.to_string())
		}
		other => index_error(other),
	}
}

fn validated_committed_payload(index: &Index) -> Result<Option<String>> {
	let payload = index.load_metas().map_err(recovery_index_error)?.payload;
	validated_payload(payload)
}

fn validated_payload(payload: Option<String>) -> Result<Option<String>> {
	if payload
		.as_ref()
		.is_some_and(|payload| payload.len() > MAX_COMMIT_PAYLOAD_BYTES)
	{
		return Err(FulltextError::new(
			"E_INDEX_CORRUPT",
			"the persisted commit payload exceeds the supported bound",
		));
	}
	Ok(payload)
}

#[cfg(test)]
mod tests {
	use super::*;
	use crate::protocol::{FieldConfig, FilterFieldConfig, Limits};
	use std::io;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::Arc;
	use tantivy::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteError};
	use tantivy::directory::{
		Directory, DirectoryLock, FileHandle, RamDirectory, WatchCallback, WatchHandle, WritePtr,
	};

	#[derive(Clone, Debug)]
	struct AmbiguousAtomicWriteDirectory {
		inner: RamDirectory,
		fail_next: Arc<AtomicBool>,
	}

	impl AmbiguousAtomicWriteDirectory {
		fn new() -> Self {
			Self {
				inner: RamDirectory::create(),
				fail_next: Arc::new(AtomicBool::new(false)),
			}
		}

		fn fail_after_next_atomic_write(&self) {
			self.fail_next.store(true, Ordering::Release);
		}
	}

	impl Directory for AmbiguousAtomicWriteDirectory {
		fn get_file_handle(&self, path: &Path) -> std::result::Result<Arc<dyn FileHandle>, OpenReadError> {
			self.inner.get_file_handle(path)
		}

		fn delete(&self, path: &Path) -> std::result::Result<(), DeleteError> {
			self.inner.delete(path)
		}

		fn exists(&self, path: &Path) -> std::result::Result<bool, OpenReadError> {
			self.inner.exists(path)
		}

		fn open_write(&self, path: &Path) -> std::result::Result<WritePtr, OpenWriteError> {
			self.inner.open_write(path)
		}

		fn atomic_read(&self, path: &Path) -> std::result::Result<Vec<u8>, OpenReadError> {
			self.inner.atomic_read(path)
		}

		fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
			self.inner.atomic_write(path, data)?;
			if path == Path::new(META_PATH) && self.fail_next.swap(false, Ordering::AcqRel) {
				return Err(io::Error::other("injected ambiguous atomic write"));
			}
			Ok(())
		}

		fn sync_directory(&self) -> io::Result<()> {
			self.inner.sync_directory()
		}

		fn acquire_lock(&self, lock: &tantivy::directory::Lock) -> std::result::Result<DirectoryLock, LockError> {
			self.inner.acquire_lock(lock)
		}

		fn watch(&self, callback: WatchCallback) -> tantivy::Result<WatchHandle> {
			self.inner.watch(callback)
		}
	}

	fn config() -> EngineConfig {
		EngineConfig {
			identity: EngineIdentityConfig {
				index_id: "products".to_owned(),
				generation: "one".to_owned(),
				fields: vec![
					FieldConfig {
						name: "title".to_owned(),
						weight: 3.0,
					},
					FieldConfig {
						name: "description".to_owned(),
						weight: 1.0,
					},
				],
				analyzer: ANALYZER_NAME.to_owned(),
				stop_words: true,
				positions: true,
				surface_terms: false,
				synonyms: Vec::new(),
				filter_fields: Vec::new(),
			},
			limits: Limits {
				max_query_terms: crate::protocol::MAX_QUERY_TERMS,
				max_query_clauses: MAX_QUERY_CLAUSES,
				indexing_threads: 1,
				search_threads: 2,
				writer_memory_bytes: 15_000_000,
				max_queued_commands: 8,
				max_queued_bytes: 1 << 20,
				max_batch_bytes: 1 << 20,
			},
		}
	}

	#[test]
	fn direct_boolean_traces_use_current_source_and_positive_branch_spans() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let request = SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: SearchExpression::And(vec![
				expression("amber", SearchMode::Any, vec!["title".into()]),
				SearchExpression::Not(Box::new(expression("stone", SearchMode::Any, vec!["title".into()]))),
			]),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 10,
			exact_total: false,
			budget_milliseconds: 0,
			traversal: false,
			after: None,
		};
		let result = engine
			.trace_matches(
				&request,
				&[TraceRecord {
					id: "one".into(),
					fields: vec![("title".into(), vec!["amber quartz".into()])],
				}],
				None,
			)
			.unwrap();
		assert!(result.complete);
		assert_eq!(result.records[0].id, "one");
		assert_eq!(result.records[0].values[0].spans, vec![TraceSpan { start: 0, end: 5 }]);
	}

	#[test]
	fn ascii_admission_bound_dominates_native_analyzer_token_counts() {
		let engine = Engine::open(RamDirectory::create(), &config()).unwrap();
		let mut analyzer = engine.analyzer.clone();
		for first in 0u8..=127 {
			for second in 0u8..=127 {
				let bytes = [b'a', first, second, b'b', b' ', first, b'c', second];
				let text = std::str::from_utf8(&bytes).unwrap();
				let mut stream = analyzer.token_stream(text);
				let mut actual = 0;
				while stream.advance() {
					actual += 1;
				}
				assert!(
					actual <= crate::protocol::ascii_query_term_upper(&bytes).unwrap(),
					"{bytes:?}: {actual}"
				);
			}
		}
		assert_eq!(crate::protocol::ascii_query_term_upper("ﷺ".as_bytes()), None);
	}

	#[test]
	fn query_capacity_defaults_and_configured_compiled_boundaries() {
		let words = |count: usize| (0..count).map(|i| format!("word{i}")).collect::<Vec<_>>().join(" ");
		let defaults = config();
		let engine = Engine::open(RamDirectory::create(), &defaults).unwrap();
		assert_eq!(engine.analyze(&words(64), false, true).unwrap().len(), 64);
		assert_eq!(
			engine.analyze(&words(65), false, true).unwrap_err().code,
			"E_INVALID_ARGUMENT"
		);
		assert_eq!(
			engine.analyze(&"word0 ".repeat(10_000), false, true).unwrap(),
			vec!["word0"]
		);
		let mut raised = config();
		raised.limits.max_query_terms = 256;
		let engine = Engine::open(RamDirectory::create(), &raised).unwrap();
		let reader = engine.reader().unwrap();
		assert_eq!(
			engine
				.expression_query(&reader.searcher(), &expression(&words(128), SearchMode::Any, vec![]))
				.unwrap()
				.clauses,
			256
		);
		assert_eq!(
			engine
				.expression_query(&reader.searcher(), &expression(&words(129), SearchMode::Any, vec![]))
				.err()
				.unwrap()
				.code,
			"E_INVALID_ARGUMENT"
		);
		raised.limits.max_query_clauses = 512;
		let engine = Engine::open(RamDirectory::create(), &raised).unwrap();
		let reader = engine.reader().unwrap();
		assert_eq!(
			engine
				.expression_query(&reader.searcher(), &expression(&words(129), SearchMode::Any, vec![]))
				.unwrap()
				.clauses,
			258
		);
	}

	#[test]
	fn direct_search_enforces_structure_before_analysis_or_recursive_construction() {
		let mut configuration = config();
		configuration.limits.max_query_clauses = 4096;
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let reader = engine.reader().unwrap();
		let searcher = reader.searcher();
		let empty = || expression("", SearchMode::Any, vec![]);
		let mut request = SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: SearchExpression::Or(vec![empty(); 4096]),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 10,
			exact_total: true,
			budget_milliseconds: 30_000,
			traversal: false,
			after: None,
		};
		assert_eq!(engine.search(&searcher, &request).unwrap().total, 0);
		request.expression = SearchExpression::Or(vec![expression("the", SearchMode::Any, vec![]); 4096]);
		assert_eq!(engine.search(&searcher, &request).unwrap().total, 0);
		request.expression = SearchExpression::Or(vec![empty(); 4097]);
		assert!(engine
			.search(&searcher, &request)
			.unwrap_err()
			.message
			.contains("structure exceeds 4096"));
		request.expression = SearchExpression::And(vec![SearchExpression::Or(vec![empty(); 2048]); 2]);
		assert!(engine
			.search(&searcher, &request)
			.unwrap_err()
			.message
			.contains("structure exceeds 4096"));
		request.expression = SearchExpression::Or(Vec::new());
		assert!(engine
			.search(&searcher, &request)
			.unwrap_err()
			.message
			.contains("at least one child"));
		request.expression = expression(
			&"x".repeat(crate::protocol::MAX_QUERY_TEXT_BYTES + 1),
			SearchMode::Any,
			vec![],
		);
		assert!(engine
			.search(&searcher, &request)
			.unwrap_err()
			.message
			.contains("65536 UTF-8 bytes"));
		request.expression = empty();
		for _ in 0..8 {
			request.expression = SearchExpression::Not(Box::new(request.expression));
		}
		assert!(engine.search(&searcher, &request).is_ok());
		for _ in 0..9992 {
			request.expression = SearchExpression::Not(Box::new(request.expression));
		}
		assert!(engine
			.search(&searcher, &request)
			.unwrap_err()
			.message
			.contains("nesting exceeds 8"));
		while let SearchExpression::Not(child) = request.expression {
			request.expression = *child;
		}
		request.expression =
			SearchExpression::Or(vec![expression("shoe", SearchMode::Any, vec!["title".to_owned()]); 300]);
		assert!(engine.search(&searcher, &request).is_ok());
		let defaults = Engine::open(RamDirectory::create(), &config()).unwrap();
		let reader = defaults.reader().unwrap();
		assert!(defaults
			.search(&reader.searcher(), &request)
			.unwrap_err()
			.message
			.contains("structure exceeds 256"));
		for clauses in [1, 2] {
			configuration.limits.max_query_clauses = clauses;
			let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
			let reader = engine.reader().unwrap();
			request.expression = SearchExpression::Not(Box::new(empty()));
			assert_eq!(engine.search(&reader.searcher(), &request).is_ok(), clauses == 2);
		}
	}

	#[test]
	fn construction_budgets_accumulate_without_exceeding_default_implied_bounds() {
		let mut budget = QueryConstructionBudget::default();
		budget.phrase(64, 256).unwrap();
		assert_eq!(budget.phrase(1, 1).unwrap_err().code, "E_INVALID_ARGUMENT");
		budget.fuzzy(16, 16, 1).unwrap();
		assert_eq!(budget.fuzzy(1, 1, 1).unwrap_err().code, "E_INVALID_ARGUMENT");
	}

	#[test]
	fn fuzzy_distance_two_charges_measured_construction_work_across_leaves() {
		for distance in [0, 1] {
			let mut budget = QueryConstructionBudget::default();
			budget.fuzzy(16, 16, distance).unwrap();
			assert_eq!(budget.fuzzy_work_units, if distance == 0 { 0 } else { 256 });
		}
		let mut budget = QueryConstructionBudget::default();
		assert!(budget
			.fuzzy(16, 16, 2)
			.unwrap_err()
			.message
			.contains("construction work units"));
		let mut budget = QueryConstructionBudget::default();
		budget.fuzzy(4, 16, 2).unwrap();
		assert_eq!(budget.fuzzy_work_units, 256);
		let mut configuration = config();
		configuration.identity.surface_terms = true;
		configuration.identity.fields = (0..16)
			.map(|index| FieldConfig {
				name: format!("field{index}"),
				weight: 1.0,
			})
			.collect();
		configuration.limits.max_query_clauses = 4096;
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let leaf = |count: usize, mode, distance| {
			SearchExpression::Clause(SearchClause {
				text: (0..count)
					.map(|index| format!("term{}", char::from(b'a' + index as u8)))
					.collect::<Vec<_>>()
					.join(" "),
				mode,
				options: QueryOptions {
					fuzzy_distance: distance,
					..QueryOptions::default()
				},
				fields: Vec::new(),
			})
		};
		engine
			.expression_query(&searcher, &leaf(16, SearchMode::Fuzzy, 1))
			.unwrap();
		assert!(engine
			.expression_query(&searcher, &leaf(16, SearchMode::Fuzzy, 2))
			.err()
			.unwrap()
			.message
			.contains("construction work units"));
		engine
			.expression_query(&searcher, &leaf(4, SearchMode::Fuzzy, 2))
			.unwrap();
		for group in [
			SearchExpression::And as fn(Vec<SearchExpression>) -> SearchExpression,
			SearchExpression::Or,
		] {
			let mut leaves = vec![
				leaf(8, SearchMode::Fuzzy, 1),
				leaf(1, SearchMode::Fuzzy, 2),
				leaf(1, SearchMode::FuzzyPrefix, 2),
			];
			engine.expression_query(&searcher, &group(leaves.clone())).unwrap();
			leaves.push(leaf(1, SearchMode::Fuzzy, 0));
			engine.expression_query(&searcher, &group(leaves.clone())).unwrap();
			leaves.push(leaf(1, SearchMode::Fuzzy, 1));
			assert!(engine
				.expression_query(&searcher, &group(leaves))
				.err()
				.unwrap()
				.message
				.contains("construction work units"));
		}
	}

	fn expression(text: &str, mode: SearchMode, fields: Vec<String>) -> SearchExpression {
		SearchExpression::Clause(SearchClause {
			text: text.to_owned(),
			mode,
			options: QueryOptions::default(),
			fields,
		})
	}

	#[test]
	fn fuzzy_query_options_preserve_exact_priority_and_native_membership() {
		use tantivy::collector::Count;
		let mut configuration = config();
		configuration.identity.stop_words = false;
		configuration.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let mut writer = engine.index.writer_with_num_threads(1, 15_000_000).unwrap();
		for text in ["abcd", "abdc", "axyd"] {
			let mut document = TantivyDocument::default();
			document.add_text(engine.fields[0].field, text);
			document.add_text(engine.fields[0].surface_field.unwrap(), text);
			writer.add_document(document).unwrap();
		}
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let fields = vec![&engine.fields[0]];
		for prefix in [false, true] {
			for distance in 0..=2 {
				for transpositions in [false, true] {
					let options = QueryOptions {
						fuzzy_distance: distance,
						fuzzy_transpositions: transpositions,
					};
					let native_field = if prefix {
						engine.fields[0].surface_field.unwrap()
					} else {
						engine.fields[0].field
					};
					let term = Term::from_field_text(native_field, "abcd");
					let native = if prefix {
						FuzzyTermQuery::new_prefix(term, distance, transpositions)
					} else {
						FuzzyTermQuery::new(term, distance, transpositions)
					};
					let (query, _, _) = if prefix {
						engine
							.prefix_query(
								&searcher,
								"abcd",
								&fields,
								true,
								options,
								&mut QueryConstructionBudget::default(),
							)
							.unwrap()
					} else {
						engine
							.fuzzy_query("abcd", &fields, options, &mut QueryConstructionBudget::default())
							.unwrap()
					};
					let expected =
						1 + usize::from(distance >= if transpositions { 1 } else { 2 }) + usize::from(distance == 2);
					let native_count = searcher.search(&native, &Count).unwrap();
					if !prefix {
						assert_eq!(native_count, expected);
					}
					assert_eq!(searcher.search(query.as_ref(), &Count).unwrap(), native_count);
					let scored = searcher
						.search(query.as_ref(), &TopDocs::with_limit(10).order_by_score())
						.unwrap();
					assert_eq!(scored[0].1.doc_id, 0);
					if scored.len() > 1 {
						assert!(scored[0].0 > scored[1].0);
					}
				}
			}
		}
	}

	#[test]
	fn zero_distance_elides_fuzzy_construction_preserving_native_score_bits() {
		let mut configuration = config();
		configuration.identity.surface_terms = true;
		configuration.identity.fields[0].weight = 2.75;
		configuration.identity.fields[1].weight = 0.5;
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let mut writer = engine.writer(&configuration).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		for records in [
			[
				("one", "waterproof trail", "waterproof"),
				("two", "waterprof trail", "catalog"),
			],
			[
				("three", "waterproof waterproof trail", "catalog"),
				("four", "catalog trail", "waterproof"),
			],
		] {
			writer
				.apply(MutationBatch {
					upserts: records
						.into_iter()
						.map(|(id, title, description)| crate::protocol::Upsert {
							id: id.to_owned(),
							version: None,
							fields: vec![
								("title".to_owned(), vec![title.to_owned()]),
								("description".to_owned(), vec![description.to_owned()]),
							],
							filters: Vec::new(),
						})
						.collect(),
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let searcher = engine.reader().unwrap().searcher();
		assert_eq!(searcher.segment_readers().len(), 2);
		let fields = engine.fields.iter().collect::<Vec<_>>();
		let exact_bonus = fields.iter().map(|field| field.weight).fold(0.0f32, f32::max) * 0.25 + 1.0;
		let exact_with_bonus = |exact: Box<dyn Query>| -> Box<dyn Query> {
			Box::new(BooleanQuery::new(vec![
				(Occur::Must, exact.box_clone()),
				(Occur::Must, Box::new(ConstScoreQuery::new(exact, exact_bonus))),
			]))
		};
		let score_bits = |query: &dyn Query| {
			let mut hits = searcher
				.search(query, &TopDocs::with_limit(20).order_by_score())
				.unwrap()
				.into_iter()
				.map(|(score, address)| (address, score.to_bits()))
				.collect::<Vec<_>>();
			hits.sort_unstable();
			hits
		};
		for transpositions in [false, true] {
			let options = QueryOptions {
				fuzzy_distance: 0,
				fuzzy_transpositions: transpositions,
			};
			let former_group = |term: &str, prefix: bool, exact: Box<dyn Query>| -> Box<dyn Query> {
				let fuzzy = DisjunctionMaxQuery::new(
					fields
						.iter()
						.map(|field| {
							let term = Term::from_field_text(
								if prefix {
									field.surface_field.unwrap()
								} else {
									field.field
								},
								term,
							);
							let query: Box<dyn Query> = if prefix {
								Box::new(FuzzyTermQuery::new_prefix(term, 0, transpositions))
							} else {
								Box::new(FuzzyTermQuery::new(term, 0, transpositions))
							};
							Box::new(BoostQuery::new(
								Box::new(ConstScoreQuery::new(query, 0.25)),
								field.weight,
							)) as Box<dyn Query>
						})
						.collect(),
				);
				Box::new(DisjunctionMaxQuery::new(vec![exact_with_bonus(exact), Box::new(fuzzy)]))
			};
			for (prefix, text) in [
				(false, "waterproof"),
				(false, "waterproof trail"),
				(true, "waterpro"),
				(true, "trail waterpro"),
			] {
				let mut budget = QueryConstructionBudget::default();
				let (current, _, _) = if prefix {
					engine
						.prefix_query(&searcher, text, &fields, true, options, &mut budget)
						.unwrap()
				} else {
					engine.fuzzy_query(text, &fields, options, &mut budget).unwrap()
				};
				let former: Box<dyn Query> = if prefix {
					let (completed, prefix) = engine.final_surface_term(text).unwrap();
					let prefix = prefix.unwrap();
					let (exact, _) = engine
						.expanded_prefix_group(
							&searcher,
							&prefix,
							&fields,
							MAX_QUERY_CLAUSES - (completed.len() + 1) * fields.len(),
						)
						.unwrap();
					let mut clauses = completed
						.iter()
						.map(|term| (Occur::Must, engine.term_group(term, &fields)))
						.collect::<Vec<_>>();
					clauses.push((Occur::Must, former_group(&prefix, true, exact)));
					Box::new(BooleanQuery::new(clauses))
				} else {
					let clauses = engine
						.analyze(text, true, true)
						.unwrap()
						.into_iter()
						.map(|surface| {
							let analyzed = engine.analyze(&surface, false, false).unwrap();
							let term = &analyzed[0];
							(
								Occur::Should,
								former_group(term, false, engine.term_group(term, &fields)),
							)
						})
						.collect();
					Box::new(BooleanQuery::new(clauses))
				};
				assert_eq!(
					score_bits(current.as_ref()),
					score_bits(former.as_ref()),
					"{text:?}, transpositions={transpositions}"
				);
				assert!(!score_bits(current.as_ref()).is_empty());
				assert_eq!(
					searcher.search(current.as_ref(), &tantivy::collector::Count).unwrap(),
					searcher.search(former.as_ref(), &tantivy::collector::Count).unwrap()
				);
				assert_eq!(
					budget.fuzzy_work_units, 0,
					"distance zero must not construct fuzzy automata: {text:?}"
				);
				assert!(format!("{former:?}").contains("FuzzyTermQuery"));
				assert!(!format!("{current:?}").contains("FuzzyTermQuery"));
				let root = current.downcast_ref::<BooleanQuery>().unwrap();
				let branches = if prefix {
					&root.clauses()[root.clauses().len() - 1..]
				} else {
					root.clauses()
				};
				for (_, branch) in branches {
					let exact_branch = branch
						.downcast_ref::<BooleanQuery>()
						.expect("distance zero must directly retain the exact bonus branch");
					assert_eq!(exact_branch.clauses().len(), 2);
					assert!(exact_branch.clauses()[1].1.is::<ConstScoreQuery>());
				}
			}
		}
		writer.close().unwrap();
	}

	#[test]
	#[ignore = "native DFA construction qualification; run optimized with --nocapture"]
	fn measure_native_fuzzy_dfa_construction() {
		let terms = [
			"abcd".to_owned(),
			"abcdefgh".to_owned(),
			"abcdefghijklmnop".to_owned(),
			"abcdefghijklmnopqrstuvwxyz0123456789abc".to_owned(),
			"a".repeat(39),
			"éκόσμος𐐨東京café".repeat(3),
		];
		for text in terms {
			for prefix in [false, true] {
				for distance in 0..=2 {
					for transpositions in [false, true] {
						let term = Term::from_field_text(Field::from_field_id(0), &text);
						let query = if prefix {
							FuzzyTermQuery::new_prefix(term, distance, transpositions)
						} else {
							FuzzyTermQuery::new(term, distance, transpositions)
						};
						let started = Instant::now();
						let automaton = query.automaton().unwrap();
						let preparation_ms = started.elapsed().as_secs_f64() * 1000.0;
						let mut states = HashSet::from([automaton.start()]);
						let mut pending = vec![automaton.start()];
						while let Some(state) = pending.pop() {
							for byte in 0..=u8::MAX {
								let next = automaton.accept(&state, byte);
								if states.insert(next) {
									pending.push(next);
								}
							}
						}
						println!("{{\"term\":{text:?},\"chars\":{},\"bytes\":{},\"distance\":{distance},\"transpositions\":{transpositions},\"prefix\":{prefix},\"preparationMs\":{preparation_ms},\"reachableStates\":{},\"approxStateBytes\":{}}}", text.chars().count(), text.len(), states.len(), states.len() * 1024);
					}
				}
			}
		}
	}

	fn admission_request(mode: SearchMode) -> SearchRequest {
		SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: expression("alpha", mode, Vec::new()),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 10,
			exact_total: false,
			budget_milliseconds: 1000,
			traversal: false,
			after: None,
		}
	}

	#[test]
	fn direct_engine_search_rejects_decoder_semantic_bypasses() {
		let mut configuration = config();
		configuration.identity.surface_terms = true;
		configuration
			.identity
			.filter_fields
			.push(crate::protocol::FilterFieldConfig {
				name: "price".to_owned(),
				field_type: FilterFieldType::Number,
			});
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let mut invalid = Vec::new();
		for (offset, limit) in [(1, 10), (0, 101)] {
			let mut request = admission_request(SearchMode::Prefix);
			request.offset = offset;
			request.limit = limit;
			invalid.push(request);
		}
		for ids in [vec!["same".to_owned(); 1025], vec!["x".repeat(4096); 257]] {
			let mut request = admission_request(SearchMode::Any);
			request.candidate_ids = Some(ids);
			invalid.push(request);
		}
		for number in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
			let mut request = admission_request(SearchMode::Any);
			request.filter = Some(FilterExpression::Clause {
				field: "price".to_owned(),
				comparator: FilterComparator::Equals,
				values: vec![FilterValue::Number(number)],
			});
			invalid.push(request);
		}
		let mut request = admission_request(SearchMode::Any);
		request.traversal = true;
		request.after = Some(crate::protocol::SearchBoundary {
			snapshot: "stale".to_owned(),
			score: f32::NAN,
			id: "one".to_owned(),
		});
		invalid.push(request);
		for request in invalid {
			assert_eq!(
				engine.search(&searcher, &request).unwrap_err().code,
				"E_INVALID_ARGUMENT",
				"{request:?}"
			);
		}
	}

	#[test]
	fn direct_engine_trace_rejects_invalid_inputs_before_empty_candidates() {
		let mut configuration = config();
		configuration.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let record = |id: String, source: String| TraceRecord {
			id,
			fields: vec![("title".to_owned(), vec![source])],
		};
		for records in [
			(0..101)
				.map(|index| record(index.to_string(), String::new()))
				.collect::<Vec<_>>(),
			vec![record("one".to_owned(), "x".repeat((1 << 20) + 1))],
			vec![record("one".to_owned(), String::new()); 2],
		] {
			let mut request = admission_request(SearchMode::Any);
			request.candidate_ids = Some(Vec::new());
			assert_eq!(
				engine.trace_matches(&request, &records, None).unwrap_err().code,
				"E_INVALID_ARGUMENT"
			);
		}
		let mut request = admission_request(SearchMode::Any);
		request.expression = expression(&"x".repeat(65_537), SearchMode::Any, Vec::new());
		request.candidate_ids = Some(Vec::new());
		assert_eq!(
			engine.trace_matches(&request, &[], None).unwrap_err().code,
			"E_INVALID_ARGUMENT"
		);
	}

	fn batch() -> MutationBatch {
		MutationBatch {
			upserts: vec![
				crate::protocol::Upsert {
					id: "one".to_owned(),
					version: None,
					fields: vec![("title".to_owned(), vec!["Running Shoes".to_owned()])],
					filters: Vec::new(),
				},
				crate::protocol::Upsert {
					id: "two".to_owned(),
					version: None,
					fields: vec![("description".to_owned(), vec!["shoe rack".to_owned()])],
					filters: Vec::new(),
				},
			],
			deletes: Vec::new(),
		}
	}

	#[test]
	fn prepared_string_filters_preserve_first_value_order_and_reject_oversized_values() {
		use tantivy::schema::Value;
		let mut config = config();
		config.identity.filter_fields.push(FilterFieldConfig {
			name: "category".to_owned(),
			field_type: FilterFieldType::String,
		});
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let writer = engine.writer(&config).unwrap();
		let prepare = |values: Vec<String>| {
			let mut mutation = batch();
			mutation.upserts.truncate(1);
			mutation.upserts[0].filters = vec![(
				"category".to_owned(),
				values.into_iter().map(FilterValue::String).collect(),
			)];
			writer.prepare(mutation)
		};
		for (values, expected) in [
			(vec!["solo"], vec!["solo"]),
			(
				vec!["zeta", "alpha", "zeta", "", "alpha", ""],
				vec!["zeta", "alpha", ""],
			),
		] {
			let prepared = prepare(values.into_iter().map(str::to_owned).collect()).unwrap();
			let actual = prepared.documents[0]
				.1
				.get_all(engine.filter_fields[0].field)
				.map(|value| value.as_str().unwrap())
				.collect::<Vec<_>>();
			assert_eq!(actual, expected);
		}
		for values in [
			vec!["x".repeat(MAX_TOKEN_LEN + 1)],
			vec!["valid".to_owned(), "x".repeat(MAX_TOKEN_LEN + 1)],
		] {
			let error = match prepare(values) {
				Ok(_) => panic!("oversized string filter accepted"),
				Err(error) => error,
			};
			assert_eq!(error.code, "E_INVALID_ARGUMENT");
		}
		writer.close().unwrap();
	}

	#[test]
	fn direct_filter_validation_bounds_depth_and_total_leaves() {
		let mut config = config();
		config.identity.filter_fields.push(FilterFieldConfig {
			name: "category".to_owned(),
			field_type: FilterFieldType::String,
		});
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let leaf = FilterExpression::Clause {
			field: "category".to_owned(),
			comparator: FilterComparator::Exists,
			values: vec![],
		};
		assert!(engine.filter_query(&leaf).is_ok());
		let mut nested = leaf.clone();
		for _ in 0..9 {
			nested = FilterExpression::Not(Box::new(nested));
		}
		assert!(engine.filter_query(&nested).is_err());
		let oversized = FilterExpression::And(vec![
			FilterExpression::Or(vec![leaf.clone(); 129]),
			FilterExpression::Or(vec![leaf.clone(); 129]),
		]);
		assert!(engine.filter_query(&oversized).is_err());
		assert!(engine.filter_query(&FilterExpression::And(vec![])).is_err());
		assert!(engine
			.filter_query(&FilterExpression::Or(vec![leaf; MAX_QUERY_CLAUSES + 1]))
			.is_err());
	}

	#[test]
	#[ignore]
	fn traversal_many_segment_qualification() {
		for segment_count in [1, 256] {
			let config = config();
			let engine = Engine::open(RamDirectory::create(), &config).unwrap();
			let mut writer = engine.writer(&config).unwrap();
			writer
				.inner
				.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
			for segment in 0..segment_count {
				let per_segment = 512 / segment_count;
				writer
					.apply(MutationBatch {
						upserts: (segment * per_segment..(segment + 1) * per_segment)
							.map(|i| crate::protocol::Upsert {
								id: format!("{i:04}"),
								version: None,
								fields: vec![("title".to_owned(), vec!["widget catalog".to_owned()])],
								filters: Vec::new(),
							})
							.collect(),
						deletes: Vec::new(),
					})
					.unwrap();
				writer.commit().unwrap();
			}
			let reader = engine.reader().unwrap();
			let searcher = reader.searcher();
			assert_eq!(searcher.segment_readers().len(), segment_count);
			let mut request = SearchRequest {
				sort: None,
				aggregations: Vec::new(),
				expression: expression("widget catalog", SearchMode::Any, vec!["title".to_owned()]),
				candidate_ids: None,
				filter: None,
				offset: 0,
				limit: 512,
				exact_total: true,
				budget_milliseconds: 30_000,
				traversal: false,
				after: None,
			};
			let expected = engine.search(&searcher, &request).unwrap().hits;
			assert_eq!(expected.len(), 512);
			request.traversal = true;
			for page_size in [1, 32] {
				request.limit = page_size;
				let mut samples = Vec::new();
				for iteration in 0..7 {
					request.after = None;
					let started = Instant::now();
					let mut hits = Vec::new();
					loop {
						let page = engine.search(&searcher, &request).unwrap();
						assert_eq!(page.total, 512);
						let (snapshot, more) = page.traversal.unwrap();
						if more {
							let last = page.hits.last().unwrap();
							request.after = Some(crate::protocol::SearchBoundary {
								snapshot,
								score: last.score,
								id: last.id.clone(),
							});
						}
						hits.extend(page.hits);
						if !more {
							break;
						}
					}
					assert_eq!(hits, expected);
					if iteration >= 2 {
						samples.push(started.elapsed().as_secs_f64() * 1000.0);
					}
				}
				samples.sort_by(f64::total_cmp);
				println!("{{\"segments\":{segment_count},\"records\":512,\"pageSize\":{page_size},\"samples\":5,\"medianMs\":{}}}", samples[2]);
			}
			writer.close().unwrap();
		}
	}

	#[test]
	fn snapshot_digest_binds_segment_ids_and_deletion_generations() {
		let mut segments = BTreeMap::new();
		assert_eq!(
			search_snapshot(&segments),
			"1962c405bdaf56d05c10327878eb6165ba017eae08c3bda932e3d36d24406361"
		);
		let segment = SegmentId::from_uuid_string("00112233445566778899aabbccddeeff").unwrap();
		segments.insert(segment, None);
		assert_eq!(
			search_snapshot(&segments),
			"7e8c34cf1637bedc14048237a152de9ddaf3866abfad53ade87447662855a4fe"
		);
		segments.insert(segment, Some(0));
		assert_eq!(
			search_snapshot(&segments),
			"99c84530ae8d4b1b7b401d1b0b9aa3dd07da7b8faf2ee4599b872647f7f3886a"
		);
		let deleted = search_snapshot(&segments);
		segments.insert(segment, Some(1));
		assert_ne!(search_snapshot(&segments), deleted);
		let replacement = SegmentId::from_uuid_string("10112233445566778899aabbccddeeff").unwrap();
		assert_ne!(search_snapshot(&BTreeMap::from([(replacement, Some(0))])), deleted);
	}

	#[test]
	fn traversal_retains_a_merged_snapshot_and_stops_collection_on_deadline() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory, &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		for mut upsert in batch().upserts {
			upsert.version = Some("original".to_owned());
			writer
				.apply(MutationBatch {
					upserts: vec![upsert],
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let reader = engine.reader().unwrap();
		let pinned = reader.searcher();
		assert_eq!(pinned.segment_readers().len(), 2);
		let mut request = SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: expression("shoes", SearchMode::Any, Vec::new()),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 1,
			exact_total: false,
			budget_milliseconds: 30_000,
			traversal: true,
			after: None,
		};
		let first = engine.search(&pinned, &request).unwrap();
		assert_eq!(first.total, 2);
		let mut invalid = request.clone();
		invalid.limit = usize::MAX;
		assert_eq!(engine.search(&pinned, &invalid).unwrap_err().code, "E_INVALID_ARGUMENT");
		invalid.limit = 1;
		invalid.offset = 1;
		assert_eq!(engine.search(&pinned, &invalid).unwrap_err().code, "E_INVALID_ARGUMENT");
		invalid.traversal = false;
		invalid.offset = usize::MAX - 1;
		invalid.limit = 2;
		assert_eq!(engine.search(&pinned, &invalid).unwrap_err().code, "E_INVALID_ARGUMENT");
		let query = engine.query(&pinned, &request).unwrap();
		let expired = TraversalScoreTieCollector {
			inner: StableScoreTieCollector::new(0..2),
			after: None,
			deadline: Some(Instant::now() - std::time::Duration::from_millis(1)),
		};
		assert_eq!(
			pinned.search(query.query.as_ref(), &expired).unwrap().unwrap_err().code,
			"E_TIMEOUT"
		);
		let segments = engine.index.searchable_segment_ids().unwrap();
		writer.inner.merge(&segments).wait().unwrap();
		let last = first.hits.last().unwrap();
		request.after = Some(crate::protocol::SearchBoundary {
			snapshot: first.traversal.unwrap().0,
			score: last.score,
			id: last.id.clone(),
		});
		let second = engine.search(&pinned, &request).unwrap();
		assert_eq!(second.total, 2);
		assert_eq!(second.hits.len(), 1);
		assert_ne!(second.hits[0].id, last.id);
		assert_eq!(second.hits[0].version.as_deref(), Some("original"));
		reader.reload().unwrap();
		assert_eq!(reader.searcher().segment_readers().len(), 1);
		assert_eq!(
			engine.search(&reader.searcher(), &request).unwrap_err().code,
			"E_CURSOR_EXPIRED"
		);
		writer.close().unwrap();
	}

	#[test]
	fn autocomplete_traversal_matches_the_ordinary_collector_beyond_one_hundred_hits() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		for segment in 0..4 {
			let upserts = (segment * 90..(segment + 1) * 90)
				.map(|i| crate::protocol::Upsert {
					id: format!("{}{:03}", ["z", "é", "中", "😀"][i % 4], i),
					version: Some("original".to_owned()),
					fields: vec![
						(
							"title".to_owned(),
							vec![if i % 4 == 0 {
								"wodget catalog".to_owned()
							} else {
								"widget ".repeat(i % 3 + 1)
							}],
						),
						(
							"description".to_owned(),
							vec![if i % 5 == 0 { "widget" } else { "catalog" }.to_owned()],
						),
					],
					filters: Vec::new(),
				})
				.collect();
			writer
				.apply(MutationBatch {
					upserts,
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let reader = engine.reader().unwrap();
		let snapshot = reader.searcher();
		for mode in [SearchMode::Prefix, SearchMode::FuzzyPrefix] {
			let mut request = SearchRequest {
				sort: None,
				aggregations: Vec::new(),
				expression: expression("widg", mode, Vec::new()),
				candidate_ids: None,
				filter: None,
				offset: 0,
				limit: 500,
				exact_total: true,
				budget_milliseconds: 30_000,
				traversal: false,
				after: None,
			};
			let query = engine.expression_query(&snapshot, &request.expression).unwrap();
			let scored_docs = snapshot
				.search(query.query.as_ref(), &TopDocs::with_limit(500).order_by_score())
				.unwrap();
			let ids = search_hit_ids(&snapshot, &scored_docs).unwrap();
			let mut ranked = scored_docs
				.into_iter()
				.zip(ids)
				.map(|((score, address), id)| ((score, id), address))
				.collect::<Vec<_>>();
			ranked.sort_unstable_by(|((left_score, left_id), left), ((right_score, right_id), right)| {
				right_score
					.total_cmp(left_score)
					.then_with(|| left_id.cmp(right_id))
					.then_with(|| left.cmp(right))
			});
			let expected = search_hits(&snapshot, ranked).unwrap();
			assert!(expected.len() > 100);
			let identities = |hits: &[SearchHit]| {
				hits.iter()
					.map(|hit| (hit.id.clone(), hit.score.to_bits(), hit.version.clone()))
					.collect::<Vec<_>>()
			};
			request.limit = 100;
			let ordinary = engine.search(&snapshot, &request).unwrap();
			assert_eq!(ordinary.total, expected.len() as u64);
			assert_eq!(identities(&ordinary.hits), identities(&expected[..100]));
			let boundary_tie_limit = (1..=100)
				.find(|&limit| expected[limit - 1].score.to_bits() == expected[limit].score.to_bits())
				.unwrap();
			request.limit = boundary_tie_limit;
			let boundary_tie_page = engine.search(&snapshot, &request).unwrap();
			assert_eq!(boundary_tie_page.total, expected.len() as u64);
			assert_eq!(
				identities(&boundary_tie_page.hits),
				identities(&expected[..boundary_tie_limit])
			);
			request.traversal = true;
			request.limit = 17;
			let mut actual = Vec::new();
			for _ in 0..expected.len().div_ceil(request.limit) {
				let page = engine.search(&snapshot, &request).unwrap();
				assert_eq!(page.total, expected.len() as u64);
				let last = page.hits.last().unwrap();
				request.after = Some(crate::protocol::SearchBoundary {
					snapshot: page.traversal.unwrap().0,
					score: last.score,
					id: last.id.clone(),
				});
				actual.extend(page.hits);
			}
			assert_eq!(actual, expected);
			assert_eq!(identities(&actual[..100]), identities(&ordinary.hits));
			assert!(engine.search(&snapshot, &request).unwrap().hits.is_empty());
		}
		writer.close().unwrap();
	}

	#[test]
	fn surface_fields_do_not_store_positions() {
		let mut config = config();
		config.identity.surface_terms = true;
		let (schema, _, _, fields, _) = build_schema(&config.identity).unwrap();
		for field in fields {
			assert_eq!(
				schema.get_field_entry(field.field).field_type().index_record_option(),
				Some(IndexRecordOption::WithFreqsAndPositions)
			);
			assert_eq!(
				schema
					.get_field_entry(field.surface_field.unwrap())
					.field_type()
					.index_record_option(),
				Some(IndexRecordOption::WithFreqs)
			);
		}
	}

	#[test]
	fn positioned_surface_schema_requires_rebuild() {
		let directory = RamDirectory::create();
		let mut config = config();
		config.identity.surface_terms = true;
		directory
			.atomic_write(Path::new(IDENTITY_PATH), &identity_bytes(&config.identity))
			.unwrap();
		let (positioned_schema, _, _, _, _) =
			build_schema_with_surface_record(&config.identity, IndexRecordOption::WithFreqsAndPositions).unwrap();
		Index::create(directory.clone(), positioned_schema, IndexSettings::default()).unwrap();
		assert_eq!(
			Engine::inspect(directory.clone(), &config.identity).unwrap_err().code,
			"E_SCHEMA_MISMATCH"
		);
		let error = match Engine::open(directory, &config) {
			Ok(_) => panic!("positioned surface schema was accepted"),
			Err(error) => error,
		};
		assert_eq!(error.code, "E_SCHEMA_MISMATCH");
	}

	#[test]
	fn single_term_phrases_respect_the_clause_limit() {
		let mut config = config();
		config.identity.fields = (0..=MAX_QUERY_CLAUSES)
			.map(|index| FieldConfig {
				name: format!("field_{index}"),
				weight: 1.0,
			})
			.collect();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let fields = engine.fields.iter().collect::<Vec<_>>();
		assert_eq!(
			engine
				.phrase_query("shoe", &fields, &mut QueryConstructionBudget::default())
				.unwrap_err()
				.code,
			"E_INVALID_ARGUMENT"
		);
	}

	#[test]
	fn any_terms_flatten_across_fields_without_changing_matches_or_scores() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let fields = engine.fields.iter().collect::<Vec<_>>();
		let (flat_query, clauses, _) = engine.term_query("waterproof trail", &fields, Occur::Should).unwrap();
		assert_eq!(clauses, 4);
		assert_eq!(format!("{flat_query:?}").matches("BooleanQuery").count(), 1);
		let (grouped_query, _, _) = engine.term_query("waterproof trail", &fields, Occur::Must).unwrap();
		assert_eq!(format!("{grouped_query:?}").matches("BooleanQuery").count(), 3);

		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![
					crate::protocol::Upsert {
						id: "both".to_owned(),
						version: None,
						fields: vec![
							("title".to_owned(), vec!["Waterproof shell".to_owned()]),
							("description".to_owned(), vec!["Trail pack".to_owned()]),
						],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "title-only".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["Waterproof jacket".to_owned()])],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "description-only".to_owned(),
						version: None,
						fields: vec![("description".to_owned(), vec!["Trail guide".to_owned()])],
						filters: Vec::new(),
					},
				],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		let searcher = reader.searcher();
		let current = engine
			.search(
				&searcher,
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("waterproof trail", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		let legacy_query = BooleanQuery::new(
			engine
				.analyze("waterproof trail", false, true)
				.unwrap()
				.into_iter()
				.map(|term| (Occur::Should, engine.term_group(&term, &fields)))
				.collect(),
		);
		let legacy_docs = searcher
			.search(&legacy_query, &TopDocs::with_limit(10).order_by_score())
			.unwrap();
		let mut legacy = legacy_docs
			.iter()
			.zip(search_hit_ids(&searcher, &legacy_docs).unwrap())
			.map(|((score, _), id)| (id, *score))
			.collect::<Vec<_>>();
		legacy.sort_by(|left, right| right.1.total_cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
		assert_eq!(
			current.hits.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>(),
			legacy.iter().map(|(id, _)| id.as_str()).collect::<Vec<_>>()
		);
		for (current, (_, legacy_score)) in current.hits.iter().zip(legacy) {
			assert!((current.score - legacy_score).abs() <= f32::EPSILON * current.score.abs().max(1.0));
		}
		let all = engine
			.search(
				&searcher,
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("waterproof trail", SearchMode::All, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(
			all.hits.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>(),
			vec!["both"]
		);
		let completed_prefix = engine
			.search(
				&searcher,
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("waterproof trail ", SearchMode::Prefix, Vec::new()),
					candidate_ids: Some(vec!["both".to_owned()]),
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(completed_prefix.hits[0].id, "both");
		writer.close().unwrap();
	}

	#[test]
	fn any_terms_respect_the_clause_limit() {
		let mut config = config();
		config.identity.fields = (0..=MAX_QUERY_CLAUSES)
			.map(|index| FieldConfig {
				name: format!("field_{index}"),
				weight: 1.0,
			})
			.collect();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let fields = engine.fields.iter().collect::<Vec<_>>();
		assert_eq!(
			engine.term_query("shoe", &fields, Occur::Should).unwrap_err().code,
			"E_INVALID_ARGUMENT"
		);
	}

	#[test]
	fn fuzzy_prefix_matcher_prepares_only_after_exact_prefix_misses() {
		let prefix = FuzzyMatcher::new(
			"shoe",
			Field::from_field_id(0),
			crate::protocol::QueryOptions::default(),
			true,
		);
		assert!(prefix.matches("shoestring").unwrap());
		assert!(prefix.automaton.get().is_none());
		assert!(prefix.matches("sjoestring").unwrap());
		assert!(prefix.automaton.get().is_some());
	}

	#[test]
	fn fuzzy_trace_matchers_skip_zero_distance_and_share_identical_terms() {
		let field = Field::from_field_id(0);
		for prefix in [false, true] {
			let matcher = FuzzyMatcher::new(
				"shoe",
				field,
				QueryOptions {
					fuzzy_distance: 0,
					fuzzy_transpositions: false,
				},
				prefix,
			);
			assert!(matcher.matches("shoe").unwrap());
			assert_eq!(matcher.matches("shoetail").unwrap(), prefix);
			assert!(!matcher.matches("shxe").unwrap());
			assert!(matcher.automaton.get().is_none());
		}
		let options = QueryOptions {
			fuzzy_distance: 2,
			fuzzy_transpositions: true,
		};
		let term = FuzzyTraceTerm::new("shoe".to_owned(), "shoe".to_owned(), field, options);
		let native = term.matchers.as_ref().unwrap();
		assert!(native.surface.is_none());
		assert!(std::ptr::eq(native.surface_matcher(), &native.analyzed));
		assert!(native.analyzed.automaton.get().is_none());
		assert!(term.matches(Some(&["shxe"]), "unmatched").unwrap());
		let prepared = native.analyzed.automaton.get().unwrap();
		assert!(term.matches(None, "shye").unwrap());
		assert!(std::ptr::eq(
			prepared,
			native.surface_matcher().automaton.get().unwrap()
		));
		let distinct = FuzzyTraceTerm::new("shoe".to_owned(), "shoes".to_owned(), field, options);
		assert!(distinct.matchers.as_ref().unwrap().surface.is_some());
		let default = FuzzyTraceTerm::new("shoe".to_owned(), "shoes".to_owned(), field, QueryOptions::default());
		assert!(default.matchers.is_none());
	}

	#[test]
	fn fuzzy_matcher_policies_agree_with_native_dictionary() {
		let candidates = [
			"shoe", "shoes", "shxe", "sxye", "sohe", "shoetail", "shxetail", "café", "cafe", "cxfé", "cxfxtail",
			"zzzz", "",
		];
		let mut schema = Schema::builder();
		let field = schema.add_text_field(
			"term",
			TextOptions::default().set_indexing_options(
				TextFieldIndexing::default()
					.set_tokenizer("raw")
					.set_index_option(IndexRecordOption::Basic),
			),
		);
		let index = Index::create_in_ram(schema.build());
		let mut writer = index.writer_with_num_threads(1, 15_000_000).unwrap();
		for candidate in candidates {
			let mut document = TantivyDocument::default();
			document.add_text(field, candidate);
			writer.add_document(document).unwrap();
		}
		writer.commit().unwrap();
		let searcher = index.reader().unwrap().searcher();
		for text in ["shoe", "café"] {
			for prefix in [false, true] {
				for distance in 0..=2 {
					for transpositions in [false, true] {
						let options = QueryOptions {
							fuzzy_distance: distance,
							fuzzy_transpositions: transpositions,
						};
						let matcher = FuzzyMatcher::new(text, field, options, prefix);
						let term = Term::from_field_text(field, text);
						let native = if prefix {
							FuzzyTermQuery::new_prefix(term, distance, transpositions)
						} else {
							FuzzyTermQuery::new(term, distance, transpositions)
						};
						let mut expected = HashSet::new();
						for segment in searcher.segment_readers() {
							let inverted = segment.inverted_index(field).unwrap();
							let mut stream = inverted
								.terms()
								.search(native.automaton().unwrap())
								.into_stream()
								.unwrap();
							while stream.advance() {
								expected.insert(String::from_utf8(stream.key().to_vec()).unwrap());
							}
						}
						for candidate in candidates {
							assert_eq!(matcher.matches(candidate).unwrap(), expected.contains(candidate), "{text:?} {candidate:?} prefix={prefix} distance={distance} transpositions={transpositions}");
						}
						if distance == 0 {
							assert!(matcher.automaton.get().is_none());
						}
					}
				}
			}
		}
	}

	#[test]
	fn trace_fuzzy_predicates_agree_with_tantivy_dictionary_automata() {
		for text in ["shoe", "café", "κόσμος", "e\u{301}lan", "𐐨abc"] {
			let chars = text.chars().collect::<Vec<_>>();
			let mut candidates = HashSet::from([text.to_owned(), "zzzzzzzzzz".to_owned(), String::new()]);
			for index in 0..=chars.len() {
				for replacement in ['x', 'é', '𐐨'] {
					let mut inserted = chars.clone();
					inserted.insert(index, replacement);
					candidates.insert(inserted.iter().collect());
					if index < chars.len() {
						let mut changed = chars.clone();
						changed[index] = replacement;
						candidates.insert(changed.iter().collect());
					}
				}
				if index < chars.len() {
					let mut removed = chars.clone();
					removed.remove(index);
					candidates.insert(removed.iter().collect());
				}
				if index + 1 < chars.len() {
					let mut swapped = chars.clone();
					swapped.swap(index, index + 1);
					candidates.insert(swapped.iter().collect());
				}
			}
			let candidates = candidates
				.into_iter()
				.flat_map(|candidate| [candidate.clone(), format!("{candidate}tail")])
				.collect::<BTreeSet<_>>();
			let mut schema = Schema::builder();
			let field = schema.add_text_field(
				"term",
				TextOptions::default().set_indexing_options(
					TextFieldIndexing::default()
						.set_tokenizer("raw")
						.set_index_option(IndexRecordOption::Basic),
				),
			);
			let index = Index::create_in_ram(schema.build());
			let mut writer = index.writer_with_num_threads(1, 15_000_000).unwrap();
			for candidate in &candidates {
				let mut document = TantivyDocument::default();
				document.add_text(field, candidate);
				writer.add_document(document).unwrap();
			}
			writer.commit().unwrap();
			let searcher = index.reader().unwrap().searcher();
			for prefix in [false, true] {
				let term = Term::from_field_text(field, text);
				let query = if prefix {
					FuzzyTermQuery::new_prefix(term, 1, true)
				} else {
					FuzzyTermQuery::new(term, 1, true)
				};
				let mut expected = HashSet::new();
				for segment in searcher.segment_readers() {
					let inverted = segment.inverted_index(field).unwrap();
					let mut stream = inverted
						.terms()
						.search(query.automaton().unwrap())
						.into_stream()
						.unwrap();
					while stream.advance() {
						expected.insert(String::from_utf8(stream.key().to_vec()).unwrap());
					}
				}
				let matcher =
					prefix.then(|| FuzzyMatcher::new(text, field, crate::protocol::QueryOptions::default(), true));
				for candidate in &candidates {
					assert_eq!(
						match &matcher {
							Some(matcher) => matcher.matches(candidate).unwrap(),
							None => within_one_edit(text, candidate),
						},
						expected.contains(candidate),
						"{text:?} {candidate:?} prefix={prefix}"
					);
				}
			}
		}
	}

	#[test]
	fn english_analyzer_normalizes_unicode_and_possessives_with_source_offsets() {
		let mut analyzer = build_analyzer(true, None).unwrap();
		let source = "Müller's ＳＨＯＥＳ re\u{301}sume\u{301} ß";
		let mut stream = analyzer.token_stream(source);
		let mut tokens = Vec::new();
		while stream.advance() {
			tokens.push(stream.token().clone());
		}
		drop(stream);
		assert_eq!(
			tokens.iter().map(|token| token.text.as_str()).collect::<Vec<_>>(),
			["muller", "shoe", "resum", "ss"]
		);
		let resume = &tokens[2];
		assert_eq!(&source[resume.offset_from..resume.offset_to], "re\u{301}sume\u{301}");
		let mut composed = analyzer.token_stream("résumé");
		assert!(composed.advance());
		assert_eq!(composed.token().text, "resum");
		assert!(!composed.advance());
		drop(composed);
		let mut possessive = analyzer.token_stream("dog＇s shoe");
		assert!(possessive.advance());
		assert_eq!(
			(possessive.token().position, possessive.token().text.as_str()),
			(0, "dog")
		);
		assert!(possessive.advance());
		assert_eq!(
			(possessive.token().position, possessive.token().text.as_str()),
			(1, "shoe")
		);
		assert!(!possessive.advance());
		for token in tokens {
			assert!(token.offset_from < token.offset_to);
			assert!(source.get(token.offset_from..token.offset_to).is_some());
		}
		assert_eq!(
			source_span(&utf16_offsets(source), source.len() + 1, source.len() + 2),
			None
		);
	}

	#[test]
	fn nfkc_tokenizer_normalizes_across_source_character_boundaries() {
		use unicode_normalization::UnicodeNormalization;

		for source in [
			"㉠ᅡ",
			"A\u{315}\u{300}",
			"\u{1100}\u{1161}\u{11a8}",
			"\u{301}A\u{30a}",
			"ﷺ\u{301}",
		] {
			let mut mapped = MappedNfkc::new(source);
			let mut actual = String::new();
			while let Some(character) = mapped.next() {
				actual.push(character.character);
				assert!(source.get(character.span.start..character.span.end).is_some());
			}
			assert_eq!(actual, source.nfkc().collect::<String>());
		}
		let source = format!("e\u{301}{}", "\u{315}".repeat(MAX_NONSTARTERS + 1));
		let mut mapped = MappedNfkc::new(&source);
		let mut actual = String::new();
		while let Some(character) = mapped.next() {
			actual.push(character.character);
		}
		assert_eq!(actual, source.stream_safe().nfkc().collect::<String>());
		assert!(actual.contains(COMBINING_GRAPHEME_JOINER));
		let source = "㉠ᅡ";
		let mut tokenizer = NfkcTokenizer::default();
		let mut stream = tokenizer.token_stream(source);
		assert!(stream.advance());
		assert_eq!(stream.token().text, "가");
		assert_eq!(stream.token().offset_from, 0);
		assert_eq!(stream.token().offset_to, source.len());
		assert!(!stream.advance());
	}

	#[test]
	fn mapped_nfkc_matches_stream_safe_unicode_normalization() {
		use unicode_normalization::UnicodeNormalization;

		let ranges = [
			(0x20, 0x7e),
			(0xa0, 0x24f),
			(0x300, 0x36f),
			(0x590, 0x6ff),
			(0x1100, 0x11ff),
			(0x1e00, 0x1eff),
			(0x2100, 0x214f),
			(0x2460, 0x24ff),
			(0xfb00, 0xfb4f),
			(0xff00, 0xffef),
			(0x1f300, 0x1f64f),
		];
		let mut random = 0x9e37_79b9_u32;
		for _ in 0..4_096 {
			random = random.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
			let length = (random as usize % 48) + 1;
			let mut source = String::new();
			for _ in 0..length {
				random = random.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
				let (start, end) = ranges[random as usize % ranges.len()];
				random = random.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
				let character = char::from_u32(start + random % (end - start + 1)).unwrap();
				source.push(character);
			}
			let mut mapped = MappedNfkc::new(&source);
			let mut actual = String::new();
			while let Some(character) = mapped.next() {
				actual.push(character.character);
				assert!(source.get(character.span.start..character.span.end).is_some());
			}
			assert_eq!(actual, source.stream_safe().nfkc().collect::<String>());
		}

		let composed = format!("\u{e9}{}", "\u{315}".repeat(MAX_NONSTARTERS + 1));
		let decomposed = format!("e\u{301}{}", "\u{315}".repeat(MAX_NONSTARTERS + 1));
		let mut tokenizer = NfkcTokenizer::default();
		let tokens = |tokenizer: &mut NfkcTokenizer, value: &str| {
			let mut stream = tokenizer.token_stream(value);
			let mut tokens = Vec::new();
			while stream.advance() {
				tokens.push(stream.token().text.clone());
			}
			tokens
		};
		assert_eq!(tokens(&mut tokenizer, &composed), tokens(&mut tokenizer, &decomposed));
	}

	#[test]
	fn prefix_length_is_checked_after_normalization() {
		let engine = Engine::open(RamDirectory::create(), &config()).unwrap();
		let query = "Ａ".repeat(14);
		let (completed, prefix) = engine.final_surface_term(&query).unwrap();
		assert!(completed.is_empty());
		assert_eq!(prefix.unwrap(), "a".repeat(14));
		let (completed, prefix) = engine.final_surface_term("½abc").unwrap();
		assert_eq!(completed, ["1"]);
		assert_eq!(prefix.unwrap(), "2abc");
		let (_, prefix) = engine.final_surface_term(&"é".repeat(39)).unwrap();
		assert_eq!(prefix.unwrap(), "e".repeat(39));
	}

	#[test]
	fn trace_token_ceiling_marks_an_unmatched_record_incomplete() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut source = "x ".repeat(MAX_TRACE_TOKENS_PER_VALUE);
		source.push_str("needle");
		let trace = engine
			.trace_matches(
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("needle", SearchMode::Any, vec!["description".to_owned()]),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
				&[TraceRecord {
					id: "one".to_owned(),
					fields: vec![("description".to_owned(), vec![source])],
				}],
				None,
			)
			.unwrap();
		assert!(!trace.complete);
		assert!(trace.records.is_empty());
	}

	#[test]
	fn index_time_synonyms_are_canonical_persisted_and_traceable() {
		let directory = RamDirectory::create();
		let mut config = config();
		config.identity.surface_terms = true;
		config.identity.synonyms = vec![SynonymRule {
			source: "ＴＶ".to_owned(),
			replacements: vec!["telly".to_owned(), "Televisions".to_owned()],
		}];
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![
					crate::protocol::Upsert {
						id: "one".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["TV stand".to_owned()])],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "two".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["monitor stand".to_owned()])],
						filters: Vec::new(),
					},
				],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		for text in ["tv", "telly", "television"] {
			let result = engine
				.search(
					&reader.searcher(),
					&SearchRequest {
						sort: None,
						aggregations: Vec::new(),
						expression: expression(text, SearchMode::Any, Vec::new()),
						candidate_ids: None,
						filter: None,
						offset: 0,
						limit: 10,
						exact_total: true,
						budget_milliseconds: 30_000,
						traversal: false,
						after: None,
					},
				)
				.unwrap();
			assert_eq!(result.hits[0].id, "one");
		}
		let trace = engine
			.trace_matches(
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("television", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
				&[TraceRecord {
					id: "one".to_owned(),
					fields: vec![("title".to_owned(), vec!["TV stand".to_owned()])],
				}],
				None,
			)
			.unwrap();
		assert_eq!(trace.records[0].values[0].spans, [TraceSpan { start: 0, end: 2 }]);
		let synonym_score = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("stand", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap()
			.hits
			.into_iter()
			.find(|hit| hit.id == "one")
			.unwrap()
			.score;
		let mut baseline_config = config.clone();
		baseline_config.identity.index_id = "baseline".to_owned();
		baseline_config.identity.synonyms.clear();
		let baseline = Engine::open(RamDirectory::create(), &baseline_config).unwrap();
		let mut baseline_writer = baseline.writer(&baseline_config).unwrap();
		baseline_writer
			.apply(MutationBatch {
				upserts: vec![
					crate::protocol::Upsert {
						id: "one".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["TV stand".to_owned()])],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "two".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["monitor stand".to_owned()])],
						filters: Vec::new(),
					},
				],
				deletes: Vec::new(),
			})
			.unwrap();
		baseline_writer.commit().unwrap();
		let baseline_reader = baseline.reader().unwrap();
		let baseline_score = baseline
			.search(
				&baseline_reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("stand", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap()
			.hits
			.into_iter()
			.find(|hit| hit.id == "one")
			.unwrap()
			.score;
		assert!(synonym_score < baseline_score);
		baseline_writer.close().unwrap();

		writer.close().unwrap();
		let mut reordered = config.clone();
		reordered.identity.synonyms[0].replacements.reverse();
		Engine::open(directory.clone(), &reordered).unwrap();
		reordered.identity.synonyms[0].replacements[0] = "display".to_owned();
		assert_eq!(
			Engine::inspect(directory, &reordered.identity).unwrap_err().code,
			"E_IDENTITY_MISMATCH"
		);
	}

	#[test]
	fn previous_stemmer_identities_require_rebuild_before_open_or_inspection() {
		for filtered in [false, true] {
			let mut config = config();
			if filtered {
				config.identity.filter_fields.push(FilterFieldConfig {
					name: "category".to_owned(),
					field_type: FilterFieldType::String,
				});
			}
			let directory = RamDirectory::create();
			let engine = Engine::open(directory.clone(), &config).unwrap();
			let mut writer = engine.writer(&config).unwrap();
			writer.apply(batch()).unwrap();
			writer.commit().unwrap();
			writer.close().unwrap();
			drop(engine);
			let mut identity = identity_bytes(&config.identity);
			identity[4] = if filtered { 5 } else { 4 };
			assert_eq!(persisted_index_id(&identity), Some("products"));
			directory.atomic_write(Path::new(IDENTITY_PATH), &identity).unwrap();
			assert_eq!(
				Engine::inspect(directory.clone(), &config.identity).unwrap_err().code,
				"E_IDENTITY_MISMATCH"
			);
			let error = match Engine::open(directory, &config) {
				Ok(_) => panic!("old stemmer identity was opened"),
				Err(error) => error,
			};
			assert_eq!(error.code, "E_IDENTITY_MISMATCH");
		}
	}

	#[test]
	fn english_stemmer_produces_frostem_tokens() {
		let engine = Engine::open(RamDirectory::create(), &config()).unwrap();
		assert_eq!(
			engine.analyze("added acarologist", false, false).unwrap(),
			vec!["add", "acarolog"]
		);
	}

	#[test]
	fn persisted_index_id_accepts_legacy_sidecars_for_reset_only() {
		let config = config();
		for version in [b"\x02\x00".as_slice(), b"\x03\x00".as_slice()] {
			let mut identity = b"HTFI".to_vec();
			identity.extend_from_slice(version);
			push_string(&mut identity, &config.identity.index_id);
			push_string(&mut identity, &config.identity.generation);
			push_string(&mut identity, &config.identity.analyzer);
			identity.extend_from_slice(&[1, 1, 0]);
			if version == b"\x03\x00" {
				identity.extend_from_slice(&0u16.to_le_bytes());
			}
			identity.extend_from_slice(&(config.identity.fields.len() as u16).to_le_bytes());
			for field in &config.identity.fields {
				push_string(&mut identity, &field.name);
			}
			assert_eq!(persisted_index_id(&identity), Some("products"));
			let directory = RamDirectory::create();
			directory.atomic_write(Path::new(IDENTITY_PATH), &identity).unwrap();
			assert_eq!(
				Engine::inspect(directory, &config.identity).unwrap_err().code,
				"E_IDENTITY_MISMATCH"
			);
		}
	}

	#[test]
	fn indexes_searches_updates_and_reopens() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		assert_eq!(
			writer
				.apply(MutationBatch {
					upserts: Vec::new(),
					deletes: vec!["x".repeat(crate::protocol::MAX_RECORD_ID_BYTES + 1)],
				})
				.unwrap_err()
				.code,
			"E_INVALID_ARGUMENT"
		);
		assert_eq!(writer.apply(batch()).unwrap(), 2);
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		reader.reload().unwrap();
		let mut request = SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: expression("shoes", SearchMode::Any, Vec::new()),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 10,
			exact_total: true,
			budget_milliseconds: 30_000,
			traversal: false,
			after: None,
		};
		let result = engine.search(&reader.searcher(), &request).unwrap();
		assert_eq!(result.total, 2);
		assert_eq!(result.hits[0].id, "one");
		writer
			.apply(MutationBatch {
				upserts: Vec::new(),
				deletes: vec!["one".to_owned()],
			})
			.unwrap();
		writer.commit().unwrap();
		reader.reload().unwrap();
		assert_eq!(engine.search(&reader.searcher(), &request).unwrap().total, 1);
		writer.close().unwrap();
		let reopened = Engine::open(directory, &config).unwrap();
		let reopened_reader = reopened.reader().unwrap();
		assert_eq!(reopened.search(&reopened_reader.searcher(), &request).unwrap().total, 1);
		request.limit = 0;
		assert_eq!(
			reopened.search(&reopened_reader.searcher(), &request).unwrap_err().code,
			"E_INVALID_ARGUMENT"
		);
	}

	#[test]
	fn supports_structured_query_modes_and_candidate_filters() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![
					crate::protocol::Upsert {
						id: "one".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["Waterproof Trail Running Shoes".to_owned()])],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "two".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["Waterproof Road Shoes".to_owned()])],
						filters: Vec::new(),
					},
					crate::protocol::Upsert {
						id: "three".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["Wireless Headphones".to_owned()])],
						filters: Vec::new(),
					},
				],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();

		let search = |text: &str, mode: SearchMode, candidates: Option<Vec<&str>>| {
			engine
				.search(
					&reader.searcher(),
					&SearchRequest {
						sort: None,
						aggregations: Vec::new(),
						expression: expression(text, mode, Vec::new()),
						candidate_ids: candidates.map(|ids| ids.into_iter().map(str::to_owned).collect()),
						filter: None,
						offset: 0,
						limit: 10,
						exact_total: true,
						budget_milliseconds: 30_000,
						traversal: false,
						after: None,
					},
				)
				.unwrap()
		};

		assert_eq!(search("trail running", SearchMode::Phrase, None).hits[0].id, "one");
		assert_eq!(search("waterproof trai", SearchMode::Prefix, None).hits[0].id, "one");
		assert_eq!(search("waterprof", SearchMode::Fuzzy, None).total, 2);
		assert_eq!(
			search("waterproof tral", SearchMode::FuzzyPrefix, None).hits[0].id,
			"one"
		);
		assert_eq!(
			search("waterproof", SearchMode::Any, Some(vec!["two"])).hits[0].id,
			"two"
		);
		let baseline = search("waterproof", SearchMode::Any, None);
		let excluded = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: SearchExpression::And(vec![
						expression("waterproof", SearchMode::Any, Vec::new()),
						SearchExpression::Not(Box::new(expression("wireless", SearchMode::Any, Vec::new()))),
					]),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(
			excluded.hits.iter().map(|hit| (&hit.id, hit.score)).collect::<Vec<_>>(),
			baseline.hits.iter().map(|hit| (&hit.id, hit.score)).collect::<Vec<_>>()
		);
		let only_negated = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: SearchExpression::Not(Box::new(expression("wireless", SearchMode::Any, Vec::new()))),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(only_negated.total, 2);
		assert_eq!(
			only_negated
				.hits
				.iter()
				.map(|hit| (hit.id.as_str(), hit.score))
				.collect::<Vec<_>>(),
			vec![("one", 0.0), ("two", 0.0)]
		);
		assert!(search("waterproof", SearchMode::Any, Some(Vec::new())).hits.is_empty());
		assert!(search("the", SearchMode::Any, None).hits.is_empty());
		writer.close().unwrap();
	}

	#[test]
	fn rejects_structured_queries_with_excessive_aggregate_expansion() {
		let engine = Engine::open(RamDirectory::create(), &config()).unwrap();
		let reader = engine.reader().unwrap();
		let expression = SearchExpression::And(
			(0..129)
				.map(|_| expression("shoe", SearchMode::Any, Vec::new()))
				.collect(),
		);
		let error = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression,
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: false,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap_err();
		assert_eq!(error.code, "E_INVALID_ARGUMENT");
	}

	#[test]
	fn dictionary_resolution_preserves_duplicate_ordinals_and_result_order() {
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: ["Shared-Version", "Other-Version", "Shared-Version"]
					.into_iter()
					.enumerate()
					.map(|(i, version)| crate::protocol::Upsert {
						id: i.to_string(),
						version: Some(version.to_owned()),
						fields: vec![("title".to_owned(), vec!["trail shoe".to_owned()])],
						filters: Vec::new(),
					})
					.collect(),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		assert_eq!(searcher.segment_readers().len(), 1);
		let column = searcher.segment_readers()[0]
			.fast_fields()
			.str(VERSION_FIELD_NAME)
			.unwrap()
			.unwrap();
		let shared = column.dictionary().term_ord(b"Shared-Version").unwrap().unwrap();
		let other = column.dictionary().term_ord(b"Other-Version").unwrap().unwrap();
		assert_eq!(
			resolve_string_ordinals(
				&column,
				vec![(shared, 2), (other, 0), (shared, 1)],
				3,
				"missing version"
			)
			.unwrap(),
			vec![
				Some(b"Other-Version".to_vec()),
				Some(b"Shared-Version".to_vec()),
				Some(b"Shared-Version".to_vec())
			]
		);
		writer.close().unwrap();
	}

	fn assert_nested_boost_reference_scores(field_weight: f32, outer: f32, inner: f32) {
		let mut config = config();
		config.identity.fields[0].weight = field_weight;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![crate::protocol::Upsert {
					id: "matched".to_owned(),
					version: None,
					fields: vec![("title".to_owned(), vec!["amber stone".to_owned()])],
					filters: Vec::new(),
				}],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		for mode in [SearchMode::Any, SearchMode::All] {
			let leaf = expression("amber", mode, vec!["title".to_owned()]);
			let query = SearchExpression::Boost {
				boost: outer,
				clause: Box::new(SearchExpression::Boost {
					boost: inner,
					clause: Box::new(leaf.clone()),
				}),
			};
			let base = engine.expression_query(&searcher, &leaf).unwrap();
			let reference = BoostQuery::new(Box::new(BoostQuery::new(base.query, inner)), outer);
			let actual = engine.expression_query(&searcher, &query).unwrap();
			let scores = |query: &dyn Query| {
				let weight = query
					.weight(tantivy::query::EnableScoring::enabled_from_searcher(&searcher))
					.unwrap();
				searcher
					.segment_readers()
					.iter()
					.flat_map(|reader| {
						let mut scorer = weight.scorer(reader, 1.0).unwrap();
						let mut scores = Vec::new();
						while scorer.doc() != tantivy::TERMINATED {
							scores.push((scorer.doc(), scorer.score().to_bits()));
							scorer.advance();
						}
						scores
					})
					.collect::<Vec<_>>()
			};
			let expected = scores(&reference);
			assert!(!expected.is_empty());
			assert_eq!(
				scores(actual.query.as_ref()),
				expected,
				"native boost order changed for {field_weight} * {inner} * {outer}"
			);
			let mut request = admission_request(mode);
			request.expression = query;
			request.exact_total = true;
			let result = engine.search(&searcher, &request).unwrap();
			assert_eq!(result.total, 1);
			assert_eq!(result.hits[0].id, "matched");
			if outer == 0.0 {
				assert_eq!(result.hits[0].score, 0.0);
			}
		}
		writer.close().unwrap();
	}

	#[test]
	fn nested_boost_zero_preserves_native_order_without_intermediate_overflow() {
		assert_nested_boost_reference_scores(3e38, 0.0, 2.0);
	}

	#[test]
	fn nested_boost_fractional_weights_preserve_native_score_bits() {
		for weight in [0.1, 0.3, 0.7, 1.3] {
			for (outer, inner) in [(0.7, 0.3), (0.1, 0.7), (1.3, 2.7)] {
				assert_nested_boost_reference_scores(weight, outer, inner);
			}
		}
	}

	#[test]
	fn simple_boosts_preserve_term_query_topology_and_collector_route() {
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let searcher = engine.reader().unwrap().searcher();
		for mode in [SearchMode::Any, SearchMode::All] {
			for boost in [0.0, 1.0, 2.5] {
				let expression = SearchExpression::Boost {
					boost,
					clause: Box::new(SearchExpression::Clause(SearchClause {
						text: "trail".to_owned(),
						mode,
						options: QueryOptions::default(),
						fields: vec!["title".to_owned()],
					})),
				};
				let built = engine.expression_query(&searcher, &expression).unwrap();
				assert_eq!(built.clauses, 2);
				assert!(!built.stable_scores);
				let topology = format!("{:?}", built.query);
				assert!(topology.starts_with("BooleanQuery"), "{topology}");
				if mode == SearchMode::Any {
					let wide = SearchExpression::Boost {
						boost,
						clause: Box::new(SearchExpression::Clause(SearchClause {
							text: "trail waterproof shoe".to_owned(),
							mode: SearchMode::Any,
							options: QueryOptions::default(),
							fields: vec!["title".to_owned()],
						})),
					};
					let built = engine.expression_query(&searcher, &wide).unwrap();
					assert_eq!(built.clauses, 4);
					assert!(built.stable_scores);
				}
			}
		}
	}

	#[test]
	fn four_term_intersection_pruning_differs_from_scorer_but_search_preserves_bits() {
		let mut configuration = config();
		configuration.identity.fields = vec![FieldConfig {
			name: "title".to_owned(),
			weight: 0.7f32 * 0.3f32,
		}];
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let mut writer = engine.writer(&configuration).unwrap();
		let terms = ["amber", "stone", "quartz", "ruby"];
		let mut upserts = (0..127)
			.map(|id| {
				let mut words = Vec::new();
				for (position, term) in terms.iter().enumerate() {
					words.extend(std::iter::repeat_n(
						*term,
						1 + ((id * (7 + position * 4)) % (11 + position * 6)),
					));
				}
				words.extend(std::iter::repeat_n("filler", 1 + ((id * 37) % 173)));
				crate::protocol::Upsert {
					id: format!("all-{id}"),
					version: None,
					fields: vec![("title".to_owned(), vec![words.join(" ")])],
					filters: Vec::new(),
				}
			})
			.collect::<Vec<_>>();
		upserts.push(crate::protocol::Upsert {
			id: "distractor".to_owned(),
			version: None,
			fields: vec![("title".to_owned(), vec!["amber stone quartz".to_owned()])],
			filters: Vec::new(),
		});
		writer
			.apply(MutationBatch {
				upserts,
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let expression = expression(&terms.join(" "), SearchMode::All, vec!["title".to_owned()]);
		let built = engine.expression_query(&searcher, &expression).unwrap();
		let weight = built
			.query
			.weight(tantivy::query::EnableScoring::enabled_from_searcher(&searcher))
			.unwrap();
		let mut iterated = Vec::new();
		for (segment, reader) in searcher.segment_readers().iter().enumerate() {
			let mut scorer = weight.scorer(reader, 1.0).unwrap();
			while scorer.doc() != tantivy::TERMINATED {
				iterated.push((scorer.score(), DocAddress::new(segment as u32, scorer.doc())));
				scorer.advance();
			}
		}
		let ids = search_hit_ids(&searcher, &iterated).unwrap();
		let mut expected = iterated
			.into_iter()
			.zip(ids)
			.map(|((score, _), id)| (id, score))
			.collect::<Vec<_>>();
		expected
			.sort_by(|(left_id, left), (right_id, right)| right.total_cmp(left).then_with(|| left_id.cmp(right_id)));
		assert_ne!(expected[4].1, expected[5].1, "fixture cutoff must avoid tie fallback");
		let raw = searcher
			.search(built.query.as_ref(), &TopDocs::with_limit(6).order_by_score())
			.unwrap();
		let raw_ids = search_hit_ids(&searcher, &raw).unwrap();
		let raw_bits = raw
			.iter()
			.zip(raw_ids)
			.take(5)
			.map(|((score, _), id)| (id, score.to_bits()))
			.collect::<Vec<_>>();
		let expected_bits = expected
			.into_iter()
			.take(5)
			.map(|(id, score)| (id, score.to_bits()))
			.collect::<Vec<_>>();
		assert_ne!(
			raw_bits, expected_bits,
			"pinned Tantivy intersection canary no longer diverges; reassess stable routing"
		);
		assert_eq!(
			raw_bits.iter().map(|(id, _)| id).collect::<Vec<_>>(),
			expected_bits.iter().map(|(id, _)| id).collect::<Vec<_>>()
		);
		let mut request = admission_request(SearchMode::All);
		request.expression = expression;
		request.limit = 5;
		let result = engine.search(&searcher, &request).unwrap();
		let actual = result
			.hits
			.into_iter()
			.map(|hit| (hit.id, hit.score.to_bits()))
			.collect::<Vec<_>>();
		assert_eq!(
			actual, expected_bits,
			"ordinary search must preserve native scorer bits"
		);
		writer.close().unwrap();
	}

	#[test]
	fn empty_direct_rust_schema_has_no_score_sum_to_stabilize() {
		let mut configuration = config();
		configuration.identity.fields.clear();
		let engine = Engine::open(RamDirectory::create(), &configuration).unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let result = engine.search(&searcher, &admission_request(SearchMode::All)).unwrap();
		assert!(result.hits.is_empty());
		assert_eq!(result.total, 0);
	}

	#[test]
	fn stable_score_routes_preserve_safe_small_queries_and_cover_term_sums() {
		let engine = Engine::open(RamDirectory::create(), &config()).unwrap();
		let mut writer = engine.writer(&config()).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![crate::protocol::Upsert {
					id: "populated-fields".to_owned(),
					version: None,
					fields: vec![
						("title".to_owned(), vec!["amber stone quartz".to_owned()]),
						("description".to_owned(), vec!["amber stone quartz".to_owned()]),
					],
					filters: Vec::new(),
				}],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let leaf = |text, mode| expression(text, mode, vec!["title".to_owned()]);
		let all_two = leaf("amber stone", SearchMode::All);
		let all_three = leaf("amber stone quartz", SearchMode::All);
		let two_leaves = vec![leaf("amber", SearchMode::Any), leaf("stone", SearchMode::Any)];
		let three_leaves = vec![
			leaf("amber", SearchMode::Any),
			leaf("stone", SearchMode::Any),
			leaf("quartz", SearchMode::Any),
		];
		let boost = |clause| SearchExpression::Boost {
			boost: 0.3,
			clause: Box::new(clause),
		};
		for (name, expression, expected_clauses, expected_stable) in [
			("single Any", leaf("amber", SearchMode::Any), 1, false),
			("two-term Any", leaf("amber stone", SearchMode::Any), 2, true),
			("two-term All", all_two.clone(), 2, false),
			("three-term All", all_three.clone(), 3, true),
			("two-leaf AND", SearchExpression::And(two_leaves.clone()), 4, false),
			("three-leaf AND", SearchExpression::And(three_leaves.clone()), 6, true),
			(
				"single-leaf OR",
				SearchExpression::Or(vec![leaf("amber", SearchMode::Any)]),
				2,
				false,
			),
			("two-leaf OR", SearchExpression::Or(two_leaves), 4, true),
			("boosted two-term All", boost(all_two), 3, false),
			("boosted three-term All", boost(all_three.clone()), 4, true),
			("nested boosted All", boost(boost(all_three)), 5, false),
			(
				"boosted AND",
				boost(SearchExpression::And(three_leaves.clone())),
				7,
				false,
			),
			(
				"identity boosted AND",
				SearchExpression::Boost {
					boost: 1.0,
					clause: Box::new(SearchExpression::And(three_leaves)),
				},
				7,
				true,
			),
			(
				"two-term multi-field All",
				expression("amber stone", SearchMode::All, vec![]),
				4,
				false,
			),
			(
				"three-term multi-field All",
				expression("amber stone quartz", SearchMode::All, vec![]),
				6,
				true,
			),
			(
				"single-term multi-field All",
				expression("amber", SearchMode::All, vec![]),
				2,
				false,
			),
			(
				"optional root",
				SearchExpression::Optional {
					required: Box::new(leaf("amber", SearchMode::Any)),
					optional: vec![leaf("stone", SearchMode::Any)],
				},
				4,
				false,
			),
			(
				"not root",
				SearchExpression::Not(Box::new(leaf("amber stone quartz", SearchMode::All))),
				5,
				false,
			),
		] {
			let built = engine.expression_query(&searcher, &expression).unwrap();
			assert_eq!(built.clauses, expected_clauses, "{name}");
			assert_eq!(built.stable_scores, expected_stable, "{name}");
		}
		writer.close().unwrap();
	}

	#[test]
	fn query_builders_report_stable_collection_for_collapsible_term_scores() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![crate::protocol::Upsert {
					id: "terms".to_owned(),
					version: None,
					fields: vec![("title".to_owned(), vec!["amber stone quartz ruby".to_owned()])],
					filters: Vec::new(),
				}],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		for (text, mode, fields, expected) in [
			("amber", SearchMode::Phrase, vec![], true),
			("amber", SearchMode::Phrase, vec!["title".to_owned()], false),
			("amber stone", SearchMode::Phrase, vec![], false),
			("xy zz", SearchMode::Fuzzy, vec![], true),
			("amber stone", SearchMode::Fuzzy, vec![], false),
			("amber stone quartz ruby ", SearchMode::Prefix, vec![], true),
			("amber stone quartz ruby ", SearchMode::FuzzyPrefix, vec![], true),
			("amber stone quartz ruby", SearchMode::Prefix, vec![], true),
			("amber stone quartz ruby", SearchMode::FuzzyPrefix, vec![], false),
		] {
			let expression = expression(text, mode, fields);
			let built = engine.expression_query(&searcher, &expression).unwrap();
			assert_eq!(built.stable_scores, expected, "{text:?} {mode:?}");
		}
		writer.close().unwrap();
	}

	#[test]
	fn scorer_and_for_each_preserve_score_bits() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		let terms = [
			"waterproof",
			"trail",
			"running",
			"shoe",
			"forest",
			"mountain",
			"river",
			"stone",
			"winter",
			"summer",
		];
		for segment in 0..3 {
			writer
				.apply(MutationBatch {
					upserts: (0..96)
						.map(|doc| crate::protocol::Upsert {
							id: format!("{segment}-{doc}"),
							version: None,
							fields: vec![
								(
									"title".to_owned(),
									vec![terms
										.iter()
										.enumerate()
										.filter(|(term, _)| (doc + term) % 4 != 0)
										.map(|(term, text)| text.repeat((doc + term) % 3 + 1))
										.collect::<Vec<_>>()
										.join(" ")],
								),
								("description".to_owned(), vec![terms.join(" ").repeat(doc % 7 + 1)]),
							],
							filters: Vec::new(),
						})
						.collect(),
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let searcher = engine.reader().unwrap().searcher();
		for (text, mode, distance) in [
			(terms.join(" "), SearchMode::Any, 1),
			(terms.join(" "), SearchMode::All, 1),
			("waterproof trail".to_owned(), SearchMode::Phrase, 1),
			("waterproo".to_owned(), SearchMode::Prefix, 1),
			("waterproff".to_owned(), SearchMode::Fuzzy, 1),
			("waterproof trail".to_owned(), SearchMode::Fuzzy, 0),
			("trail waterpro".to_owned(), SearchMode::FuzzyPrefix, 0),
		] {
			let clause = SearchExpression::Clause(SearchClause {
				text: text.clone(),
				mode,
				options: QueryOptions {
					fuzzy_distance: distance,
					..QueryOptions::default()
				},
				fields: Vec::new(),
			});
			for clause in [
				clause.clone(),
				SearchExpression::Boost {
					boost: 0.0,
					clause: Box::new(clause.clone()),
				},
				SearchExpression::Boost {
					boost: 2.5,
					clause: Box::new(clause.clone()),
				},
				SearchExpression::Optional {
					required: Box::new(clause.clone()),
					optional: vec![clause.clone()],
				},
			] {
				let built = engine.expression_query(&searcher, &clause).unwrap();
				let reference = if let SearchExpression::Boost { boost, clause } = &clause {
					let child = engine.expression_query(&searcher, clause).unwrap();
					Some(boosted(child.query, *boost))
				} else {
					None
				};
				let weight = built
					.query
					.weight(tantivy::query::EnableScoring::enabled_from_searcher(&searcher))
					.unwrap();
				for reader in searcher.segment_readers() {
					let mut scorer = weight.scorer(reader, 1.0).unwrap();
					let mut scored = Vec::new();
					while scorer.doc() != tantivy::TERMINATED {
						scored.push((scorer.doc(), scorer.score().to_bits()));
						scorer.advance();
					}
					let mut enumerated = Vec::new();
					weight
						.for_each(reader, &mut |doc, score| enumerated.push((doc, score.to_bits())))
						.unwrap();
					assert!(!scored.is_empty(), "fixture must match {text}");
					if let Some(reference) = &reference {
						let reference_weight = reference
							.weight(tantivy::query::EnableScoring::enabled_from_searcher(&searcher))
							.unwrap();
						let mut reference_scorer = reference_weight.scorer(reader, 1.0).unwrap();
						let mut expected = Vec::new();
						while reference_scorer.doc() != tantivy::TERMINATED {
							expected.push((reference_scorer.doc(), reference_scorer.score().to_bits()));
							reference_scorer.advance();
						}
						assert_eq!(scored, expected, "lowered boost changed score bits for {text}");
					}
					scored.sort_unstable();
					enumerated.sort_unstable();
					assert_eq!(scored, enumerated, "scoring topology changed for {text}");
				}
			}
		}
		writer.close().unwrap();
	}

	#[test]
	fn orders_equal_scores_by_id_across_segments_and_pages() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		let inserted_ids = [
			"catalog-0009",
			"catalog-café",
			"catalog-0001",
			"catalog-0010",
			"catalog-café",
			"catalog-0004",
			"catalog-0007",
			"catalog-cafe",
			"catalog-0002",
			"catalog-0008",
			"catalog-0003",
			"catalog-0006",
			"catalog-0005",
			"Catalog-UPPER",
			"catalog-upper",
			"catalog-0004",
		];
		for (chunk_index, ids) in inserted_ids.chunks(7).enumerate() {
			writer
				.apply(MutationBatch {
					upserts: ids
						.iter()
						.enumerate()
						.map(|(index, id)| crate::protocol::Upsert {
							id: (*id).to_owned(),
							version: (*id == "catalog-0004").then(|| {
								if chunk_index * 7 + index + 1 == inserted_ids.len() {
									"latest"
								} else {
									"original"
								}
								.to_owned()
							}),
							fields: vec![("title".to_owned(), vec!["identical catalog text".to_owned()])],
							filters: Vec::new(),
						})
						.collect(),
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let reader = engine.reader().unwrap();
		for segment in reader.searcher().segment_readers() {
			let column = segment.fast_fields().str(ID_FIELD_NAME).unwrap().unwrap();
			let mut ids = Vec::new();
			for doc in 0..segment.max_doc() {
				let ordinal = column.ords().first(doc).unwrap();
				let mut id = String::new();
				assert!(column.ord_to_str(ordinal, &mut id).unwrap());
				ids.push((ordinal, id));
			}
			assert!(ids.len() > 1);
			ids.sort_unstable_by_key(|(ordinal, _)| *ordinal);
			assert!(ids.windows(2).all(|pair| pair[0].1.as_bytes() <= pair[1].1.as_bytes()));
		}
		let page = |offset, limit| {
			engine
				.search(
					&reader.searcher(),
					&SearchRequest {
						sort: None,
						aggregations: Vec::new(),
						expression: expression("identical catalog text", SearchMode::Any, Vec::new()),
						candidate_ids: None,
						filter: None,
						offset,
						limit,
						exact_total: false,
						budget_milliseconds: 30_000,
						traversal: false,
						after: None,
					},
				)
				.unwrap()
				.hits
		};
		let expected_ids = inserted_ids
			.into_iter()
			.collect::<BTreeSet<_>>()
			.into_iter()
			.collect::<Vec<_>>();
		let bounded = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("identical catalog text", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 3,
					exact_total: false,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(bounded.total, 3);
		assert_eq!(bounded.total_relation, TotalRelation::LowerBound);
		let exact = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("identical catalog text", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 3,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(exact.total, expected_ids.len() as u64);
		assert_eq!(exact.total_relation, TotalRelation::Exact);
		let exact_boundary_tie = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("identical", SearchMode::Any, vec!["title".to_owned()]),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 3,
					exact_total: true,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(exact_boundary_tie.total, expected_ids.len() as u64);
		assert_eq!(exact_boundary_tie.total_relation, TotalRelation::Exact);
		assert_eq!(
			exact_boundary_tie
				.hits
				.iter()
				.map(|hit| hit.id.as_str())
				.collect::<Vec<_>>(),
			expected_ids[..3]
		);
		let full_page = page(0, expected_ids.len());
		let paged_hits = (0..expected_ids.len())
			.flat_map(|offset| page(offset, 1))
			.collect::<Vec<_>>();
		assert_eq!(
			paged_hits.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>(),
			full_page.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>()
		);
		assert_eq!(
			paged_hits.iter().map(|hit| hit.score.to_bits()).collect::<Vec<_>>(),
			full_page.iter().map(|hit| hit.score.to_bits()).collect::<Vec<_>>()
		);
		assert_eq!(
			full_page.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>(),
			expected_ids
		);
		assert_eq!(
			paged_hits
				.iter()
				.find(|hit| hit.id == "catalog-0004")
				.unwrap()
				.version
				.as_deref(),
			Some("latest")
		);
		let past_end = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("identical catalog text", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 20,
					limit: 5,
					exact_total: false,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert!(past_end.hits.is_empty());
		assert_eq!(past_end.total, expected_ids.len() as u64);
		assert_eq!(past_end.total_relation, TotalRelation::Exact);
		writer.close().unwrap();
		let reopened = Engine::open(directory, &config).unwrap();
		let result = reopened
			.search(
				&reopened.reader().unwrap().searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("identical catalog", SearchMode::Any, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 3,
					exact_total: false,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap();
		assert_eq!(
			result.hits.iter().map(|hit| hit.id.as_str()).collect::<Vec<_>>(),
			expected_ids[..3].to_vec()
		);
	}

	#[test]
	fn resolves_hit_ids_and_versions_in_document_order_with_duplicate_and_missing_versions() {
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: [("one", Some("shared")), ("two", None), ("three", Some("shared"))]
					.into_iter()
					.map(|(id, version)| crate::protocol::Upsert {
						id: id.to_owned(),
						version: version.map(str::to_owned),
						fields: vec![("title".to_owned(), vec!["catalog".to_owned()])],
						filters: Vec::new(),
					})
					.collect(),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let scored_docs = [2, 0, 1]
			.into_iter()
			.map(|doc_id| (1.0, DocAddress::new(0, doc_id)))
			.collect::<Vec<_>>();
		assert_eq!(
			search_hit_ids(&searcher, &scored_docs).unwrap(),
			vec!["three", "one", "two"]
		);
		assert_eq!(
			search_hit_versions(&searcher, scored_docs.iter().map(|(_, address)| *address)).unwrap(),
			vec![Some("shared".to_owned()), Some("shared".to_owned()), None]
		);
		writer.close().unwrap();
	}

	#[test]
	fn offset_pages_preserve_versions_scores_and_totals_inside_score_ties() {
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		for group in 0..3 {
			writer
				.apply(MutationBatch {
					upserts: ["z", "A", "é", "a"]
						.into_iter()
						.enumerate()
						.map(|(i, suffix)| crate::protocol::Upsert {
							id: format!("{group}-{suffix}"),
							version: match i % 3 {
								0 => None,
								1 => Some("shared".to_owned()),
								_ => Some("long".repeat(512)),
							},
							fields: vec![(
								"title".to_owned(),
								vec![format!("catalog {}", "other ".repeat(group * 100))],
							)],
							filters: Vec::new(),
						})
						.collect(),
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit().unwrap();
		}
		let searcher = engine.reader().unwrap().searcher();
		let mut request = SearchRequest {
			sort: None,
			aggregations: Vec::new(),
			expression: expression("catalog", SearchMode::All, vec!["title".to_owned()]),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 12,
			exact_total: true,
			budget_milliseconds: 30_000,
			traversal: false,
			after: None,
		};
		let full = engine.search(&searcher, &request).unwrap();
		assert_eq!(
			full.hits.iter().map(|hit| hit.id.clone()).collect::<Vec<_>>(),
			(0..3)
				.flat_map(|group| ["A", "a", "z", "é"].map(|suffix| format!("{group}-{suffix}")))
				.collect::<Vec<_>>()
		);
		assert_eq!(full.hits[0].score.to_bits(), full.hits[3].score.to_bits());
		assert!(full.hits[3].score > full.hits[4].score);
		assert!(full.hits[7].score > full.hits[8].score);
		let hits = |values: &[SearchHit]| {
			values
				.iter()
				.map(|hit| (hit.id.clone(), hit.score.to_bits(), hit.version.clone()))
				.collect::<Vec<_>>()
		};
		for exact in [false, true] {
			for (offset, limit) in [(1, 7), (1, 2), (9, 5), (12, 5)] {
				request.offset = offset;
				request.limit = limit;
				request.exact_total = exact;
				let page = engine.search(&searcher, &request).unwrap();
				assert_eq!(hits(&page.hits), hits(&full.hits[offset..12.min(offset + limit)]));
				let lower_bound = !exact && offset + limit < 12;
				assert_eq!(page.total, if lower_bound { (offset + limit) as u64 } else { 12 });
				assert_eq!(
					page.total_relation,
					if lower_bound {
						TotalRelation::LowerBound
					} else {
						TotalRelation::Exact
					}
				);
			}
		}
		writer.close().unwrap();
	}

	#[test]
	fn rejects_prefix_expansion_past_the_release_ceiling() {
		let directory = RamDirectory::create();
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(directory, &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.apply(MutationBatch {
				upserts: (0..=MAX_PREFIX_EXPANSIONS)
					.map(|id| crate::protocol::Upsert {
						id: id.to_string(),
						version: None,
						fields: vec![("title".to_owned(), vec![format!("catalog{id:03}")])],
						filters: Vec::new(),
					})
					.collect(),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		let error = engine
			.search(
				&reader.searcher(),
				&SearchRequest {
					sort: None,
					aggregations: Vec::new(),
					expression: expression("cat", SearchMode::Prefix, Vec::new()),
					candidate_ids: None,
					filter: None,
					offset: 0,
					limit: 10,
					exact_total: false,
					budget_milliseconds: 30_000,
					traversal: false,
					after: None,
				},
			)
			.unwrap_err();
		assert_eq!(error.code, "E_PREFIX_TOO_BROAD");
		assert!(error.message.contains("more than 50 terms"));
		writer.close().unwrap();
	}

	#[test]
	fn rejects_identity_mismatch() {
		let directory = RamDirectory::create();
		let config = config();
		Engine::open(directory.clone(), &config).unwrap();
		let mut different = config;
		different.identity.generation = "two".to_owned();
		let error = match Engine::open(directory, &different) {
			Ok(_) => panic!("identity mismatch was accepted"),
			Err(error) => error,
		};
		assert_eq!(error.code, "E_IDENTITY_MISMATCH");
	}

	#[test]
	fn validates_the_complete_persisted_identity() {
		let config = config();
		let mut identity = identity_bytes(&config.identity);
		assert_eq!(&identity[4..6], b"\x06\x00");
		assert_eq!(persisted_index_id(&identity), Some("products"));

		identity.pop();
		assert_eq!(persisted_index_id(&identity), None);

		let mut identity = identity_bytes(&config.identity);
		identity.push(0);
		assert_eq!(persisted_index_id(&identity), None);

		let mut filtered = config.identity;
		filtered.filter_fields.push(FilterFieldConfig {
			name: "category".to_owned(),
			field_type: FilterFieldType::String,
		});
		let identity = identity_bytes(&filtered);
		assert_eq!(&identity[4..6], b"\x08\x00");
		assert_eq!(persisted_index_id(&identity), Some("products"));
	}

	#[test]
	fn inspection_is_read_only_and_reports_committed_payload() {
		let directory = RamDirectory::create();
		let config = config();
		assert_eq!(
			Engine::inspect(directory.clone(), &config.identity).unwrap(),
			InspectionResult::Missing
		);
		assert!(!directory.exists(Path::new(IDENTITY_PATH)).unwrap());
		assert!(!directory.exists(Path::new(META_PATH)).unwrap());

		let engine = Engine::open(directory.clone(), &config).unwrap();
		assert_eq!(
			Engine::inspect(directory.clone(), &config.identity).unwrap(),
			InspectionResult::Cursorless
		);
		let mut writer = engine.writer(&config).unwrap();
		writer.commit_with_payload(Some("cursor-v1")).unwrap();
		assert_eq!(
			Engine::inspect(directory.clone(), &config.identity).unwrap(),
			InspectionResult::Payload("cursor-v1".to_owned())
		);
		writer.close().unwrap();

		let mut different = config;
		different.identity.generation = "two".to_owned();
		assert_eq!(
			Engine::inspect(directory, &different.identity).unwrap_err().code,
			"E_IDENTITY_MISMATCH"
		);
	}

	#[test]
	fn recovery_maps_tantivy_format_incompatibility() {
		let error =
			tantivy::TantivyError::IncompatibleIndex(tantivy::directory::error::Incompatibility::CompressionMismatch {
				library_compression_format: "zstd".to_owned(),
				index_compression_format: "lz4".to_owned(),
			});
		assert_eq!(recovery_index_error(error).code, "E_INDEX_FORMAT_INCOMPATIBLE");
		let error = tantivy::TantivyError::DataCorruption(tantivy::error::DataCorruption::comment_only("broken meta"));
		assert_eq!(recovery_index_error(error).code, "E_INDEX_CORRUPT");
		let missing = || {
			tantivy::TantivyError::OpenReadError(OpenReadError::FileDoesNotExist(
				Path::new("missing.term").to_path_buf(),
			))
		};
		assert_eq!(recovery_index_error(missing()).code, "E_INDEX_CORRUPT");
		assert_eq!(index_error(missing()).code, "E_STORAGE");
	}

	#[test]
	fn tantivy_plain_commit_erases_a_prior_payload_without_the_guard() {
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer.commit_with_payload(Some("checkpoint")).unwrap();
		writer.inner.commit().unwrap();
		assert_eq!(engine.committed_payload().unwrap(), None);
	}

	#[test]
	fn checkpoint_guard_survives_reopen_and_rejected_commits_preserve_mutations() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer.commit_with_payload(Some("")).unwrap();
		writer.apply(batch()).unwrap();
		assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
		assert_eq!(engine.committed_payload().unwrap().as_deref(), Some(""));
		writer.commit_with_payload(Some("next")).unwrap();
		assert_eq!(engine.reader().unwrap().searcher().num_docs(), 2);
		writer.close().unwrap();
		let reopened = Engine::open(directory, &config).unwrap();
		let (mut writer, payload) = reopened.writer_with_payload(&config).unwrap();
		assert_eq!(payload.as_deref(), Some("next"));
		assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
		writer.rollback().unwrap();
		assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
	}

	#[test]
	fn writer_reads_checkpoint_after_another_owner_publishes() {
		let directory = RamDirectory::create();
		let config = config();
		let old_view = Engine::open(directory.clone(), &config).unwrap();
		assert_eq!(old_view.committed_payload().unwrap(), None);
		let other = Engine::open(directory, &config).unwrap();
		let mut owner = other.writer(&config).unwrap();
		owner.commit_with_payload(Some("published-by-other-owner")).unwrap();
		owner.close().unwrap();
		let (mut writer, payload) = old_view.writer_with_payload(&config).unwrap();
		assert_eq!(payload.as_deref(), Some("published-by-other-owner"));
		assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
	}

	#[test]
	fn committed_payload_requires_the_same_searcher_snapshot() {
		let directory = RamDirectory::create();
		let config = config();
		let reader_engine = Engine::open(directory.clone(), &config).unwrap();
		let reader = reader_engine.reader().unwrap();
		let writer_engine = Engine::open(directory, &config).unwrap();
		let mut writer = writer_engine.writer(&config).unwrap();
		writer.apply(batch()).unwrap();
		writer.commit_with_payload(Some("cursor-v1")).unwrap();
		assert_eq!(
			reader_engine
				.committed_payload_for_searcher(&reader.searcher())
				.unwrap_err()
				.code,
			"E_RELOAD_FAILED"
		);
		reader.reload().unwrap();
		assert_eq!(
			reader_engine
				.committed_payload_for_searcher(&reader.searcher())
				.unwrap()
				.as_deref(),
			Some("cursor-v1")
		);
	}

	#[test]
	fn failed_checkpoint_commit_keeps_the_guard_conservative() {
		let directory = AmbiguousAtomicWriteDirectory::new();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		directory.fail_after_next_atomic_write();
		let error = writer.commit_with_payload(Some("uncertain")).unwrap_err();
		assert_eq!(error.code, "E_STORAGE");
		assert!(error.message.starts_with("Tantivy commit: "), "{error}");
		assert!(error.message.contains("injected ambiguous atomic write"));
		assert_eq!(engine.committed_payload().unwrap().as_deref(), Some("uncertain"));
		assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
	}

	#[test]
	fn commit_payload_survives_merge_and_reopen() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		writer.apply(batch()).unwrap();
		writer.commit_with_payload(Some("cursor-v1:42")).unwrap();
		writer
			.apply(MutationBatch {
				upserts: vec![crate::protocol::Upsert {
					id: "three".to_owned(),
					version: None,
					fields: vec![("title".to_owned(), vec!["Hiking Boots".to_owned()])],
					filters: Vec::new(),
				}],
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit_with_payload(Some("cursor-v1:43")).unwrap();
		let segments = engine.index.searchable_segment_ids().unwrap();
		assert_eq!(segments.len(), 2);
		writer.inner.merge(&segments).wait().unwrap();
		assert_eq!(engine.committed_payload().unwrap().as_deref(), Some("cursor-v1:43"));
		writer.close().unwrap();

		let reopened = Engine::open(directory, &config).unwrap();
		assert_eq!(reopened.committed_payload().unwrap().as_deref(), Some("cursor-v1:43"));
	}

	#[test]
	fn delete_aware_merges_preserve_live_documents_and_checkpoint() {
		let directory = RamDirectory::create();
		let config = config();
		let engine = Engine::open(directory.clone(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		let upserts = |range: Range<usize>, version: &str| {
			range
				.map(|id| crate::protocol::Upsert {
					id: format!("product-{id}"),
					version: Some(version.to_owned()),
					fields: vec![("title".to_owned(), vec![format!("Trail shoe {id}")])],
					filters: Vec::new(),
				})
				.collect()
		};
		writer
			.apply(MutationBatch {
				upserts: upserts(0..100, "one"),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit_with_payload(Some("cursor:100")).unwrap();
		writer
			.apply(MutationBatch {
				upserts: upserts(0..51, "two"),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit_with_payload(Some("cursor:151")).unwrap();
		writer.close().unwrap();

		let meta = engine.index.load_metas().unwrap();
		assert_eq!(meta.segments.iter().map(|segment| segment.num_docs()).sum::<u32>(), 100);
		assert_eq!(
			meta.segments
				.iter()
				.map(|segment| segment.num_deleted_docs())
				.sum::<u32>(),
			0
		);
		let reopened = Engine::open(directory, &config).unwrap();
		assert_eq!(reopened.committed_payload().unwrap().as_deref(), Some("cursor:151"));
	}

	#[test]
	fn completes_an_interrupted_sidecar_first_create() {
		let directory = RamDirectory::create();
		let config = config();
		directory
			.atomic_write(Path::new(IDENTITY_PATH), &identity_bytes(&config.identity))
			.unwrap();
		Engine::open(directory.clone(), &config).unwrap();
		assert!(directory.exists(Path::new(META_PATH)).unwrap());
	}

	#[test]
	fn rejects_meta_without_an_identity_sidecar() {
		let directory = RamDirectory::create();
		let config = config();
		let (schema, _, _, _, _) = build_schema(&config.identity).unwrap();
		Index::create(directory.clone(), schema, IndexSettings::default()).unwrap();
		let error = match Engine::open(directory, &config) {
			Ok(_) => panic!("meta without an identity sidecar was accepted"),
			Err(error) => error,
		};
		assert_eq!(error.code, "E_INCOMPLETE_CREATE");
	}
	#[test]
	fn ordinary_stable_collection_aborts_before_visiting_all_matches() {
		use std::sync::atomic::{AtomicUsize, Ordering};
		use tantivy::query::{EnableScoring, Explanation, Scorer, Weight};
		use tantivy::DocSet;
		#[derive(Debug)]
		struct ObservedQuery {
			inner: Box<dyn Query>,
			visited: Arc<AtomicUsize>,
			deadline: Instant,
		}
		impl Clone for ObservedQuery {
			fn clone(&self) -> Self {
				Self {
					inner: self.inner.box_clone(),
					visited: self.visited.clone(),
					deadline: self.deadline,
				}
			}
		}
		struct ObservedWeight {
			inner: Box<dyn Weight>,
			visited: Arc<AtomicUsize>,
			deadline: Instant,
		}
		struct ObservedScorer {
			inner: Box<dyn Scorer>,
			visited: Arc<AtomicUsize>,
			deadline: Instant,
		}
		impl Query for ObservedQuery {
			fn weight(&self, scoring: EnableScoring<'_>) -> tantivy::Result<Box<dyn Weight>> {
				Ok(Box::new(ObservedWeight {
					inner: self.inner.weight(scoring)?,
					visited: self.visited.clone(),
					deadline: self.deadline,
				}))
			}
		}
		impl Weight for ObservedWeight {
			fn scorer(&self, reader: &SegmentReader, boost: Score) -> tantivy::Result<Box<dyn Scorer>> {
				Ok(Box::new(ObservedScorer {
					inner: self.inner.scorer(reader, boost)?,
					visited: self.visited.clone(),
					deadline: self.deadline,
				}))
			}
			fn explain(&self, reader: &SegmentReader, doc: DocId) -> tantivy::Result<Explanation> {
				self.inner.explain(reader, doc)
			}
		}
		impl DocSet for ObservedScorer {
			fn advance(&mut self) -> DocId {
				self.visited.fetch_add(1, Ordering::Relaxed);
				self.inner.advance()
			}
			fn doc(&self) -> DocId {
				self.inner.doc()
			}
			fn size_hint(&self) -> u32 {
				self.inner.size_hint()
			}
		}
		impl Scorer for ObservedScorer {
			fn score(&mut self) -> Score {
				if self.visited.load(Ordering::Relaxed) == 0 {
					while Instant::now() < self.deadline {
						std::thread::yield_now();
					}
				}
				self.inner.score()
			}
		}
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		writer
			.inner
			.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
		writer
			.apply(MutationBatch {
				upserts: (0..4096)
					.map(|id| crate::protocol::Upsert {
						id: format!("doc-{id}"),
						version: None,
						fields: vec![("title".to_owned(), vec!["amber stone quartz ruby".to_owned()])],
						filters: Vec::new(),
					})
					.collect(),
				deletes: Vec::new(),
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		let mut request = admission_request(SearchMode::All);
		request.expression = expression("amber stone quartz ruby", SearchMode::All, vec!["title".to_owned()]);
		let mut query = engine.query(&searcher, &request).unwrap();
		assert!(query.stable_scores);
		let visited = Arc::new(AtomicUsize::new(0));
		let deadline = Instant::now() + std::time::Duration::from_secs(1);
		query.query = Box::new(ObservedQuery {
			inner: query.query,
			visited: visited.clone(),
			deadline,
		});
		let error = engine
			.collect_search(&searcher, &request, Some(deadline), None, None, query)
			.unwrap_err();
		assert_eq!(error.code, "E_TIMEOUT");
		assert!(visited.load(Ordering::Relaxed) > 0);
		assert!(
			visited.load(Ordering::Relaxed) <= 1024,
			"collection must abort before a full scan"
		);
		assert_eq!(engine.search(&searcher, &request).unwrap().hits.len(), request.limit);
		writer
			.apply(MutationBatch {
				upserts: Vec::new(),
				deletes: vec!["doc-0".to_owned()],
			})
			.unwrap();
		writer.commit().unwrap();
		let searcher = engine.reader().unwrap().searcher();
		assert!(searcher
			.segment_readers()
			.iter()
			.any(|segment| segment.alive_bitset().is_some()));
		let query = engine.query(&searcher, &request).unwrap();
		let (_, count, expected) = searcher
			.search(query.query.as_ref(), &StableScoreTieCollector::new(0..request.limit))
			.unwrap()
			.unwrap();
		assert_eq!(count, 4095);
		let expected = search_hits(&searcher, expected).unwrap();
		request.exact_total = true;
		let actual = engine
			.search_with_deadline(&searcher, &request, Instant::now() + std::time::Duration::from_secs(10))
			.unwrap();
		assert_eq!(actual.total, 4095);
		assert_eq!(
			actual
				.hits
				.iter()
				.map(|hit| (&hit.id, hit.score.to_bits()))
				.collect::<Vec<_>>(),
			expected
				.iter()
				.map(|hit| (&hit.id, hit.score.to_bits()))
				.collect::<Vec<_>>()
		);
		assert!(!actual.hits.iter().any(|hit| hit.id == "doc-0"));
		writer.close().unwrap();
	}

	#[test]
	fn ordinary_exact_count_preserves_native_error_classification() {
		use tantivy::query::{EnableScoring, Weight};
		#[derive(Clone, Copy, Debug)]
		enum FailureKind {
			Storage,
			Query,
			Timeout,
		}
		#[derive(Debug)]
		struct CountFailure {
			inner: Box<dyn Query>,
			kind: FailureKind,
		}
		impl Clone for CountFailure {
			fn clone(&self) -> Self {
				Self {
					inner: self.inner.box_clone(),
					kind: self.kind,
				}
			}
		}
		impl Query for CountFailure {
			fn weight(&self, scoring: EnableScoring<'_>) -> tantivy::Result<Box<dyn Weight>> {
				if scoring.is_scoring_enabled() {
					return self.inner.weight(scoring);
				}
				Err(match self.kind {
					FailureKind::Storage => {
						tantivy::TantivyError::IoError(std::sync::Arc::new(std::io::Error::other("count read failed")))
					}
					FailureKind::Query => tantivy::TantivyError::InvalidArgument("count query failed".into()),
					FailureKind::Timeout => {
						tantivy::TantivyError::InvalidArgument(crate::error::SEARCH_TIMEOUT_MESSAGE.into())
					}
				})
			}
		}
		let config = config();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		let mut mutation = batch();
		mutation.upserts[0].fields = vec![("title".into(), vec!["shoe".into()])];
		mutation.upserts[1].fields = vec![(
			"title".into(),
			vec!["shoe padding padding padding padding padding".into()],
		)];
		writer.apply(mutation).unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		reader.reload().unwrap();
		let searcher = reader.searcher();
		let request = SearchRequest {
			sort: None,
			aggregations: vec![],
			expression: expression("shoe", SearchMode::Any, vec!["title".into()]),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 1,
			exact_total: true,
			budget_milliseconds: 30000,
			traversal: false,
			after: None,
		};
		let real = engine.query(&searcher, &request).unwrap();
		let scores = searcher
			.search(real.query.as_ref(), &TopDocs::with_limit(2).order_by_score())
			.unwrap();
		assert_ne!(scores[0].0, scores[1].0);
		for (kind, expected) in [
			(FailureKind::Storage, "E_STORAGE"),
			(FailureKind::Query, "E_NATIVE_FAILURE"),
			(FailureKind::Timeout, "E_TIMEOUT"),
		] {
			let result = engine.collect_search(
				&searcher,
				&request,
				None,
				None,
				None,
				BuiltQuery {
					query: Box::new(CountFailure {
						inner: real.query.box_clone(),
						kind,
					}),
					clauses: real.clauses,
					stable_scores: real.stable_scores,
				},
			);
			assert_eq!(result.unwrap_err().code, expected);
		}
		assert_eq!(engine.search(&searcher, &request).unwrap().total, 2);
	}
	#[test]
	fn exact_total_with_aggregations_uses_two_native_query_passes() {
		use std::sync::{
			atomic::{AtomicUsize, Ordering},
			Arc,
		};
		use tantivy::query::{EnableScoring, Weight};
		#[derive(Debug)]
		struct CountingQuery {
			inner: Box<dyn Query>,
			weights: Arc<AtomicUsize>,
		}
		impl Clone for CountingQuery {
			fn clone(&self) -> Self {
				Self {
					inner: self.inner.box_clone(),
					weights: self.weights.clone(),
				}
			}
		}

		impl Query for CountingQuery {
			fn weight(&self, scoring: EnableScoring<'_>) -> tantivy::Result<Box<dyn Weight>> {
				self.weights.fetch_add(1, Ordering::Relaxed);
				self.inner.weight(scoring)
			}
		}
		let mut config = config();
		config.identity.filter_fields.push(FilterFieldConfig {
			name: "price".into(),
			field_type: FilterFieldType::Number,
		});
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut writer = engine.writer(&config).unwrap();
		let texts = [
			"shoe shoe shoe shoe",
			"shoe shoe",
			"shoe",
			"shoe padding padding padding padding padding",
		];
		writer
			.apply(MutationBatch {
				upserts: texts
					.into_iter()
					.enumerate()
					.map(|(index, text)| crate::protocol::Upsert {
						id: format!("record-{index}"),
						version: None,
						fields: vec![("title".into(), vec![text.into()])],
						filters: vec![("price".into(), vec![FilterValue::Number((index + 1) as f64)])],
					})
					.collect(),
				deletes: vec![],
			})
			.unwrap();
		writer.commit().unwrap();
		let reader = engine.reader().unwrap();
		reader.reload().unwrap();
		let searcher = reader.searcher();
		let request = SearchRequest {
			sort: None,
			aggregations: vec![catalog::AggregationRequest {
				aggregations: Vec::new(),
				name: "prices".into(),
				field: "price".into(),
				kind: catalog::AggregationKind::Stats,
			}],
			expression: expression("shoe", SearchMode::Any, vec!["title".into()]),
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 1,
			exact_total: true,
			budget_milliseconds: 30000,
			traversal: false,
			after: None,
		};
		let query = engine.query(&searcher, &request).unwrap();
		let scores = searcher
			.search(query.query.as_ref(), &TopDocs::with_limit(2).order_by_score())
			.unwrap();
		assert_ne!(scores[0].0, scores[1].0);
		let weights = Arc::new(AtomicUsize::new(0));
		let query = BuiltQuery {
			query: Box::new(CountingQuery {
				inner: query.query,
				weights: weights.clone(),
			}),
			clauses: query.clauses,
			stable_scores: query.stable_scores,
		};
		let result = engine
			.collect_search(
				&searcher,
				&request,
				None,
				Some(catalog::validate_requests(None, &request.aggregations, &config.identity.filter_fields).unwrap()),
				None,
				query,
			)
			.unwrap();
		assert_eq!(
			weights.load(Ordering::Relaxed),
			2,
			"expected hits and combined count/aggregation query passes"
		);
		assert_eq!(result.total, 4);
		assert_eq!(result.total_relation, TotalRelation::Exact);
		assert_eq!(result.hits.len(), 1);
		assert_eq!(
			result.aggregations[0].value,
			catalog::AggregationValue::Stats {
				count: 4,
				sum: 10.0,
				min: Some(1.0),
				max: Some(4.0),
				avg: Some(2.5)
			}
		);
		let control_weights = Arc::new(AtomicUsize::new(0));
		let control = CountingQuery {
			inner: engine.query(&searcher, &request).unwrap().query,
			weights: control_weights.clone(),
		};
		searcher
			.search(&control, &TopDocs::with_limit(2).order_by_score())
			.unwrap();
		let control_aggregations = catalog::aggregate(
			&searcher,
			&control,
			&request.aggregations,
			&config.identity.filter_fields,
			engine.index.tokenizers().clone(),
			None,
		)
		.unwrap();
		let control_total = catalog::count_matches(&searcher, &control, None).unwrap();
		assert_eq!(control_weights.load(Ordering::Relaxed), 3);
		assert_eq!(control_total, result.total);
		assert_eq!(control_aggregations, result.aggregations);
	}
	#[cfg(feature = "node-api")]
	mod close_quiescence {
		use super::*;
		use std::sync::{mpsc, Condvar, Mutex};
		use std::time::Duration;
		#[derive(Debug, Default)]
		struct GateState {
			armed: bool,
			entered: bool,
			released: bool,
		}
		#[derive(Debug, Default)]
		struct MergeGate {
			state: Mutex<GateState>,
			changed: Condvar,
		}
		impl MergeGate {
			fn arm(&self) {
				self.state.lock().unwrap().armed = true;
			}
			fn block(&self) {
				let mut state = self.state.lock().unwrap();
				if !state.armed {
					return;
				}
				state.entered = true;
				self.changed.notify_all();
				while !state.released {
					state = self.changed.wait(state).unwrap();
				}
			}
			fn wait_entered(&self) {
				let state = self.state.lock().unwrap();
				let (state, _) = self
					.changed
					.wait_timeout_while(state, Duration::from_secs(5), |state| !state.entered)
					.unwrap();
				assert!(state.entered, "native merge did not reach the directory gate");
			}
			fn release(&self) {
				self.state.lock().unwrap().released = true;
				self.changed.notify_all();
			}
		}
		struct ReleaseGate(Arc<MergeGate>);
		impl Drop for ReleaseGate {
			fn drop(&mut self) {
				self.0.release();
			}
		}
		#[derive(Clone, Debug)]
		struct MergeGateDirectory {
			inner: RamDirectory,
			gate: Arc<MergeGate>,
			index_gate: Option<Arc<MergeGate>>,
			fail_index_write: Arc<AtomicBool>,
			worker_gates: HashMap<String, Arc<MergeGate>>,
		}
		impl Directory for MergeGateDirectory {
			fn get_file_handle(&self, path: &Path) -> std::result::Result<Arc<dyn FileHandle>, OpenReadError> {
				self.inner.get_file_handle(path)
			}
			fn delete(&self, path: &Path) -> std::result::Result<(), DeleteError> {
				self.inner.delete(path)
			}
			fn exists(&self, path: &Path) -> std::result::Result<bool, OpenReadError> {
				self.inner.exists(path)
			}
			fn open_write(&self, path: &Path) -> std::result::Result<WritePtr, OpenWriteError> {
				let current = std::thread::current();
				let name = current.name().unwrap_or("");
				if name.starts_with("merge_thread") {
					self.gate.block();
				}
				if name.starts_with("thrd-tantivy-index") {
					if let Some(gate) = self.worker_gates.get(name) {
						gate.block();
					}
					if let Some(gate) = &self.index_gate {
						gate.block();
					}
					if self.fail_index_write.swap(false, Ordering::AcqRel) {
						return Err(OpenWriteError::wrap_io_error(
							io::Error::other("injected indexing write failure during close"),
							path.to_path_buf(),
						));
					}
				}
				self.inner.open_write(path)
			}
			fn atomic_read(&self, path: &Path) -> std::result::Result<Vec<u8>, OpenReadError> {
				self.inner.atomic_read(path)
			}
			fn atomic_write(&self, path: &Path, bytes: &[u8]) -> io::Result<()> {
				self.inner.atomic_write(path, bytes)
			}
			fn sync_directory(&self) -> io::Result<()> {
				self.inner.sync_directory()
			}
			fn acquire_lock(&self, lock: &tantivy::directory::Lock) -> std::result::Result<DirectoryLock, LockError> {
				self.inner.acquire_lock(lock)
			}
			fn watch(&self, callback: WatchCallback) -> tantivy::Result<WatchHandle> {
				self.inner.watch(callback)
			}
		}
		#[test]
		fn discard_close_waits_for_original_merge_and_preserves_checkpoint() {
			let gate = Arc::new(MergeGate::default());
			let _release = ReleaseGate(gate.clone());
			let directory = MergeGateDirectory {
				inner: RamDirectory::create(),
				gate: gate.clone(),
				index_gate: None,
				fail_index_write: Arc::new(AtomicBool::new(false)),
				worker_gates: HashMap::new(),
			};
			let config = config();
			let engine = Engine::open(directory.clone(), &config).unwrap();
			let mut writer = engine.writer(&config).unwrap();
			writer
				.inner
				.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
			writer.apply(batch()).unwrap();
			writer.commit_with_payload(Some("published")).unwrap();
			let upsert = |id: &str| crate::protocol::Upsert {
				id: id.to_owned(),
				version: None,
				fields: vec![("title".to_owned(), vec!["shoe".to_owned()])],
				filters: Vec::new(),
			};
			writer
				.apply(MutationBatch {
					upserts: vec![upsert("three")],
					deletes: Vec::new(),
				})
				.unwrap();
			writer.commit_with_payload(Some("published")).unwrap();
			let segments = engine.index.searchable_segment_ids().unwrap();
			assert_eq!(segments.len(), 2);
			gate.arm();
			let merging = writer.inner.merge(&segments);
			gate.wait_entered();
			writer
				.apply(MutationBatch {
					upserts: vec![upsert("pending")],
					deletes: vec!["one".to_owned()],
				})
				.unwrap();
			let (finished, completion) = mpsc::channel();
			let (started, start) = mpsc::channel();
			let close_gate = gate.clone();
			let closer = std::thread::spawn(move || {
				let mut writer = Some(writer);
				started.send(()).unwrap();
				let result = crate::native::finish_writer_close(&mut writer, false);
				let released_at_return = close_gate.state.lock().unwrap().released;
				finished
					.send((result.quiesced, result.error, released_at_return))
					.unwrap();
			});
			start.recv_timeout(Duration::from_secs(5)).unwrap();
			let premature = completion.recv_timeout(Duration::from_millis(250)).ok();
			let returned_before_release = premature.is_some();
			gate.release();
			let completed = premature
				.or_else(|| completion.recv_timeout(Duration::from_secs(5)).ok())
				.expect("close did not complete after releasing its native merge");
			closer.join().unwrap();
			let merged = merging.wait();
			assert!(
				!returned_before_release,
				"discard close completed while its original native merge was blocked"
			);
			assert!(merged.is_ok());
			let (quiesced, error, released_at_return) = completed;
			assert!(
				released_at_return,
				"close returned before the native merge gate was released"
			);
			assert!(quiesced);
			assert!(error.is_none());
			let reopened = Engine::open(directory, &config).unwrap();
			let searcher = reopened.reader().unwrap().searcher();
			assert_eq!(searcher.num_docs(), 3);
			let count = |id| {
				searcher
					.search(
						&tantivy::query::TermQuery::new(
							Term::from_field_text(reopened.id_field, id),
							IndexRecordOption::Basic,
						),
						&tantivy::collector::Count,
					)
					.unwrap()
			};
			assert_eq!(count("one"), 1);
			assert_eq!(count("pending"), 0);
			assert_eq!(reopened.committed_payload().unwrap().as_deref(), Some("published"));
			let mut writer = reopened.writer(&config).unwrap();
			assert_eq!(writer.commit().unwrap_err().code, "E_CHECKPOINT_REQUIRED");
			writer.close().unwrap();
		}
		#[test]
		fn discard_close_reports_unproven_quiescence_on_native_pending_write_failure() {
			let index_gate = Arc::new(MergeGate::default());
			let _release = ReleaseGate(index_gate.clone());
			let failure = Arc::new(AtomicBool::new(false));
			let directory = MergeGateDirectory {
				inner: RamDirectory::create(),
				gate: Arc::new(MergeGate::default()),
				index_gate: Some(index_gate.clone()),
				fail_index_write: failure.clone(),
				worker_gates: HashMap::new(),
			};
			let config = config();
			let engine = Engine::open(directory.clone(), &config).unwrap();
			let mut writer = engine.writer(&config).unwrap();
			writer
				.inner
				.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
			writer.apply(batch()).unwrap();
			writer.commit_with_payload(Some("published")).unwrap();
			index_gate.arm();
			writer
				.apply(MutationBatch {
					upserts: vec![crate::protocol::Upsert {
						id: "pending".to_owned(),
						version: None,
						fields: vec![("title".to_owned(), vec!["shoe".to_owned()])],
						filters: Vec::new(),
					}],
					deletes: vec!["one".to_owned()],
				})
				.unwrap();
			index_gate.wait_entered();
			assert_eq!(engine.committed_payload().unwrap().as_deref(), Some("published"));
			let (started, start) = mpsc::channel();
			let (finished, completion) = mpsc::channel();
			let closer = std::thread::spawn(move || {
				let mut writer = Some(writer);
				failure.store(true, Ordering::Release);
				started.send(()).unwrap();
				let result = crate::native::finish_writer_close(&mut writer, false);
				finished.send((result.quiesced, result.error)).unwrap();
			});
			start.recv_timeout(Duration::from_secs(5)).unwrap();
			index_gate.release();
			let (quiesced, error) = completion.recv_timeout(Duration::from_secs(5)).unwrap();
			closer.join().unwrap();
			assert!(!quiesced);
			assert_eq!(error.unwrap().code, "E_NATIVE_FAILURE");
			assert!(
				!directory.fail_index_write.load(Ordering::Acquire),
				"native indexing open_write failure was not exercised"
			);
			drop(engine);
			let reopened = Engine::open(directory, &config).unwrap();
			let searcher = reopened.reader().unwrap().searcher();
			assert_eq!(searcher.num_docs(), 2);
			let count = |id| {
				searcher
					.search(
						&tantivy::query::TermQuery::new(
							Term::from_field_text(reopened.id_field, id),
							IndexRecordOption::Basic,
						),
						&tantivy::collector::Count,
					)
					.unwrap()
			};
			assert_eq!(count("one"), 1);
			assert_eq!(count("pending"), 0);
			assert_eq!(reopened.committed_payload().unwrap().as_deref(), Some("published"));
		}
		#[test]
		fn failed_prepare_commit_cannot_prove_close_while_another_native_worker_is_blocked() {
			let failing = Arc::new(MergeGate::default());
			let blocked = Arc::new(MergeGate::default());
			let _release_failing = ReleaseGate(failing.clone());
			let _release_blocked = ReleaseGate(blocked.clone());
			let failure = Arc::new(AtomicBool::new(false));
			let directory = MergeGateDirectory {
				inner: RamDirectory::create(),
				gate: Arc::new(MergeGate::default()),
				index_gate: None,
				fail_index_write: failure.clone(),
				worker_gates: HashMap::from([
					("thrd-tantivy-index0".to_owned(), failing.clone()),
					("thrd-tantivy-index1".to_owned(), blocked.clone()),
				]),
			};
			let mut config = config();
			config.limits.indexing_threads = 2;
			config.limits.writer_memory_bytes = 30_000_000;
			let engine = Engine::open(directory, &config).unwrap();
			let mut seed = engine.writer(&config).unwrap();
			seed.apply(batch()).unwrap();
			seed.commit_with_payload(Some("published")).unwrap();
			seed.close().unwrap();
			let writer = engine.writer(&config).unwrap();
			writer
				.inner
				.set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy));
			failing.arm();
			blocked.arm();
			let upsert = |id: &str| crate::protocol::Upsert {
				id: id.to_owned(),
				version: None,
				fields: vec![("title".to_owned(), vec!["shoe".to_owned()])],
				filters: Vec::new(),
			};
			writer
				.apply(MutationBatch {
					upserts: vec![upsert("pending-a"), upsert("pending-b")],
					deletes: vec!["one".to_owned()],
				})
				.unwrap();
			failing.wait_entered();
			blocked.wait_entered();
			failure.store(true, Ordering::Release);
			failing.release();
			let (finished, completion) = mpsc::channel();
			let closer = std::thread::spawn(move || {
				let mut writer = writer;
				let preparation_error = writer.commit_with_payload(Some("not-published")).unwrap_err();
				let mut writer = Some(writer);
				let result = crate::native::finish_writer_close(&mut writer, false);
				finished
					.send((preparation_error, result.quiesced, result.error))
					.unwrap();
			});
			let (preparation_error, quiesced, error) = completion
				.recv_timeout(Duration::from_secs(5))
				.expect("failed prepare/close did not return while the detached worker was gated");
			let blocked_at_close_return = !blocked.state.lock().unwrap().released;
			failure.store(true, Ordering::Release);
			blocked.release();
			closer.join().unwrap();
			assert_eq!(preparation_error.code, "E_STORAGE");
			assert!(
				preparation_error.message.starts_with("Tantivy prepare_commit: "),
				"{preparation_error}"
			);
			assert!(blocked_at_close_return);
			assert!(
				!quiesced,
				"close claimed quiescence after failed prepare discarded a still-blocked worker handle"
			);
			assert_eq!(error.unwrap().code, "E_QUIESCENCE_FAILED");
			let searcher = engine.reader().unwrap().searcher();
			assert_eq!(searcher.num_docs(), 2);
			let count = |id| {
				searcher
					.search(
						&tantivy::query::TermQuery::new(
							Term::from_field_text(engine.id_field, id),
							IndexRecordOption::Basic,
						),
						&tantivy::collector::Count,
					)
					.unwrap()
			};
			assert_eq!(count("one"), 1);
			assert_eq!(count("pending-a"), 0);
			assert_eq!(count("pending-b"), 0);
			assert_eq!(engine.committed_payload().unwrap().as_deref(), Some("published"));
		}
	}
	fn recovery_request(expression: SearchExpression) -> SearchRequest {
		SearchRequest {
			sort: None,
			aggregations: vec![],
			expression,
			candidate_ids: None,
			filter: None,
			offset: 0,
			limit: 100,
			exact_total: false,
			budget_milliseconds: 30_000,
			traversal: false,
			after: None,
		}
	}

	fn recovery_record(id: &str, values: Vec<String>) -> TraceRecord {
		TraceRecord {
			id: id.into(),
			fields: vec![("description".into(), values)],
		}
	}

	#[test]
	fn recovery_pages_preserve_whole_records_and_legacy_partial_output() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		for boolean in [false, true] {
			let leaf = expression("needle", SearchMode::Any, vec!["description".into()]);
			let request = recovery_request(if boolean {
				SearchExpression::And(vec![leaf])
			} else {
				leaf
			});
			let records: Vec<_> = (0..3)
				.map(|i| recovery_record(&i.to_string(), vec!["needle ".repeat(600)]))
				.collect();
			let legacy = engine.trace_matches(&request, &records, None).unwrap();
			assert!(!legacy.complete);
			assert_eq!(
				legacy
					.records
					.iter()
					.flat_map(|r| &r.values)
					.map(|v| v.spans.len())
					.sum::<usize>(),
				1024
			);
			for start in 0..3 {
				let page = engine.trace_matches_page(&request, &records[start..], None).unwrap();
				assert_eq!(page.next_record, 1);
				assert_eq!(page.trace.complete, start == 2);
				assert_eq!(
					page.states,
					[TraceRecordState {
						membership: Some(true),
						analysis_complete: true,
						spans_complete: true
					}]
				);
				assert_eq!(page.trace.records[0].id, start.to_string());
				assert_eq!(page.trace.records[0].values[0].spans.len(), 600);
			}
			let oversized = engine
				.trace_matches_page(
					&request,
					&[recovery_record("large", vec!["needle ".repeat(1100)])],
					None,
				)
				.unwrap();
			assert_eq!(oversized.next_record, 1);
			assert_eq!(
				oversized.states[0],
				TraceRecordState {
					membership: Some(true),
					analysis_complete: true,
					spans_complete: false
				}
			);
			assert_eq!(oversized.trace.records[0].values[0].spans.len(), 1024);
			assert!(!oversized.trace.complete);
		}
	}

	#[test]
	fn recovery_simple_preserves_value_limits_and_ordered_nonmatches() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let mut request = recovery_request(expression("needle", SearchMode::Any, vec!["description".into()]));
		let mut values = vec!["stone".into(); 4096];
		values.push("needle".into());
		let records = [
			recovery_record("excluded", vec!["needle".into()]),
			recovery_record("absent", vec!["stone".into()]),
			recovery_record("found", values),
		];
		request.candidate_ids = Some(vec!["absent".into(), "found".into()]);
		let page = engine.trace_matches_page(&request, &records, None).unwrap();
		assert_eq!(page.next_record, 3);
		assert!(page.trace.complete);
		assert_eq!(
			page.states.iter().map(|s| s.membership).collect::<Vec<_>>(),
			[Some(false), Some(false), Some(true)]
		);
		assert_eq!(page.trace, engine.trace_matches(&request, &records, None).unwrap());
		assert_eq!(page.trace.records[0].values[0].value_index, 4096);
		let empty = engine.trace_matches_page(&request, &[], None).unwrap();
		assert!(empty.trace.complete);
		assert_eq!(empty.next_record, 0);
		assert!(empty.states.is_empty());
		assert_eq!(
			engine
				.trace_matches_page(&request, &records, Some(Instant::now()))
				.unwrap_err()
				.code,
			"E_TIMEOUT"
		);
	}

	#[test]
	fn recovery_boolean_preserves_unknown_not_and_known_or_proofs() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let missing = expression("needle", SearchMode::Any, vec!["description".into()]);
		let records = [recovery_record(
			"one",
			vec!["x ".repeat(MAX_TRACE_TOKENS_PER_VALUE + 1), "marker".into()],
		)];
		let request = recovery_request(SearchExpression::Not(Box::new(missing.clone())));
		let unknown = engine.trace_matches_page(&request, &records, None).unwrap();
		assert_eq!(unknown.next_record, 1);
		assert_eq!(unknown.states[0].membership, None);
		assert!(!unknown.states[0].analysis_complete);
		assert!(!unknown.states[0].spans_complete);
		assert!(unknown.trace.records.is_empty());
		let request = recovery_request(SearchExpression::Or(vec![
			SearchExpression::Not(Box::new(missing)),
			expression("marker", SearchMode::Any, vec!["description".into()]),
		]));
		let known = engine.trace_matches_page(&request, &records, None).unwrap();
		assert_eq!(known.states[0].membership, Some(true));
		assert!(!known.states[0].analysis_complete);
		assert_eq!(known.trace.records.len(), 1);
		let request = recovery_request(SearchExpression::Not(Box::new(expression(
			"needle",
			SearchMode::Any,
			vec!["description".into()],
		))));
		let zero = engine
			.trace_matches_page(&request, &[recovery_record("zero", vec!["stone".into()])], None)
			.unwrap();
		assert!(zero.trace.complete);
		assert_eq!(zero.states[0].membership, Some(true));
		assert!(zero.trace.records[0].values.is_empty());
	}

	#[test]
	fn recovery_late_shared_work_exhaustion_defers_a_whole_nonmatch() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let request = recovery_request(SearchExpression::And(
			(0..32)
				.map(|_| expression("absent", SearchMode::Any, vec!["description".into()]))
				.collect(),
		));
		let records: Vec<_> = (0..14)
			.map(|i| recovery_record(&i.to_string(), vec!["x ".repeat(20_000)]))
			.collect();
		let page = engine.trace_matches_page(&request, &records, None).unwrap();
		assert_eq!(page.next_record, MAX_TRACE_WORK_TOKENS / (32 * 20_000));
		assert!(!page.trace.complete);
		assert!(page
			.states
			.iter()
			.all(|s| s.membership == Some(false) && s.analysis_complete));
		let resumed = engine
			.trace_matches_page(&request, &records[page.next_record..], None)
			.unwrap();
		assert!(resumed.trace.complete);
		assert_eq!(resumed.next_record, records.len() - page.next_record);
	}

	#[test]
	fn recovery_late_span_work_exhaustion_defers_a_whole_match() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let request = recovery_request(SearchExpression::Or(
			(0..32)
				.map(|_| expression("needle", SearchMode::Any, vec!["description".into()]))
				.collect(),
		));
		let records: Vec<_> = (0..7)
			.map(|i| recovery_record(&i.to_string(), vec![format!("needle {}", "x ".repeat(20_000))]))
			.collect();
		let page = engine.trace_matches_page(&request, &records, None).unwrap();
		assert_eq!(page.next_record, 6);
		assert_eq!(page.trace.records.len(), 6);
		assert!(!page.trace.complete);
		assert!(page
			.states
			.iter()
			.all(|state| state.membership == Some(true) && state.analysis_complete && state.spans_complete));
		let resumed = engine
			.trace_matches_page(&request, &records[page.next_record..], None)
			.unwrap();
		assert_eq!(resumed.next_record, 1);
		assert!(resumed.trace.complete);
		assert_eq!(resumed.trace.records[0].id, "6");
	}

	#[test]
	fn recovery_response_boundary_never_skips_a_later_large_record() {
		let field = "f".repeat(9000);
		let mut config = config();
		config.identity.surface_terms = true;
		config.identity.fields[1].name = field.clone();
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		for boolean in [false, true] {
			let leaf = expression("needle", SearchMode::Any, vec![field.clone()]);
			let request = recovery_request(if boolean {
				SearchExpression::And(vec![leaf])
			} else {
				leaf
			});
			let record = |id: &str, count| TraceRecord {
				id: id.into(),
				fields: vec![(field.clone(), vec!["needle".into(); count])],
			};
			let records = [record("first", 500), record("boundary", 500), record("small", 1)];
			let page = engine.trace_matches_page(&request, &records, None).unwrap();
			assert_eq!(page.next_record, 1);
			assert_eq!(page.trace.records.len(), 1);
			assert!(page.states[0].spans_complete);
			let resumed = engine.trace_matches_page(&request, &records[1..], None).unwrap();
			assert!(resumed.trace.complete);
			assert_eq!(resumed.next_record, 2);
			let singleton = engine
				.trace_matches_page(&request, &[record("large", 1000)], None)
				.unwrap();
			assert_eq!(singleton.next_record, 1);
			assert!(!singleton.states[0].spans_complete);
			assert!(singleton.states[0].analysis_complete);
			let byte_size = 3
				+ 2 + 1 + 6 + "large".len()
				+ singleton.trace.records[0]
					.values
					.iter()
					.map(|v| 10 + v.field.len() + v.spans.len() * 8)
					.sum::<usize>();
			assert!(byte_size <= MAX_SEARCH_RESPONSE_BYTES);
		}
	}
	#[test]
	fn recovery_simple_retains_per_value_analysis_and_exact_span_boundary() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let request = recovery_request(expression("needle", SearchMode::Prefix, vec!["description".into()]));
		let records = [recovery_record(
			"many",
			vec!["x ".repeat(150_000), format!("{}needle", "x ".repeat(150_000))],
		)];
		let page = engine.trace_matches_page(&request, &records, None).unwrap();
		assert!(page.trace.complete);
		assert_eq!(page.trace, engine.trace_matches(&request, &records, None).unwrap());
		assert_eq!(
			page.states[0],
			TraceRecordState {
				membership: Some(true),
				analysis_complete: true,
				spans_complete: true
			}
		);
		let request = recovery_request(expression("needle", SearchMode::Any, vec!["description".into()]));
		let exact = engine
			.trace_matches_page(
				&request,
				&[recovery_record("exact", vec!["needle ".repeat(MAX_TRACE_SPANS)])],
				None,
			)
			.unwrap();
		assert!(exact.trace.complete);
		assert!(exact.states[0].spans_complete);
		let source = format!("{}needle", "x ".repeat(MAX_TRACE_TOKENS_PER_VALUE));
		let unknown = engine
			.trace_matches_page(&request, &[recovery_record("unknown", vec![source])], None)
			.unwrap();
		assert_eq!(
			unknown.states[0],
			TraceRecordState {
				membership: None,
				analysis_complete: false,
				spans_complete: false
			}
		);
		assert_eq!(unknown.next_record, 1);
		assert!(!unknown.trace.complete);
		assert!(unknown.trace.records.is_empty());
	}

	#[test]
	fn recovery_boolean_first_work_exhaustion_and_value_limit_are_terminal() {
		let mut config = config();
		config.identity.surface_terms = true;
		let engine = Engine::open(RamDirectory::create(), &config).unwrap();
		let request = recovery_request(SearchExpression::Or(
			(0..33)
				.map(|_| expression("absent", SearchMode::Any, vec!["description".into()]))
				.collect(),
		));
		let page = engine
			.trace_matches_page(
				&request,
				&[recovery_record("large", vec!["x ".repeat(MAX_TRACE_TOKENS_PER_VALUE)])],
				None,
			)
			.unwrap();
		assert_eq!(page.next_record, 1);
		assert_eq!(page.states[0].membership, None);
		assert!(!page.states[0].analysis_complete);
		assert!(!page.trace.complete);
		let request = recovery_request(SearchExpression::Not(Box::new(expression(
			"needle",
			SearchMode::Any,
			vec!["description".into()],
		))));
		let mut values = vec!["stone".into(); MAX_TRACE_EXPRESSION_VALUES];
		values.push("needle".into());
		let page = engine
			.trace_matches_page(&request, &[recovery_record("values", values)], None)
			.unwrap();
		assert_eq!(page.next_record, 1);
		assert_eq!(page.states[0].membership, None);
		assert!(!page.states[0].analysis_complete);
		assert!(page.trace.records.is_empty());
	}
}