fathomdb 0.5.2

//! Tethered query builders for the Phase 1 adaptive search surface.
//!
//! These builders wrap the AST-only [`fathomdb_query::QueryBuilder`] and carry
//! a borrow of the [`Engine`] so that a zero-arg `.execute()` terminal can
//! route to the right coordinator entry point by type. Non-search chains
//! return [`QueryRows`]; `.text_search(...).execute()` returns [`SearchRows`].

use fathomdb_engine::{EngineError, GroupedQueryRows, QueryRows};
use fathomdb_query::{
    BuilderValidationError, CompileError, CompiledGroupedQuery, CompiledQuery,
    CompiledRetrievalPlan, CompiledSearchPlan, CompiledVectorSearch, QueryAst, QueryBuilder,
    QueryStep, SearchRows, TextQuery, compile_grouped_query, compile_search,
    compile_search_plan_from_queries, compile_vector_search,
};

use crate::Engine;

/// Shared fusion gate implementation used by every tethered builder.
///
/// Resolves the FTS property schema for `kind` via the engine's admin
/// handle and verifies that `path` is in the registered include list.
/// On success the caller may safely append a fused predicate variant to
/// the underlying AST builder.
///
/// # Errors
/// - [`BuilderValidationError::KindRequiredForFusion`] if `kind` is empty.
/// - [`BuilderValidationError::MissingPropertyFtsSchema`] if no schema
///   is registered for the kind.
/// - [`BuilderValidationError::PathNotIndexed`] if a schema exists but
///   does not include `path`.
fn validate_fusable_property_path(
    engine: &Engine,
    kind: &str,
    path: &str,
    method: &str,
) -> Result<(), BuilderValidationError> {
    if kind.is_empty() {
        return Err(BuilderValidationError::KindRequiredForFusion {
            method: method.to_owned(),
        });
    }
    let schema = engine.describe_fts_property_schema(kind).map_err(|_| {
        BuilderValidationError::MissingPropertyFtsSchema {
            kind: kind.to_owned(),
        }
    })?;
    let schema = schema.ok_or_else(|| BuilderValidationError::MissingPropertyFtsSchema {
        kind: kind.to_owned(),
    })?;
    if !schema.property_paths.iter().any(|p| p == path) {
        return Err(BuilderValidationError::PathNotIndexed {
            kind: kind.to_owned(),
            path: path.to_owned(),
        });
    }
    Ok(())
}

/// Resolve the kind currently bound on a filter accumulator's `KindEq`
/// predicate, if any. The adaptive `FallbackSearchBuilder` path has no
/// explicit `root_kind` field — its kind comes from a chained
/// `filter_kind_eq`. This helper walks the accumulator's AST steps and
/// returns the most recent `KindEq` value.
fn filter_builder_kind(builder: &QueryBuilder) -> Option<&str> {
    for step in &builder.ast().steps {
        if let QueryStep::Filter(fathomdb_query::Predicate::KindEq(kind)) = step {
            return Some(kind.as_str());
        }
    }
    None
}

/// Tethered node query builder.
///
/// Returned by [`Engine::query`]. Carries an `&Engine` so that terminal
/// methods can dispatch directly to the coordinator. The underlying AST is
/// the same [`QueryBuilder`] the query crate has always produced — this is
/// purely an execution tether, not a new AST.
#[must_use]
pub struct NodeQueryBuilder<'e> {
    engine: &'e Engine,
    inner: QueryBuilder,
}

impl<'e> NodeQueryBuilder<'e> {
    pub(crate) fn new(engine: &'e Engine, kind: impl Into<String>) -> Self {
        Self {
            engine,
            inner: QueryBuilder::nodes(kind),
        }
    }

