khive_db/stores/

text.rs

//! FTS5-backed `TextSearch`: one virtual table per model, scores normalized to `(0.05, 1.0]`.

use std::sync::Arc;

use async_trait::async_trait;
use chrono::{DateTime, TimeZone, Utc};
use uuid::Uuid;

use khive_score::DeterministicScore;
use khive_storage::error::StorageError;
use khive_storage::types::{
    BatchWriteSummary, IndexRebuildScope, SqlStatement, SqlValue, TextDocument, TextFilter,
    TextGatherMode, TextIndexStats, TextQueryMode, TextSearchHit, TextSearchOptions,
    TextSearchRequest, TextTermStats, TextTermStatsRequest,
};
use khive_storage::usage::{UsageContext, UsageUnit};
use khive_storage::StorageCapability;
use khive_storage::TextSearch;
use khive_types::SubstrateKind;

use crate::error::SqliteError;
use crate::pool::ConnectionPool;
use crate::sql_bridge::bind_params;
use crate::writer_task::WriterTaskHandle;

/// Name of the sidecar rowid map for a given FTS table: `{namespace,
/// subject_id} -> rowid`, `WITHOUT ROWID`, `PRIMARY KEY (namespace,
/// subject_id)`.
///
/// `namespace` and `subject_id` are declared `UNINDEXED` in every FTS5 DDL
/// (trigram-tokenized production tables cannot index them without polluting
/// text queries with UUID fragments), so a point lookup on either column is a
/// full virtual-table scan of `table` — this sidecar turns it into a
/// primary-key lookup instead. `table` must already be a trusted, sanitized
/// table name.
pub fn rowid_map_table(table: &str) -> String {
    format!("{table}_rowids")
}

/// Name of [`rowid_map_table`]'s own state sidecar: a `key -> value` table
/// recording whether the map's one-time legacy backfill has completed, so
/// completion is a durable, transactionally-written fact instead of being
/// inferred from the map's row count.
pub fn rowid_map_state_table(table: &str) -> String {
    format!("{}_state", rowid_map_table(table))
}

/// Value of the `backfill` key in [`rowid_map_state_table`] once
/// `ensure_fts_rowid_map_backfilled` has completed a full reconciliation
/// pass over `table`.
pub const ROWID_MAP_BACKFILL_COMPLETE: &str = "complete";

/// DDL for [`rowid_map_table`]'s sidecar and its [`rowid_map_state_table`].
/// `IF NOT EXISTS` so every creation site (backend ensure-path, migrations,
/// test helpers) can call it unconditionally. `table` must already be a
/// trusted, sanitized table name.
pub fn rowid_map_ddl(table: &str) -> String {
    let map = rowid_map_table(table);
    let state = rowid_map_state_table(table);
    format!(
        "CREATE TABLE IF NOT EXISTS {map} (\
         namespace TEXT NOT NULL, \
         subject_id TEXT NOT NULL, \
         rowid INTEGER NOT NULL, \
         PRIMARY KEY (namespace, subject_id)\
         ) WITHOUT ROWID; \
         CREATE TABLE IF NOT EXISTS {state} (\
         key TEXT PRIMARY KEY, \
         value TEXT NOT NULL\
         ) WITHOUT ROWID"
    )
}

/// The exact `DELETE` this store's `delete_document` issues, for a given
/// FTS table (ADR-099 B3 r6 structural cut — see `entity.rs`'s sibling
/// block). Looks the target row's rowid up in [`rowid_map_table`] first (a
/// primary-key lookup) instead of scanning `table` for `namespace`/
/// `subject_id`, which are `UNINDEXED` FTS5 columns. `table` must already be
/// a trusted, sanitized table name (this mirrors `delete_document`'s own
/// pre-existing lack of a placeholder for table names — `format!` is
/// required since table identifiers cannot be bound as SQL parameters).
///
/// The trailing `AND namespace = ?1 AND subject_id = ?2` re-checks the key
/// on the (already rowid-narrowed) candidate row itself: if the map ever
/// held a stale entry — e.g. a crash between an earlier delete's FTS-row
/// removal and its own map-row removal, before FTS5 reused that rowid for a
/// different document — this statement would otherwise delete whatever
/// unrelated document now lives at that rowid instead of affecting zero
/// rows. Cheap: the `rowid IN (...)` subquery already narrows to at most one
/// candidate row via the map's primary key, so this is a single-row
/// post-filter, not a second table scan.
///
/// Callers that also need the sidecar's own row removed (a real delete, not
/// half of an upsert) must additionally run [`delete_document_map_statement`]
/// — see [`delete_document_statements`] for the paired, order-safe form.
pub fn delete_document_statement(table: &str, namespace: &str, subject_id: Uuid) -> SqlStatement {
    let map = rowid_map_table(table);
    SqlStatement {
        sql: format!(
            "DELETE FROM {table} WHERE rowid IN \
             (SELECT rowid FROM {map} WHERE namespace = ?1 AND subject_id = ?2) \
             AND namespace = ?1 AND subject_id = ?2"
        ),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(subject_id.to_string()),
        ],
        label: Some(format!("fts-delete-{table}")),
    }
}

/// Pre-map fallback: the literal delete this store issued before the rowid
/// map existed, scanning `table`'s `UNINDEXED` `namespace`/`subject_id`
/// columns directly. Used only by [`Fts5TextSearch`] instances constructed
/// in scan-fallback mode (a read-only snapshot whose FTS table predates the
/// sidecar map — see `StorageBackend::text_with_tokenizer`'s read-only
/// branch), where the map table does not exist to route through.
fn delete_document_statement_scan_fallback(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
) -> SqlStatement {
    SqlStatement {
        sql: format!("DELETE FROM {table} WHERE namespace = ?1 AND subject_id = ?2"),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(subject_id.to_string()),
        ],
        label: Some(format!("fts-delete-scan-fallback-{table}")),
    }
}

/// Remove `(namespace, subject_id)`'s row from `table`'s [`rowid_map_table`].
///
/// Order matters relative to [`delete_document_statement`]: that statement's
/// subquery still needs this row, so run it first, then this one, in the same
/// transaction. Use [`delete_document_statements`] to get both in the correct
/// order.
pub fn delete_document_map_statement(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
) -> SqlStatement {
    let map = rowid_map_table(table);
    SqlStatement {
        sql: format!("DELETE FROM {map} WHERE namespace = ?1 AND subject_id = ?2"),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(subject_id.to_string()),
        ],
        label: Some(format!("fts-delete-map-{table}")),
    }
}

/// The full, order-safe deletion of one FTS document: index 0 (the FTS row,
/// looked up through the map) MUST execute before index 1 (the map row
/// itself) in the same transaction/connection — index 0's subquery reads the
/// row index 1 removes. Callers that only need the FTS row gone as half of a
/// delete-then-insert upsert (the map row gets overwritten by the following
/// insert's `INSERT OR REPLACE`, so removing it first is redundant) can use
/// [`delete_document_statement`] alone.
pub fn delete_document_statements(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
) -> [SqlStatement; 2] {
    [
        delete_document_statement(table, namespace, subject_id),
        delete_document_map_statement(table, namespace, subject_id),
    ]
}

/// Build the `INSERT` half of the FTS delete-then-insert upsert.
///
/// `table` must be a trusted, sanitized table name because SQL identifiers
/// cannot be bound as parameters.
///
/// This statement alone leaves [`rowid_map_table`] out of date — pair it with
/// [`insert_document_map_statement`] (or use [`insert_document_statements`]
/// for the order-safe combination) so the sidecar tracks the new rowid.
pub fn insert_document_statement(table: &str, document: &TextDocument) -> SqlStatement {
    let tags_json = tags_to_json(&document.tags);
    let metadata_json = document.metadata.as_ref().map(|v| v.to_string());
    SqlStatement {
        sql: format!(
            "INSERT INTO {table} \
             (subject_id, kind, title, body, tags, namespace, metadata, updated_at, record_kind) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"
        ),
        params: vec![
            SqlValue::Text(document.subject_id.to_string()),
            SqlValue::Text(document.kind.to_string()),
            SqlValue::Text(document.title.clone().unwrap_or_default()),
            SqlValue::Text(document.body.clone()),
            SqlValue::Text(tags_json),
            SqlValue::Text(document.namespace.clone()),
            match metadata_json {
                Some(m) => SqlValue::Text(m),
                None => SqlValue::Null,
            },
            SqlValue::Integer(dt_to_micros(&document.updated_at)),
            match &document.record_kind {
                Some(kind) => SqlValue::Text(kind.clone()),
                None => SqlValue::Null,
            },
        ],
        label: Some(format!("fts-insert-{table}")),
    }
}

