frankensearch 0.4.0

//! # frankensearch
//!
//! Two-tier hybrid search for Rust: sub-millisecond initial results,
//! quality-refined rankings in ~150ms.
//!
//! frankensearch combines **lexical** (native Quill or the Tantivy oracle) and
//! **semantic** (vector cosine similarity) search via [Reciprocal Rank
//! Fusion][rrf], with a two-tier
//! progressive embedding model that delivers results in two phases:
//!
//! 1. **Phase 1 (Initial):** Fast embedder (potion-128M, 256d, ~0.57ms) produces
//!    results immediately via brute-force vector search + optional BM25 fusion.
//! 2. **Phase 2 (Refined):** Quality embedder (MiniLM-L6-v2, 384d, ~128ms)
//!    re-scores the top candidates for higher relevance.
//!
//! Consumers receive results progressively via [`SearchPhase`] callbacks, so UIs
//! can display fast results while quality refinement runs in the background.
//!
//! [rrf]: https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf
//!
//! # Quick Start
//!
//! Production indexing and search need a verified semantic embedder.
//! `HashEmbedder` is an explicit control double, not a semantic engine;
//! `auto_detect` / `auto_detect_with` still return that control stack when
//! no model is present.
//!
//! ```rust,no_run
//! use std::path::Path;
//! use std::sync::Arc;
//! use frankensearch::prelude::*;
//! use frankensearch::{EmbedderStack, IndexBuilder, TwoTierIndex};
//!
//! asupersync::test_utils::run_test_with_cx(|cx| async move {
//!     let stack = EmbedderStack::auto_detect_semantic_with(Some(Path::new("./models")))
//!         .expect("production search needs a verified semantic embedder");
//!
//!     IndexBuilder::new("./my_index")
//!         .with_embedder_stack(stack)
//!         .add_document("doc-1", "Rust ownership and borrowing")
//!         .add_document("doc-2", "Python garbage collection")
//!         .build(&cx)
//!         .await
//!         .expect("build index");
//!
//!     let stack = EmbedderStack::auto_detect_semantic_with(Some(Path::new("./models")))
//!         .expect("search must use the same semantic family the index was built with");
//!     let index = Arc::new(
//!         TwoTierIndex::open(Path::new("./my_index"), TwoTierConfig::default()).unwrap(),
//!     );
//!     let mut searcher = TwoTierSearcher::new(index, stack.fast_arc(), TwoTierConfig::default());
//!     if let Some(quality) = stack.quality_arc() {
//!         searcher = searcher.with_quality_embedder(quality);
//!     }
//!     let (results, _metrics) = searcher
//!         .search_collect(&cx, "memory management", 10)
//!         .await
//!         .expect("search");
//!
//!     for result in &results {
//!         println!("{}: {:.4}", result.doc_id, result.score);
//!     }
//!
//!     // With `--features quill`, IndexBuilder also emits a bulk-finalized
//!     // native lexical index. `open_hybrid` reopens every arm and attaches
//!     // the active lexical reader — the wiring earlier examples dropped.
//!     #[cfg(feature = "quill")]
//!     {
//!         use frankensearch::LexicalRead;
//!
//!         let parts = frankensearch::open_hybrid(
//!             &cx,
//!             "./my_index",
//!             frankensearch::TwoTierConfig::default(),
//!         )
//!         .await
//!         .expect("open hybrid index");
//!         let lexical = parts.lexical.expect("lexical arm attached");
//!         let lexical_hits = lexical
//!             .search(&cx, "ownership", 10)
//!             .await
//!             .expect("search lexical arm");
//!         assert!(!lexical_hits.is_empty());
//!     }
//! });
//! ```
//!
//! # Architecture
//!
//! ```text
//!  Query ─┬─► Fast Embed (256d) ─► Vector Search ─┐
//!         │                                         ├─► RRF Fusion ─► Phase 1 Results
//!         └─► Quill / Tantivy-oracle BM25 ─────────┘
//!                                                        │
//!                                              Quality Embed (384d)
//!                                                        │
//!                                                   Score Blend
//!                                                        │
//!                                                  Phase 2 Results
//! ```
//!
//! ## Crate Layout
//!
//! | Crate | Purpose |
//! |-------|---------|
//! | [`frankensearch-core`](core) | Types, traits, errors, config |
//! | [`frankensearch-embed`](embed) | Embedder implementations (hash, model2vec, fastembed) |
//! | [`frankensearch-index`](index) | FSVI vector index format, brute-force + HNSW search |
//! | [`frankensearch-fusion`](fusion) | RRF fusion, blending, [`TwoTierSearcher`] orchestration |
//! | `frankensearch-quill` | Native lexical backend (`quill`) |
//! | `frankensearch-lexical` | Tantivy oracle backend (`lexical-tantivy`) |
//! | `frankensearch-rerank` | `FlashRank` cross-encoder (feature-gated) |
//!
//! ## Key Types
//!
//! - [`IndexBuilder`] — Build a search index from documents
//! - [`TwoTierSearcher`] — Progressive two-phase search orchestrator
//! - [`TwoTierConfig`] — Search configuration (blend factor, budgets, fast-only mode)
//! - [`TwoTierMetrics`] — Per-search timing and diagnostic metrics
//! - [`SearchPhase`] — Progressive result delivery (Initial / Refined / `RefinementFailed`)
//! - [`EmbedderStack`] — Fast + optional quality embedder pair
//! - [`VectorIndex`] — Low-level FSVI vector index reader
//!
//! # Performance
//!
//! Measured on a single core (no GPU), 10K document corpus:
//!
//! | Operation | Embedder | Latency |
//! |-----------|----------|---------|
//! | Hash embed (256d) | FNV-1a | ~11 μs |
//! | Fast embed (256d) | potion-128M | ~0.57 ms |
//! | Quality embed (384d) | MiniLM-L6-v2 | ~128 ms |
//! | Vector search (10K, top-10) | brute-force | ~2 ms |
//! | RRF fusion (500+500) | - | ~1 ms |
//! | Full pipeline (hash, 10K) | hash only | ~3 ms |
//!
//! # Feature Flags
//!
//! | Feature      | Description                                            |
//! |--------------|--------------------------------------------------------|
//! | `hash`       | FNV-1a hash embedder (default, zero dependencies)      |
//! | `model2vec`  | potion-128M static embedder (fast tier, ~0.57ms)       |
//! | `fastembed`  | MiniLM-L6-v2 ONNX embedder (quality tier, ~128ms)      |
//! | `lexical`    | Native Quill production lexical backend                |
//! | `quill`      | Native Quill lexical engine and builder integration     |
//! | `lexical-tantivy` | Explicit Tantivy oracle/comparator surface       |
//! | `cass-compat` | External CASS schema-v8 Tantivy-format interoperability |
//! | `rerank`     | `FlashRank` cross-encoder reranking                    |
//! | `ann`        | HNSW approximate nearest-neighbor index                |
//! | `download`   | Model auto-download from `HuggingFace` via asupersync  |
//! | `storage`    | `FrankenSQLite` document metadata + embedding queue     |
//! | `durability` | `RaptorQ` self-healing for persistent index artifacts   |
//! | `fts5`       | Enables `FrankenSQLite` FTS5 lexical backend wiring     |
//! | `semantic`   | `hash` + `model2vec` + `fastembed`                     |
//! | `hybrid`     | `semantic` + `lexical`                                 |
//! | `persistent` | `hybrid` + `storage`                                   |
//! | `durable`    | `persistent` + `durability`                            |
//! | `full`       | `durable` + `rerank` + `ann` + `download`             |
//! | `full-fts5`  | `full` + `fts5`                                        |
//!
//! ## Recommended Feature Combinations
//!
//! - **Development/testing:** `default` (hash only, no downloads)
//! - **Production semantic:** `semantic` + `download`
//! - **Persistent hybrid search:** `persistent`
//! - **Maximum durability:** `durable` or `full`
//!
//! # Async Runtime
//!
//! frankensearch uses [asupersync](https://docs.rs/asupersync) exclusively — **not
//! tokio**. All async methods take `&Cx` (capability context) as their first
//! parameter. The `Cx` is provided by the consumer's asupersync runtime;
//! frankensearch never creates its own runtime.

