nabu-core 0.1.6

//! History search: lexical FTS, hybrid/auto dispatch, reciprocal-rank fusion,
//! snippet/term helpers, and result hydration. The git corroboration engine is
//! the `corroborate` submodule; the feature-gated vector-read path stays in
//! lib.rs (the semantic module, final phase) and is reached via `crate::`.

pub(crate) mod corroborate;

use crate::{
    expand_query_terms, native_jsonl_line_command, normalize_date_or_duration, open_index,
    open_raw_offset_reader, raw_envelope_for_line_scan, read_raw_envelope_at_offset,
    resolved_payload_for_envelope, semantic_search_available, sha256_hex, vector_search_results,
    CanonicalType, Error, RankedSearchResult, Result, SearchContinuation, SearchMode,
    SearchOptions, SearchPage, SearchResult, Tool, MAX_SEARCH_LIMIT, MAX_SEARCH_SNIPPET_CHARS,
};
pub(crate) use corroborate::corroborate_text;
use rusqlite::params_from_iter;
use rusqlite::types::Value as SqlValue;
use rusqlite::Connection;
use serde_json::Value;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::str::FromStr;

pub fn search_history(home: &Path, query: &str, limit: usize) -> Result<Vec<SearchResult>> {
    search_history_filtered(
        home,
        query,
        SearchOptions {
            limit,
            ..SearchOptions::default()
        },
    )
}

pub fn search_history_filtered(
    home: &Path,
    query: &str,
    options: SearchOptions,
) -> Result<Vec<SearchResult>> {
    Ok(search_history_page(home, query, options)?.results)
}

pub fn search_history_page(home: &Path, query: &str, options: SearchOptions) -> Result<SearchPage> {
    if query.trim().is_empty() {
        return Err(Error::Validation("query must not be empty".to_string()));
    }
    let mode_requested = options.mode;
    let semantic_available = semantic_search_available(home);
    let mut mode_applied = match mode_requested {
        SearchMode::Auto if semantic_available => SearchMode::Hybrid,
        SearchMode::Auto => SearchMode::Lexical,
        SearchMode::Lexical => SearchMode::Lexical,
        SearchMode::Hybrid if semantic_available => SearchMode::Hybrid,
        SearchMode::Hybrid => {
            return Err(Error::SemanticUnavailable(
                "local embedding model and vector index are not available; run lexical mode or install the semantic model explicitly".to_string(),
            ))
        }
    };
    if mode_applied == SearchMode::Hybrid {
        match search_history_hybrid_page(home, query, options.clone(), semantic_available) {
            Ok(page) => return Ok(page),
            Err(Error::SemanticUnavailable(_)) if mode_requested == SearchMode::Auto => {
                mode_applied = SearchMode::Lexical;
            }
            Err(error) => return Err(error),
        }
    }
    let query_terms = effective_search_terms(query, options.expand_concepts)?;
    let fts_query = quoted_fts_terms(&query_terms);
    let limit = options.limit.clamp(1, MAX_SEARCH_LIMIT);
    let offset = options.offset;
    let max_snippet_chars = options.max_snippet_chars.clamp(1, MAX_SEARCH_SNIPPET_CHARS);
    let raw_fetch_limit = search_overfetch_limit(offset, limit);
    let (results, has_more_raw_rows) = lexical_search_ranked_results(
        home,
        &options,
        &query_terms,
        &fts_query,
        raw_fetch_limit,
        max_snippet_chars,
    )?;

    finalize_search_page(
        &options,
        SearchPageAssembly {
            results,
            has_more_raw_rows,
            mode_requested,
            mode_applied,
            semantic_available,
            limit,
            offset,
            max_snippet_chars,
        },
    )
}

/// Ranked results plus the pagination facts a page needs before its shared
/// finalization. Both the lexical and hybrid paths produce one of these; the
/// only path-specific inputs are `mode_applied` and the ranked list itself.
pub(crate) struct SearchPageAssembly {
    pub(crate) results: Vec<RankedSearchResult>,
    /// The ranked list was overfetch-truncated (more matching rows exist than
    /// were retrieved), so the total is unknown and `total_estimated` is `None`.
    pub(crate) has_more_raw_rows: bool,
    pub(crate) mode_requested: SearchMode,
    pub(crate) mode_applied: SearchMode,
    pub(crate) semantic_available: bool,
    pub(crate) limit: usize,
    pub(crate) offset: usize,
    pub(crate) max_snippet_chars: usize,
}