/// Upsert `(namespace, subject_id)`'s [`rowid_map_table`] row to point at the
/// rowid the connection's most recent successful `INSERT` produced.
///
/// Must run immediately after the FTS `INSERT` it tracks, on the same
/// connection, with nothing else written in between: `last_insert_rowid()` is
/// connection-scoped and reports whichever `INSERT` ran last. `INSERT OR
/// REPLACE` means this is safe to call whether or not a map row for this key
/// already exists (the ordinary upsert case does: the old row is overwritten
/// with the new rowid rather than requiring a prior delete).
pub fn insert_document_map_statement(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
) -> SqlStatement {
    let map = rowid_map_table(table);
    SqlStatement {
        sql: format!(
            "INSERT OR REPLACE INTO {map} (namespace, subject_id, rowid) \
             VALUES (?1, ?2, last_insert_rowid())"
        ),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(subject_id.to_string()),
        ],
        label: Some(format!("fts-insert-map-{table}")),
    }
}

/// The full, order-safe insertion of one FTS document: index 0 (the FTS
/// `INSERT`) MUST execute immediately before index 1 (the rowid-map upsert)
/// on the same connection, with no other write between them — see
/// [`insert_document_map_statement`]'s adjacency contract. Every insert path
/// (create, and the insert half of an upsert) must go through this pair, or
/// the sidecar map silently falls out of sync with the FTS table it indexes.
pub fn insert_document_statements(table: &str, document: &TextDocument) -> [SqlStatement; 2] {
    [
        insert_document_statement(table, document),
        insert_document_map_statement(table, &document.namespace, document.subject_id),
    ]
}

/// Ensure the FTS5 virtual table for `table_key` (and its [`rowid_map_table`]
/// sidecar) exist.
///
/// Used in tests to set up an in-memory FTS5 table without the full `StorageBackend`.
#[cfg(test)]
pub(crate) fn ensure_fts5_schema(
    conn: &rusqlite::Connection,
    table_key: &str,
) -> Result<(), rusqlite::Error> {
    let table_name = format!("fts_{}", table_key);
    let ddl = format!(
        "CREATE VIRTUAL TABLE IF NOT EXISTS {} USING fts5(\
         subject_id UNINDEXED, \
         kind UNINDEXED, \
         title, \
         body, \
         tags UNINDEXED, \
         namespace UNINDEXED, \
         metadata UNINDEXED, \
         updated_at UNINDEXED, \
         record_kind\
         )",
        table_name
    );
    conn.execute_batch(&ddl)?;
    conn.execute_batch(&rowid_map_ddl(&table_name))
}

fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Text, op, e)
}

fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
    e.into_storage_error(StorageCapability::Text, op)
}

fn count_fts_pass(context: Option<&UsageContext>) {
    if let Some(context) = context {
        context.add(UsageUnit::FtsPasses, 1);
    }
}

/// A TextSearch backed by SQLite FTS5 virtual tables.
///
/// Each instance manages one table: `fts_{table_key}`. Documents are stored
/// with their metadata in UNINDEXED columns. `title`, `body`, and the optional
/// granular `record_kind` classifier are indexed; corpus filters use the
/// classifier inside MATCH and retain exact row predicates.
pub struct Fts5TextSearch {
    pool: Arc<ConnectionPool>,
    is_file_backed: bool,
    table_name: String,
    writer_task: Option<WriterTaskHandle>,
    /// Set only for a read-only snapshot whose FTS table predates the
    /// [`rowid_map_table`] sidecar (see `StorageBackend::text_with_tokenizer`'s
    /// read-only branch, which cannot create/backfill the map on a read-only
    /// connection). `get_document`/`delete_document` fall back to the
    /// pre-map `namespace`/`subject_id` scan predicates in this mode instead
    /// of joining/subquerying a map table that does not exist.
    scan_fallback: bool,
}

impl Fts5TextSearch {
    /// Create a new FTS5 text search instance.
    ///
    /// The FTS5 virtual table (and its [`rowid_map_table`] sidecar) must
    /// already exist (created by `StorageBackend::text()`).
    pub(crate) fn new(pool: Arc<ConnectionPool>, is_file_backed: bool, table_key: String) -> Self {
        Self::new_with_mode(pool, is_file_backed, table_key, false)
    }

    /// Create an instance in scan-fallback mode: the sidecar map is assumed
    /// absent, so every point lookup/delete falls back to the pre-map
    /// `namespace`/`subject_id` scan predicates. Reserved for a read-only
    /// snapshot whose FTS table predates the map (see the [`scan_fallback`]
    /// field doc and `StorageBackend::text_with_tokenizer`'s read-only
    /// branch).
    ///
    /// [`scan_fallback`]: Self::scan_fallback
    pub(crate) fn new_scan_fallback(
        pool: Arc<ConnectionPool>,
        is_file_backed: bool,
        table_key: String,
    ) -> Self {
        Self::new_with_mode(pool, is_file_backed, table_key, true)
    }

    fn new_with_mode(
        pool: Arc<ConnectionPool>,
        is_file_backed: bool,
        table_key: String,
        scan_fallback: bool,
    ) -> Self {
        let table_name = format!("fts_{}", table_key);
        // Enabled by default for file-backed pools; explicit off/degraded
        // construction remains synchronous (ADR-067 Component A, mirrors
        // entity.rs policy): a missing writer task is cached without failing
        // construction. Every write re-resolves it and applies
        // strict/compatibility policy then.
        let writer_task = pool.writer_task_handle().ok().flatten();
        Self {
            pool,
            is_file_backed,
            table_name,
            writer_task,
            scan_fallback,
        }
    }

    /// Re-derive writer-task availability at write time instead of trusting
    /// only the field cached at construction (ADR-136 D1 gate 3 amendment).
    /// `self.writer_task` permanently caches `None` when this store was
    /// constructed outside a Tokio runtime (`writer_task_handle()` returns
    /// `Err(WriterTaskNoRuntime)`, which construction collapses via
    /// `.ok().flatten()`) — every later write, even ones running inside a
    /// runtime, would otherwise silently keep bypassing an enabled queue.
    /// The pool helper also enforces strict fail-closed routing and preserves
    /// a typed `WriterTaskNoRuntime` error when no runtime is available.
    fn current_writer_task(
        &self,
        operation: &'static str,
    ) -> Result<Option<WriterTaskHandle>, StorageError> {
        self.pool
            .writer_task_for_write(self.writer_task.as_ref(), operation)
    }

    /// Route a single-row write through the pool-wide `WriterTask` when
    /// the write queue is enabled and a handle is available. Strict mode
    /// refuses a missing handle; compatibility mode falls back to the legacy
    /// standalone-connection / pool-mutex path (ADR-067 Component A, Fork C
    /// slice 2). See `crates/khive-db/docs/api/text.md`
    /// for the per-caller routing rules (which methods bypass this via
    /// `with_writer_unmanaged` and why).
    async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = self.current_writer_task(op)? {
            return writer_task
                .send_bounded(move |conn| f(conn).map_err(|e| map_err(e, op)))
                .await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteFtsGeneralWrite);
        self.with_writer_unmanaged(op, f).await
    }

    /// Direct standalone or pooled typed transaction path, bypassing the
    /// WriterTask channel after routing has selected a compatibility fallback.
    /// Callbacks contain DML only; this seam owns BEGIN and settlement.
    async fn with_writer_unmanaged<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        let pool = Arc::clone(&self.pool);
        let result = tokio::task::spawn_blocking(move || {
            pool.execute_direct_transaction(StorageCapability::Text, op, move |conn| {
                f(conn).map_err(|error| map_err(error, op))
            })
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Text, op, e))?;
        if let Err(err) = &result {
            let msg = err.to_string();
            if msg.contains("locked") || msg.contains("busy") {
                if self.is_file_backed {
                    // Only the standalone-connection path: the pool-mutex
                    // (`try_writer`) path's own timeout is already recorded
                    // at `Site::PoolAdmission` inside `ConnectionPool::writer`
                    // — emitting here too would double-count the same event.
                    crate::timeout_sink::emit_timeout(
                        &crate::timeout_sink::db_label(&self.pool),
                        crate::timeout_sink::Site::StandaloneText,
                        &msg,
                        None,
                    );
                }
                let open: Vec<String> = khive_storage::tx_registry::snapshot()
                    .into_iter()
                    .map(|(age, label)| {
                        format!(
                            "{}@{}ms",
                            label.as_deref().unwrap_or("unlabeled"),
                            age.as_millis()
                        )
                    })
                    .collect();
                tracing::warn!(
                    op,
                    open_tx_count = open.len(),
                    open_txs = %open.join(","),
                    "text write starved on the SQLite write lock; open registered \
                     transactions listed (empty means the holder issued no \
                     registered BEGIN IMMEDIATE in this process)"
                );
            }
        }
        result
    }

    async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        super::run_pooled_store_read(
            Arc::clone(&self.pool),
            StorageCapability::Text,
            op,
            move |conn| f(conn).map_err(|error| map_err(error, op)),
        )
        .await
    }
}

// -- Helper functions --

fn tags_to_json(tags: &[String]) -> String {
    serde_json::to_string(tags).unwrap_or_else(|_| "[]".to_string())
}