#[cfg(all(feature = "fts5", not(feature = "storage")))]
compile_error!("feature `fts5` requires feature `storage`");

#[cfg(all(
    feature = "persistent",
    not(all(feature = "hybrid", feature = "storage"))
))]
compile_error!("feature `persistent` requires both `hybrid` and `storage`");

#[cfg(all(
    feature = "durable",
    not(all(feature = "persistent", feature = "durability"))
))]
compile_error!("feature `durable` requires both `persistent` and `durability`");

#[cfg(all(feature = "full-fts5", not(all(feature = "full", feature = "fts5"))))]
compile_error!("feature `full-fts5` requires both `full` and `fts5`");

// ─── Sub-crate module aliases (advanced access) ─────────────────────────────

/// Core types, traits, and error definitions.
pub use frankensearch_core as core;
/// Embedding model implementations and auto-detection.
pub use frankensearch_embed as embed;
/// RRF fusion, blending, search orchestration, and queue management.
pub use frankensearch_fusion as fusion;
/// Vector index I/O (FSVI format) and brute-force/HNSW search.
pub use frankensearch_index as index;

#[cfg(feature = "lexical")]
/// Native Quill production lexical backend.
///
/// Tantivy remains available only from [`lexical_tantivy`] when the explicit
/// `lexical-tantivy` feature is selected.
pub use frankensearch_quill as lexical;

#[cfg(feature = "lexical-tantivy")]
/// Explicit Tantivy-native backend for oracle and foreign-index consumers.
///
/// Unlike `lexical`, this namespace is stable across the facade's default
/// lexical-backend transition. Consumers that read or write Tantivy-native
/// formats must opt into `lexical-tantivy` (or `cass-compat`) and import this
/// namespace explicitly.
pub use frankensearch_lexical as lexical_tantivy;

#[cfg(feature = "quill")]
/// Native Quill lexical backend.
pub use frankensearch_quill as quill;

#[cfg(feature = "rerank")]
/// `FlashRank` cross-encoder reranking.
pub use frankensearch_rerank as rerank;

#[cfg(feature = "storage")]
/// `FrankenSQLite` storage backend.
pub use frankensearch_storage as storage;

#[cfg(feature = "durability")]
/// `RaptorQ` self-healing durability layer.
pub use frankensearch_durability as durability;

// ─── Feature-gated facade exports (flat import surface) ────────────────────

#[cfg(feature = "storage")]
pub use frankensearch_storage::{
    BatchResult, ContentHasher, DeduplicationDecision, DocumentRecord, IndexMetadata, IngestAction,
    IngestResult, JobQueueConfig, JobQueueMetrics, PersistentJobQueue, StalenessCheck,
    StalenessReason, Storage, StorageBackedJobRunner, StorageConfig,
};

#[cfg(feature = "fts5")]
pub use frankensearch_storage::{
    Fts5AdapterConfig as Fts5Config, Fts5ContentMode, Fts5LexicalSearch,
    Fts5TokenizerChoice as Fts5Tokenizer,
};

#[cfg(feature = "durability")]
pub use frankensearch_durability::{
    DefaultSymbolCodec, DurabilityConfig, DurabilityMetrics, FileHealth,
    FileProtectionResult as ProtectionResult, FileProtector, FileRepairOutcome as RepairResult,
    FsviProtector, RepairCodec, RepairCodecConfig, VerifyResult as RepairCodecVerifyResult,
};

// ─── Async runtime re-exports ───────────────────────────────────────────────

/// Capability context for structured concurrency (from asupersync).
///
/// All async search methods take `&Cx` as their first parameter. The `Cx` flows
/// down from the consumer's asupersync runtime — frankensearch does not create
/// its own runtime.
pub use asupersync::Cx;

// ─── Core types (always available) ──────────────────────────────────────────

// Error types
pub use frankensearch_core::error::{SearchError, SearchResult};

// Configuration
pub use frankensearch_core::config::{TwoTierConfig, TwoTierMetrics};

// Search result types
pub use frankensearch_core::types::{
    FusedHit, IndexableDocument, PhaseMetrics, RankChanges, ScoreSource, ScoredResult, SearchMode,
    SearchPhase, VectorHit,
};

// Telemetry types
pub use frankensearch_core::types::{EmbeddingMetrics, IndexMetrics, SearchMetrics};

// Traits
pub use frankensearch_core::traits::{
    Embedder, LexicalCandidateBatch, LexicalHydrationContext, LexicalRead, LexicalWrite,
    MetricsExporter, ModelCategory, ModelInfo, ModelTier, NoOpMetricsExporter, Reranker,
    SearchFuture, SharedMetricsExporter, SyncEmbed, SyncEmbedderAdapter, SyncRerank,
    SyncRerankerAdapter,
};
pub use frankensearch_core::{
    AttestedDaemonEmbeddingResponseV1, DAEMON_ATTESTATION_SCHEMA_V1, DAEMON_CHALLENGE_SCHEMA_V1,
    DAEMON_CONNECTION_IDENTITY_SCHEMA_V1, DaemonChallengeV1, DaemonClient,
    DaemonConnectionIdentityV1, DaemonEmbeddingAttestationV1, DaemonError, DaemonOperationV1,
    DaemonRetryConfig, MIN_DAEMON_ATTESTATION_KEY_BYTES, daemon_embedding_payload_sha256,
    daemon_endpoint_fingerprint, daemon_executable_fingerprint, daemon_ordered_request_sha256,
};