/// Shared page assembly for every search path: dedupe, total-estimate
/// semantics, window slicing, payload hydration, corroboration, and the
/// pagination cursor. Extracting this keeps the lexical and hybrid paths from
/// drifting — every result path applies the same dedupe/total/corroboration
/// rules, so citation completeness and `--corroborate` behave identically.
fn finalize_search_page(
    options: &SearchOptions,
    assembly: SearchPageAssembly,
) -> Result<SearchPage> {
    let SearchPageAssembly {
        mut results,
        has_more_raw_rows,
        mode_requested,
        mode_applied,
        semantic_available,
        limit,
        offset,
        max_snippet_chars,
    } = assembly;

    if options.dedupe {
        results = dedupe_ranked_search_results(results)?;
    }
    let advisory = search_advisory(&results, mode_applied, semantic_available, options);

    let total_estimated = if has_more_raw_rows {
        None
    } else {
        Some(results.len())
    };
    let has_more_logical_rows = results.len() > offset.saturating_add(limit);
    let mut page_results = results
        .into_iter()
        .skip(offset)
        .take(limit)
        .map(|ranked| ranked.result)
        .collect::<Vec<_>>();
    if options.include_payload {
        hydrate_search_result_payloads(&mut page_results)?;
    }
    if options.corroborate {
        annotate_search_results_with_corroboration(&mut page_results);
    }
    let returned = page_results.len();
    let continuation = if returned > 0 && (has_more_raw_rows || has_more_logical_rows) {
        Some(SearchContinuation {
            next_offset: offset.saturating_add(returned),
        })
    } else {
        None
    };

    Ok(SearchPage {
        results: page_results,
        truncated: continuation.is_some(),
        returned,
        total_estimated,
        continuation,
        mode_requested,
        mode_applied,
        semantic_available,
        limit_applied: limit,
        offset_applied: offset,
        max_snippet_chars_applied: max_snippet_chars,
        include_payload: options.include_payload,
        include_deltas: options.include_deltas,
        dedupe: options.dedupe,
        expand_concepts: options.expand_concepts,
        advisory,
    })
}

/// Scatter window for the weak-page advisory: how many top-ranked hits are
/// inspected, and the minimum among them before scatter is judged at all.
const SEARCH_ADVISORY_TOP_K: usize = 10;
const SEARCH_ADVISORY_MIN_TOP: usize = 5;

/// Retrieval hint for weak lexical-only pages. Fires only when the query ran
/// lexical-only with neither semantic rescue nor concept expansion available,
/// and the ranked set is empty or its top hits are spread across sessions with
/// no concentration — the signature of a concept-worded query matching generic
/// tokens. Hybrid pages and expanded queries never carry it, so the two score
/// scales (BM25 vs RRF) never need comparing.
fn search_advisory(
    results: &[RankedSearchResult],
    mode_applied: SearchMode,
    semantic_available: bool,
    options: &SearchOptions,
) -> Option<String> {
    match (mode_applied, semantic_available, options.expand_concepts) {
        (SearchMode::Lexical, false, false) => {}
        _ => return None,
    }
    if results.is_empty() {
        return Some(
            "No lexical matches and semantic search is unavailable. Retry with distinctive \
             literal tokens (identifiers, filenames, error strings, command fragments) or set \
             expand_concepts=true."
                .to_string(),
        );
    }
    let top: Vec<&str> = results
        .iter()
        .take(SEARCH_ADVISORY_TOP_K)
        .map(|ranked| ranked.result.session_id.as_str())
        .collect();
    let distinct = top.iter().copied().collect::<HashSet<_>>().len();
    let scattered = top.len() >= SEARCH_ADVISORY_MIN_TOP && distinct * 3 >= top.len() * 2;
    scattered.then(|| {
        "Lexical-only search with top hits scattered across sessions; semantic search is \
         unavailable. If these look off-target, retry with distinctive literal tokens \
         (identifiers, filenames, error strings) or set expand_concepts=true."
            .to_string()
    })
}

/// How a `ref=` search filter compares against the stored `event_refs` rows.
pub(crate) struct RefFilterMatch {
    /// `pr` or `commit`.
    pub(crate) kind: &'static str,
    /// A `LIKE` pattern (with `\` as the escape character) applied to
    /// `event_refs.ref_value`.
    pub(crate) value_pattern: String,
}

/// Resolve a raw `ref=` filter string into the `event_refs` row it should match.
///
/// PR forms (`#54`, `54`, `PR #54`, `org/repo#54`) normalize to an exact match
/// on the `pr` kind value `#54`. A bare hex token is treated as a `commit` SHA
/// and matched by case-insensitive prefix (`abc123` matches `abc123def...`), so
/// abbreviated SHAs resolve the way they are written in transcripts. Any `%`/`_`
/// in the input is escaped so it cannot act as a wildcard.
pub(crate) fn normalize_ref_filter(raw: &str) -> RefFilterMatch {
    let trimmed = raw.trim();
    let pr_digits = trimmed
        .strip_prefix('#')
        .or_else(|| trimmed.rsplit_once('#').map(|(_, tail)| tail))
        .map(str::trim)
        .filter(|tail| !tail.is_empty() && tail.bytes().all(|byte| byte.is_ascii_digit()));
    if let Some(digits) = pr_digits.or_else(|| {
        // `ref=54` (no `#`) is a PR number when it is purely decimal.
        (!trimmed.is_empty() && trimmed.bytes().all(|byte| byte.is_ascii_digit()))
            .then_some(trimmed)
    }) {
        let normalized = digits.trim_start_matches('0');
        let normalized = if normalized.is_empty() {
            "0"
        } else {
            normalized
        };
        return RefFilterMatch {
            kind: "pr",
            value_pattern: escape_like(&format!("#{normalized}")),
        };
    }
    // Otherwise treat the input as a commit SHA (or SHA prefix), matched by
    // prefix against the lowercased stored value.
    RefFilterMatch {
        kind: "commit",
        value_pattern: format!("{}%", escape_like(&trimmed.to_ascii_lowercase())),
    }
}