    /// Transition this chain into the unified Phase 12 retrieval builder.
    ///
    /// `search()` is the primary client-facing retrieval entry point per
    /// `dev/design-adaptive-text-search-surface-addendum-1-vec.md` §Public
    /// Surface. Subsequent filters accumulate on the returned
    /// [`SearchBuilder`] and `.execute()` returns [`SearchRows`] populated
    /// from the unified retrieval planner: text strict, optional text
    /// relaxed, and (in a future phase) vector retrieval, fused under the
    /// addendum's block precedence rules.
    ///
    /// **v1 scope**: the planner's vector branch slot is wired
    /// architecturally but never fires through `search()` because read-time
    /// embedding of natural-language queries is deferred. Callers who need
    /// vector retrieval today should use the advanced `vector_search()`
    /// override directly with a caller-provided vector literal.
    pub fn search(self, query: impl Into<String>, limit: usize) -> SearchBuilder<'e> {
        SearchBuilder::new(
            self.engine,
            self.inner.ast().root_kind.clone(),
            query,
            limit,
        )
    }

    /// Transition this chain into a text-search builder. Subsequent filters
    /// accumulate on the search builder and `.execute()` returns
    /// [`SearchRows`] rather than [`QueryRows`].
    pub fn text_search(self, query: impl Into<String>, limit: usize) -> TextSearchBuilder<'e> {
        TextSearchBuilder {
            engine: self.engine,
            inner: self.inner.text_search(query, limit),
            attribution_requested: false,
        }
    }

    /// Transition this chain into a vector-search builder. Subsequent
    /// filters accumulate on the vector-search builder and `.execute()`
    /// returns [`SearchRows`] populated with the vector retrieval block.
    ///
    /// Phase 11 (HITL-Q5 closure): this method switches to a type-state
    /// terminal returning [`VectorSearchBuilder`], mirroring
    /// [`NodeQueryBuilder::text_search`]. The old self-returning form is
    /// no longer available on the facade surface; advanced callers that
    /// need the flat `vector_search` AST step alongside other pipeline
    /// steps can still reach it via [`QueryBuilder::vector_search`] on
    /// the untethered builder.
    pub fn vector_search(self, query: impl Into<String>, limit: usize) -> VectorSearchBuilder<'e> {
        VectorSearchBuilder::new(
            self.engine,
            self.inner.ast().root_kind.clone(),
            query,
            limit,
        )
    }

    /// Add a graph traversal step.
    pub fn traverse(
        mut self,
        direction: fathomdb_query::TraverseDirection,
        label: impl Into<String>,
        max_depth: usize,
    ) -> Self {
        self.inner = self.inner.traverse(direction, label, max_depth);
        self
    }

    /// Filter results to a single logical ID.
    pub fn filter_logical_id_eq(mut self, logical_id: impl Into<String>) -> Self {
        self.inner = self.inner.filter_logical_id_eq(logical_id);
        self
    }

    /// Filter results to nodes matching the given kind.
    pub fn filter_kind_eq(mut self, kind: impl Into<String>) -> Self {
        self.inner = self.inner.filter_kind_eq(kind);
        self
    }

    /// Filter results to nodes matching the given `source_ref`.
    pub fn filter_source_ref_eq(mut self, source_ref: impl Into<String>) -> Self {
        self.inner = self.inner.filter_source_ref_eq(source_ref);
        self
    }

    /// Filter results to nodes where `content_ref` is not NULL.
    pub fn filter_content_ref_not_null(mut self) -> Self {
        self.inner = self.inner.filter_content_ref_not_null();
        self
    }

    /// Filter results to nodes matching the given `content_ref` URI.
    pub fn filter_content_ref_eq(mut self, content_ref: impl Into<String>) -> Self {
        self.inner = self.inner.filter_content_ref_eq(content_ref);
        self
    }

    /// Filter results where a JSON property at `path` equals the given text value.
    pub fn filter_json_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.inner = self.inner.filter_json_text_eq(path, value);
        self
    }

    /// Filter results where a JSON property at `path` equals the given boolean value.
    pub fn filter_json_bool_eq(mut self, path: impl Into<String>, value: bool) -> Self {
        self.inner = self.inner.filter_json_bool_eq(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than `value`.
    pub fn filter_json_integer_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_gt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than or equal to `value`.
    pub fn filter_json_integer_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_gte(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than `value`.
    pub fn filter_json_integer_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_lt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than or equal to `value`.
    pub fn filter_json_integer_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_lte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is after `value`.
    pub fn filter_json_timestamp_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_gt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or after `value`.
    pub fn filter_json_timestamp_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_gte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is before `value`.
    pub fn filter_json_timestamp_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_lt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or before `value`.
    pub fn filter_json_timestamp_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_lte(path, value);
        self
    }

    /// Filter results where a JSON text property at `path` equals
    /// `value`, pushing the predicate into the inner search CTE so the
    /// CTE `LIMIT` applies *after* the filter runs.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError`] if the root kind has no
    /// registered property-FTS schema or the schema does not cover
    /// `path`.
    pub fn filter_json_fused_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_eq")?;
        self.inner = self.inner.filter_json_fused_text_eq_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError`] if the root kind has no
    /// registered property-FTS schema or the schema does not cover
    /// `path`.
    pub fn filter_json_fused_text_in(
        mut self,
        path: impl Into<String>,
        values: Vec<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_in")?;
        self.inner = self.inner.filter_json_fused_text_in_unchecked(path, values);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`. Non-fused; no FTS schema required.
    pub fn filter_json_text_in(mut self, path: impl Into<String>, values: Vec<String>) -> Self {
        self.inner = self.inner.filter_json_text_in(path, values);
        self
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// greater than `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gt",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_gt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or after
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gte",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_gte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// before `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lt",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_lt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or before
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lte",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_lte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON boolean property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_bool_eq(
        mut self,
        path: impl Into<String>,
        value: bool,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_bool_eq")?;
        self.inner = self.inner.filter_json_fused_bool_eq_unchecked(path, value);
        Ok(self)
    }

    /// Add an expansion slot that traverses edges per root result.
    ///
    /// Pass `filter: None` to preserve the existing behavior. `filter: Some(_)` is
    /// accepted by the AST but the compilation is not yet implemented (Pack 3).
    /// Pass `edge_filter: None` to preserve pre-Pack-D behavior.
    pub fn expand(
        mut self,
        slot: impl Into<String>,
        direction: fathomdb_query::TraverseDirection,
        label: impl Into<String>,
        max_depth: usize,
        filter: Option<fathomdb_query::Predicate>,
        edge_filter: Option<fathomdb_query::Predicate>,
    ) -> Self {
        self.inner = self
            .inner
            .expand(slot, direction, label, max_depth, filter, edge_filter);
        self
    }

    /// Set the final row limit.
    pub fn limit(mut self, limit: usize) -> Self {
        self.inner = self.inner.limit(limit);
        self
    }

    /// Borrow the underlying [`QueryBuilder`].
    #[must_use]
    pub fn as_builder(&self) -> &QueryBuilder {
        &self.inner
    }

    /// Consume the tether and return the underlying AST-only builder.
    #[must_use]
    pub fn into_builder(self) -> QueryBuilder {
        self.inner
    }

    /// Consume the tether and return the underlying [`QueryAst`].
    #[must_use]
    pub fn into_ast(self) -> fathomdb_query::QueryAst {
        self.inner.into_ast()
    }

    /// Compile this query to a [`CompiledQuery`]. Mirrors
    /// [`QueryBuilder::compile`].
    ///
    /// # Errors
    /// Returns [`CompileError`] if compilation fails.
    pub fn compile(&self) -> Result<CompiledQuery, CompileError> {
        self.inner.compile()
    }

    /// Compile this query into a grouped plan. Mirrors
    /// [`QueryBuilder::compile_grouped`].
    ///
    /// # Errors
    /// Returns [`CompileError`] if grouped compilation fails.
    pub fn compile_grouped(&self) -> Result<CompiledGroupedQuery, CompileError> {
        self.inner.compile_grouped()
    }

    /// Execute the query and return matching node rows.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute(&self) -> Result<QueryRows, EngineError> {
        let compiled = self
            .inner
            .compile()
            .map_err(|e| EngineError::InvalidConfig(format!("query compilation failed: {e}")))?;
        self.engine.coordinator().execute_compiled_read(&compiled)
    }

    /// Execute the grouped query and return root rows plus named expansion slots.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute_grouped(self) -> Result<GroupedQueryRows, EngineError> {
        let compiled = self.inner.compile_grouped().map_err(|e| {
            EngineError::InvalidConfig(format!("grouped query compilation failed: {e}"))
        })?;
        self.engine
            .coordinator()
            .execute_compiled_grouped_read(&compiled)
    }
}

/// Tethered text-search builder returned from
/// [`NodeQueryBuilder::text_search`].
///
/// Accumulates filter predicates alongside the text-search step and dispatches
/// `.execute()` through [`fathomdb_engine::ExecutionCoordinator::execute_compiled_search`],
/// returning [`SearchRows`] populated with score, source, snippet, and
/// active-version `written_at` values.
#[must_use]
pub struct TextSearchBuilder<'e> {
    engine: &'e Engine,
    inner: QueryBuilder,
    attribution_requested: bool,
}