// Reranker support types
pub use frankensearch_core::traits::{RerankDocument, RerankScore};

// Query classification
pub use frankensearch_core::query_class::QueryClass;

// Text canonicalization
pub use frankensearch_core::canonicalize::{Canonicalizer, DefaultCanonicalizer};
pub use frankensearch_core::fingerprint::{
    DEFAULT_SEMANTIC_CHANGE_THRESHOLD, DocumentFingerprint, SIGNIFICANT_CHAR_COUNT_CHANGE_THRESHOLD,
};

// IR evaluation metrics
pub use frankensearch_core::metrics_eval::{
    BootstrapCi, BootstrapComparison, QualityComparison, QualityMetric, QualityMetricComparison,
    QualityMetricSamples, bootstrap_ci, bootstrap_compare, map_at_k, mrr, ndcg_at_k,
    quality_comparison, recall_at_k,
};

// Utility functions
pub use frankensearch_core::traits::{cosine_similarity, l2_normalize, truncate_embedding};

// ─── Embedder stack (always available) ──────────────────────────────────────

pub use frankensearch_embed::auto_detect::{DimReduceEmbedder, EmbedderStack, TwoTierAvailability};
pub use frankensearch_embed::model_registry::{EmbedderRegistry, RegisteredEmbedder};

// ─── Vector index (always available) ────────────────────────────────────────

pub use frankensearch_index::{
    InMemoryTwoTierIndex, InMemoryVectorIndex, TwoTierIndex, TwoTierIndexBuilder,
    TwoTierIndexPaths, VectorIndex, VectorIndexWriter,
};

#[cfg(feature = "ann")]
pub use frankensearch_index::{AnnSearchStats, HnswConfig, HnswIndex, HnswLoadDisposition};

// ─── Fusion and search orchestration (always available) ─────────────────────

pub use frankensearch_fusion::{
    AssumedDaemonClient, AssumedDaemonEmbeddingBatchV1, DaemonFallbackEmbedder,
    DaemonFallbackReranker, DaemonTrustLevelV1, FederatedConfig, FederatedCoverage,
    FederatedFusion, FederatedHit, FederatedResponse, FederatedSearcher, FederatedShardError,
    NoopDaemonClient, PinnedDaemonVerifierV1, RrfConfig, SyncLexicalSearch, SyncSearchIterator,
    SyncTwoTierSearcher, TwoTierSearcher, blend_two_tier, candidate_count, rrf_fuse,
};

#[cfg(feature = "graph")]
pub use frankensearch_fusion::GraphRanker;

// ─── Feature-gated embedder re-exports ──────────────────────────────────────

#[cfg(feature = "hash")]
pub use frankensearch_embed::hash_embedder::{HashAlgorithm, HashEmbedder};

#[cfg(feature = "model2vec")]
pub use frankensearch_embed::model2vec_embedder::Model2VecEmbedder;

#[cfg(feature = "fastembed")]
pub use frankensearch_embed::fastembed_embedder::FastEmbedEmbedder;

// ─── Feature-gated lexical re-exports ───────────────────────────────────────

#[cfg(feature = "lexical-tantivy")]
pub use frankensearch_lexical::TantivyIndex;

#[cfg(feature = "cass-compat")]
pub use frankensearch_lexical::cass_compat;

#[cfg(feature = "quill")]
pub use frankensearch_quill::{
    QueryExplanation, QuillConfig, QuillIndex, QuillSearchIndex, QuillSearchResult, SegmentStats,
    SegmentStatsProvider, SnippetConfig,
};

#[cfg(feature = "quill")]
pub use frankensearch_quill::{QuillHit as LexicalIdHit, QuillSnippetHit as LexicalHit};

#[cfg(feature = "quill")]
pub use frankensearch_fusion::QuillSyncLexicalSearch;

// ─── Feature-gated reranker re-exports ──────────────────────────────────────

#[cfg(feature = "rerank")]
pub use frankensearch_rerank::rerank_step;

#[cfg(feature = "native")]
pub use frankensearch_rerank::NativeReranker;

#[cfg(feature = "native")]
pub use frankensearch_rerank::NativeEmbedder;

#[cfg(feature = "fastembed-reranker")]
pub use frankensearch_rerank::FastEmbedReranker;

// ─── IndexBuilder convenience API ────────────────────────────────────────────

mod index_builder;
pub use index_builder::{
    HybridIndexParts, IndexBuildStats, IndexBuilder, IndexProgress, IndexSizeBreakdown,
    LexicalArmReceipt, LexicalReaderBackend, open_admitted_v2_sync_with_residual_sidecar_cache,
    open_hybrid,
};

// ─── CASS engine-equivalence gate (dev only) ────────────────────────────────

#[cfg(feature = "cass-equivalence")]
#[doc(hidden)]
pub mod cass_equivalence;

// ─── Prelude ────────────────────────────────────────────────────────────────

/// Convenience re-exports for common usage.
///
/// ```rust,ignore
/// use frankensearch::prelude::*;
/// ```
pub mod prelude {
    pub use asupersync::Cx;

    pub use crate::{
        DocumentFingerprint, Embedder, FederatedConfig, FederatedSearcher, LexicalRead, Reranker,
        ScoreSource, ScoredResult, SearchError, SearchPhase, SearchResult, SyncTwoTierSearcher,
        TwoTierConfig, TwoTierMetrics, TwoTierSearcher,
    };

    #[cfg(feature = "storage")]
    pub use crate::{IngestAction, IngestResult, Storage, StorageBackedJobRunner};

    #[cfg(feature = "durability")]
    pub use crate::{FileProtector, FsviProtector, RepairCodec};

    #[cfg(feature = "graph")]
    pub use crate::GraphRanker;
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn core_types_accessible() {
        // Verify key types are re-exported and usable.
        let _config = TwoTierConfig::default();
        let _metrics = TwoTierMetrics::default();
        let _rrf = RrfConfig::default();
    }

    #[test]
    fn error_types_accessible() {
        let err: SearchError = SearchError::DurabilityDisabled;
        let result: SearchResult<()> = Err(err);
        assert!(result.is_err());
    }