fn tags_from_json(s: &str) -> Vec<String> {
    serde_json::from_str(s).unwrap_or_default()
}

fn dt_to_micros(dt: &DateTime<Utc>) -> i64 {
    dt.timestamp_micros()
}

fn micros_to_dt(micros: i64) -> DateTime<Utc> {
    Utc.timestamp_micros(micros)
        .single()
        .unwrap_or_else(Utc::now)
}

/// Sanitize an FTS5 query string to prevent driver errors from special
/// chars: replace grouping/separator chars with spaces (splitting
/// punctuated identifiers into terms), strip remaining FTS5 operator
/// characters, then drop FTS5 keyword tokens (AND, OR, NOT, NEAR). See
/// `crates/khive-db/docs/api/text.md` for the exact char sets and the
/// issues (#388 and others) that grew them.
fn sanitize_fts5_query(query: &str) -> String {
    // Pass 1: replace grouping/separator chars with spaces to isolate tokens.
    // Colon, hyphen, and dot are included here (not in Pass 2) so punctuated
    // identifiers become separate terms rather than merged tokens.
    let spaced: String = query
        .chars()
        .map(|c| {
            if matches!(c, '(' | ')' | ',' | ':' | '-' | '.' | '/') {
                ' '
            } else {
                c
            }
        })
        .collect();

    // Pass 2: remove remaining FTS5 special chars and control characters.
    // Single quote (apostrophe) is included because FTS5 Plain-mode queries treat
    // it as a string-literal delimiter causing "syntax error near '''".
    // Dollar sign is included (#388) because FTS5's MATCH parser rejects it
    // unconditionally — `syntax error near "$"` — wherever it appears in the
    // expression, e.g. "$prev.id" (a common agent query for DSL docs).
    let sanitized: String = spaced
        .chars()
        .filter(|c| {
            !matches!(c, '*' | '"' | '\'' | '+' | '^' | '~' | '!' | '$' | '\0') && !c.is_control()
        })
        .collect();

    // Pass 3: filter FTS5 operator keywords.
    sanitized
        .split_whitespace()
        .filter(|t| {
            !matches!(
                t.to_ascii_uppercase().as_str(),
                "AND" | "OR" | "NOT" | "NEAR"
            )
        })
        .collect::<Vec<_>>()
        .join(" ")
}

/// Legacy (pre-#397) sanitization: hyphen and dot are stripped outright
/// (not space-split), so `khive-pack-memory` normalizes to the single merged
/// bareword `khivepackmemory`. Used only to build the merged-form
/// OR-alternative in [`sanitize_fts5_token_group`] — never as the sole
/// sanitized query — so pre-#397-indexed content stays reachable. See
/// `crates/khive-db/docs/api/text.md`.
fn sanitize_fts5_query_legacy_merged(query: &str) -> String {
    let spaced: String = query
        .chars()
        .map(|c| {
            if matches!(c, '(' | ')' | ',' | ':' | '/') {
                ' '
            } else {
                c
            }
        })
        .collect();

    let sanitized: String = spaced
        .chars()
        .filter(|c| {
            !matches!(
                c,
                '*' | '"' | '\'' | '+' | '-' | '^' | '.' | '~' | '!' | '$' | '\0'
            ) && !c.is_control()
        })
        .collect();

    sanitized
        .split_whitespace()
        .filter(|t| {
            !matches!(
                t.to_ascii_uppercase().as_str(),
                "AND" | "OR" | "NOT" | "NEAR"
            )
        })
        .collect::<Vec<_>>()
        .join(" ")
}

/// Escape a raw token for use inside a double-quoted FTS5 phrase. Embedded
/// double quotes are doubled; the trailing-`*` prefix-query trigger and
/// control characters are removed. Everything else, including punctuation,
/// passes through literally, since FTS5 phrase text is matched
/// against the column's own tokenization of that literal text (word-exact
/// under `unicode61`, substring-exact under `trigram`), not re-sanitized.
///
/// Returns `None` if nothing survives the filter.
fn sanitize_fts5_phrase_literal(token: &str) -> Option<String> {
    let mut literal = String::with_capacity(token.len());
    for c in token
        .chars()
        .filter(|c| !matches!(c, '*' | '\0') && !c.is_control())
    {
        if c == '"' {
            literal.push_str("\"\"");
        } else {
            literal.push(c);
        }
    }
    if literal.is_empty() {
        None
    } else {
        Some(literal)
    }
}

/// Below this length, FTS5's built-in `trigram` tokenizer (the production
/// default — `backend.rs::StorageBackend::text()`) produces zero tokens for
/// a bareword MATCH term. Verified against a live `tokenize='trigram'`
/// table: a term this short silently drops out of its AND clause instead of
/// constraining the match, e.g. `2026 07 10` matches any row containing
/// `2026` regardless of month/day. `sanitize_fts5_token_group` treats any
/// split segment at or below this length as trigram-unsafe.
const FTS5_TRIGRAM_MIN_SAFE_LEN: usize = 3;