impl TextSearchBuilder<'_> {
    /// Request per-hit match attribution.
    ///
    /// When set, the coordinator populates
    /// [`SearchHit::attribution`](fathomdb_query::SearchHit::attribution) on
    /// every hit with the set of property paths (or `"text_content"` for
    /// chunk hits) that contributed to the match. Without this flag (the
    /// default), attribution stays `None` and the Phase 4 position map is not
    /// read at all — it is a pay-as-you-go feature.
    pub fn with_match_attribution(mut self) -> Self {
        self.attribution_requested = true;
        self
    }

    /// Filter results to a single logical ID.
    pub fn filter_logical_id_eq(mut self, logical_id: impl Into<String>) -> Self {
        self.inner = self.inner.filter_logical_id_eq(logical_id);
        self
    }

    /// Filter results to nodes matching the given kind.
    pub fn filter_kind_eq(mut self, kind: impl Into<String>) -> Self {
        self.inner = self.inner.filter_kind_eq(kind);
        self
    }

    /// Filter results to nodes matching the given `source_ref`.
    pub fn filter_source_ref_eq(mut self, source_ref: impl Into<String>) -> Self {
        self.inner = self.inner.filter_source_ref_eq(source_ref);
        self
    }

    /// Filter results to nodes where `content_ref` is not NULL.
    pub fn filter_content_ref_not_null(mut self) -> Self {
        self.inner = self.inner.filter_content_ref_not_null();
        self
    }

    /// Filter results to nodes matching the given `content_ref` URI.
    pub fn filter_content_ref_eq(mut self, content_ref: impl Into<String>) -> Self {
        self.inner = self.inner.filter_content_ref_eq(content_ref);
        self
    }

    /// Filter results where a JSON property at `path` equals the given text value.
    pub fn filter_json_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.inner = self.inner.filter_json_text_eq(path, value);
        self
    }

    /// Filter results where a JSON property at `path` equals the given boolean value.
    pub fn filter_json_bool_eq(mut self, path: impl Into<String>, value: bool) -> Self {
        self.inner = self.inner.filter_json_bool_eq(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than `value`.
    pub fn filter_json_integer_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_gt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than or equal to `value`.
    pub fn filter_json_integer_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_gte(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than `value`.
    pub fn filter_json_integer_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_lt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than or equal to `value`.
    pub fn filter_json_integer_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_integer_lte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is after `value`.
    pub fn filter_json_timestamp_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_gt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or after `value`.
    pub fn filter_json_timestamp_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_gte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is before `value`.
    pub fn filter_json_timestamp_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_lt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or before `value`.
    pub fn filter_json_timestamp_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.inner = self.inner.filter_json_timestamp_lte(path, value);
        self
    }

    /// Filter results where a JSON text property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`NodeQueryBuilder::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError`] if the root kind has no
    /// registered property-FTS schema or the schema does not cover
    /// `path`.
    pub fn filter_json_fused_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_eq")?;
        self.inner = self.inner.filter_json_fused_text_eq_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_text_in(
        mut self,
        path: impl Into<String>,
        values: Vec<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_in")?;
        self.inner = self.inner.filter_json_fused_text_in_unchecked(path, values);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`. Non-fused; no FTS schema required.
    pub fn filter_json_text_in(mut self, path: impl Into<String>, values: Vec<String>) -> Self {
        self.inner = self.inner.filter_json_text_in(path, values);
        self
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// greater than `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gt",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_gt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or after
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gte",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_gte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// before `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lt",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_lt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or before
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lte",
        )?;
        self.inner = self
            .inner
            .filter_json_fused_timestamp_lte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON boolean property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_bool_eq(
        mut self,
        path: impl Into<String>,
        value: bool,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = self.inner.ast().root_kind.clone();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_bool_eq")?;
        self.inner = self.inner.filter_json_fused_bool_eq_unchecked(path, value);
        Ok(self)
    }

    /// Set the final row limit on the underlying AST.
    ///
    /// Phase 1 note: [`CompiledSearch`](fathomdb_query::CompiledSearch) derives
    /// its effective limit from the `text_search` step, not from this field.
    /// `limit` is still delegated to the inner builder so callers that later
    /// fall back to [`TextSearchBuilder::compile_query`] keep the same shape.
    pub fn limit(mut self, limit: usize) -> Self {
        self.inner = self.inner.limit(limit);
        self
    }

    /// Add a graph traversal step. Applied after the text-search step when
    /// the inner AST is compiled via [`TextSearchBuilder::compile_query`].
    /// The Phase 1 [`TextSearchBuilder::execute`] path ignores traversals.
    pub fn traverse(
        mut self,
        direction: fathomdb_query::TraverseDirection,
        label: impl Into<String>,
        max_depth: usize,
    ) -> Self {
        self.inner = self.inner.traverse(direction, label, max_depth);
        self
    }

    /// Add an expansion slot. Applied when compiling via
    /// [`TextSearchBuilder::compile_grouped_query`]; ignored by
    /// [`TextSearchBuilder::execute`] in Phase 1.
    ///
    /// Pass `filter: None` to preserve the existing behavior. `filter: Some(_)` is
    /// accepted by the AST but the compilation is not yet implemented (Pack 3).
    /// Pass `edge_filter: None` to preserve pre-Pack-D behavior.
    pub fn expand(
        mut self,
        slot: impl Into<String>,
        direction: fathomdb_query::TraverseDirection,
        label: impl Into<String>,
        max_depth: usize,
        filter: Option<fathomdb_query::Predicate>,
        edge_filter: Option<fathomdb_query::Predicate>,
    ) -> Self {
        self.inner = self
            .inner
            .expand(slot, direction, label, max_depth, filter, edge_filter);
        self
    }

    /// Borrow the underlying [`QueryBuilder`].
    #[must_use]
    pub fn as_builder(&self) -> &QueryBuilder {
        &self.inner
    }

    /// Compile the underlying AST as a flat [`CompiledQuery`]. Provided for
    /// call sites that mix search with traversal steps and still need to run
    /// the flat node-row pipeline.
    ///
    /// # Errors
    /// Returns [`CompileError`] if compilation fails.
    pub fn compile(&self) -> Result<CompiledQuery, CompileError> {
        self.inner.compile()
    }

    /// Compile the underlying AST as a [`CompiledGroupedQuery`].
    ///
    /// # Errors
    /// Returns [`CompileError`] if compilation fails.
    pub fn compile_grouped(&self) -> Result<CompiledGroupedQuery, CompileError> {
        self.inner.compile_grouped()
    }

    /// Consume the tether and return the underlying [`QueryAst`].
    #[must_use]
    pub fn into_ast(self) -> fathomdb_query::QueryAst {
        self.inner.into_ast()
    }

    /// Execute the text search and return matching hits.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute(&self) -> Result<SearchRows, EngineError> {
        let mut compiled = compile_search(self.inner.ast())
            .map_err(|e| EngineError::InvalidConfig(format!("search compilation failed: {e}")))?;
        compiled.attribution_requested = self.attribution_requested;
        self.engine.coordinator().execute_compiled_search(&compiled)
    }
}