    #[test]
    fn prelude_provides_essentials() {
        fn _takes_cx(_cx: &Cx) {}

        use crate::prelude::*;

        let _config = TwoTierConfig::default();
        let _metrics = TwoTierMetrics::default();
    }

    #[test]
    fn score_source_accessible() {
        assert_ne!(ScoreSource::Hybrid, ScoreSource::Lexical);
    }

    #[test]
    fn query_class_accessible() {
        let class = QueryClass::classify("hello world");
        assert!(matches!(
            class,
            QueryClass::NaturalLanguage | QueryClass::ShortKeyword
        ));
    }

    #[test]
    fn traits_are_object_safe() {
        fn _takes_embedder(_: &dyn Embedder) {}
        fn _takes_reranker(_: &dyn Reranker) {}
        fn _takes_lexical_read(_: &dyn LexicalRead) {}
        fn _takes_lexical_write(_: &dyn LexicalWrite) {}
        fn _takes_metrics(_: &dyn MetricsExporter) {}
    }

    #[test]
    fn utility_functions_accessible() {
        let v = l2_normalize(&[3.0, 4.0]);
        let norm: f32 = v.iter().map(|x| x * x).sum::<f32>().sqrt();
        assert!((norm - 1.0).abs() < 1e-6);

        let sim = cosine_similarity(&[1.0, 0.0], &[1.0, 0.0]);
        assert!((sim - 1.0).abs() < 1e-6);
    }

    #[test]
    fn indexable_document_accessible() {
        let doc = IndexableDocument::new("id", "content").with_title("title");
        assert_eq!(doc.id, "id");
    }

    #[test]
    fn embedder_stack_accessible() {
        assert!(matches!(
            TwoTierAvailability::HashOnly,
            TwoTierAvailability::HashOnly
        ));
    }

    #[test]
    fn sub_crate_modules_accessible() {
        // Advanced users can access sub-crate modules directly.
        let _ = core::error::SearchError::DurabilityDisabled;
        let _ = fusion::rrf::RrfConfig::default();
    }

    #[cfg(feature = "hash")]
    #[test]
    fn hash_embedder_accessible() {
        assert!(matches!(
            HashAlgorithm::FnvModular,
            HashAlgorithm::FnvModular
        ));
    }

    #[cfg(feature = "storage")]
    #[test]
    fn storage_reexports_accessible() {
        let schema_version = storage::SCHEMA_VERSION;
        assert!(schema_version >= 1);

        let _cfg = StorageConfig::default();
        let _queue = JobQueueConfig::default();
        assert!(matches!(IngestAction::New, IngestAction::New));
        let _ = std::mem::size_of::<StorageBackedJobRunner>();
    }

    #[cfg(feature = "durability")]
    #[test]
    fn durability_reexports_accessible() {
        let trailer_version = durability::REPAIR_TRAILER_VERSION;
        assert!(trailer_version >= 1);

        let _ = std::mem::size_of::<DurabilityConfig>();
        let _ = std::mem::size_of::<ProtectionResult>();
        let _ = std::mem::size_of::<RepairResult>();
        let _ = std::mem::size_of::<RepairCodecVerifyResult>();
    }
}

#[cfg(test)]
mod feature_matrix_smoke {
    #[cfg(feature = "durability")]
    use std::sync::Arc;

    use super::*;

    fn emit_evidence(lane: &str, behavior: &str, observations: &serde_json::Value) {
        eprintln!(
            "{}",
            serde_json::json!({
                "schema": "frankensearch-feature-behavior-v2",
                "lane": lane,
                "behavior": behavior,
                "status": "pass",
                "observations": observations,
            })
        );
    }