/// FTS5 reserves punctuation in barewords for query syntax, including some
/// forms that change query meaning without producing an error. Keep bareword
/// alternatives to the conservative ASCII set and quote everything else.
fn is_fts5_bareword_safe(s: &str) -> bool {
    !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// Sanitize a single whitespace-isolated raw token into an FTS5
/// match-expression fragment, emitting additive OR-alternatives (split
/// AND-group, legacy-merged form, quoted literal phrase) so the result is
/// never a narrower match than any single safe form alone. Returns `None` if
/// the token sanitizes to nothing. See
/// `crates/khive-db/docs/api/text.md` for the full #397 trigram-safety
/// rationale behind which alternatives get emitted and when.
fn sanitize_fts5_token_group(token: &str) -> Option<String> {
    let split = sanitize_fts5_query(token);
    let split_terms: Vec<&str> = split.split_whitespace().collect();
    if split_terms.is_empty() {
        return None;
    }

    let all_bareword_safe = split_terms.iter().all(|t| is_fts5_bareword_safe(t));
    if split_terms.len() == 1 && is_fts5_bareword_safe(token) {
        return Some(split_terms[0].to_string());
    }

    let has_trigram_unsafe_segment = split_terms
        .iter()
        .any(|t| t.chars().count() < FTS5_TRIGRAM_MIN_SAFE_LEN);

    let mut alternatives = Vec::new();
    if all_bareword_safe && !has_trigram_unsafe_segment {
        alternatives.push(format!("({})", split_terms.join(" ")));
    }

    // An operator-bearing token (e.g. `NEAR(alpha-beta,5)`) can make the
    // legacy merge itself collapse to multiple space-separated terms rather
    // than one bareword: pass 1 of `sanitize_fts5_query_legacy_merged` spaces
    // out `(`, `)`, and `,` while pass 2 removes `-`/`.` outright, so
    // `NEAR(alpha-beta,5)` merges to `"alphabeta 5"`, not one word. Pushed
    // unguarded, that multi-term fragment carries the same trigram-unsafe
    // `5` the split-group check above exists to exclude, and under FTS5's
    // implicit-AND adjacency it silently drops, broadening the OR-alternative
    // to any row containing `alphabeta`. Apply the same trigram-safety gate
    // here whenever the merge is multi-term.
    let merged = sanitize_fts5_query_legacy_merged(token);
    let merged_terms: Vec<&str> = merged.split_whitespace().collect();
    let merged_all_bareword_safe =
        !merged_terms.is_empty() && merged_terms.iter().all(|t| is_fts5_bareword_safe(t));
    let merged_has_unsafe_segment = merged_terms.len() > 1
        && merged_terms
            .iter()
            .any(|t| t.chars().count() < FTS5_TRIGRAM_MIN_SAFE_LEN);
    // Compare against the split AND-group's own space-joined content — the
    // form it actually contributes to the expression — not the bare
    // concatenation of its terms. For an ordinary punctuated identifier the
    // concatenation always matches the merged bareword (both simply drop the
    // separators), which suppressed this alternative unconditionally and cut
    // off legacy-indexed content; a real duplicate only exists when the
    // merge itself produced the same space-separated content the split group
    // already emits (e.g. an operator token whose merge stays multi-term).
    let phrase = sanitize_fts5_phrase_literal(token);
    let duplicates_split = merged == split_terms.join(" ");
    let duplicates_phrase = phrase.as_deref() == Some(merged.as_str());
    if merged_all_bareword_safe
        && !merged.is_empty()
        && !duplicates_split
        && !duplicates_phrase
        && !merged_has_unsafe_segment
    {
        alternatives.push(merged);
    }

    if let Some(phrase) = phrase {
        alternatives.push(format!("\"{}\"", phrase));
    }

    match alternatives.len() {
        0 => None,
        1 => alternatives.into_iter().next(),
        _ => Some(format!("({})", alternatives.join(" OR "))),
    }
}

/// Join Plain-mode per-token groups into one MATCH expression.
///
/// FTS5's implicit-AND adjacency rule (a bare space between two terms)
/// applies only between two *plain* terms — verified against a live table:
/// `GQA KV cache` (all bare) matches, but `GQA AND KV AND cache` does not,
/// because FTS5's explicit `AND` treats a trigram-unsafe short bareword
/// (`KV`, 2 chars, zero trigrams) as an unsatisfiable operand, while
/// implicit adjacency instead lets it drop out harmlessly. So: use a bare
/// space between two plain (unparenthesized) groups to preserve that
/// existing leniency, and fall back to explicit `AND` only where at least
/// one side is a parenthesized OR-group (`sanitize_fts5_token_group`'s
/// output for punctuated tokens) — adjacency there without the operator is
/// a MATCH-expression syntax error (`(a OR b) c` fails; `(a OR b) AND c`
/// does not).
fn join_plain_groups(groups: &[String]) -> String {
    let mut expr = String::new();
    for (i, group) in groups.iter().enumerate() {
        if i > 0 {
            let prev_compound = groups[i - 1].starts_with('(');
            let this_compound = group.starts_with('(');
            expr.push_str(if prev_compound || this_compound {
                " AND "
            } else {
                " "
            });
        }
        expr.push_str(group);
    }
    expr
}

/// Build the FTS5 MATCH expression for a query string under a given mode.
///
/// Centralizes the AnyTerm/Plain/Phrase branching previously duplicated
/// across `search()`, `search_unranked()`, and `search_rank_within_cap()`.
///
/// AnyTerm and Plain both process the query token-by-token (split on
/// whitespace) through [`sanitize_fts5_token_group`], so a punctuated
/// identifier anywhere in the query gets its split/merged OR-alternative;
/// AnyTerm joins the per-token groups with `OR`; Plain joins them via
/// [`join_plain_groups`] (implicit-AND space where safe, explicit `AND`
/// where a group is parenthesized). Phrase mode keeps the single-string
/// literal behavior — a double-quoted FTS5 phrase cannot contain
/// `OR`/parenthesized groups.
///
/// Returns `None` when the sanitized query is empty (caller short-circuits
/// to an empty result set rather than sending an invalid MATCH expression).
fn build_match_expr(query: &str, mode: TextQueryMode) -> Option<String> {
    match mode {
        TextQueryMode::AnyTerm => {
            let groups: Vec<String> = query
                .split_whitespace()
                .filter_map(sanitize_fts5_token_group)
                .collect();
            if groups.is_empty() {
                None
            } else {
                Some(groups.join(" OR "))
            }
        }
        TextQueryMode::Plain => {
            let groups: Vec<String> = query
                .split_whitespace()
                .filter_map(sanitize_fts5_token_group)
                .collect();
            if groups.is_empty() {
                None
            } else {
                Some(join_plain_groups(&groups))
            }
        }
        TextQueryMode::Phrase => {
            sanitize_fts5_phrase_literal(query).map(|literal| format!("\"{}\"", literal))
        }
    }
}

fn quote_fts5_phrase(value: &str) -> String {
    format!("\"{}\"", value.replace('"', "\"\""))
}

/// Build an indexed classifier predicate for FTS5's MATCH expression.
///
/// The production tokenizer is trigram, so values shorter than three
/// characters have no postings. In that uncommon case the caller keeps only
/// the exact row predicate from `build_filter_clause`; correctness is
/// preserved while the indexed optimization deliberately declines.
fn record_kind_match_expr(filter: Option<&TextFilter>) -> Option<String> {
    let record_kinds = &filter?.record_kinds;
    if record_kinds.is_empty() || record_kinds.iter().any(|kind| kind.chars().count() < 3) {
        return None;
    }

    let clauses: Vec<String> = record_kinds
        .iter()
        .map(|kind| format!("record_kind : {}", quote_fts5_phrase(kind)))
        .collect();
    if clauses.len() == 1 {
        clauses.into_iter().next()
    } else {
        Some(format!("({})", clauses.join(" OR ")))
    }
}

fn build_filtered_match_expr(
    query: &str,
    mode: TextQueryMode,
    filter: Option<&TextFilter>,
) -> Option<String> {
    // Keep lexical terms confined to the two historically indexed text
    // columns. Once record_kind became indexed, an unqualified query would
    // otherwise make `memory` match every memory row solely via its classifier.
    let query = format!("{{title body}} : ({})", build_match_expr(query, mode)?);
    match record_kind_match_expr(filter) {
        Some(classifier) => Some(format!("{classifier} AND ({query})")),
        None => Some(query),
    }
}

/// Custom FTS5 rank configuration that keeps the classifier out of lexical
/// relevance while retaining the optimized hidden-`rank` ORDER BY path.
///
/// `record_kind` participates in MATCH for candidate pruning, but its weight is
/// zero so the classifier cannot become a relevance signal. Subject metadata
/// columns were already UNINDEXED; title/body retain their default unit weight.
const LEXICAL_BM25_RANK: &str = "bm25(0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0)";

/// Build a WHERE clause fragment and params for a `TextFilter`.
///
/// Returns `(clause, params)` where clause is empty if no filters are active.
/// Parameter indices start at `?{start_idx}`.
fn build_filter_clause(
    filter: &TextFilter,
    table: &str,
    start_idx: usize,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut conditions: Vec<String> = Vec::new();
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
    let mut idx = start_idx;

    if !filter.ids.is_empty() {
        let placeholders: Vec<String> = filter
            .ids
            .iter()
            .map(|_| {
                let p = format!("?{}", idx);
                idx += 1;
                p
            })
            .collect();
        conditions.push(format!(
            "{}.subject_id IN ({})",
            table,
            placeholders.join(", ")
        ));
        for id in &filter.ids {
            params.push(Box::new(id.to_string()));
        }
    }

    if !filter.kinds.is_empty() {
        let placeholders: Vec<String> = filter
            .kinds
            .iter()
            .map(|_| {
                let p = format!("?{}", idx);
                idx += 1;
                p
            })
            .collect();
        conditions.push(format!("{}.kind IN ({})", table, placeholders.join(", ")));
        for kind in &filter.kinds {
            params.push(Box::new(kind.to_string()));
        }
    }

    if !filter.record_kinds.is_empty() {
        let placeholders: Vec<String> = filter
            .record_kinds
            .iter()
            .map(|_| {
                let p = format!("?{}", idx);
                idx += 1;
                p
            })
            .collect();
        conditions.push(format!(
            "{}.record_kind IN ({})",
            table,
            placeholders.join(", ")
        ));
        for kind in &filter.record_kinds {
            params.push(Box::new(kind.clone()));
        }
    }

    if !filter.namespaces.is_empty() {
        let placeholders: Vec<String> = filter
            .namespaces
            .iter()
            .map(|_| {
                let p = format!("?{}", idx);
                idx += 1;
                p
            })
            .collect();
        conditions.push(format!(
            "{}.namespace IN ({})",
            table,
            placeholders.join(", ")
        ));
        for ns in &filter.namespaces {
            params.push(Box::new(ns.clone()));
        }
    }

    if conditions.is_empty() {
        (String::new(), params)
    } else {
        (format!(" AND {}", conditions.join(" AND ")), params)
    }
}

/// Test-only seam for [`delete_document_dml`]'s unmanaged (no-writer-task)
/// path: lets a test force the map-row delete to fail after the FTS-row
/// delete has already run, so the enclosing `BEGIN IMMEDIATE` / `ROLLBACK`
/// block's atomicity can be exercised without a real process crash
/// mid-transaction. Compiles to nothing outside `#[cfg(test)]` — production
/// code never reads this flag.
///
/// A process-wide `Mutex<HashSet<(String, Uuid)>>` rather than a
/// thread-local: the unmanaged path runs the DML closure inside
/// `tokio::task::spawn_blocking` (`with_writer_unmanaged`), on a worker
/// thread distinct from the test's own thread, so a thread-local set by the
/// test would never be observed by the closure. Each `(namespace,
/// subject_id)` target is armed/disarmed independently in the shared set —
/// `arm` inserts its own tuple and `disarm` removes only that same tuple —
/// so two tests running concurrently, each targeting a different key, cannot
/// clobber each other's armed state the way a single shared `Option` slot
/// would (one test's `disarm` would otherwise clear a target a second test
/// armed after the first but before the first's own disarm ran).
/// `delete_document_dml` only forces the failure when its own arguments are
/// a member of the set, so an unrelated delete running concurrently (e.g.
/// `test_delete_document`, which is not `#[serial]` against this seam) on a
/// key that was never armed is unaffected regardless of what else is armed
/// at the time.
#[cfg(test)]
pub(crate) mod delete_dml_test_seam {
    use std::collections::HashSet;
    use std::sync::Mutex;
    use uuid::Uuid;

    pub(crate) static TARGETS: Mutex<Option<HashSet<(String, Uuid)>>> = Mutex::new(None);

    pub(crate) fn arm(namespace: &str, subject_id: Uuid) {
        TARGETS
            .lock()
            .unwrap()
            .get_or_insert_with(HashSet::new)
            .insert((namespace.to_string(), subject_id));
    }

    pub(crate) fn disarm(namespace: &str, subject_id: Uuid) {
        if let Some(targets) = TARGETS.lock().unwrap().as_mut() {
            targets.remove(&(namespace.to_string(), subject_id));
        }
    }

    pub(crate) fn matches(namespace: &str, subject_id: Uuid) -> bool {
        TARGETS
            .lock()
            .unwrap()
            .as_ref()
            .is_some_and(|targets| targets.contains(&(namespace.to_string(), subject_id)))
    }
}

/// DML-only single-document delete shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `delete_document` paths.
///
/// Issues no `BEGIN` / `COMMIT` / `ROLLBACK` itself — the caller owns the
/// enclosing transaction. Both statements (the FTS row, looked up through
/// the map, then the map row itself) must commit atomically: a crash
/// between them would leave the map pointing at a rowid FTS5 may later
/// reuse for a different document.
fn delete_document_dml(
    conn: &rusqlite::Connection,
    table: &str,
    namespace: &str,
    subject_id: Uuid,
) -> Result<bool, SqliteError> {
    let [fts_statement, map_statement] = delete_document_statements(table, namespace, subject_id);

    let mut stmt = conn.prepare(&fts_statement.sql)?;
    bind_params(&mut stmt, &fts_statement.params)?;
    let deleted = stmt.raw_execute()? > 0;
    drop(stmt);

    #[cfg(test)]
    if delete_dml_test_seam::matches(namespace, subject_id) {
        return Err(SqliteError::InvalidData(
            "delete_document_dml test seam: forced failure between the FTS-row \
             delete and the map-row delete"
                .to_string(),
        ));
    }

    let mut map_stmt = conn.prepare(&map_statement.sql)?;
    bind_params(&mut map_stmt, &map_statement.params)?;
    map_stmt.raw_execute()?;

    Ok(deleted)
}

/// DML-only single-document upsert shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `upsert_document` paths (ADR-067 Component A).
///
/// Issues no `BEGIN` / `COMMIT` / `ROLLBACK` itself — the caller owns the
/// enclosing transaction.
fn upsert_document_dml(
    conn: &rusqlite::Connection,
    table: &str,
    document: &TextDocument,
) -> Result<(), rusqlite::Error> {
    // Delete (old rowid, via the map) -> insert (new rowid) -> map upsert
    // (new rowid). The map's own row for this key needs no separate delete:
    // `INSERT OR REPLACE` in the third statement overwrites it in place.
    let statement = delete_document_statement(table, &document.namespace, document.subject_id);
    let mut stmt = conn.prepare(&statement.sql)?;
    bind_params(&mut stmt, &statement.params)?;
    stmt.raw_execute()?;

    for statement in insert_document_statements(table, document) {
        let mut stmt = conn.prepare(&statement.sql)?;
        bind_params(&mut stmt, &statement.params)?;
        stmt.raw_execute()?;
    }
    Ok(())
}

/// DML-only batch upsert loop shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `upsert_documents` paths (ADR-067 Component A).
///
/// Issues no OUTER `BEGIN` / `COMMIT` / `ROLLBACK` — the caller owns the
/// enclosing transaction. The per-row named `SAVEPOINT fts_upsert_doc` is
/// preserved unchanged: it is what gives this loop its partial-success
/// semantics (one bad document does not abort the whole batch) independent
/// of which outer transaction wraps the loop.
fn batch_upsert_documents_dml(
    conn: &rusqlite::Connection,
    table: &str,
    documents: &[TextDocument],
    attempted: u64,
) -> Result<BatchWriteSummary, rusqlite::Error> {
    let map = rowid_map_table(table);
    // Delete via the map's primary key instead of a `namespace`/`subject_id`
    // scan of the (UNINDEXED on those columns) FTS table itself. The trailing
    // `AND namespace = ?1 AND subject_id = ?2` re-checks the key on the
    // rowid-narrowed candidate row: a stale map entry pointing at a rowid
    // FTS5 has since reused for a different document must delete zero rows,
    // not that unrelated document (mirrors `delete_document_statement`).
    let del_sql = format!(
        "DELETE FROM {table} WHERE rowid IN \
         (SELECT rowid FROM {map} WHERE namespace = ?1 AND subject_id = ?2) \
         AND namespace = ?1 AND subject_id = ?2"
    );
    let ins_sql = format!(
        "INSERT INTO {} \
         (subject_id, kind, title, body, tags, namespace, metadata, updated_at, record_kind) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
        table
    );
    // `INSERT OR REPLACE` — no separate delete of the map's own row needed;
    // see `upsert_document_dml`'s matching comment.
    let map_ins_sql = format!(
        "INSERT OR REPLACE INTO {map} (namespace, subject_id, rowid) \
         VALUES (?1, ?2, last_insert_rowid())"
    );

    let mut summary = BatchWriteSummary {
        attempted,
        ..BatchWriteSummary::default()
    };

    for (index, doc) in documents.iter().enumerate() {
        conn.execute_batch("SAVEPOINT fts_upsert_doc")?;
        let id_str = doc.subject_id.to_string();
        let namespace = &doc.namespace;
        let result = (|| {
            conn.execute(&del_sql, rusqlite::params![namespace, &id_str])?;

            let tags_json = tags_to_json(&doc.tags);
            let metadata_json: Option<String> = doc.metadata.as_ref().map(|v| v.to_string());

            conn.execute(
                &ins_sql,
                rusqlite::params![
                    &id_str,
                    &doc.kind.to_string(),
                    doc.title.as_deref().unwrap_or(""),
                    &doc.body,
                    &tags_json,
                    namespace,
                    &metadata_json,
                    dt_to_micros(&doc.updated_at),
                    &doc.record_kind,
                ],
            )?;
            conn.execute(&map_ins_sql, rusqlite::params![namespace, &id_str])?;
            Ok::<(), rusqlite::Error>(())
        })();

        match result {
            Ok(()) => {
                conn.execute_batch("RELEASE SAVEPOINT fts_upsert_doc")?;
                summary.affected = summary.affected.saturating_add(1);
            }
            Err(e) => {
                let _ = conn.execute_batch("ROLLBACK TO SAVEPOINT fts_upsert_doc");
                let _ = conn.execute_batch("RELEASE SAVEPOINT fts_upsert_doc");
                let (class, retryability) = super::classify_batch_sqlite_error(&e);
                summary.record_failure(index, Some(id_str), class, retryability, e.to_string());
            }
        }
    }

    Ok(summary)
}

#[async_trait]
impl TextSearch for Fts5TextSearch {
    async fn upsert_document(&self, document: TextDocument) -> Result<(), StorageError> {
        let table = self.table_name.clone();

        // ADR-067 Component A: when the write queue is enabled, route
        // through the pool-wide WriterTask. DML-only closure — no BEGIN
        // IMMEDIATE/COMMIT/ROLLBACK here, since the WriterTask's run loop
        // owns the transaction. `current_writer_task("fts_upsert")`
        // (ADR-136 D1 gate 3
        // amendment) re-checks past a construction-time `None` cache so a
        // handle that only became available later is still used here.
        if let Some(writer_task) = self.current_writer_task("fts_upsert")? {
            let table2 = table.clone();
            return writer_task
                .send_bounded(move |conn| {
                    upsert_document_dml(conn, &table2, &document)
                        .map_err(|e| map_err(e, "fts_upsert"))
                })
                .await;
        }

        // The direct typed unit owns the transaction around this DML body.
        self.with_writer("fts_upsert", move |conn| {
            upsert_document_dml(conn, &table, &document)
        })
        .await
    }

    async fn upsert_documents(
        &self,
        documents: Vec<TextDocument>,
    ) -> Result<BatchWriteSummary, StorageError> {
        let table = self.table_name.clone();
        let attempted = documents.len() as u64;

        // ADR-067 Component A: when the write queue is enabled, route
        // through the pool-wide WriterTask. DML-only closure (the per-row
        // `SAVEPOINT fts_upsert_doc` is preserved unchanged — only the OUTER
        // BEGIN IMMEDIATE/COMMIT is removed, since the WriterTask's run loop
        // owns the enclosing transaction). `current_writer_task` re-checks
        // past a construction-time `None` cache — see `upsert_document`'s
        // matching note.
        if let Some(writer_task) = self.current_writer_task("fts_upsert_batch")? {
            let table2 = table.clone();
            return writer_task
                .send_bounded(move |conn| {
                    batch_upsert_documents_dml(conn, &table2, &documents, attempted)
                        .map_err(|e| map_err(e, "fts_upsert_batch"))
                })
                .await;
        }

        // The direct typed unit owns the transaction around the complete batch.
        self.with_writer("fts_upsert_batch", move |conn| {
            batch_upsert_documents_dml(conn, &table, &documents, attempted)
        })
        .await
    }

    async fn delete_document(
        &self,
        namespace: &str,
        subject_id: Uuid,
    ) -> Result<bool, StorageError> {
        let table = self.table_name.clone();
        let namespace = namespace.to_string();

        if self.scan_fallback {
            let statement = delete_document_statement_scan_fallback(&table, &namespace, subject_id);
            return self
                .with_writer("fts_delete_scan_fallback", move |conn| {
                    let mut stmt = conn.prepare(&statement.sql)?;
                    bind_params(&mut stmt, &statement.params)?;
                    Ok(stmt.raw_execute()? > 0)
                })
                .await;
        }

        // Route through the shared WriterTask (DML-only closure — the
        // queue's run loop owns the transaction) when available, exactly
        // like `upsert_document`'s matching branch. The fallback below wraps
        // both statements in one explicit transaction: a crash between the
        // FTS-row delete and the map-row delete must never persist only one
        // of the two, or a later insert could reuse the freed rowid while
        // the map still points at it.
        if let Some(writer_task) = self.current_writer_task("fts_delete")? {
            let table2 = table.clone();
            let namespace2 = namespace.clone();
            return writer_task
                .send_bounded(move |conn| {
                    delete_document_dml(conn, &table2, &namespace2, subject_id)
                        .map_err(|e| map_sqlite_err(e, "fts_delete"))
                })
                .await;
        }

        self.with_writer("fts_delete", move |conn| {
            delete_document_dml(conn, &table, &namespace, subject_id).map_err(|error| match error {
                SqliteError::Rusqlite(inner) => inner,
                other => rusqlite::Error::InvalidParameterName(other.to_string()),
            })
        })
        .await
    }

    async fn get_document(
        &self,
        namespace: &str,
        subject_id: Uuid,
    ) -> Result<Option<TextDocument>, StorageError> {
        let namespace = namespace.to_string();
        let table = self.table_name.clone();
        let scan_fallback = self.scan_fallback;

        self.with_reader("fts_get", move |conn| {
            let sql = if scan_fallback {
                format!(
                    "SELECT subject_id, kind, title, body, tags, namespace, \
                     metadata, updated_at, record_kind \
                     FROM {table} WHERE namespace = ?1 AND subject_id = ?2"
                )
            } else {
                let map = rowid_map_table(&table);
                format!(
                    "SELECT t.subject_id, t.kind, t.title, t.body, t.tags, t.namespace, \
                     t.metadata, t.updated_at, t.record_kind \
                     FROM {table} AS t JOIN {map} AS m ON m.rowid = t.rowid \
                     WHERE m.namespace = ?1 AND m.subject_id = ?2 \
                     AND t.namespace = ?1 AND t.subject_id = ?2"
                )
            };
            let mut stmt = conn.prepare(&sql)?;
            let mut rows = stmt.query(rusqlite::params![namespace, subject_id.to_string()])?;

            match rows.next()? {
                Some(row) => {
                    let id_str: String = row.get(0)?;
                    let kind_str: String = row.get(1)?;
                    let title: String = row.get(2)?;
                    let body: String = row.get(3)?;
                    let tags_json: String = row.get(4)?;
                    let ns: String = row.get(5)?;
                    let metadata_json: Option<String> = row.get(6)?;
                    let updated_at_micros: i64 = row.get(7)?;
                    let record_kind: Option<String> = row.get(8)?;

                    let sid = Uuid::parse_str(&id_str).map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            0,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;

                    let kind = kind_str.parse::<SubstrateKind>().map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            1,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;

                    Ok(Some(TextDocument {
                        subject_id: sid,
                        kind,
                        record_kind,
                        title: if title.is_empty() { None } else { Some(title) },
                        body,
                        tags: tags_from_json(&tags_json),
                        namespace: ns,
                        metadata: metadata_json.and_then(|s| serde_json::from_str(&s).ok()),
                        updated_at: micros_to_dt(updated_at_micros),
                    }))
                }
                None => Ok(None),
            }
        })
        .await
    }

    async fn search(&self, request: TextSearchRequest) -> Result<Vec<TextSearchHit>, StorageError> {
        let table = self.table_name.clone();
        let usage = khive_storage::usage::current();

        let result = self
            .with_reader("fts_search", move |conn| {
                let match_expr = match build_filtered_match_expr(
                    &request.query,
                    request.mode,
                    request.filter.as_ref(),
                ) {
                    Some(expr) => expr,
                    None => return Ok(Vec::new()),
                };

                // Snippet column index 3 = body in the FTS5 schema.
                // snippet_chars == 0 is the sentinel for "no snippet" — skip the
                // snippet(...) call entirely and return NULL instead.  This avoids
                // the ~12ms BM25 snippet computation on the hot recall path where
                // snippets are unused.  Callers that need snippets (diagnostics) pass
                // snippet_chars > 0 and get the same behaviour as before.
                let snippet_expr = if request.snippet_chars == 0 {
                    "NULL AS snippet".to_string()
                } else {
                    let chars = i32::try_from(request.snippet_chars).unwrap_or(i32::MAX);
                    format!("snippet({table}, 3, '', '', '...', {chars})")
                };

                let (filter_clause, filter_params) = if let Some(ref filter) = request.filter {
                    build_filter_clause(filter, &table, 3)
                } else {
                    (String::new(), Vec::new())
                };

                let sql = format!(
                    "SELECT subject_id, rank, title, {snippet_expr} \
                     FROM {table} WHERE {table} MATCH ?1 \
                     AND rank MATCH '{LEXICAL_BM25_RANK}'{filter_clause} \
                     ORDER BY rank LIMIT ?2",
                );

                let mut stmt = conn.prepare(&sql)?;
                count_fts_pass(usage.as_ref());
                stmt.raw_bind_parameter(1, &match_expr)?;
                stmt.raw_bind_parameter(2, request.top_k as i64)?;

                for (i, param) in filter_params.iter().enumerate() {
                    param
                        .to_sql()
                        .map(|val| stmt.raw_bind_parameter(3 + i, val))
                        .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
                }

                let mut hits = Vec::new();
                let mut rows = stmt.raw_query();
                let mut rank_idx = 0u32;

                while let Some(row) = rows.next()? {
                    let id_str: String = row.get(0)?;
                    let fts_rank: f64 = row.get(1)?;
                    let title: String = row.get(2)?;
                    let snippet: Option<String> = row.get(3)?;

                    let subject_id = Uuid::parse_str(&id_str).map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            0,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;

                    rank_idx += 1;
                    hits.push((subject_id, fts_rank, rank_idx, title, snippet));
                }

                // Normalize scores within the result set to (0.05, 1.0].
                // Best rank (most negative) maps to 1.0, worst to 0.05.
                let min_rank = hits.iter().map(|h| h.1).fold(f64::INFINITY, f64::min);
                let max_rank = hits.iter().map(|h| h.1).fold(f64::NEG_INFINITY, f64::max);
                let range = max_rank - min_rank;

                let results = hits
                    .into_iter()
                    .map(|(subject_id, raw_rank, rank, title, snippet)| {
                        let score = if range.abs() < 1e-12 {
                            1.0
                        } else {
                            let t = (max_rank - raw_rank) / range;
                            0.05 + 0.95 * t
                        };
                        TextSearchHit {
                            subject_id,
                            score: DeterministicScore::from_f64(score),
                            rank,
                            title: if title.is_empty() { None } else { Some(title) },
                            snippet: snippet.filter(|s| !s.is_empty()),
                        }
                    })
                    .collect();

                Ok(results)
            })
            .await;

        result
    }

    async fn count(&self, filter: TextFilter) -> Result<u64, StorageError> {
        let table = self.table_name.clone();

        self.with_reader("fts_count", move |conn| {
            let indexed_classifier = record_kind_match_expr(Some(&filter));
            let filter_start = if indexed_classifier.is_some() { 2 } else { 1 };
            let (filter_clause, filter_params) = build_filter_clause(&filter, &table, filter_start);

            let sql = if indexed_classifier.is_some() {
                format!("SELECT COUNT(*) FROM {table} WHERE {table} MATCH ?1{filter_clause}")
            } else if filter_clause.is_empty() {
                format!("SELECT COUNT(*) FROM {table}")
            } else {
                let where_part = filter_clause.trim_start_matches(" AND ");
                format!("SELECT COUNT(*) FROM {table} WHERE {where_part}")
            };

            let mut stmt = conn.prepare(&sql)?;

            if let Some(classifier) = indexed_classifier {
                stmt.raw_bind_parameter(1, classifier)?;
            }

            for (i, param) in filter_params.iter().enumerate() {
                param
                    .to_sql()
                    .map(|val| stmt.raw_bind_parameter(filter_start + i, val))
                    .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
            }

            let mut rows = stmt.raw_query();
            match rows.next()? {
                Some(row) => {
                    let count: i64 = row.get(0)?;
                    Ok(count as u64)
                }
                None => Ok(0),
            }
        })
        .await
    }

    async fn stats(&self) -> Result<TextIndexStats, StorageError> {
        let table = self.table_name.clone();

        self.with_reader("fts_stats", move |conn| {
            let sql = format!("SELECT COUNT(*) FROM {}", table);
            let count: i64 = conn.query_row(&sql, [], |row| row.get(0))?;

            Ok(TextIndexStats {
                document_count: count as u64,
                needs_rebuild: false,
                last_rebuild_at: None,
            })
        })
        .await
    }

    async fn search_with_options(
        &self,
        request: TextSearchRequest,
        options: TextSearchOptions,
    ) -> Result<Vec<TextSearchHit>, StorageError> {
        match options.gather_mode {
            TextGatherMode::Ranked => self.search(request).await,
            TextGatherMode::Unranked => self.search_unranked(request).await,
            TextGatherMode::RankWithinCap => {
                let gather_limit = options
                    .gather_limit
                    .unwrap_or(request.top_k)
                    .max(request.top_k);
                self.search_rank_within_cap(request, gather_limit).await
            }
        }
    }

    async fn term_stats(
        &self,
        request: TextTermStatsRequest,
    ) -> Result<Vec<TextTermStats>, StorageError> {
        let table = self.table_name.clone();

        self.with_reader("fts_term_stats", move |conn| {
            let filter = request.filter.as_ref();
            let indexed_classifier = record_kind_match_expr(filter);

            let count_filter_start = if indexed_classifier.is_some() { 2 } else { 1 };
            let (count_filter_clause, count_filter_params) = if let Some(f) = filter {
                build_filter_clause(f, &table, count_filter_start)
            } else {
                (String::new(), Vec::new())
            };

            let document_count: u64 = {
                let count_sql = if indexed_classifier.is_some() {
                    format!(
                        "SELECT COUNT(*) FROM {table} \
                         WHERE {table} MATCH ?1{count_filter_clause}"
                    )
                } else if count_filter_clause.is_empty() {
                    format!("SELECT COUNT(*) FROM {table}")
                } else {
                    let where_part = count_filter_clause.trim_start_matches(" AND ");
                    format!("SELECT COUNT(*) FROM {table} WHERE {where_part}")
                };
                let mut stmt = conn.prepare(&count_sql)?;
                if let Some(ref classifier) = indexed_classifier {
                    stmt.raw_bind_parameter(1, classifier)?;
                }
                for (i, param) in count_filter_params.iter().enumerate() {
                    param
                        .to_sql()
                        .map(|val| stmt.raw_bind_parameter(count_filter_start + i, val))
                        .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
                }
                let mut rows = stmt.raw_query();
                match rows.next()? {
                    Some(row) => {
                        let c: i64 = row.get(0)?;
                        c as u64
                    }
                    None => 0,
                }
            };

            let mut results = Vec::with_capacity(request.terms.len());
            for term in &request.terms {
                // Keep document frequency aligned with the per-token search expression.
                let sanitized = sanitize_fts5_token_group(term).unwrap_or_default();
                if sanitized.is_empty() {
                    results.push(TextTermStats {
                        term: term.clone(),
                        sanitized_term: sanitized,
                        document_frequency: 0,
                        document_count,
                        inverse_document_frequency: 0.0,
                    });
                    continue;
                }

                // Per-term count: MATCH is ?1, so filter params start at ?2.
                let (term_filter_clause, term_filter_params) = if let Some(f) = filter {
                    build_filter_clause(f, &table, 2)
                } else {
                    (String::new(), Vec::new())
                };

                let text_term = format!("{{title body}} : ({sanitized})");
                let term_match = match &indexed_classifier {
                    Some(classifier) => format!("{classifier} AND ({text_term})"),
                    None => text_term,
                };
                let count_sql = format!(
                    "SELECT COUNT(*) FROM {table} WHERE {table} MATCH ?1{term_filter_clause}"
                );
                let mut stmt = conn.prepare(&count_sql)?;
                stmt.raw_bind_parameter(1, &term_match)?;
                for (i, param) in term_filter_params.iter().enumerate() {
                    param
                        .to_sql()
                        .map(|val| stmt.raw_bind_parameter(2 + i, val))
                        .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
                }

                let df: u64 = {
                    let mut rows = stmt.raw_query();
                    match rows.next()? {
                        Some(row) => {
                            let c: i64 = row.get(0)?;
                            c as u64
                        }
                        None => 0,
                    }
                };

                let idf = Fts5TextSearch::bm25_idf(df, document_count);
                results.push(TextTermStats {
                    term: term.clone(),
                    sanitized_term: sanitized,
                    document_frequency: df,
                    document_count,
                    inverse_document_frequency: idf,
                });
            }

            Ok(results)
        })
        .await
    }

    async fn rebuild(&self, _scope: IndexRebuildScope) -> Result<TextIndexStats, StorageError> {
        let table = self.table_name.clone();

        self.with_writer("fts_rebuild", move |conn| {
            // FTS5 rebuild command: repopulates the internal index structures.
            let sql = format!("INSERT INTO {}({}) VALUES('rebuild')", table, table);
            conn.execute(&sql, [])?;

            let count_sql = format!("SELECT COUNT(*) FROM {}", table);
            let count: i64 = conn.query_row(&count_sql, [], |row| row.get(0))?;

            Ok(TextIndexStats {
                document_count: count as u64,
                needs_rebuild: false,
                last_rebuild_at: Some(Utc::now()),
            })
        })
        .await
    }
}

impl Fts5TextSearch {
    /// Robertson-Walker BM25 IDF: ln(((N - df + 0.5) / (df + 0.5)) + 1)
    fn bm25_idf(df: u64, document_count: u64) -> f64 {
        let n = document_count as f64;
        let f = df as f64;
        ((n - f + 0.5) / (f + 0.5) + 1.0).ln()
    }

    /// Gather candidates without BM25 ranking; return with uniform score 1.0.
    async fn search_unranked(
        &self,
        request: TextSearchRequest,
    ) -> Result<Vec<TextSearchHit>, StorageError> {
        let table = self.table_name.clone();
        let usage = khive_storage::usage::current();

        self.with_reader("fts_search_unranked", move |conn| {
            let match_expr = match build_filtered_match_expr(
                &request.query,
                request.mode,
                request.filter.as_ref(),
            ) {
                Some(expr) => expr,
                None => return Ok(Vec::new()),
            };

            let (filter_clause, filter_params) = if let Some(ref filter) = request.filter {
                build_filter_clause(filter, &table, 3)
            } else {
                (String::new(), Vec::new())
            };

            // No rank column, no ORDER BY — avoids BM25 computation entirely.
            let sql = format!(
                "SELECT subject_id, title \
                 FROM {table} WHERE {table} MATCH ?1{filter_clause} \
                 LIMIT ?2",
            );

            let mut stmt = conn.prepare(&sql)?;
            count_fts_pass(usage.as_ref());
            stmt.raw_bind_parameter(1, &match_expr)?;
            stmt.raw_bind_parameter(2, request.top_k as i64)?;

            for (i, param) in filter_params.iter().enumerate() {
                param
                    .to_sql()
                    .map(|val| stmt.raw_bind_parameter(3 + i, val))
                    .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
            }

            let mut results = Vec::new();
            let mut rows = stmt.raw_query();
            let mut rank_idx = 0u32;

            while let Some(row) = rows.next()? {
                let id_str: String = row.get(0)?;
                let title: String = row.get(1)?;

                let subject_id = Uuid::parse_str(&id_str).map_err(|e| {
                    rusqlite::Error::FromSqlConversionFailure(
                        0,
                        rusqlite::types::Type::Text,
                        Box::new(e),
                    )
                })?;

                rank_idx += 1;
                results.push(TextSearchHit {
                    subject_id,
                    score: DeterministicScore::from_f64(1.0),
                    rank: rank_idx,
                    title: if title.is_empty() { None } else { Some(title) },
                    snippet: None,
                });
            }

            Ok(results)
        })
        .await
    }

    /// Two-stage gather: cheap unranked LIMIT gather_limit, then BM25-rank the subset.
    async fn search_rank_within_cap(
        &self,
        request: TextSearchRequest,
        gather_limit: u32,
    ) -> Result<Vec<TextSearchHit>, StorageError> {
        let table = self.table_name.clone();
        let usage = khive_storage::usage::current();

        self.with_reader("fts_search_rank_within_cap", move |conn| {
            let match_expr = match build_filtered_match_expr(
                &request.query,
                request.mode,
                request.filter.as_ref(),
            ) {
                Some(expr) => expr,
                None => return Ok(Vec::new()),
            };

            let (filter_clause, filter_params) = if let Some(ref filter) = request.filter {
                build_filter_clause(filter, &table, 3)
            } else {
                (String::new(), Vec::new())
            };

            // Stage 1: cheap unranked gather of rowids.
            let gather_sql = format!(
                "SELECT subject_id FROM {table} WHERE {table} MATCH ?1{filter_clause} LIMIT ?2"
            );

            let mut stmt = conn.prepare(&gather_sql)?;
            count_fts_pass(usage.as_ref());
            stmt.raw_bind_parameter(1, &match_expr)?;
            stmt.raw_bind_parameter(2, gather_limit as i64)?;
            for (i, param) in filter_params.iter().enumerate() {
                param
                    .to_sql()
                    .map(|val| stmt.raw_bind_parameter(3 + i, val))
                    .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))??;
            }

            let mut gathered_ids: Vec<String> = Vec::new();
            let mut rows = stmt.raw_query();
            while let Some(row) = rows.next()? {
                gathered_ids.push(row.get::<_, String>(0)?);
            }

            if gathered_ids.is_empty() {
                return Ok(Vec::new());
            }

            // Stage 2: BM25-rank only the gathered subset via subject_id IN (...).
            let snippet_expr = if request.snippet_chars == 0 {
                "NULL AS snippet".to_string()
            } else {
                let chars = i32::try_from(request.snippet_chars).unwrap_or(i32::MAX);
                format!("snippet({table}, 3, '', '', '...', {chars})")
            };

            // Build IN clause for the gathered IDs.
            let id_placeholders: Vec<String> = gathered_ids
                .iter()
                .enumerate()
                .map(|(i, _)| format!("?{}", 3 + i))
                .collect();
            let in_clause = id_placeholders.join(", ");

            let rank_sql = format!(
                "SELECT subject_id, rank, title, {snippet_expr} \
                 FROM {table} WHERE {table} MATCH ?1 \
                 AND rank MATCH '{LEXICAL_BM25_RANK}' \
                 AND subject_id IN ({in_clause}) \
                 ORDER BY rank LIMIT ?2",
            );

            let mut stmt2 = conn.prepare(&rank_sql)?;
            count_fts_pass(usage.as_ref());
            stmt2.raw_bind_parameter(1, &match_expr)?;
            stmt2.raw_bind_parameter(2, request.top_k as i64)?;
            for (i, id_str) in gathered_ids.iter().enumerate() {
                stmt2.raw_bind_parameter(3 + i, id_str.as_str())?;
            }

            let mut hits = Vec::new();
            let mut rows2 = stmt2.raw_query();
            let mut rank_idx = 0u32;

            while let Some(row) = rows2.next()? {
                let id_str: String = row.get(0)?;
                let fts_rank: f64 = row.get(1)?;
                let title: String = row.get(2)?;
                let snippet: Option<String> = row.get(3)?;

                let subject_id = Uuid::parse_str(&id_str).map_err(|e| {
                    rusqlite::Error::FromSqlConversionFailure(
                        0,
                        rusqlite::types::Type::Text,
                        Box::new(e),
                    )
                })?;

                rank_idx += 1;
                hits.push((subject_id, fts_rank, rank_idx, title, snippet));
            }

            // Normalize scores within the ranked subset (same formula as search()).
            let min_rank = hits.iter().map(|h| h.1).fold(f64::INFINITY, f64::min);
            let max_rank = hits.iter().map(|h| h.1).fold(f64::NEG_INFINITY, f64::max);
            let range = max_rank - min_rank;

            let results = hits
                .into_iter()
                .map(|(subject_id, raw_rank, rank, title, snippet)| {
                    let score = if range.abs() < 1e-12 {
                        1.0
                    } else {
                        let t = (max_rank - raw_rank) / range;
                        0.05 + 0.95 * t
                    };
                    TextSearchHit {
                        subject_id,
                        score: DeterministicScore::from_f64(score),
                        rank,
                        title: if title.is_empty() { None } else { Some(title) },
                        snippet: snippet.filter(|s| !s.is_empty()),
                    }
                })
                .collect();

            Ok(results)
        })
        .await
    }

    /// Move all FTS5 documents from `old_namespace` to `new_namespace` in a
    /// single transaction.
    ///
    /// FTS5 virtual tables do not support updating indexed columns (`title`,
    /// `body`) via UPDATE. The correct approach is read-then-delete-then-reinsert.
    ///
    /// Callers must invoke this after any SQL-level namespace change on the
    /// backing entity table so that FTS5 keyword search stays consistent with
    /// the entity store.
    // REASON: reserved for namespace migration operations
    #[allow(dead_code)]
    pub(crate) async fn rename_namespace(
        &self,
        old_namespace: &str,
        new_namespace: &str,
    ) -> Result<u64, StorageError> {
        if old_namespace == new_namespace {
            return Ok(0);
        }
        let table = self.table_name.clone();
        let old_ns = old_namespace.to_string();
        let new_ns = new_namespace.to_string();

        // ADR-136 D1 gate 2: queue-first, DML-only closure. `run_writer_task`
        // already owns the enclosing `BEGIN IMMEDIATE`/`COMMIT`/`ROLLBACK` for
        // this request, so the SELECT enumerating rows-to-move runs INSIDE
        // that same transaction as the DELETE+INSERT that consumes it —
        // closing the TOCTOU window the pre-migration path had (its SELECT
        // ran before its own `BEGIN IMMEDIATE`, so a writer landing between
        // the two could resurrect a stale row or lose one mid-rename).
        if let Some(writer_task) = self.current_writer_task("fts_rename_namespace")? {
            let table2 = table.clone();
            return writer_task
                .send_bounded(move |conn| {
                    rename_namespace_dml(conn, &table2, &old_ns, &new_ns)
                        .map_err(|e| map_err(e, "fts_rename_namespace"))
                })
                .await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteFtsRenameNamespace);

        self.with_writer_unmanaged("fts_rename_namespace", move |conn| {
            // The SELECT and writes share the typed unit's transaction.
            rename_namespace_dml(conn, &table, &old_ns, &new_ns)
        })
        .await
    }
}