fn lexical_search_ranked_results(
    home: &Path,
    options: &SearchOptions,
    query_terms: &[String],
    fts_query: &str,
    fetch_limit: usize,
    max_snippet_chars: usize,
) -> Result<(Vec<RankedSearchResult>, bool)> {
    let db_path = home.join("index").join("harness.db");
    let conn = open_index(&db_path)?;
    let mut sql = String::from(
        "SELECT
           e.id,
           e.tool,
           e.session_id,
           e.canonical_type,
           e.captured_at,
           -bm25(events_fts, 8.0, 6.0, 4.0, 1.0, 0.5) AS score,
           NULL AS snippet,
           e.searchable_text,
           e.raw_file,
           e.raw_line,
           e.raw_offset,
           e.compaction_state,
           e.cwd,
           e.project_root,
           e.tool_invocation_id
         FROM events_fts
         JOIN events e ON e.id = events_fts.rowid
         WHERE events_fts MATCH ?",
    );
    let mut params = vec![SqlValue::Text(fts_query.to_string())];

    if let Some(tool) = options.tool {
        sql.push_str(" AND e.tool = ?");
        params.push(SqlValue::Text(tool.as_str().to_string()));
    }
    if let Some(session_id) = options.session_id.as_deref() {
        let resolved = resolve_session_filter_ids(&conn, &db_path, options.tool, session_id)?;
        let placeholders = vec!["?"; resolved.len()].join(", ");
        sql.push_str(&format!(" AND e.session_id IN ({placeholders})"));
        for id in resolved {
            params.push(SqlValue::Text(id));
        }
    }
    if let Some(cwd) = options.cwd.as_deref() {
        sql.push_str(" AND e.cwd = ?");
        params.push(SqlValue::Text(cwd.to_string()));
    }
    if let Some(since) = options.since.as_deref() {
        sql.push_str(" AND e.captured_at >= ?");
        params.push(SqlValue::Text(normalize_date_or_duration(since, "since")?));
    }
    if let Some(canonical_type) = options.canonical_type.as_deref() {
        sql.push_str(" AND e.canonical_type = ?");
        params.push(SqlValue::Text(canonical_type.to_string()));
    }
    if !options.include_deltas {
        sql.push_str(" AND e.canonical_type != 'assistant.delta'");
    }
    if let Some(file) = options.file.as_deref() {
        sql.push_str(
            " AND EXISTS (
                SELECT 1
                FROM event_files ef
                JOIN files f ON f.id = ef.file_id
                WHERE ef.event_id = e.id
                  AND (f.path = ? OR f.path LIKE ?)
              )",
        );
        params.push(SqlValue::Text(file.to_string()));
        params.push(SqlValue::Text(format!("%{file}%")));
    }
    if let Some(command) = options.command.as_deref() {
        sql.push_str(
            " AND EXISTS (
                SELECT 1
                FROM tool_events te
                WHERE te.event_id = e.id
                  AND te.command LIKE ?
              )",
        );
        params.push(SqlValue::Text(format!("%{command}%")));
    }
    if let Some(ref_filter) = options.ref_filter.as_deref() {
        let ref_match = normalize_ref_filter(ref_filter);
        sql.push_str(
            " AND EXISTS (
                SELECT 1
                FROM event_refs er
                WHERE er.event_id = e.id
                  AND er.ref_kind = ?
                  AND er.ref_value LIKE ? ESCAPE '\\'
              )",
        );
        params.push(SqlValue::Text(ref_match.kind.to_string()));
        params.push(SqlValue::Text(ref_match.value_pattern));
    }
    sql.push_str(
        " ORDER BY bm25(events_fts, 8.0, 6.0, 4.0, 1.0, 0.5), e.captured_at DESC, e.raw_line ASC
          LIMIT ?",
    );
    params.push(SqlValue::Integer(fetch_limit.saturating_add(1) as i64));

    let mut statement = conn.prepare(&sql).map_err(|source| Error::Sqlite {
        path: db_path.clone(),
        source,
    })?;
    let rows = statement
        .query_map(params_from_iter(params), |row| {
            let tool_text: String = row.get(1)?;
            let searchable_text = row.get::<_, String>(7).unwrap_or_default();
            let canonical_type: String = row.get(3)?;
            let summary_kind = crate::summary_kind_for_canonical_str(&canonical_type);
            let raw_file: String = row.get(8)?;
            let raw_line: i64 = row.get(9)?;
            Ok(RankedSearchResult {
                event_id: row.get(0)?,
                tool_invocation_id: row.get(14)?,
                result: SearchResult {
                    tool: Tool::from_str(&tool_text).map_err(|_| rusqlite::Error::InvalidQuery)?,
                    session_id: row.get(2)?,
                    canonical_type,
                    summary_kind,
                    timestamp: row.get(4)?,
                    score: row.get(5)?,
                    snippet: match_centered_snippet(
                        row.get::<_, Option<String>>(6)?,
                        searchable_text.clone(),
                        query_terms,
                        max_snippet_chars,
                    ),
                    native_command: native_jsonl_line_command(&raw_file, raw_line),
                    raw_file,
                    raw_line,
                    raw_offset: row.get(10)?,
                    compaction_state: row.get(11)?,
                    payload: Value::Null,
                    also_at: Vec::new(),
                    corroboration: None,
                    retrieval_key: retrieval_key_for_text(&searchable_text),
                    corroboration_text: searchable_text,
                    cwd: row.get(12)?,
                    project_root: row.get(13)?,
                },
            })
        })
        .map_err(|source| Error::Sqlite {
            path: db_path.clone(),
            source,
        })?;

    let mut results = Vec::new();
    for row in rows {
        results.push(row.map_err(|source| Error::Sqlite {
            path: db_path.clone(),
            source,
        })?);
    }
    let has_more_raw_rows = results.len() > fetch_limit;
    if has_more_raw_rows {
        results.truncate(fetch_limit);
    }
    Ok((results, has_more_raw_rows))
}