    #[cfg(feature = "hash")]
    fn hash_embed_roundtrip(lane: &str) {
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let embedder = HashEmbedder::default_256();
            let vector = embedder
                .embed(&cx, "feature matrix deterministic fixture")
                .await
                .expect("hash embedding");
            assert_eq!(vector.len(), 256);
            assert!(vector.iter().any(|value| *value != 0.0));
            emit_evidence(
                lane,
                "hash_embed_roundtrip",
                &serde_json::json!({"dimension": vector.len()}),
            );
        });
    }

    #[cfg(feature = "hash")]
    #[test]
    fn default_lane_behavior() {
        hash_embed_roundtrip("default");
    }

    #[cfg(feature = "semantic")]
    #[test]
    fn semantic_lane_behavior() {
        hash_embed_roundtrip("semantic");
        assert!(matches!(
            TwoTierAvailability::HashOnly,
            TwoTierAvailability::HashOnly
        ));
    }

    #[cfg(feature = "hybrid")]
    #[test]
    fn hybrid_lane_behavior() {
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let dir = tempfile::tempdir().expect("hybrid lexical tempdir");
            let index = QuillIndex::create(
                &cx,
                dir.path(),
                QuillConfig {
                    bulk_load_mode: true,
                    deterministic_ingest: true,
                    max_ingest_shards: 1,
                    ..QuillConfig::default()
                },
            )
            .await
            .expect("create hybrid Quill index");
            let document = IndexableDocument::new("doc-hybrid", "hybrid quill lexical fixture");
            index
                .index_document(&cx, &document)
                .await
                .expect("index hybrid Quill document");
            index
                .finish_bulk_load(&cx)
                .await
                .expect("finalize hybrid Quill index");
            let hits = index
                .search_results(&cx, "hybrid", 5)
                .expect("search hybrid Quill index");
            assert_eq!(hits.len(), 1);
            assert_eq!(hits[0].doc_id, "doc-hybrid");
            emit_evidence(
                "hybrid",
                "quill_lexical_build_search",
                &serde_json::json!({
                    "documents": 1,
                    "hits": hits.len(),
                    "lexical_backend": "quill",
                    "selected_backend": "quill",
                }),
            );
        });
    }

    #[cfg(feature = "storage")]
    #[test]
    fn persistent_lane_behavior() {
        let storage = Storage::open_in_memory().expect("open in-memory storage");
        let schema_version = storage::SCHEMA_VERSION;
        assert!(schema_version >= 1);
        emit_evidence(
            "persistent",
            "real_in_memory_storage",
            &serde_json::json!({"schema_version": schema_version}),
        );
        drop(storage);
    }

    #[cfg(feature = "durability")]
    #[test]
    fn durable_lane_behavior() {
        let dir = tempfile::tempdir().expect("durability tempdir");
        let source = dir.path().join("feature-matrix.fsvi");
        std::fs::write(&source, vec![0x5a_u8; 1024]).expect("write durability source");
        let protector =
            FsviProtector::new(Arc::new(DefaultSymbolCodec), DurabilityConfig::default())
                .expect("construct protector");
        let protection = protector.protect_atomic(&source).expect("protect source");
        assert!(protection.sidecar_path.exists());
        assert!(protector.verify_and_repair(&source).expect("verify source"));
        emit_evidence(
            "durable",
            "protect_verify_roundtrip",
            &serde_json::json!({
                "source_bytes": protection.source_size,
                "repair_bytes": protection.repair_size,
            }),
        );
    }

    #[cfg(feature = "ann")]
    #[test]
    fn ann_lane_behavior() {
        let dir = tempfile::tempdir().expect("ann tempdir");
        let path = dir.path().join("feature-matrix.fsvi");
        let mut writer = VectorIndex::create_with_revision(
            &path,
            "feature-matrix",
            "v1",
            4,
            frankensearch_index::Quantization::F16,
        )
        .expect("create vector index");
        writer
            .write_record("doc-axis-x", &[1.0, 0.0, 0.0, 0.0])
            .expect("write x-axis vector");
        writer
            .write_record("doc-axis-y", &[0.0, 1.0, 0.0, 0.0])
            .expect("write y-axis vector");
        writer.finish().expect("finish vector index");

        let index = VectorIndex::open(&path).expect("reopen vector index");
        let ann =
            HnswIndex::build_from_vector_index(&index, HnswConfig::default()).expect("build ann");
        let ann_path = dir.path().join("feature-matrix.hnsw");
        ann.save(&ann_path).expect("save native ann");
        let (ann, disposition): (HnswIndex, HnswLoadDisposition) =
            HnswIndex::load_with_disposition(&ann_path, &index)
                .expect("load native ann through facade");
        assert_eq!(disposition, HnswLoadDisposition::Native);
        let (hits, stats): (Vec<VectorHit>, AnnSearchStats) = ann
            .knn_search_with_stats(&[1.0, 0.0, 0.0, 0.0], 1, 16)
            .expect("query ann");
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].doc_id, "doc-axis-x");
        assert_eq!(stats.index_size, 2);
        assert_eq!(stats.k_returned, 1);
        assert!(stats.is_approximate);
        emit_evidence(
            "ann",
            "real_hnsw_build_query",
            &serde_json::json!({
                "dimension": stats.dimension,
                "documents": stats.index_size,
                "hits": stats.k_returned,
            }),
        );
    }

    #[cfg(feature = "full")]
    #[test]
    fn full_lane_behavior() {
        let config = TwoTierConfig::default();
        assert!(!config.fast_only);
        assert!(std::mem::size_of::<rerank::NativeEmbedder>() > 0);
        assert!(std::mem::size_of::<HnswConfig>() > 0);
        emit_evidence(
            "full",
            "full_surface_exports",
            &serde_json::json!({"native": true, "ann": true, "download": true}),
        );
    }

    #[cfg(feature = "full-fts5")]
    #[test]
    fn full_fts5_lane_behavior() {
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let adapter = Fts5LexicalSearch::new(Fts5Config::default());
            let document =
                IndexableDocument::new("doc-fts5", "fts5 feature matrix integration fixture");
            LexicalWrite::index_document(&adapter, &cx, &document)
                .await
                .expect("index FTS5 document");
            let hits = LexicalRead::search(&adapter, &cx, "integration", 5)
                .await
                .expect("search FTS5 document");
            assert_eq!(hits.len(), 1);
            assert_eq!(hits[0].doc_id, "doc-fts5");
            emit_evidence(
                "full-fts5",
                "real_fts5_index_search",
                &serde_json::json!({
                    "documents": LexicalRead::doc_count(&adapter)
                        .expect("read FTS5 document count"),
                    "hits": hits.len(),
                }),
            );
        });
    }

    #[cfg(feature = "quill")]
    #[test]
    fn quill_lane_behavior() {
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let dir = tempfile::tempdir().expect("Quill feature tempdir");
            let index = QuillIndex::create(
                &cx,
                dir.path(),
                QuillConfig {
                    bulk_load_mode: true,
                    deterministic_ingest: true,
                    max_ingest_shards: 1,
                    ..QuillConfig::default()
                },
            )
            .await
            .expect("create Quill index");
            let documents = [
                IndexableDocument::new("doc-alpha", "alpha quill feature matrix"),
                IndexableDocument::new("doc-beta", "beta consumer integration"),
            ];
            index
                .index_documents(&cx, &documents)
                .await
                .expect("index Quill documents");
            index
                .finish_bulk_load(&cx)
                .await
                .expect("finalize Quill index");
            let hits = index
                .search_results(&cx, "alpha", 5)
                .expect("search Quill index");
            assert_eq!(hits.len(), 1);
            assert_eq!(hits[0].doc_id, "doc-alpha");
            emit_evidence(
                "quill",
                "real_index_build_search",
                &serde_json::json!({
                    "documents": documents.len(),
                    "hits": hits.len(),
                    "lexical_backend": "quill",
                    "selected_backend": "quill",
                }),
            );
        });
    }

    #[cfg(feature = "lexical")]
    #[test]
    fn lexical_lane_behavior() {
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let dir = tempfile::tempdir().expect("lexical Quill feature tempdir");
            let index = lexical::QuillIndex::create(
                &cx,
                dir.path(),
                lexical::QuillConfig {
                    bulk_load_mode: true,
                    deterministic_ingest: true,
                    max_ingest_shards: 1,
                    ..lexical::QuillConfig::default()
                },
            )
            .await
            .expect("create lexical Quill index");
            let documents = [
                IndexableDocument::new("doc-alpha", "alpha quill feature matrix"),
                IndexableDocument::new("doc-beta", "beta consumer integration"),
            ];
            index
                .index_documents(&cx, &documents)
                .await
                .expect("index lexical Quill document");
            index
                .finish_bulk_load(&cx)
                .await
                .expect("finalize lexical Quill index");
            let hits = index
                .search_results(&cx, "alpha", 5)
                .expect("search lexical Quill index");
            assert_eq!(hits.len(), 1);
            assert_eq!(hits[0].doc_id, "doc-alpha");
            emit_evidence(
                "lexical",
                "quill_index_build_search",
                &serde_json::json!({
                    "documents": documents.len(),
                    "hits": hits.len(),
                    "lexical_backend": "quill",
                    "selected_backend": "quill",
                }),
            );
        });
    }

    /// bd-8nqz.4: BOTH backends compiled at once must SELECT DETERMINISTICALLY.
    ///
    /// This is the cell the flip endangered and the one the feature matrix did
    /// not cover: every other lane compiles exactly one lexical engine, so no
    /// lane could observe the `lexical` alias resolving to the wrong one.
    /// d117ce1f repointed `lexical` at Quill while `lexical-tantivy` kept
    /// Tantivy under its own name, and only a build with BOTH present can show
    /// that those two namespaces stayed distinct rather than collapsing.
    ///
    /// The assertion is on the RESOLVED BACKEND IDENTITY, not on whether the
    /// names merely compile: `lexical` re-exporting Tantivy would still
    /// compile, and would still pass any test that only checked the paths
    /// exist.
    #[cfg(all(feature = "lexical", feature = "lexical-tantivy"))]
    #[test]
    fn both_backends_select_deterministically() {
        // `lexical` is the engine-neutral alias. Under the flip it must be
        // Quill, and it must say so in its own words.
        assert_eq!(
            lexical::QUILL_LEXICAL_BACKEND,
            "quill",
            "the `lexical` alias must resolve to Quill when both backends are compiled"
        );

        // Tantivy remains reachable, but only through its explicit namespace.
        let tantivy_schema = lexical_tantivy::CASS_SCHEMA_VERSION;

        // And the two must be DIFFERENT surfaces. If `lexical` were aliased to
        // Tantivy, the CASS schema identity would be reachable through it.
        emit_evidence(
            "both",
            "deterministic_backend_selection",
            &serde_json::json!({
                "lexical_backend": lexical::QUILL_LEXICAL_BACKEND,
                "lexical_tantivy_cass_schema_version": tantivy_schema,
                "quill": cfg!(feature = "quill"),
                "lexical_tantivy": cfg!(feature = "lexical-tantivy"),
            }),
        );
    }

    /// bd-8nqz.4: the PUBLIC FIELD INVENTORY of the two lexical hit surfaces,
    /// with every Quill/Tantivy difference explicitly reviewed.
    ///
    /// The acceptance requires that the public field/serde inventory has "no
    /// unreviewed Quill/Tantivy gap". The engine-neutral facade alias
    /// `LexicalHit` IS Quill's `QuillSnippetHit`, while `lexical_tantivy`
    /// keeps Tantivy's own `LexicalHit`, so a consumer migrating between them
    /// meets four real differences. Leaving them undocumented is exactly the
    /// "unreviewed gap" this clause forbids.
    ///
    /// # What actually enforces this
    ///
    /// The EXHAUSTIVE DESTRUCTURING below, not the table. Neither pattern uses
    /// `..`, so adding or removing a field on either engine's hit type makes
    /// this test fail to COMPILE until someone updates the inventory and
    /// classifies the change. A runtime key-set comparison would only catch a
    /// field that changed its serialized name, and would silently pass a newly
    /// added one on whichever engine the test happened not to serialize.
    #[cfg(all(feature = "lexical", feature = "lexical-tantivy"))]
    #[test]
    fn the_public_lexical_hit_inventory_has_no_unreviewed_gap() {
        /// (quill field, tantivy field, reviewed correspondence).
        const INVENTORY: &[(&str, &str, &str)] = &[
            (
                "document_id",
                "doc_id",
                "RENAMED. A source-compatibility break for any consumer moving \
                 off the Tantivy type; the facade exposes Quill's spelling.",
            ),
            (
                "score",
                "bm25_score",
                "RENAMED. Same value class (exhaustive BM25), different name; \
                 the facade exposes Quill's spelling.",
            ),
            (
                "rank",
                "rank",
                "IDENTICAL. Zero-based, usize, both engines.",
            ),
            (
                "snippet",
                "snippet",
                "IDENTICAL SHAPE. Option<String> on both, and both preserve \
                 None as distinct from Some(\"\").",
            ),
            (
                "query_type",
                "query_type",
                "SAME NAME, TWO INDEPENDENTLY DEFINED ENUMS. \
                 frankensearch_quill::QueryExplanation and \
                 frankensearch_lexical::QueryExplanation are separate types \
                 with no From/Into between them. Their variant sets agree \
                 TODAY by coincidence, not by contract, so they must never be \
                 compared by type equality.",
            ),
            (
                "metadata",
                "metadata",
                "SAME NAME, DIFFERENT REPRESENTATION. Quill wraps in \
                 Option<Arc<Value>>, Tantivy uses Option<Value>. Serde output \
                 is identical; the difference is a source-compatibility one \
                 for callers that name the type.",
            ),
        ];

        // Quill side. No `..`: a new field breaks this build.
        let quill_hit = LexicalHit {
            document_id: "doc-alpha".to_owned(),
            score: 1.5,
            rank: 0,
            snippet: Some("<b>alpha</b>".to_owned()),
            query_type: quill::QueryExplanation::Simple,
            metadata: None,
        };
        let LexicalHit {
            document_id,
            score,
            rank,
            snippet,
            query_type,
            metadata,
        } = quill_hit;
        let quill_fields = [
            ("document_id", !document_id.is_empty()),
            ("score", score > 0.0),
            ("rank", rank == 0),
            ("snippet", snippet.is_some()),
            ("query_type", query_type == quill::QueryExplanation::Simple),
            ("metadata", metadata.is_none()),
        ];

        // Tantivy side. Same discipline, its own spelling.
        let tantivy_hit = lexical_tantivy::LexicalHit {
            doc_id: "doc-alpha".to_owned(),
            bm25_score: 1.5,
            rank: 0,
            snippet: Some("<b>alpha</b>".to_owned()),
            query_type: lexical_tantivy::QueryExplanation::Simple,
            metadata: None,
        };
        let lexical_tantivy::LexicalHit {
            doc_id,
            bm25_score,
            rank: tantivy_rank,
            snippet: tantivy_snippet,
            query_type: tantivy_query_type,
            metadata: tantivy_metadata,
        } = tantivy_hit;
        let tantivy_fields = [
            ("doc_id", !doc_id.is_empty()),
            ("bm25_score", bm25_score > 0.0),
            ("rank", tantivy_rank == 0),
            ("snippet", tantivy_snippet.is_some()),
            (
                "query_type",
                tantivy_query_type == lexical_tantivy::QueryExplanation::Simple,
            ),
            ("metadata", tantivy_metadata.is_none()),
        ];

        // The inventory must cover EXACTLY the destructured fields on both
        // sides -- no entry without a field, no field without an entry.
        assert_eq!(
            INVENTORY.len(),
            quill_fields.len(),
            "every Quill field must have exactly one inventory entry"
        );
        assert_eq!(
            INVENTORY.len(),
            tantivy_fields.len(),
            "every Tantivy field must have exactly one inventory entry"
        );
        for (index, (quill_name, tantivy_name, review)) in INVENTORY.iter().enumerate() {
            assert_eq!(
                *quill_name, quill_fields[index].0,
                "inventory row {index} does not name the Quill field it reviews"
            );
            assert_eq!(
                *tantivy_name, tantivy_fields[index].0,
                "inventory row {index} does not name the Tantivy field it reviews"
            );
            assert!(
                quill_fields[index].1 && tantivy_fields[index].1,
                "inventory row {index} names a field whose value was not actually observed"
            );
            assert!(
                !review.trim().is_empty(),
                "field {quill_name} has no recorded review"
            );
        }

        // The four real gaps, asserted rather than merely described, so the
        // table cannot drift into claiming a correspondence that stopped
        // being true.
        assert_ne!(
            INVENTORY[0].0, INVENTORY[0].1,
            "document_id/doc_id must remain recorded as a rename"
        );
        assert_ne!(
            INVENTORY[1].0, INVENTORY[1].1,
            "score/bm25_score must remain recorded as a rename"
        );

        emit_evidence(
            "both",
            "public_lexical_hit_inventory",
            &serde_json::json!({
                "fields_reviewed": INVENTORY.len(),
                "renamed": [[INVENTORY[0].0, INVENTORY[0].1], [INVENTORY[1].0, INVENTORY[1].1]],
                "same_name_different_type": ["query_type", "metadata"],
            }),
        );
    }

    #[cfg(feature = "lexical-tantivy")]
    #[test]
    fn lexical_tantivy_lane_behavior() {
        let _explicit_tantivy_namespace = lexical_tantivy::CASS_SCHEMA_VERSION;
        asupersync::test_utils::run_test_with_cx(|cx| async move {
            let dir = tempfile::tempdir().expect("Tantivy feature tempdir");
            let index = TantivyIndex::create(dir.path()).expect("create Tantivy index");
            let documents = [
                IndexableDocument::new("doc-alpha", "alpha tantivy oracle matrix"),
                IndexableDocument::new("doc-beta", "beta consumer integration"),
            ];
            LexicalWrite::index_documents(&index, &cx, &documents)
                .await
                .expect("index Tantivy documents");
            LexicalWrite::commit(&index, &cx)
                .await
                .expect("commit Tantivy index");
            let hits = LexicalRead::search(&index, &cx, "alpha", 5)
                .await
                .expect("search Tantivy index");
            assert_eq!(hits.len(), 1);
            assert_eq!(hits[0].doc_id, "doc-alpha");
            emit_evidence(
                "lexical-tantivy",
                "real_index_build_search",
                &serde_json::json!({"documents": documents.len(), "hits": hits.len()}),
            );
        });
    }

    #[cfg(feature = "cass-compat")]
    #[test]
    fn cass_compat_lane_behavior() {
        assert_eq!(
            lexical_tantivy::CASS_SCHEMA_VERSION,
            cass_compat::CASS_SCHEMA_VERSION
        );
        let dir = tempfile::tempdir().expect("CASS feature tempdir");
        let mut index =
            cass_compat::CassTantivyIndex::open_or_create(dir.path()).expect("create CASS index");
        let documents = [cass_compat::CassDocument {
            agent: "RoseMaple".to_owned(),
            workspace: Some("frankensearch".to_owned()),
            workspace_original: Some("/data/projects/frankensearch".to_owned()),
            source_path: "fixtures/consumer-e2e.jsonl".to_owned(),
            msg_idx: 1,
            created_at: Some(1_753_307_200),
            title: Some("Consumer integration".to_owned()),
            content: "CASS compatibility feature matrix document".to_owned(),
            source_id: "consumer-e2e-1".to_owned(),
            origin_kind: "test".to_owned(),
            origin_host: Some("ci".to_owned()),
            conversation_id: Some(7),
        }];
        index
            .add_cass_documents(&documents)
            .expect("index CASS document");
        index.commit().expect("commit CASS index");
        assert!(index.segment_count() >= 1);
        assert!(cass_compat::cass_schema_hash_matches(
            cass_compat::CASS_SCHEMA_HASH
        ));
        emit_evidence(
            "cass-compat",
            "real_cass_index_commit",
            &serde_json::json!({
                "documents": documents.len(),
                "segments": index.segment_count(),
            }),
        );
    }

    // --- bd-8nqz.5: CASS lexical compatibility receipt for the Quill flip ---

    /// The cass-compat lane must resolve Tantivy IN and Quill OUT.
    ///
    /// This is the negative half of the bead's acceptance and nothing asserted
    /// it before: `cargo check --features cass-compat` proves the lane
    /// COMPILES, never that Quill stayed out of the resolved graph. Because
    /// the facade's `quill` feature is the only thing that pulls
    /// `dep:frankensearch-quill`, its absence here is exactly "Quill is not in
    /// this lane's dependency graph".
    ///
    /// Deliberately asserted on the LANE, not as a `compile_error!` on the
    /// feature pair: the contract permits cass-compat together with Quill when
    /// a consumer independently requests Quill, so forbidding the combination
    /// outright would be stricter than the contract and would break that
    /// consumer.
    #[cfg(all(feature = "cass-compat", not(feature = "quill")))]
    #[test]
    fn cass_compat_lane_namespace_keeps_schema_v8_identity() {
        // NOTE ON WHERE THE QUILL-ABSENCE PROOF LIVES. It is deliberately NOT
        // here. Inside `#[cfg(all(cass-compat, not(quill)))]` an assertion on
        // `cfg!(feature = "quill")` is a compile-time constant, and the gate
        // presupposes exactly what such an assertion would claim to prove —
        // circular, and clippy's assertions_on_constants says so. The real
        // proof asks the resolver instead and lives in
        // scripts/check_feature_matrix.sh::validate_cass_compat_backend_graph,
        // which fails if `cargo tree --features cass-compat` ever resolves
        // frankensearch-quill.
        //
        // What this test CAN prove at runtime is the part the flip endangered:
        // the explicit namespace the contract requires CASS to import through
        // must still resolve, and must still agree on the schema-v8 identity,
        // after the d117ce1f flip and the 327d264a trait split.
        assert_eq!(
            lexical_tantivy::CASS_SCHEMA_VERSION,
            cass_compat::CASS_SCHEMA_VERSION,
            "the lexical_tantivy namespace must expose the same schema-v8 identity as cass_compat"
        );
        emit_evidence(
            "cass-compat",
            "lane_namespace_identity",
            &serde_json::json!({
                "lexical_tantivy": cfg!(feature = "lexical-tantivy"),
                "quill": cfg!(feature = "quill"),
                "cass_schema_version": cass_compat::CASS_SCHEMA_VERSION,
            }),
        );
    }

    /// Full schema-v8 lifecycle on the REAL cass-compat surface.
    ///
    /// The pre-existing lane test covers create/ingest/commit only; the bead
    /// requires create/ingest/commit/reopen/merge/query/restart. The two flip
    /// commits touched these seams directly — d117ce1f repointed the `lexical`
    /// feature at Quill and 327d264a removed the combined `LexicalSearch`
    /// trait — so a lane that merely compiles proves nothing about whether a
    /// foreign CASS index still round-trips.
    ///
    /// Every step drives `cass_compat::CassTantivyIndex` itself. No fixture
    /// stand-in: the query goes through the real CASS boolean parser and
    /// returns real ranked hits, and the restart arm reopens the same on-disk
    /// path in a fresh index handle.
    #[cfg(feature = "cass-compat")]
    #[test]
    fn cass_compat_schema_v8_survives_reopen_merge_query_and_restart() {
        let dir = tempfile::tempdir().expect("CASS lifecycle tempdir");
        let document = |source_id: &str, msg_idx: u64, content: &str| cass_compat::CassDocument {
            agent: "TopazCat".to_owned(),
            workspace: Some("frankensearch".to_owned()),
            workspace_original: Some("/data/projects/frankensearch".to_owned()),
            source_path: "fixtures/cass-lifecycle.jsonl".to_owned(),
            msg_idx,
            created_at: Some(1_753_307_200 + i64::try_from(msg_idx).expect("msg_idx fits i64")),
            title: Some("CASS lifecycle".to_owned()),
            content: content.to_owned(),
            source_id: source_id.to_owned(),
            origin_kind: "test".to_owned(),
            origin_host: Some("ci".to_owned()),
            conversation_id: Some(11),
        };

        // CREATE + INGEST + COMMIT.
        let mut index =
            cass_compat::CassTantivyIndex::open_or_create(dir.path()).expect("create CASS index");
        index
            .add_cass_documents(&[document(
                "lifecycle-1",
                1,
                "quill flip compatibility receipt",
            )])
            .expect("ingest first CASS batch");
        index.commit().expect("commit first CASS batch");
        let segments_after_first = index.segment_count();
        assert!(segments_after_first >= 1);

        // REOPEN: drop the handle and open the same path again.
        drop(index);
        let mut reopened = cass_compat::CassTantivyIndex::open_or_create(dir.path())
            .expect("reopen the committed CASS index");
        assert!(
            cass_compat::cass_schema_hash_matches(cass_compat::CASS_SCHEMA_HASH),
            "reopened index must still match the schema-v8 hash"
        );

        // MERGE: a second committed batch, then a real force_merge. Asserting
        // the segment count actually collapses is what proves the merge ran,
        // rather than merely reading merge_status back.
        reopened
            .add_cass_documents(&[document("lifecycle-2", 2, "second batch tantivy interop")])
            .expect("ingest second CASS batch");
        reopened.commit().expect("commit second CASS batch");
        let segments_before_merge = reopened.segment_count();
        assert!(
            segments_before_merge > segments_after_first,
            "a second committed batch must add a segment, observed {segments_before_merge}"
        );
        reopened.force_merge().expect("force merge the CASS index");
        let segments_after_merge = reopened.segment_count();
        assert!(
            segments_after_merge < segments_before_merge,
            "force_merge must collapse segments: {segments_before_merge} -> {segments_after_merge}"
        );
        assert_eq!(
            reopened.merge_status().segment_count,
            segments_after_merge,
            "merge status must report the post-merge segment count"
        );

        // QUERY through the real CASS reader and schema — the same
        // cass_open_search_reader / cass_fields_from_schema surface a CASS
        // consumer uses, not a facade convenience wrapper.
        // CassTantivyIndex::open_or_create writes the Tantivy directory AT the
        // path it is given. cass_index_dir is a different convention
        // (base/index/<schema-version>) that CASS uses when laying out its own
        // tree, so the reader must be pointed at the same path the writer used.
        let index_dir = dir.path().to_path_buf();
        let search_one = |needle: &str| -> (usize, usize) {
            // Tantivy-native types come through the facade's explicit
            // compatibility namespace, which is exactly the migration this
            // bead requires of CASS: never through the overloaded `lexical`
            // name. If lexical_tantivy ever stops re-exporting what a CASS
            // consumer needs, this test breaks — a direct tantivy dependency
            // would have hidden that.
            let (reader, fields) = cass_compat::cass_open_search_reader(
                &index_dir,
                lexical_tantivy::ReloadPolicy::Manual,
            )
            .expect("open the CASS search reader");
            let searcher = reader.searcher();
            let live = searcher.num_docs();
            let query = lexical_tantivy::TermQuery::new(
                lexical_tantivy::Term::from_field_text(fields.content, needle),
                lexical_tantivy::IndexRecordOption::WithFreqs,
            );
            let hits = searcher
                .search(&query, &lexical_tantivy::Count)
                .expect("run the CASS term query");
            (hits, usize::try_from(live).expect("live docs fit usize"))
        };

        let (hits_before_restart, live_before_restart) = search_one("compatibility");
        assert_eq!(
            hits_before_restart, 1,
            "the CASS query path must find exactly the matching document"
        );
        assert_eq!(
            live_before_restart, 2,
            "both committed documents must be live after the merge"
        );

        // RESTART: drop every handle, then reopen the same on-disk path.
        drop(reopened);
        let restarted = cass_compat::CassTantivyIndex::open_or_create(dir.path())
            .expect("restart against the committed CASS index");
        let (hits_after_restart, live_after_restart) = search_one("compatibility");
        assert_eq!(
            hits_after_restart, hits_before_restart,
            "restart must preserve the CASS query result"
        );
        assert_eq!(
            live_after_restart, live_before_restart,
            "restart must preserve the live document count"
        );

        emit_evidence(
            "cass-compat",
            "schema_v8_lifecycle",
            &serde_json::json!({
                "segments_after_first": segments_after_first,
                "segments_before_merge": segments_before_merge,
                "segments_after_merge": segments_after_merge,
                "segments_after_restart": restarted.segment_count(),
                "query_hits": hits_after_restart,
                "live_docs": live_after_restart,
                "schema_version": cass_compat::CASS_SCHEMA_VERSION,
            }),
        );
    }
}