/// Tethered two-shape fallback search builder returned from
/// [`Engine::fallback_search`].
///
/// `fallback_search(strict, Some(relaxed))` is the "advanced caller who
/// wants explicit control over the relaxed shape" surface. The strict and
/// relaxed queries are both caller-provided — neither is passed through
/// [`fathomdb_query::derive_relaxed`] — and the resulting
/// [`SearchRows`] flows through the same retrieval, merge, and dedup
/// machinery as the adaptive [`TextSearchBuilder`] path.
///
/// `fallback_search(strict, None)` is the strict-only "dedup-on-write"
/// form: it runs the strict branch through the same plan shape (with no
/// relaxed sibling) so callers share the same retrieval and result surface
/// as adaptive `text_search()` rather than an ad hoc path.
///
/// Filters mirror [`TextSearchBuilder`]. There is intentionally no `.nodes`
/// or `.traverse` entry point — this helper is narrow. Its only job is to
/// run one or two search shapes through the shared policy.
#[must_use]
pub struct FallbackSearchBuilder<'e> {
    engine: &'e Engine,
    strict: TextQuery,
    relaxed: Option<TextQuery>,
    limit: usize,
    attribution_requested: bool,
    // Reuse a QueryBuilder as a filter accumulator so the fusion helper
    // partitions exactly the same predicates as TextSearchBuilder.
    filter_builder: QueryBuilder,
}

impl<'e> FallbackSearchBuilder<'e> {
    pub(crate) fn new(
        engine: &'e Engine,
        strict: impl Into<String>,
        relaxed: Option<&str>,
        limit: usize,
    ) -> Self {
        let strict = TextQuery::parse(&strict.into());
        let relaxed = relaxed.map(TextQuery::parse);
        // The filter accumulator's root kind is a placeholder — fallback_search
        // is kind-agnostic until the caller adds `.filter_kind_eq(...)`. We
        // pick an empty string so `partition_search_filters` ignores it (it
        // only inspects Filter steps).
        //
        // The accumulator is seeded with a no-op `text_search` step so that
        // `partition_search_filters` treats subsequent `.filter_*` calls as
        // post-search filters (the partitioner only fuses predicates that
        // appear after a TextSearch/VectorSearch step). The dummy step's
        // query text and limit are never executed — `compile_plan` pulls
        // the real strict/relaxed text queries and limit from the
        // `FallbackSearchBuilder` fields directly when it assembles the
        // `CompiledSearchPlan`.
        let filter_builder = QueryBuilder::nodes(String::new()).text_search("", 0);
        Self {
            engine,
            strict,
            relaxed,
            limit,
            attribution_requested: false,
            filter_builder,
        }
    }

    /// Request per-hit match attribution. Mirrors
    /// [`TextSearchBuilder::with_match_attribution`].
    pub fn with_match_attribution(mut self) -> Self {
        self.attribution_requested = true;
        self
    }

    /// Filter results to a single logical ID.
    pub fn filter_logical_id_eq(mut self, logical_id: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_logical_id_eq(logical_id);
        self
    }

    /// Filter results to nodes matching the given kind.
    ///
    /// P6-P2-4: unlike the adaptive `TextSearchBuilder` path (which pins
    /// `root_kind` from `Engine::query(kind)`), the narrow fallback helper
    /// applies the kind check through the fusable filter list only. The
    /// fusion pass pushes the resulting `KindEq` predicate into the
    /// `search_hits` CTE's WHERE clause, which is sufficient to narrow
    /// the result set and keeps the emitted SQL free of the redundant
    /// `src.kind = ?` / `fp.kind = ?` checks inside the inner UNION arms.
    pub fn filter_kind_eq(mut self, kind: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_kind_eq(kind);
        self
    }

    /// Filter results to nodes matching the given `source_ref`.
    pub fn filter_source_ref_eq(mut self, source_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_source_ref_eq(source_ref);
        self
    }

    /// Filter results to nodes where `content_ref` is not NULL.
    pub fn filter_content_ref_not_null(mut self) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_not_null();
        self
    }

    /// Filter results to nodes matching the given `content_ref` URI.
    pub fn filter_content_ref_eq(mut self, content_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_eq(content_ref);
        self
    }

    /// Filter results where a JSON property at `path` equals the given text value.
    pub fn filter_json_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_eq(path, value);
        self
    }

    /// Filter results where a JSON property at `path` equals the given boolean value.
    pub fn filter_json_bool_eq(mut self, path: impl Into<String>, value: bool) -> Self {
        self.filter_builder = self.filter_builder.filter_json_bool_eq(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than `value`.
    pub fn filter_json_integer_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than or equal to `value`.
    pub fn filter_json_integer_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gte(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than `value`.
    pub fn filter_json_integer_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than or equal to `value`.
    pub fn filter_json_integer_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is after `value`.
    pub fn filter_json_timestamp_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or after `value`.
    pub fn filter_json_timestamp_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is before `value`.
    pub fn filter_json_timestamp_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or before `value`.
    pub fn filter_json_timestamp_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lte(path, value);
        self
    }

    /// Filter results where a JSON text property at `path` equals
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError::KindRequiredForFusion`] if no
    /// `filter_kind_eq` has been chained on this builder. The fallback
    /// builder is kind-agnostic by default and cannot resolve a
    /// property-FTS schema without an explicit kind binding.
    pub fn filter_json_fused_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_text_eq".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_eq")?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_eq_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_text_in(
        mut self,
        path: impl Into<String>,
        values: Vec<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_text_in".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_text_in")?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_in_unchecked(path, values);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`. Non-fused; no FTS schema required.
    pub fn filter_json_text_in(mut self, path: impl Into<String>, values: Vec<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_in(path, values);
        self
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// greater than `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_timestamp_gt".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or after
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_timestamp_gte".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_gte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// before `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_timestamp_lt".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or before
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_timestamp_lte".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(
            self.engine,
            &kind,
            &path,
            "filter_json_fused_timestamp_lte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON boolean property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_bool_eq(
        mut self,
        path: impl Into<String>,
        value: bool,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        let kind = filter_builder_kind(&self.filter_builder)
            .ok_or_else(|| BuilderValidationError::KindRequiredForFusion {
                method: "filter_json_fused_bool_eq".to_owned(),
            })?
            .to_owned();
        validate_fusable_property_path(self.engine, &kind, &path, "filter_json_fused_bool_eq")?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_bool_eq_unchecked(path, value);
        Ok(self)
    }

    /// Compile the builder into a [`CompiledSearchPlan`] without executing
    /// it. Useful for tests and introspection.
    ///
    /// # Errors
    /// Returns [`CompileError`] if filter partitioning fails.
    pub fn compile_plan(&self) -> Result<CompiledSearchPlan, CompileError> {
        // `FallbackSearchBuilder` is kind-agnostic at the UNION level:
        // when `root_kind` is empty, the coordinator's `run_search_branch`
        // omits the `src.kind = ?` / `fp.kind = ?` predicates from the
        // inner UNION arms entirely, so the search runs across all node
        // kinds. Callers that want kind filtering chain
        // `.filter_kind_eq(kind)`, which adds a fusable `KindEq`
        // predicate (P6-P2-4: the fusion pass then pushes the check into
        // the outer `search_hits` CTE's WHERE clause — a single kind
        // check, not three). The narrow fallback helper therefore always
        // uses an empty root kind on this path.
        let mut ast = self.filter_builder.clone().into_ast();
        ast.root_kind = String::new();
        compile_search_plan_from_queries(
            &ast,
            self.strict.clone(),
            self.relaxed.clone(),
            self.limit,
            self.attribution_requested,
        )
    }

    /// Execute the fallback search and return matching hits.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute(&self) -> Result<SearchRows, EngineError> {
        let plan = self
            .compile_plan()
            .map_err(|e| EngineError::InvalidConfig(format!("search compilation failed: {e}")))?;
        self.engine
            .coordinator()
            .execute_compiled_search_plan(&plan)
    }
}