fn search_history_hybrid_page(
    home: &Path,
    query: &str,
    options: SearchOptions,
    semantic_available: bool,
) -> Result<SearchPage> {
    let query_terms = effective_search_terms(query, options.expand_concepts)?;
    let limit = options.limit.clamp(1, MAX_SEARCH_LIMIT);
    let offset = options.offset;
    let max_snippet_chars = options.max_snippet_chars.clamp(1, MAX_SEARCH_SNIPPET_CHARS);
    let raw_fetch_limit = search_overfetch_limit(offset, limit);
    let fts_query = quoted_fts_terms(&query_terms);

    let (lexical_results, lexical_has_more_raw_rows) = lexical_search_ranked_results(
        home,
        &options,
        &query_terms,
        &fts_query,
        raw_fetch_limit,
        max_snippet_chars,
    )?;
    let vector_results = vector_search_results(
        home,
        query,
        &options,
        raw_fetch_limit,
        &query_terms,
        max_snippet_chars,
    )?;
    let results = reciprocal_rank_fuse(lexical_results, vector_results);

    finalize_search_page(
        &options,
        SearchPageAssembly {
            results,
            // The fused set inherits the lexical overfetch truncation: if the
            // lexical side dropped rows, the true total is unknown, so
            // total_estimated must be None (matching the lexical path) rather
            // than the fused length.
            has_more_raw_rows: lexical_has_more_raw_rows,
            mode_requested: options.mode,
            mode_applied: SearchMode::Hybrid,
            semantic_available,
            limit,
            offset,
            max_snippet_chars,
        },
    )
}

fn reciprocal_rank_fuse(
    lexical_results: Vec<RankedSearchResult>,
    vector_results: Vec<RankedSearchResult>,
) -> Vec<RankedSearchResult> {
    const RRF_K: f64 = 60.0;
    let lexical_results = unique_ranked_results_by_event(lexical_results);
    let vector_results = unique_ranked_results_by_event(vector_results);
    let mut fused: HashMap<i64, (RankedSearchResult, f64)> = HashMap::new();

    for (rank, result) in lexical_results.into_iter().enumerate() {
        let key = result.event_id;
        let entry = fused.entry(key).or_insert((result, 0.0));
        entry.1 += 1.0 / (RRF_K + rank as f64 + 1.0);
    }
    for (rank, result) in vector_results.into_iter().enumerate() {
        let key = result.event_id;
        let entry = fused.entry(key).or_insert((result, 0.0));
        entry.1 += 1.0 / (RRF_K + rank as f64 + 1.0);
    }

    let mut results = fused
        .into_values()
        .map(|(mut result, score)| {
            result.result.score = score;
            result
        })
        .collect::<Vec<_>>();
    results.sort_by(|left, right| {
        right
            .result
            .score
            .total_cmp(&left.result.score)
            .then_with(|| right.result.timestamp.cmp(&left.result.timestamp))
            .then_with(|| left.result.raw_line.cmp(&right.result.raw_line))
    });
    results
}