struct FtsRenameRow {
    subject_id: String,
    kind: String,
    title: String,
    body: String,
    tags: String,
    metadata: Option<String>,
    updated_at: i64,
    record_kind: Option<String>,
}

/// Move every FTS5 document row from `old_ns` to `new_ns` in `table`.
/// DML-only — issues no `BEGIN`/`COMMIT`/`ROLLBACK` of its own, so it is safe
/// to call both inside an already-open transaction (the `WriterTask` queue
/// path) and wrapped by a caller-managed one (the legacy standalone path).
/// The `SELECT` enumerating rows-to-move and the `DELETE`+`INSERT` that
/// consumes it always run inside the SAME transaction the caller opened —
/// see `Fts5TextSearch::rename_namespace`'s TOCTOU note.
fn rename_namespace_dml(
    conn: &rusqlite::Connection,
    table: &str,
    old_ns: &str,
    new_ns: &str,
) -> Result<u64, rusqlite::Error> {
    let sel_sql = format!(
        "SELECT subject_id, kind, title, body, tags, metadata, updated_at, record_kind \
         FROM {} WHERE namespace = ?1",
        table
    );
    let rows: Vec<FtsRenameRow> = {
        let mut stmt = conn.prepare(&sel_sql)?;
        let iter = stmt.query_map(rusqlite::params![old_ns], |row| {
            Ok(FtsRenameRow {
                subject_id: row.get(0)?,
                kind: row.get(1)?,
                title: row.get(2)?,
                body: row.get(3)?,
                tags: row.get(4)?,
                metadata: row.get(5)?,
                updated_at: row.get(6)?,
                record_kind: row.get(7)?,
            })
        })?;
        iter.collect::<Result<Vec<_>, _>>()?
    };
    let moved = rows.len() as u64;
    if moved == 0 {
        return Ok(0);
    }

    let del_sql = format!("DELETE FROM {} WHERE namespace = ?1", table);
    conn.execute(&del_sql, rusqlite::params![old_ns])?;

    // The rows just deleted above are gone from `table` entirely (whole
    // namespace, not a single subject), so their map entries are pure
    // leftovers now — clear them before reinserting under `new_ns`.
    let map = rowid_map_table(table);
    let map_del_sql = format!("DELETE FROM {map} WHERE namespace = ?1");
    conn.execute(&map_del_sql, rusqlite::params![old_ns])?;

    let ins_sql = format!(
        "INSERT INTO {} \
         (subject_id, kind, title, body, tags, namespace, metadata, updated_at, record_kind) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
        table
    );
    let map_ins_sql = format!(
        "INSERT OR REPLACE INTO {map} (namespace, subject_id, rowid) \
         VALUES (?1, ?2, last_insert_rowid())"
    );
    for row in &rows {
        conn.execute(
            &ins_sql,
            rusqlite::params![
                row.subject_id,
                row.kind,
                row.title,
                row.body,
                row.tags,
                new_ns,
                row.metadata,
                row.updated_at,
                row.record_kind,
            ],
        )?;
        conn.execute(&map_ins_sql, rusqlite::params![new_ns, row.subject_id])?;
    }

    Ok(moved)
}

#[cfg(test)]
#[path = "text_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "text_busy_tests.rs"]
mod direct_busy_tests;