/// Tethered vector-search builder returned from
/// [`NodeQueryBuilder::vector_search`].
///
/// Accumulates filter predicates alongside a caller-provided vector query
/// and dispatches `.execute()` through
/// [`fathomdb_engine::ExecutionCoordinator::execute_compiled_vector_search`],
/// returning [`SearchRows`] whose hits carry
/// `modality = RetrievalModality::Vector`, `source = SearchHitSource::Vector`,
/// `match_mode = None`, and `vector_distance = Some(raw_distance)`. The
/// higher-is-better `score` field is the negated distance.
///
/// See `dev/design-adaptive-text-search-surface-addendum-1-vec.md` §Public
/// Surface for the full surface contract and degradation semantics.
#[must_use]
pub struct VectorSearchBuilder<'e> {
    engine: &'e Engine,
    root_kind: String,
    query: String,
    limit: usize,
    attribution_requested: bool,
    // Reuse a QueryBuilder as a filter accumulator so the fusion helper
    // partitions exactly the same predicates as TextSearchBuilder.
    filter_builder: QueryBuilder,
}

impl<'e> VectorSearchBuilder<'e> {
    pub(crate) fn new(
        engine: &'e Engine,
        root_kind: impl Into<String>,
        query: impl Into<String>,
        limit: usize,
    ) -> Self {
        let root_kind = root_kind.into();
        // Mirror FallbackSearchBuilder: the filter accumulator is seeded
        // with a no-op `vector_search("", 0)` step so that
        // `partition_search_filters` treats subsequent `.filter_*` calls
        // as post-search predicates. The P2-N2 fix tightened the
        // partitioner to only collect filters AFTER a search-step marker;
        // without this seed, `.filter_kind_eq("Goal")` would land in
        // neither bucket and would be silently dropped. The dummy step's
        // query text and limit are never executed — `compile_plan` pulls
        // the real vector query string and limit from the builder's
        // fields directly when it assembles the `CompiledVectorSearch`.
        let filter_builder = QueryBuilder::nodes(root_kind.clone()).vector_search("", 0);
        Self {
            engine,
            root_kind,
            query: query.into(),
            limit,
            attribution_requested: false,
            filter_builder,
        }
    }

    /// Request per-hit match attribution.
    ///
    /// When set, every returned hit carries
    /// `attribution: Some(HitAttribution { matched_paths: vec![] })` per
    /// addendum 1 §Attribution on vector hits. The empty `matched_paths`
    /// list is intentional — vector matches have no per-field provenance
    /// to attribute, but the `Some(...)` sentinel lets downstream code
    /// distinguish "attribution was requested and produced no paths" from
    /// "attribution was not requested at all".
    pub fn with_match_attribution(mut self) -> Self {
        self.attribution_requested = true;
        self
    }

    /// Filter results to a single logical ID.
    pub fn filter_logical_id_eq(mut self, logical_id: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_logical_id_eq(logical_id);
        self
    }

    /// Filter results to nodes matching the given kind.
    pub fn filter_kind_eq(mut self, kind: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_kind_eq(kind);
        self
    }

    /// Filter results to nodes matching the given `source_ref`.
    pub fn filter_source_ref_eq(mut self, source_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_source_ref_eq(source_ref);
        self
    }