pub(crate) fn unique_ranked_results_by_event(
    results: Vec<RankedSearchResult>,
) -> Vec<RankedSearchResult> {
    let mut seen = HashSet::new();
    let mut unique = Vec::new();
    for result in results {
        if seen.insert(result.event_id) {
            unique.push(result);
        }
    }
    unique
}

fn annotate_search_results_with_corroboration(results: &mut [SearchResult]) {
    for result in results {
        result.corroboration = Some(corroborate_text(
            result.cwd.as_deref(),
            result.project_root.as_deref(),
            &result.corroboration_text,
        ));
    }
}

fn search_overfetch_limit(offset: usize, limit: usize) -> usize {
    let requested_window = offset.saturating_add(limit);
    let extra = requested_window.min(500).max(limit);
    requested_window.saturating_add(extra)
}

fn bounded_snippet(snippet: String, max_chars: usize) -> String {
    truncate_chars(snippet.trim().to_string(), max_chars)
}

pub(crate) fn match_centered_snippet(
    sqlite_snippet: Option<String>,
    searchable_text: String,
    query_terms: &[String],
    max_chars: usize,
) -> String {
    if let Some(snippet) = sqlite_snippet.filter(|snippet| !snippet.trim().is_empty()) {
        return bounded_snippet(snippet, max_chars);
    }
    if searchable_text.chars().count() <= max_chars {
        return searchable_text.trim().to_string();
    }
    let lower_text = searchable_text.to_lowercase();
    let first_match = query_terms
        .iter()
        .filter_map(|term| lower_text.find(&term.to_lowercase()))
        .min()
        .unwrap_or(0);
    let half_window = max_chars.saturating_div(2);
    let mut start = first_match.saturating_sub(half_window);
    while start > 0 && !searchable_text.is_char_boundary(start) {
        start -= 1;
    }
    let mut end = start.saturating_add(max_chars).min(searchable_text.len());
    while end > start && !searchable_text.is_char_boundary(end) {
        end -= 1;
    }
    searchable_text[start..end].trim().to_string()
}

fn truncate_chars(mut value: String, max_chars: usize) -> String {
    if value.chars().count() <= max_chars {
        return value;
    }
    let mut cutoff = 0usize;
    for (count, (index, character)) in value.char_indices().enumerate() {
        if count == max_chars {
            break;
        }
        cutoff = index + character.len_utf8();
    }
    value.truncate(cutoff);
    value
}

fn dedupe_ranked_search_results(
    results: Vec<RankedSearchResult>,
) -> Result<Vec<RankedSearchResult>> {
    let mut seen: HashMap<RetrievalTwinKey, usize> = HashMap::new();
    let mut deduped: Vec<RankedSearchResult> = Vec::new();
    for result in results {
        let key = retrieval_twin_key(&result);
        if let Some(existing) = seen.get(&key).copied() {
            deduped[existing]
                .result
                .also_at
                .push(result.result.raw_line);
        } else {
            seen.insert(key, deduped.len());
            deduped.push(result);
        }
    }
    Ok(deduped)
}

// Retrieval-layer dedupe identity for a hit's text. Hashing the rendered
// `searchable_text` verbatim made the key sensitive to whitespace that carries
// no semantic content: two captures of the same answer that differ only by an
// embedded newline vs. a space (e.g. a native `output_text` block vs. its
// `agent_message` twin) hashed to different keys and survived dedupe as adjacent
// duplicates. Normalizing collapses any run of Unicode whitespace to a single
// space and trims the ends, so whitespace-only divergence yields one key while
// genuinely distinct text stays distinct. The normalization is applied to the
// dedupe key only; `searchable_text`, snippets, and `corroboration_text` keep
// their original bytes.
pub(crate) fn retrieval_key_for_text(searchable_text: &str) -> String {
    let normalized = searchable_text
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ");
    sha256_hex(normalized.as_bytes())
}

/// Retrieval-layer dedupe identity for a ranked hit. `canonical_type` is part
/// of the text key, so a `tool.call` and its `tool.result` can never collapse
/// by content — they collapse only through the invocation key, which requires
/// the harness-assigned invocation id both events share. The highest-ranked
/// event of an invocation keeps the slot; the others become `also_at` lines.
/// Rows indexed before `tool_invocation_id` existed have NULL there and keep
/// the text-key behavior.
#[derive(Debug, PartialEq, Eq, Hash)]
enum RetrievalTwinKey {
    ToolInvocation {
        session_id: String,
        tool_invocation_id: String,
    },
    Text {
        session_id: String,
        canonical_type: String,
        retrieval_key: String,
    },
}

fn retrieval_twin_key(ranked: &RankedSearchResult) -> RetrievalTwinKey {
    let result = &ranked.result;
    let is_tool_event = result.canonical_type == CanonicalType::ToolCall.as_str()
        || result.canonical_type == CanonicalType::ToolResult.as_str();
    match ranked.tool_invocation_id.as_deref() {
        Some(tool_invocation_id) if is_tool_event && !tool_invocation_id.is_empty() => {
            RetrievalTwinKey::ToolInvocation {
                session_id: result.session_id.clone(),
                tool_invocation_id: tool_invocation_id.to_string(),
            }
        }
        _ => RetrievalTwinKey::Text {
            session_id: result.session_id.clone(),
            canonical_type: result.canonical_type.clone(),
            retrieval_key: result.retrieval_key.clone(),
        },
    }
}

// Resolve the literal query into the term list used for FTS matching and
// snippet centering. With `expand_concepts` set, the literal terms are
// OR-extended with curated synonyms (originals kept up front); off, this is
// exactly `searchable_terms`.
fn effective_search_terms(query: &str, expand_concepts: bool) -> Result<Vec<String>> {
    let terms = searchable_terms(query)?;
    if expand_concepts {
        Ok(expand_query_terms(&terms))
    } else {
        Ok(terms)
    }
}

fn searchable_terms(query: &str) -> Result<Vec<String>> {
    let mut terms = Vec::new();
    let mut current = String::new();

    for character in query.chars() {
        if character.is_alphanumeric() || character == '_' {
            current.push(character);
        } else if !current.is_empty() {
            terms.push(std::mem::take(&mut current));
        }
    }
    if !current.is_empty() {
        terms.push(current);
    }

    if terms.is_empty() {
        return Err(Error::Validation(
            "query must contain searchable text".to_string(),
        ));
    }

    Ok(terms)
}