    /// Filter results to nodes where `content_ref` is not NULL.
    pub fn filter_content_ref_not_null(mut self) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_not_null();
        self
    }

    /// Filter results to nodes matching the given `content_ref` URI.
    pub fn filter_content_ref_eq(mut self, content_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_eq(content_ref);
        self
    }

    /// Filter results where a JSON property at `path` equals the given text value.
    pub fn filter_json_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_eq(path, value);
        self
    }

    /// Filter results where a JSON property at `path` equals the given boolean value.
    pub fn filter_json_bool_eq(mut self, path: impl Into<String>, value: bool) -> Self {
        self.filter_builder = self.filter_builder.filter_json_bool_eq(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than `value`.
    pub fn filter_json_integer_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than or equal to `value`.
    pub fn filter_json_integer_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gte(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than `value`.
    pub fn filter_json_integer_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than or equal to `value`.
    pub fn filter_json_integer_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is after `value`.
    pub fn filter_json_timestamp_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or after `value`.
    pub fn filter_json_timestamp_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is before `value`.
    pub fn filter_json_timestamp_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or before `value`.
    pub fn filter_json_timestamp_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lte(path, value);
        self
    }

    /// Filter results where a JSON text property at `path` equals
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError`] if the root kind has no
    /// registered property-FTS schema or the schema does not cover
    /// `path`.
    pub fn filter_json_fused_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_text_eq",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_eq_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_text_in(
        mut self,
        path: impl Into<String>,
        values: Vec<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_text_in",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_in_unchecked(path, values);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`. Non-fused; no FTS schema required.
    pub fn filter_json_text_in(mut self, path: impl Into<String>, values: Vec<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_in(path, values);
        self
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// greater than `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_gt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or after
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_gte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// before `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_lt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or before
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_lte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON boolean property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_bool_eq(
        mut self,
        path: impl Into<String>,
        value: bool,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_bool_eq",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_bool_eq_unchecked(path, value);
        Ok(self)
    }

    /// Compile the builder into a [`CompiledVectorSearch`] without executing
    /// it. Useful for tests and introspection.
    ///
    /// # Errors
    /// Returns [`CompileError`] if filter partitioning fails.
    pub fn compile_plan(&self) -> Result<CompiledVectorSearch, CompileError> {
        let mut ast = self.filter_builder.clone().into_ast();
        ast.root_kind.clone_from(&self.root_kind);
        let mut compiled = compile_vector_search(&ast)?;
        // The seed `.vector_search("", 0)` step on the filter accumulator
        // is an artifact of the partition workaround; `compile_plan` pulls
        // the caller's real query text and limit from `self` directly.
        compiled.query_text.clone_from(&self.query);
        compiled.limit = self.limit;
        compiled.attribution_requested = self.attribution_requested;
        Ok(compiled)
    }

    /// Execute the vector search and return matching hits.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails. A
    /// capability miss (sqlite-vec unavailable) is NOT an error: it
    /// returns an empty [`SearchRows`] with `was_degraded = true`.
    pub fn execute(&self) -> Result<SearchRows, EngineError> {
        let plan = self
            .compile_plan()
            .map_err(|e| EngineError::InvalidConfig(format!("search compilation failed: {e}")))?;
        self.engine
            .coordinator()
            .execute_compiled_vector_search(&plan)
    }
}

/// Tethered unified retrieval builder returned from
/// [`NodeQueryBuilder::search`].
///
/// `SearchBuilder` is the Phase 12 primary retrieval entry point per
/// `dev/design-adaptive-text-search-surface-addendum-1-vec.md` §Public
/// Surface. It accumulates filter predicates alongside a caller-provided
/// raw query string, compiles into a [`CompiledRetrievalPlan`] via
/// [`fathomdb_query::compile_retrieval_plan`], and dispatches `.execute()`
/// through [`fathomdb_engine::ExecutionCoordinator::execute_retrieval_plan`]
/// to return [`SearchRows`] with the strict/relaxed/vector blocks fused
/// under the addendum's block precedence rules.
///
/// **v1 scope**: the unified planner's vector branch slot is wired
/// architecturally but never fires through `search()` because read-time
/// embedding of natural-language queries is deferred. Until that future
/// phase lands, every `SearchBuilder::execute()` result has
/// `vector_hit_count == 0` regardless of vector capability availability.
/// Callers who want vector retrieval today must use the advanced
/// `vector_search()` override directly with a caller-provided vector
/// literal.
#[must_use]
pub struct SearchBuilder<'e> {
    engine: &'e Engine,
    root_kind: String,
    query: String,
    limit: usize,
    attribution_requested: bool,
    // Reuse a QueryBuilder as a filter accumulator so the fusion helper
    // partitions exactly the same predicates as TextSearchBuilder /
    // FallbackSearchBuilder / VectorSearchBuilder. The accumulator is
    // seeded with a no-op `text_search("", 0)` step so `partition_search_filters`
    // treats subsequent `.filter_*` calls as post-search predicates;
    // without that seed the predicates would land in neither bucket and
    // would be silently dropped. The dummy step's query text and limit
    // are never executed — `compile_plan` rewrites the AST with the real
    // `Search { query, limit }` step before calling
    // `compile_retrieval_plan`.
    filter_builder: QueryBuilder,
}

impl<'e> SearchBuilder<'e> {
    pub(crate) fn new(
        engine: &'e Engine,
        root_kind: impl Into<String>,
        query: impl Into<String>,
        limit: usize,
    ) -> Self {
        let root_kind = root_kind.into();
        let filter_builder = QueryBuilder::nodes(root_kind.clone()).text_search("", 0);
        Self {
            engine,
            root_kind,
            query: query.into(),
            limit,
            attribution_requested: false,
            filter_builder,
        }
    }

    /// Request per-hit match attribution on the resulting [`SearchRows`].
    /// Mirrors [`TextSearchBuilder::with_match_attribution`] semantics for
    /// text hits and [`VectorSearchBuilder::with_match_attribution`] for
    /// vector hits.
    pub fn with_match_attribution(mut self) -> Self {
        self.attribution_requested = true;
        self
    }

    /// Filter results to a single logical ID.
    pub fn filter_logical_id_eq(mut self, logical_id: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_logical_id_eq(logical_id);
        self
    }

    /// Filter results to nodes matching the given kind.
    pub fn filter_kind_eq(mut self, kind: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_kind_eq(kind);
        self
    }

    /// Filter results to nodes matching the given `source_ref`.
    pub fn filter_source_ref_eq(mut self, source_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_source_ref_eq(source_ref);
        self
    }

    /// Filter results to nodes where `content_ref` is not NULL.
    pub fn filter_content_ref_not_null(mut self) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_not_null();
        self
    }

    /// Filter results to nodes matching the given `content_ref` URI.
    pub fn filter_content_ref_eq(mut self, content_ref: impl Into<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_content_ref_eq(content_ref);
        self
    }

    /// Filter results where a JSON property at `path` equals the given text value.
    pub fn filter_json_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_eq(path, value);
        self
    }

    /// Filter results where a JSON property at `path` equals the given boolean value.
    pub fn filter_json_bool_eq(mut self, path: impl Into<String>, value: bool) -> Self {
        self.filter_builder = self.filter_builder.filter_json_bool_eq(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than `value`.
    pub fn filter_json_integer_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is greater than or equal to `value`.
    pub fn filter_json_integer_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_gte(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than `value`.
    pub fn filter_json_integer_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lt(path, value);
        self
    }

    /// Filter results where a JSON integer at `path` is less than or equal to `value`.
    pub fn filter_json_integer_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_integer_lte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is after `value`.
    pub fn filter_json_timestamp_gt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or after `value`.
    pub fn filter_json_timestamp_gte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_gte(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is before `value`.
    pub fn filter_json_timestamp_lt(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lt(path, value);
        self
    }

    /// Filter results where a JSON timestamp at `path` is at or before `value`.
    pub fn filter_json_timestamp_lte(mut self, path: impl Into<String>, value: i64) -> Self {
        self.filter_builder = self.filter_builder.filter_json_timestamp_lte(path, value);
        self
    }

    /// Filter results where a JSON text property at `path` equals
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// Returns [`BuilderValidationError`] if the root kind has no
    /// registered property-FTS schema or the schema does not cover
    /// `path`.
    pub fn filter_json_fused_text_eq(
        mut self,
        path: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_text_eq",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_eq_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_text_in(
        mut self,
        path: impl Into<String>,
        values: Vec<String>,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_text_in",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_text_in_unchecked(path, values);
        Ok(self)
    }

    /// Filter results where a JSON text property at `path` is one of
    /// `values`. Non-fused; no FTS schema required.
    pub fn filter_json_text_in(mut self, path: impl Into<String>, values: Vec<String>) -> Self {
        self.filter_builder = self.filter_builder.filter_json_text_in(path, values);
        self
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// greater than `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_gt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or after
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_gte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_gte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_gte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is strictly
    /// before `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lt(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_lt",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lt_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON timestamp at `path` is at or before
    /// `value`, with fusion semantics.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_timestamp_lte(
        mut self,
        path: impl Into<String>,
        value: i64,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_timestamp_lte",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_timestamp_lte_unchecked(path, value);
        Ok(self)
    }

    /// Filter results where a JSON boolean property at `path` equals
    /// `value`, with fusion semantics. See
    /// [`Self::filter_json_fused_text_eq`] for the contract.
    ///
    /// # Errors
    /// See [`Self::filter_json_fused_text_eq`].
    pub fn filter_json_fused_bool_eq(
        mut self,
        path: impl Into<String>,
        value: bool,
    ) -> Result<Self, BuilderValidationError> {
        let path = path.into();
        validate_fusable_property_path(
            self.engine,
            &self.root_kind,
            &path,
            "filter_json_fused_bool_eq",
        )?;
        self.filter_builder = self
            .filter_builder
            .filter_json_fused_bool_eq_unchecked(path, value);
        Ok(self)
    }

    /// Compile the builder into a [`CompiledRetrievalPlan`] without executing
    /// it. Useful for tests and introspection.
    ///
    /// # Errors
    /// Returns [`CompileError`] if filter partitioning or text-query parsing
    /// fails.
    pub fn compile_plan(&self) -> Result<CompiledRetrievalPlan, CompileError> {
        // Take the filter accumulator AST and rewrite the seed
        // `text_search("", 0)` step into the real `Search { query, limit }`
        // step. Rewriting in place (rather than appending) preserves the
        // post-search position so that the filter partitioner classifies
        // every chained `.filter_*` predicate the same way the text/vector
        // builders do.
        let mut ast: QueryAst = self.filter_builder.clone().into_ast();
        ast.root_kind.clone_from(&self.root_kind);
        let mut replaced = false;
        for step in &mut ast.steps {
            if let QueryStep::TextSearch {
                query: TextQuery::Empty,
                limit: 0,
            } = step
            {
                *step = QueryStep::Search {
                    query: self.query.clone(),
                    limit: self.limit,
                };
                replaced = true;
                break;
            }
        }
        debug_assert!(
            replaced,
            "SearchBuilder filter accumulator must contain the seed TextSearch step"
        );
        let mut plan = fathomdb_query::compile_retrieval_plan(&ast)?;
        plan.text.strict.attribution_requested = self.attribution_requested;
        if let Some(relaxed) = plan.text.relaxed.as_mut() {
            relaxed.attribution_requested = self.attribution_requested;
        }
        Ok(plan)
    }

    /// Execute the unified retrieval plan and return matching hits.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute(&self) -> Result<SearchRows, EngineError> {
        let plan = self
            .compile_plan()
            .map_err(|e| EngineError::InvalidConfig(format!("search compilation failed: {e}")))?;
        self.engine
            .coordinator()
            .execute_retrieval_plan(&plan, &self.query)
    }

    /// Add an expansion slot that traverses edges per root result.
    /// Pass `edge_filter: None` to preserve pre-Pack-D behavior.
    pub fn expand(
        mut self,
        slot: impl Into<String>,
        direction: fathomdb_query::TraverseDirection,
        label: impl Into<String>,
        max_depth: usize,
        filter: Option<fathomdb_query::Predicate>,
        edge_filter: Option<fathomdb_query::Predicate>,
    ) -> Self {
        self.filter_builder =
            self.filter_builder
                .expand(slot, direction, label, max_depth, filter, edge_filter);
        self
    }

    /// Compile this builder's AST into an executable grouped query.
    ///
    /// Rewrites the seed `TextSearch` step into a real text-search step before
    /// delegating to [`compile_grouped_query`].
    ///
    /// # Errors
    /// Returns [`CompileError`] if grouped compilation fails.
    pub fn compile_grouped(&self) -> Result<CompiledGroupedQuery, CompileError> {
        let mut ast: QueryAst = self.filter_builder.clone().into_ast();
        ast.root_kind.clone_from(&self.root_kind);
        let mut replaced = false;
        for step in &mut ast.steps {
            if let QueryStep::TextSearch {
                query: TextQuery::Empty,
                limit: 0,
            } = step
            {
                *step = QueryStep::TextSearch {
                    query: TextQuery::parse(&self.query),
                    limit: self.limit,
                };
                replaced = true;
                break;
            }
        }
        debug_assert!(
            replaced,
            "SearchBuilder filter accumulator must contain the seed TextSearch step"
        );
        compile_grouped_query(&ast)
    }

    /// Execute the grouped query and return root rows plus named expansion slots.
    ///
    /// # Errors
    /// Returns [`EngineError`] if compilation or execution fails.
    pub fn execute_grouped(self) -> Result<GroupedQueryRows, EngineError> {
        let compiled = self.compile_grouped().map_err(|e| {
            EngineError::InvalidConfig(format!("grouped query compilation failed: {e}"))
        })?;
        self.engine
            .coordinator()
            .execute_compiled_grouped_read(&compiled)
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::panic)]
mod tests {
    use super::{FallbackSearchBuilder, VectorSearchBuilder};
    use crate::{BuilderValidationError, Engine, EngineOptions};
    use fathomdb_query::Predicate;
    use tempfile::NamedTempFile;

    fn open_engine_with_schema(register: bool) -> (NamedTempFile, Engine) {
        let db = NamedTempFile::new().expect("temporary db");
        let engine = Engine::open(EngineOptions::new(db.path())).expect("engine opens");
        if register {
            engine
                .register_fts_property_schema(
                    "Note",
                    &["$.title".to_owned(), "$.body".to_owned()],
                    None,
                )
                .expect("register fts property schema");
        }
        (db, engine)
    }

    #[test]
    fn node_query_fused_text_eq_requires_registered_schema() {
        let (_db, engine) = open_engine_with_schema(false);
        let result = engine
            .query("Note")
            .filter_json_fused_text_eq("$.title", "hello");
        let Err(err) = result else {
            panic!("must reject fused filter without schema");
        };
        assert!(
            matches!(err, BuilderValidationError::MissingPropertyFtsSchema { ref kind } if kind == "Note"),
            "expected MissingPropertyFtsSchema, got {err:?}"
        );
    }

    #[test]
    fn node_query_fused_text_eq_rejects_path_not_in_schema() {
        let (_db, engine) = open_engine_with_schema(true);
        let result = engine
            .query("Note")
            .filter_json_fused_text_eq("$.not_covered", "hello");
        let Err(err) = result else {
            panic!("path not in schema must be rejected");
        };
        assert!(
            matches!(err, BuilderValidationError::PathNotIndexed { ref kind, ref path } if kind == "Note" && path == "$.not_covered"),
            "expected PathNotIndexed, got {err:?}"
        );
    }

    #[test]
    fn node_query_fused_text_eq_succeeds_with_registered_schema() {
        let (_db, engine) = open_engine_with_schema(true);
        let builder = engine
            .query("Note")
            .filter_json_fused_text_eq("$.title", "hello")
            .expect("fused filter with registered schema must succeed");
        let compiled = builder.compile().expect("compile");
        // The fused predicate must land inside base_candidates, with a
        // src.properties json_extract clause.
        assert!(
            compiled.sql.contains("json_extract(src.properties, ?"),
            "fused filter must emit against src.properties, got {}",
            compiled.sql
        );
    }

    #[test]
    fn text_search_fused_timestamp_gt_validates_and_compiles() {
        let (_db, engine) = open_engine_with_schema(true);
        // Path must be present in schema.
        engine
            .register_fts_property_schema("Note2", &["$.written_at".to_owned()], None)
            .expect("register Note2 schema");
        let builder = engine
            .query("Note2")
            .text_search("budget", 5)
            .filter_json_fused_timestamp_gt("$.written_at", 1_700_000_000)
            .expect("fused timestamp gt must succeed with schema");
        let _ = builder.compile().expect("compile succeeds");
    }

    #[test]
    fn vector_search_fused_text_eq_validates() {
        let (_db, engine) = open_engine_with_schema(true);
        let result = VectorSearchBuilder::new(&engine, "NoSchema", "q", 5)
            .filter_json_fused_text_eq("$.title", "hello");
        let Err(err) = result else {
            panic!("missing schema must error");
        };
        assert!(
            matches!(err, BuilderValidationError::MissingPropertyFtsSchema { .. }),
            "expected MissingPropertyFtsSchema, got {err:?}"
        );
        let ok = VectorSearchBuilder::new(&engine, "Note", "q", 5)
            .filter_json_fused_text_eq("$.title", "hello");
        assert!(ok.is_ok(), "registered kind must succeed");
    }

    #[test]
    fn fallback_search_fused_text_eq_requires_kind_binding() {
        let (_db, engine) = open_engine_with_schema(true);
        let result = FallbackSearchBuilder::new(&engine, "budget", None, 10)
            .filter_json_fused_text_eq("$.title", "hello");
        let Err(err) = result else {
            panic!("no kind binding must error");
        };
        assert!(
            matches!(err, BuilderValidationError::KindRequiredForFusion { .. }),
            "expected KindRequiredForFusion, got {err:?}"
        );
        let ok = FallbackSearchBuilder::new(&engine, "budget", None, 10)
            .filter_kind_eq("Note")
            .filter_json_fused_text_eq("$.title", "hello");
        assert!(ok.is_ok(), "kind-bound fallback fused filter must succeed");
    }

    #[test]
    fn unified_search_fused_text_eq_validates() {
        let (_db, engine) = open_engine_with_schema(true);
        let ok = engine
            .query("Note")
            .search("hello", 5)
            .filter_json_fused_text_eq("$.title", "hello");
        assert!(
            ok.is_ok(),
            "unified search builder must accept fused filter"
        );
        let result = engine
            .query("Unknown")
            .search("hello", 5)
            .filter_json_fused_text_eq("$.title", "hello");
        let Err(err) = result else {
            panic!("missing schema must error");
        };
        assert!(matches!(
            err,
            BuilderValidationError::MissingPropertyFtsSchema { .. }
        ));
    }

    #[test]
    fn existing_filter_json_text_eq_still_compiles_unchanged_regression() {
        // Regression guard for D3: the post-filter filter_json_* family
        // must retain its original non-fused semantics. On the text
        // search path this means the predicate lands in the outer WHERE
        // against `n.properties`, not inside the CTE.
        let (_db, engine) = open_engine_with_schema(false);
        let compiled = engine
            .query("Note")
            .text_search("budget", 5)
            .filter_json_text_eq("$.status", "active")
            .compile()
            .expect("compile");
        assert!(
            compiled
                .sql
                .contains("\n  AND json_extract(n.properties, ?"),
            "filter_json_text_eq must emit into outer WHERE, got {}",
            compiled.sql
        );
    }

    /// P7.5-N1: pin the dummy-step workaround invariant.
    ///
    /// `FallbackSearchBuilder` seeds an inert `text_search("", 0)` step
    /// into its internal filter accumulator so that
    /// `partition_search_filters` treats subsequent `.filter_*` calls as
    /// post-search predicates (the partitioner only classifies filters
    /// that appear AFTER a `TextSearch` or `VectorSearch` step in source
    /// order). Without that seed, `.filter_kind_eq("Goal")` would land in
    /// neither the fusable nor the residual bucket and would be silently
    /// dropped. This test compiles a plan via the public builder API and
    /// verifies the kind predicate ends up in `fusable_filters`.
    #[test]
    fn fallback_builder_filter_kind_eq_fuses_without_explicit_text_search_step() {
        let db = NamedTempFile::new().expect("temporary db");
        let engine =
            Engine::open(EngineOptions::new(db.path())).expect("engine opens for unit test");

        let builder = FallbackSearchBuilder::new(&engine, "budget", Some("budget OR nothing"), 10)
            .filter_kind_eq("Goal");
        let plan = builder.compile_plan().expect("compile plan");

        assert!(
            plan.strict
                .fusable_filters
                .iter()
                .any(|p| matches!(p, Predicate::KindEq(k) if k == "Goal")),
            "KindEq(\"Goal\") must land in strict.fusable_filters (got {:?})",
            plan.strict.fusable_filters
        );
        assert!(
            plan.strict.residual_filters.is_empty(),
            "strict.residual_filters should be empty for a single kind filter (got {:?})",
            plan.strict.residual_filters
        );

        let relaxed = plan
            .relaxed
            .as_ref()
            .expect("relaxed branch present when caller supplied a relaxed query");
        assert!(
            relaxed
                .fusable_filters
                .iter()
                .any(|p| matches!(p, Predicate::KindEq(k) if k == "Goal")),
            "KindEq(\"Goal\") must also land in relaxed.fusable_filters (got {:?})",
            relaxed.fusable_filters
        );
    }
}