// Join terms with OR, not AND. AND made specificity collapse recall: a
// longer, more-specific query required every term to occur in one event, so
// adding terms could only ever shrink the match set to zero. With OR the FTS
// match set is the union of per-term hits and bm25 (weighted by term rarity
// and the column weights in lexical_search_ranked_results) floats events that
// satisfy more of the query to the top — specificity now improves ranking
// instead of erasing recall.
fn quoted_fts_terms(terms: &[String]) -> String {
    terms
        .iter()
        .map(|term| format!("\"{}\"", term.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" OR ")
}

fn hydrate_search_result_payloads(results: &mut [SearchResult]) -> Result<()> {
    let mut grouped = BTreeMap::<String, Vec<usize>>::new();
    for (index, result) in results.iter().enumerate() {
        grouped
            .entry(result.raw_file.clone())
            .or_default()
            .push(index);
    }

    for (raw_file, mut indexes) in grouped {
        indexes.sort_by_key(|index| {
            (
                results[*index].raw_offset.unwrap_or(i64::MAX),
                results[*index].raw_line,
            )
        });
        let raw_path = PathBuf::from(&raw_file);
        let mut offset_reader = None;
        for index in indexes {
            let raw_line = results[index].raw_line;
            let raw_offset = results[index].raw_offset;
            let envelope = if let Some(raw_offset) = raw_offset {
                if offset_reader.is_none() {
                    offset_reader = Some(open_raw_offset_reader(&raw_path)?);
                }
                match read_raw_envelope_at_offset(
                    &raw_path,
                    offset_reader.as_mut().expect("offset reader initialized"),
                    raw_offset,
                )? {
                    Some(envelope) => envelope,
                    None => raw_envelope_for_line_scan(&raw_path, raw_line)?,
                }
            } else {
                raw_envelope_for_line_scan(&raw_path, raw_line)?
            };
            results[index].payload = resolved_payload_for_envelope(&raw_path, &envelope)?;
        }
    }
    Ok(())
}

// Resolve a caller-supplied session identifier to the canonical stored
// session_id(s) so the filter fails open instead of closed. An exact
// `session_id = ?` filter returned empty whenever a session was referenced by
// anything other than its full id — most commonly a short id prefix (the
// natural handle, like a short git SHA) or the filename-sanitized form. Those
// are present sessions, so empty was a confidently-wrong answer.
//
// Resolution is tiered, most-specific first: exact session_id, then
// filename_session_id, then id prefix — every session whose id starts with the
// input, so an ambiguous prefix widens the filter rather than emptying it
// (fail-open, never fail-closed). The first non-empty tier wins.
// `tool` is optional — when absent, resolution spans every tool, so the filter
// never silently requires a tool arg. When nothing resolves the literal input
// is returned unchanged: a genuinely-absent session then yields an empty
// result, which is correct rather than false-empty.
pub(crate) fn resolve_session_filter_ids(
    conn: &Connection,
    db_path: &Path,
    tool: Option<Tool>,
    session_id: &str,
) -> Result<Vec<String>> {
    let tiers: [(&str, String); 3] = [
        (
            "SELECT session_id FROM sessions WHERE session_id = ?1",
            session_id.to_string(),
        ),
        (
            "SELECT session_id FROM sessions WHERE filename_session_id = ?1",
            session_id.to_string(),
        ),
        (
            "SELECT session_id FROM sessions WHERE session_id LIKE ?1 ESCAPE '\\'",
            format!("{}%", escape_like(session_id)),
        ),
    ];

    for (base_sql, needle) in tiers {
        let mut sql = base_sql.to_string();
        let mut params = vec![SqlValue::Text(needle)];
        if let Some(tool) = tool {
            sql.push_str(" AND tool = ?2");
            params.push(SqlValue::Text(tool.as_str().to_string()));
        }
        let mut statement = conn.prepare(&sql).map_err(|source| Error::Sqlite {
            path: db_path.to_path_buf(),
            source,
        })?;
        let mut ids = Vec::new();
        let mut seen = HashSet::new();
        let rows = statement
            .query_map(params_from_iter(params), |row| row.get::<_, String>(0))
            .map_err(|source| Error::Sqlite {
                path: db_path.to_path_buf(),
                source,
            })?;
        for row in rows {
            let id = row.map_err(|source| Error::Sqlite {
                path: db_path.to_path_buf(),
                source,
            })?;
            if seen.insert(id.clone()) {
                ids.push(id);
            }
        }
        if !ids.is_empty() {
            return Ok(ids);
        }
    }

    Ok(vec![session_id.to_string()])
}

// Escape SQL LIKE metacharacters so a prefix match treats `%` and `_` in a
// session id literally (paired with `ESCAPE '\'` on the query).
fn escape_like(value: &str) -> String {
    let mut escaped = String::with_capacity(value.len());
    for character in value.chars() {
        match character {
            '\\' | '%' | '_' => {
                escaped.push('\\');
                escaped.push(character);
            }
            _ => escaped.push(character),
        }
    }
    escaped
}

#[cfg(test)]
mod page_assembly_tests {
    //! Exercises the finalization the lexical and hybrid paths share. Before the
    //! extraction, the hybrid page carried its own copy that skipped
    //! corroboration and set `total_estimated` to the fused length even after
    //! overfetch truncation; these tests pin the shared function so the two paths
    //! cannot drift apart again.
    use super::*;

    fn ranked(event_id: i64, raw_line: i64) -> RankedSearchResult {
        RankedSearchResult {
            event_id,
            tool_invocation_id: None,
            result: SearchResult {
                tool: Tool::Claude,
                session_id: "session".to_string(),
                canonical_type: "assistant.message".to_string(),
                summary_kind: None,
                timestamp: "2026-01-01T00:00:00Z".to_string(),
                score: 1.0,
                snippet: "snippet".to_string(),
                native_command: None,
                raw_file: "/tmp/nabu-page-assembly.jsonl".to_string(),
                raw_line,
                raw_offset: None,
                compaction_state: "none".to_string(),
                payload: Value::Null,
                also_at: Vec::new(),
                corroboration: None,
                retrieval_key: format!("retrieval-key-{event_id}"),
                // A PR ref resolves without touching the filesystem or git.
                corroboration_text: "resolved in PR #42".to_string(),
                cwd: None,
                project_root: None,
            },
        }
    }

    fn ranked_tool_event(
        event_id: i64,
        raw_line: i64,
        canonical_type: &str,
        tool_invocation_id: Option<&str>,
    ) -> RankedSearchResult {
        let mut ranked = ranked(event_id, raw_line);
        ranked.tool_invocation_id = tool_invocation_id.map(str::to_string);
        ranked.result.canonical_type = canonical_type.to_string();
        ranked
    }

    #[test]
    fn dedupe_collapses_call_and_results_of_one_invocation() {
        let results = vec![
            ranked_tool_event(1, 1, "tool.call", Some("call-7")),
            ranked_tool_event(2, 2, "tool.result", Some("call-7")),
            ranked_tool_event(3, 3, "tool.result", Some("call-7")),
        ];
        let deduped = dedupe_ranked_search_results(results).unwrap();
        assert_eq!(
            deduped.len(),
            1,
            "call and results sharing an invocation id must occupy one slot"
        );
        assert_eq!(
            deduped[0].result.raw_line, 1,
            "the highest-ranked event of the invocation keeps the slot"
        );
        assert_eq!(
            deduped[0].result.also_at,
            vec![2, 3],
            "collapsed twins must stay cited via also_at"
        );
    }

    #[test]
    fn dedupe_keeps_distinct_invocations_and_id_less_tool_events_separate() {
        let results = vec![
            ranked_tool_event(1, 1, "tool.call", Some("call-7")),
            ranked_tool_event(2, 2, "tool.call", Some("call-8")),
            ranked_tool_event(3, 3, "tool.result", None),
            ranked_tool_event(4, 4, "tool.result", None),
        ];
        let deduped = dedupe_ranked_search_results(results).unwrap();
        assert_eq!(
            deduped.len(),
            4,
            "distinct invocation ids and id-less tool events with distinct text must not merge"
        );
    }

    fn ranked_in_session(event_id: i64, raw_line: i64, session_id: &str) -> RankedSearchResult {
        let mut ranked = ranked(event_id, raw_line);
        ranked.result.session_id = session_id.to_string();
        ranked
    }

    fn lexical_assembly(results: Vec<RankedSearchResult>) -> SearchPageAssembly {
        SearchPageAssembly {
            results,
            has_more_raw_rows: false,
            mode_requested: SearchMode::Auto,
            mode_applied: SearchMode::Lexical,
            semantic_available: false,
            limit: 10,
            offset: 0,
            max_snippet_chars: 200,
        }
    }

    #[test]
    fn advisory_fires_on_empty_lexical_only_page() {
        let page =
            finalize_search_page(&SearchOptions::default(), lexical_assembly(Vec::new())).unwrap();
        let advisory = page
            .advisory
            .expect("empty lexical-only page must carry the advisory");
        assert!(advisory.contains("expand_concepts=true"), "{advisory}");
    }

    #[test]
    fn advisory_fires_when_top_hits_scatter_across_sessions() {
        let results = (0..6)
            .map(|index| ranked_in_session(index, index, &format!("session-{index}")))
            .collect();
        let page =
            finalize_search_page(&SearchOptions::default(), lexical_assembly(results)).unwrap();
        assert!(
            page.advisory.is_some(),
            "six hits in six sessions is a scattered page"
        );
    }

    fn scattered_results() -> Vec<RankedSearchResult> {
        (0..6)
            .map(|index| ranked_in_session(index, index, &format!("session-{index}")))
            .collect()
    }

    #[test]
    fn advisory_stays_absent_for_concentrated_expanded_or_hybrid_pages() {
        let concentrated = (0..6)
            .map(|index| {
                ranked_in_session(
                    index,
                    index,
                    if index < 5 { "session-a" } else { "session-b" },
                )
            })
            .collect();
        let page = finalize_search_page(&SearchOptions::default(), lexical_assembly(concentrated))
            .unwrap();
        assert_eq!(
            page.advisory, None,
            "session-concentrated hits are not weak"
        );

        let expanded = finalize_search_page(
            &SearchOptions {
                expand_concepts: true,
                ..SearchOptions::default()
            },
            lexical_assembly(scattered_results()),
        )
        .unwrap();
        assert_eq!(
            expanded.advisory, None,
            "expanded queries never carry the advisory"
        );

        let hybrid = finalize_search_page(
            &SearchOptions::default(),
            hybrid_assembly(scattered_results(), false),
        )
        .unwrap();
        assert_eq!(
            hybrid.advisory, None,
            "hybrid pages never carry the advisory"
        );
    }

    // A hybrid-shaped assembly: `mode_applied` is Hybrid, matching what
    // `search_history_hybrid_page` builds.
    fn hybrid_assembly(
        results: Vec<RankedSearchResult>,
        has_more_raw_rows: bool,
    ) -> SearchPageAssembly {
        SearchPageAssembly {
            results,
            has_more_raw_rows,
            mode_requested: SearchMode::Auto,
            mode_applied: SearchMode::Hybrid,
            semantic_available: true,
            limit: 10,
            offset: 0,
            max_snippet_chars: 200,
        }
    }

    #[test]
    fn hybrid_page_honors_corroborate() {
        let options = SearchOptions {
            corroborate: true,
            dedupe: false,
            ..SearchOptions::default()
        };
        let page = finalize_search_page(
            &options,
            hybrid_assembly(vec![ranked(1, 1), ranked(2, 2)], false),
        )
        .unwrap();
        assert_eq!(page.mode_applied, SearchMode::Hybrid);
        assert_eq!(page.results.len(), 2);
        for result in &page.results {
            assert!(
                result.corroboration.is_some(),
                "corroborate=true must annotate every hybrid result"
            );
        }
    }

    #[test]
    fn page_without_corroborate_leaves_it_unset() {
        let options = SearchOptions {
            corroborate: false,
            dedupe: false,
            ..SearchOptions::default()
        };
        let page =
            finalize_search_page(&options, hybrid_assembly(vec![ranked(1, 1)], false)).unwrap();
        assert!(page.results[0].corroboration.is_none());
    }

    #[test]
    fn hybrid_total_estimated_is_none_when_overfetch_truncated() {
        let options = SearchOptions {
            dedupe: false,
            ..SearchOptions::default()
        };
        let truncated = finalize_search_page(
            &options,
            hybrid_assembly(vec![ranked(1, 1), ranked(2, 2)], true),
        )
        .unwrap();
        assert_eq!(
            truncated.total_estimated, None,
            "a truncated overfetch means the true total is unknown"
        );
        assert!(
            truncated.continuation.is_some(),
            "truncation must advance the pagination cursor"
        );

        let complete = finalize_search_page(
            &options,
            hybrid_assembly(vec![ranked(1, 1), ranked(2, 2)], false),
        )
        .unwrap();
        assert_eq!(complete.total_estimated, Some(2));
    }
}