# session_search_db.py — SQLCipher-encrypted FTS5 search index over chat history.
# Derived, disposable index that can be rebuilt from chat_storage.py at any time.
# Self-healing: automatically rebuilds on corruption or deletion.
from __future__ import annotations

import logging
import os
import re
import sqlite3
import tempfile
import struct
import threading
import time
from datetime import UTC, datetime
from hashlib import sha256
from math import sqrt
from typing import Any

from utils.fts5_query import safe_fts_query as _safe_fts_query

_log = logging.getLogger(__name__)

try:
    from persistence import MUGI_DIR
except ImportError:
    MUGI_DIR = os.path.expanduser("~/.mugi")

SEARCH_DB_PATH = os.path.join(MUGI_DIR, "search.db")

# Thread IDs used for internal / background agents; optional to hide from session search.
_REINDEX_STAGING_SUFFIX = ":reindex-staging"
_BACKGROUND_THREAD_ID_PREFIXES = (
    "heartbeat_",
    "background_heartbeat",
    "drive_discovery_",
    "skill_eval_",
)

_conn_lock = threading.RLock()
_rebuilding = threading.Event()
# Serializes the intra-process check-and-set of ``_rebuilding`` so two threads in the same
# process cannot both pass the guard and run concurrent rebuilds (the cross-process fcntl
# lock does not conflict within a single process, so it cannot provide this guarantee).
_rebuild_start_lock = threading.Lock()
_search_index_plain_sqlite: bool | None = None
_vec_needs_rebuild = False
_vec_backfill_in_progress = False

_KEYRING_SERVICE = "Mugi"
_KEYRING_ACCOUNT = "chat_encryption_key"

_DEFAULT_VECTOR_DIM = 256
_DEFAULT_SEMANTIC_MAX_HITS = 25
_DEFAULT_SEMANTIC_CANDIDATE_LIMIT = 2000
_DEFAULT_SEMANTIC_QUERY_MIN_CHARS = 3
_DEFAULT_EMBED_BATCH_MAX = 64
_DEFAULT_EMBED_MAX_CHUNKS_PER_CALL = 256
_DEFAULT_RRF_K = 60


def _get_key() -> str:
    """Retrieve the encryption key from macOS Keychain (same key as chat_storage)."""
    try:
        import keyring
        key = keyring.get_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT)
        if key:
            return key
    except Exception:
        pass
    raise RuntimeError("Cannot retrieve encryption key from Keychain for search DB.")


def _open_search_index(
    db_path: str,
    *,
    schedule_vec_backfill: bool = False,
) -> sqlite3.Connection:
    """Open (or create) a search index database at *db_path*.

    Falls back to plain SQLite when SQLCipher is unavailable or fails to open.
    """
    ddir = os.path.dirname(db_path)
    if ddir:
        os.makedirs(ddir, exist_ok=True, mode=0o700)
    global _search_index_plain_sqlite
    try:
        import sqlcipher3
        conn = sqlcipher3.connect(db_path)
        key = _get_key()
        esc = (key or "").replace("'", "''")
        conn.execute(f"PRAGMA key = '{esc}'")
        _search_index_plain_sqlite = False
    except ImportError:
        _log.debug("sqlcipher3 not available; using plain SQLite for search index.")
        conn = sqlite3.connect(db_path)
        _search_index_plain_sqlite = True
    except Exception as e:
        _log.warning("SQLCipher connection failed (%s); falling back to plain SQLite.", e)
        conn = sqlite3.connect(db_path)
        _search_index_plain_sqlite = True

    try:
        conn.execute(
            """CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
                thread_id UNINDEXED,
                role UNINDEXED,
                content,
                created_at UNINDEXED
            )"""
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS vector_chunks (
                thread_id TEXT NOT NULL,
                message_row_id INTEGER NOT NULL,
                chunk_index INTEGER NOT NULL,
                role TEXT NOT NULL,
                created_at TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                text_preview TEXT NOT NULL,
                embedding BLOB NOT NULL,
                PRIMARY KEY (thread_id, message_row_id, chunk_index)
            )
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS search_meta (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            )
            """
        )
        conn.execute("CREATE INDEX IF NOT EXISTS idx_vector_chunks_thread_id ON vector_chunks(thread_id)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_vector_chunks_content_hash ON vector_chunks(content_hash)")
        _wire_vec_index_on_open(conn, schedule_backfill=schedule_vec_backfill)
        conn.commit()
    except Exception as e:
        conn.close()
        raise RuntimeError(f"Failed to initialize FTS5 table: {e}") from e
    return conn


def _get_connection(*, schedule_vec_backfill: bool = False) -> sqlite3.Connection:
    """Open (or create) the SQLCipher search database at the canonical path."""
    return _open_search_index(SEARCH_DB_PATH, schedule_vec_backfill=schedule_vec_backfill)


def _is_sqlite_locked_error(exc: BaseException) -> bool:
    msg = str(exc).lower()
    return "database is locked" in msg or (
        isinstance(exc, sqlite3.OperationalError) and "locked" in msg
    )


def _safe_get_connection() -> sqlite3.Connection | None:
    """Get a connection; rebuild only on real open/init failures (not transient locks)."""
    last_exc: BaseException | None = None
    for attempt in range(5):
        try:
            return _get_connection(schedule_vec_backfill=False)
        except Exception as e:
            last_exc = e
            if _is_sqlite_locked_error(e) and attempt < 4:
                time.sleep(0.05 * (2**attempt))
                continue
            break
    if last_exc is None:
        return None
    if _is_sqlite_locked_error(last_exc):
        _log.warning(
            "Search DB temporarily locked (%s); skipping work this call (no rebuild).",
            last_exc,
        )
        return None
    _log.warning("Search DB connection failed (%s); triggering rebuild.", last_exc)
    _trigger_rebuild_async()
    return None


def _session_search_config() -> dict[str, Any]:
    try:
        from core.config import get_session_search_config

        return get_session_search_config() or {}
    except Exception:
        return {}


def _semantic_enabled(explicit: bool | None = None) -> bool:
    if explicit is not None:
        return bool(explicit)
    return bool(_session_search_config().get("semantic_search_enabled", False))


def _semantic_hybrid_enabled(explicit: bool | None = None) -> bool:
    if explicit is not None:
        return bool(explicit)
    return bool(_session_search_config().get("hybrid_search_enabled", True))


def _vector_dim() -> int:
    try:
        from utils.embedding_model import embedding_dim, is_real

        if is_real():
            return embedding_dim()
    except Exception:
        pass
    raw = _session_search_config().get("semantic_embedding_dimensions", _DEFAULT_VECTOR_DIM)
    try:
        return max(32, min(int(raw), 2048))
    except Exception:
        return _DEFAULT_VECTOR_DIM


def _semantic_candidate_limit() -> int:
    raw = _session_search_config().get("semantic_candidate_limit", _DEFAULT_SEMANTIC_CANDIDATE_LIMIT)
    try:
        return max(50, min(int(raw), 50000))
    except Exception:
        return _DEFAULT_SEMANTIC_CANDIDATE_LIMIT


def _semantic_max_hits() -> int:
    raw = _session_search_config().get("semantic_max_hits", _DEFAULT_SEMANTIC_MAX_HITS)
    try:
        return max(1, min(int(raw), 500))
    except Exception:
        return _DEFAULT_SEMANTIC_MAX_HITS


def _semantic_query_min_chars() -> int:
    raw = _session_search_config().get("semantic_query_min_chars", _DEFAULT_SEMANTIC_QUERY_MIN_CHARS)
    try:
        return max(1, min(int(raw), 200))
    except Exception:
        return _DEFAULT_SEMANTIC_QUERY_MIN_CHARS


def _embed_batch_max() -> int:
    raw = _session_search_config().get("semantic_embed_batch_max", _DEFAULT_EMBED_BATCH_MAX)
    try:
        return max(1, min(int(raw), 256))
    except Exception:
        return _DEFAULT_EMBED_BATCH_MAX


def _rrf_k() -> int:
    raw = _session_search_config().get("semantic_rrf_k", _DEFAULT_RRF_K)
    try:
        return max(1, min(int(raw), 500))
    except Exception:
        return _DEFAULT_RRF_K


def _embed_max_chunks_per_call() -> int:
    raw = _session_search_config().get("semantic_embed_max_chunks_per_call", _DEFAULT_EMBED_MAX_CHUNKS_PER_CALL)
    try:
        return max(1, min(int(raw), 5000))
    except Exception:
        return _DEFAULT_EMBED_MAX_CHUNKS_PER_CALL


def _semantic_ann_enabled() -> bool:
    return bool(_session_search_config().get("semantic_ann_enabled", True))


def _semantic_ann_k() -> int:
    raw = _session_search_config().get("semantic_ann_k")
    if raw is None:
        return min(200, _semantic_candidate_limit())
    try:
        return max(10, min(int(raw), _semantic_candidate_limit()))
    except Exception:
        return min(200, _semantic_candidate_limit())


def _semantic_ann_overfetch() -> int:
    raw = _session_search_config().get("semantic_ann_overfetch", 3)
    try:
        return max(1, min(int(raw), 10))
    except Exception:
        return 3


def _vec_available(conn: sqlite3.Connection) -> bool:
    try:
        from utils import sqlite_vec_index as svi

        return svi.vec_available(conn)
    except Exception:
        return False


def get_vec_backend_status() -> str:
    """Read-only label for operator status (vec0 vs bruteforce)."""
    if not _semantic_ann_enabled() or _vec_needs_rebuild:
        return "bruteforce"
    try:
        from utils import sqlite_vec_index as svi

        if not svi.vec_available():
            return "bruteforce"
        with _conn_lock:
            try:
                conn = _open_search_index(SEARCH_DB_PATH, schedule_vec_backfill=False)
            except Exception:
                return "bruteforce"
        try:
            if not svi.vec_available(conn):
                return "bruteforce"
            if _vec_needs_rebuild:
                return "bruteforce"
            dim = _vector_dim()
            if not svi.ensure_vec_table(conn, dim):
                return "bruteforce"
            if (
                svi.embedding_blob_mismatch_count(conn, dim) > 0
                or svi.unmirrored_row_count(conn, expected_dim=dim) > 0
            ):
                return "bruteforce"
            return svi.vec_backend_label()
        finally:
            conn.close()
    except Exception:
        return "bruteforce"


def _wire_vec_index_on_open(
    conn: sqlite3.Connection,
    *,
    schedule_backfill: bool = False,
) -> None:
    """Load sqlite-vec, ensure vec_chunks, validate meta; schedule backfill when needed."""
    global _vec_needs_rebuild
    try:
        from utils import sqlite_vec_index as svi
    except Exception:
        return
    svi.reset_table_cache()
    if not svi.vec_available(conn):
        return
    dim = _vector_dim()
    if not svi.ensure_vec_table(conn, dim):
        return
    backend = svi.probe_backend_string()
    stored_dim = _get_meta(conn, "vec_dim")
    stored_backend = _get_meta(conn, "vec_backend")
    if stored_dim:
        try:
            if int(stored_dim) != dim:
                _vec_needs_rebuild = True
        except ValueError:
            _vec_needs_rebuild = True
    if stored_backend and backend and stored_backend != backend:
        _vec_needs_rebuild = True
    if not _vec_needs_rebuild and svi.embedding_blob_mismatch_count(conn, dim) > 0:
        _vec_needs_rebuild = True
    if not _vec_needs_rebuild:
        _set_meta(conn, "vec_dim", str(dim))
        if backend:
            _set_meta(conn, "vec_backend", backend)
        try:
            purged = svi.purge_orphan_vec_mirrors(conn)
            if purged:
                _log.info("[search] vec_chunks purged %d orphan mirror rows", purged)
        except Exception:
            pass
        if schedule_backfill:
            _maybe_schedule_vec_backfill()


def _vec_mirror_delete_rowids(conn: sqlite3.Connection, rowids: list[int]) -> None:
    """Delete vec mirror rows (runs even when write mirroring is disabled)."""
    if not rowids:
        return
    try:
        from utils import sqlite_vec_index as svi

        if svi.vec_available(conn):
            svi.mirror_delete_rowids(conn, rowids)
    except Exception as exc:
        _log.warning("[search] vec mirror_delete_rowids failed: %s", str(exc)[:120])


def _vec_mirror_delete_thread(conn: sqlite3.Connection, thread_id: str) -> None:
    """Delete all vec mirror rows for a thread (runs even when write mirroring is disabled)."""
    try:
        from utils import sqlite_vec_index as svi

        if svi.vec_available(conn):
            svi.mirror_delete_thread(conn, thread_id)
    except Exception as exc:
        _log.warning(
            "[search] vec mirror_delete_thread failed for %s: %s",
            thread_id,
            str(exc)[:120],
        )


def _maybe_schedule_vec_backfill() -> None:
    global _vec_backfill_in_progress
    if _vec_needs_rebuild or _vec_backfill_in_progress:
        return
    _vec_backfill_in_progress = True

    def _runner() -> None:
        global _vec_backfill_in_progress
        try:
            with _conn_lock:
                if _rebuilding.is_set() or _vec_needs_rebuild:
                    return
                conn = _safe_get_connection()
                if conn is None:
                    return
                try:
                    from utils import sqlite_vec_index as svi

                    svi.reset_table_cache()
                    if not svi.vec_available(conn) or not svi.ensure_vec_table(conn, _vector_dim()):
                        return
                    vc = conn.execute("SELECT COUNT(*) FROM vector_chunks").fetchone()
                    vc_n = int(vc[0] if vc else 0)
                    if vc_n == 0:
                        return
                    dim = _vector_dim()
                    if (
                        svi.embedding_blob_mismatch_count(conn, dim) > 0
                        or svi.unmirrored_row_count(conn, expected_dim=dim) == 0
                    ):
                        return
                    added = svi.backfill(conn, expected_dim=dim)
                    conn.commit()
                    if added:
                        _log.info("[search] vec_chunks backfill mirrored %d rows", added)
                finally:
                    conn.close()
        except Exception as exc:
            _log.warning("[search] vec backfill failed: %s", exc)
        finally:
            _vec_backfill_in_progress = False

    threading.Thread(target=_runner, daemon=True, name="search-vec-backfill").start()


def _vec_scope_fully_mirrored(conn: sqlite3.Connection, thread_id: str) -> bool:
    tid = (thread_id or "").strip()
    if not tid:
        return True
    try:
        from utils import sqlite_vec_index as svi

        vc = conn.execute(
            "SELECT COUNT(*) FROM vector_chunks WHERE thread_id = ?",
            (tid,),
        ).fetchone()
        if int(vc[0] if vc else 0) == 0:
            return True
        dim = _vector_dim()
        if svi.embedding_blob_mismatch_count(conn, dim, tid) > 0:
            return False
        return svi.unmirrored_row_count(conn, tid, expected_dim=dim) == 0
    except Exception:
        return False


def _vec_ann_active(conn: sqlite3.Connection, *, scope_thread_id: str = "") -> bool:
    if not _semantic_ann_enabled() or _vec_needs_rebuild:
        return False
    if not _vec_available(conn):
        return False
    try:
        from utils import sqlite_vec_index as svi

        dim = _vector_dim()
        if not svi.ensure_vec_table(conn, dim):
            return False
        if svi.embedding_blob_mismatch_count(conn, dim) > 0:
            return False
        scope = (scope_thread_id or "").strip()
        if scope:
            if not _vec_scope_fully_mirrored(conn, scope):
                _maybe_schedule_vec_backfill()
                return False
            return True
        if svi.unmirrored_row_count(conn, expected_dim=dim) > 0:
            _maybe_schedule_vec_backfill()
            return False
        return True
    except Exception:
        return False


def _vector_chunk_rowid(
    conn: sqlite3.Connection,
    thread_id: str,
    message_row_id: int,
    chunk_index: int,
) -> int | None:
    row = conn.execute(
        "SELECT rowid FROM vector_chunks WHERE thread_id = ? AND message_row_id = ? AND chunk_index = ?",
        (thread_id, message_row_id, chunk_index),
    ).fetchone()
    if not row:
        return None
    try:
        return int(row[0])
    except (TypeError, ValueError):
        return None


def _rowids_for_chunk_range(
    conn: sqlite3.Connection,
    thread_id: str,
    message_row_id: int,
    from_chunk_index: int,
) -> list[int]:
    return [
        int(r[0])
        for r in conn.execute(
            "SELECT rowid FROM vector_chunks WHERE thread_id = ? AND message_row_id = ? AND chunk_index >= ?",
            (thread_id, message_row_id, from_chunk_index),
        ).fetchall()
    ]


def _mirror_vector_chunk_writes(conn: sqlite3.Connection) -> bool:
    if _vec_needs_rebuild:
        return False
    try:
        from utils import sqlite_vec_index as svi

        return svi.vec_available(conn) and svi.ensure_vec_table(conn, _vector_dim())
    except Exception:
        return False


def _purge_vector_chunks_range(
    conn: sqlite3.Connection,
    thread_id: str,
    message_row_id: int,
    from_chunk_index: int,
) -> None:
    """Delete trailing vector_chunks and vec mirror rows for one message."""
    doomed = _rowids_for_chunk_range(conn, thread_id, message_row_id, from_chunk_index)
    conn.execute(
        "DELETE FROM vector_chunks WHERE thread_id = ? AND message_row_id = ? AND chunk_index >= ?",
        (thread_id, message_row_id, from_chunk_index),
    )
    _vec_mirror_delete_rowids(conn, doomed)


def _purge_vector_chunks_thread(conn: sqlite3.Connection, thread_id: str) -> None:
    """Delete all vector_chunks for a thread and the vec mirror rows."""
    _vec_mirror_delete_thread(conn, thread_id)
    conn.execute("DELETE FROM vector_chunks WHERE thread_id = ?", (thread_id,))


def _chunk_text(text: str, *, size: int = 700, overlap: int = 120) -> list[str]:
    """Chat-session chunking (legacy 700/120); knowledge corpora use embedding_chunking + pre_chunked."""
    t = (text or "").strip()
    if not t:
        return []
    if len(t) <= size:
        return [t]
    chunks: list[str] = []
    start = 0
    step = max(1, size - overlap)
    while start < len(t):
        chunk = t[start : start + size].strip()
        if chunk:
            chunks.append(chunk)
        start += step
    return chunks


def _content_to_text(content: Any) -> str:
    if content is None:
        return ""
    if isinstance(content, list):
        text_parts: list[str] = []
        for part in content:
            if isinstance(part, dict) and part.get("type") == "text":
                text_parts.append(str(part.get("text", "")))
            elif isinstance(part, str):
                text_parts.append(part)
        return " ".join(text_parts).strip()
    return str(content).strip()


def _encode_embedding(vec: list[float]) -> bytes:
    return struct.pack("<" + "f" * len(vec), *vec)


def _decode_embedding(blob: bytes, *, dimensions: int) -> list[float]:
    if not blob:
        return []
    expected = 4 * dimensions
    if len(blob) != expected:
        return []
    return list(struct.unpack("<" + "f" * dimensions, blob))


def _cosine_similarity(a: list[float], b: list[float]) -> float:
    if not a or not b:
        return 0.0
    n = min(len(a), len(b))
    dot = 0.0
    na = 0.0
    nb = 0.0
    for i in range(n):
        av = float(a[i])
        bv = float(b[i])
        dot += av * bv
        na += av * av
        nb += bv * bv
    if na <= 0.0 or nb <= 0.0:
        return 0.0
    return dot / (sqrt(na) * sqrt(nb))


def _utility_embed_texts(
    texts: list[str],
    *,
    dimensions: int,
    kind: str = "document",
) -> list[list[float]] | None:
    try:
        from utility_model import utility_embed_texts

        return utility_embed_texts(
            texts,
            dimensions=dimensions,
            kind=kind,
            activity_operation="session_search_embedding",
        )
    except Exception as e:
        _log.debug("Utility embedding unavailable: %s", e)
        return None


def _set_meta(conn: sqlite3.Connection, key: str, value: str) -> None:
    conn.execute(
        "INSERT INTO search_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        (key, value),
    )


def _embed_identity_meta_key(thread_id: str) -> str:
    return f"embed_identity:{(thread_id or '').strip()}"


def _chunker_version_meta_key(thread_id: str) -> str:
    return f"thread_chunker_version:{(thread_id or '').strip()}"


def _get_meta(conn: sqlite3.Connection, key: str) -> str:
    row = conn.execute("SELECT value FROM search_meta WHERE key = ?", (key,)).fetchone()
    return str(row[0] or "") if row else ""


def _clear_thread_semantic_meta_conn(conn: sqlite3.Connection, thread_id: str) -> None:
    """Drop per-thread embed stamps (call when clearing knowledge vectors)."""
    tid = (thread_id or "").strip()
    if not tid:
        return
    for key in (_embed_identity_meta_key(tid), _chunker_version_meta_key(tid)):
        conn.execute("DELETE FROM search_meta WHERE key = ?", (key,))


def knowledge_semantic_index_complete_for_thread(thread_id: str) -> bool:
    """Thread-safe wrapper for :func:`knowledge_semantic_index_complete`."""
    tid = (thread_id or "").strip()
    if not tid.startswith("knowledge:"):
        return False
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return False
        try:
            return knowledge_semantic_index_complete(conn, tid)
        finally:
            conn.close()


def knowledge_semantic_index_complete(conn: sqlite3.Connection, thread_id: str) -> bool:
    """True when every FTS row for a knowledge thread has at least one vector chunk."""
    tid = (thread_id or "").strip()
    if not tid.startswith("knowledge:"):
        return False
    fts_row = conn.execute(
        "SELECT COUNT(*) FROM messages_fts WHERE thread_id = ?",
        (tid,),
    ).fetchone()
    fts_n = int(fts_row[0] or 0) if fts_row else 0
    if fts_n <= 0:
        return False
    vec_row = conn.execute(
        "SELECT COUNT(DISTINCT message_row_id) FROM vector_chunks WHERE thread_id = ?",
        (tid,),
    ).fetchone()
    distinct = int(vec_row[0] or 0) if vec_row else 0
    return distinct >= fts_n


def get_thread_chunker_version(thread_id: str) -> str:
    """Per-thread chunker stamp (knowledge corpora); empty if unset."""
    tid = (thread_id or "").strip()
    if not tid:
        return ""
    conn = _safe_get_connection()
    if conn is None:
        return ""
    try:
        return _get_meta(conn, _chunker_version_meta_key(tid))
    finally:
        conn.close()


def get_thread_embed_identity(thread_id: str) -> str:
    """Read per-thread embedding identity stamp (empty if unset)."""
    tid = (thread_id or "").strip()
    if not tid:
        return ""
    conn = _safe_get_connection()
    if conn is None:
        return ""
    try:
        return _get_meta(conn, _embed_identity_meta_key(tid))
    finally:
        conn.close()


def _stamp_thread_embed_identity_conn(conn: sqlite3.Connection, thread_id: str) -> None:
    """Write embedding identity into an open search DB connection (no commit)."""
    tid = (thread_id or "").strip()
    if not tid:
        return
    try:
        from utility_model import utility_embedding_identity

        identity = utility_embedding_identity()
    except Exception:
        identity = ""
    if not identity:
        return
    if tid.startswith("knowledge:"):
        if not knowledge_semantic_index_complete(conn, tid):
            return
        # Never stamp hash fallback identity on knowledge corpora (causes reindex loops).
        if identity.startswith("hash:"):
            return
    _set_meta(conn, _embed_identity_meta_key(tid), identity)
    _set_meta(conn, "vector_dimensions", str(_vector_dim()))
    if tid.startswith("knowledge:"):
        try:
            from utils.embedding_chunking import CHUNKER_VERSION

            _set_meta(conn, _chunker_version_meta_key(tid), CHUNKER_VERSION())
        except Exception:
            pass


def get_thread_chunk_vectors(thread_id: str) -> dict[int, tuple[str, list[float]]]:
    """Read per-chunk vectors from vector_chunks (chunk_index=0) for graph reuse.

    Returns empty when the DB is unavailable, the thread identity is stale, or
    decode fails. Fail-open: callers fall through to residual embedding.
    """
    tid = (thread_id or "").strip()
    if not tid:
        return {}
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return {}
        try:
            if not _semantic_identity_ok(conn, tid):
                return {}
            if tid.startswith("knowledge:") and not _semantic_search_ok(conn, tid):
                return {}
            dim = _vector_dim()
            rows = conn.execute(
                """
                SELECT message_row_id, content_hash, embedding
                FROM vector_chunks
                WHERE thread_id = ? AND chunk_index = 0
                """,
                (tid,),
            ).fetchall()
            out: dict[int, tuple[str, list[float]]] = {}
            for message_row_id, content_hash, blob in rows:
                try:
                    rid = int(message_row_id)
                except (TypeError, ValueError):
                    continue
                vec = _decode_embedding(bytes(blob or b""), dimensions=dim)
                if vec:
                    out[rid] = (str(content_hash or ""), vec)
            return out
        except Exception as exc:
            _log.debug("[search] get_thread_chunk_vectors failed for %s: %s", tid, exc)
            return {}
        finally:
            conn.close()


def get_thread_chunk_previews(thread_id: str) -> dict[int, str]:
    """Read text_preview per message row (chunk_index=0) for labeling / theme discovery.

    Returns empty when the DB is unavailable or semantic identity is stale (same gate
    as get_thread_chunk_vectors) so labels are not paired with mismatched embeddings.
    """
    tid = (thread_id or "").strip()
    if not tid:
        return {}
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return {}
        try:
            if not _semantic_identity_ok(conn, tid):
                return {}
            if tid.startswith("knowledge:") and not _semantic_search_ok(conn, tid):
                return {}
            rows = conn.execute(
                """
                SELECT message_row_id, text_preview
                FROM vector_chunks
                WHERE thread_id = ? AND chunk_index = 0
                """,
                (tid,),
            ).fetchall()
            out: dict[int, str] = {}
            for message_row_id, preview in rows:
                try:
                    rid = int(message_row_id)
                except (TypeError, ValueError):
                    continue
                text = str(preview or "").strip()
                if text:
                    out[rid] = text
            return out
        except Exception as exc:
            _log.debug("[search] get_thread_chunk_previews failed for %s: %s", tid, exc)
            return {}
        finally:
            conn.close()


def count_thread_fts_messages(thread_id: str) -> int:
    """Count FTS-indexed messages for a thread (0 when DB unavailable)."""
    tid = (thread_id or "").strip()
    if not tid:
        return 0
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return 0
        try:
            row = conn.execute(
                "SELECT COUNT(*) FROM messages_fts WHERE thread_id = ?",
                (tid,),
            ).fetchone()
            return int(row[0] or 0) if row else 0
        except Exception:
            return 0
        finally:
            conn.close()


def count_thread_vector_chunks(thread_id: str) -> int:
    """Count semantic vector rows for a thread (0 when DB unavailable)."""
    tid = (thread_id or "").strip()
    if not tid:
        return 0
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return 0
        try:
            row = conn.execute(
                "SELECT COUNT(*) FROM vector_chunks WHERE thread_id = ?",
                (tid,),
            ).fetchone()
            return int(row[0] or 0) if row else 0
        except Exception:
            return 0
        finally:
            conn.close()


def stamp_thread_embed_identity(thread_id: str) -> None:
    """Write current embedding identity for a thread after (re)indexing vectors."""
    tid = (thread_id or "").strip()
    if not tid:
        return
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return
        try:
            _stamp_thread_embed_identity_conn(conn, tid)
            conn.commit()
        finally:
            conn.close()


def _utility_embed_identity() -> str:
    try:
        from utility_model import utility_embedding_identity

        return utility_embedding_identity()
    except Exception:
        return "hash:256"


def _semantic_identity_ok(conn: sqlite3.Connection, thread_id: str) -> bool:
    if not thread_id:
        return True
    stored = _get_meta(conn, _embed_identity_meta_key(thread_id))
    if not stored:
        try:
            from utils.embedding_model import is_real

            # Unstamped index with MLX loaded: vectors may be hash-era or missing.
            if is_real():
                return False
        except Exception:
            pass
        return True
    current = _utility_embed_identity()
    if stored == current:
        return True
    try:
        from utils.embedding_model import embedding_identity_stale

        return not embedding_identity_stale(stored)
    except Exception:
        return stored == current


def _semantic_chunker_ok(conn: sqlite3.Connection, thread_id: str) -> bool:
    """Knowledge threads require per-thread chunker_version to match active policy."""
    tid = (thread_id or "").strip()
    if not tid.startswith("knowledge:"):
        return True
    try:
        from utils.embedding_chunking import thread_chunker_version_stale

        return not thread_chunker_version_stale(conn, tid)
    except Exception:
        return False


def _semantic_search_ok(conn: sqlite3.Connection, thread_id: str) -> bool:
    tid = (thread_id or "").strip()
    if tid.startswith("knowledge:") and not knowledge_semantic_index_complete(conn, tid):
        return False
    return _semantic_identity_ok(conn, thread_id) and _semantic_chunker_ok(conn, thread_id)


def _stamp_chunker_version_conn(conn: sqlite3.Connection) -> None:
    try:
        from utils.embedding_chunking import CHUNKER_VERSION

        _set_meta(conn, "chunker_version", CHUNKER_VERSION())
    except Exception:
        pass


def _iter_knowledge_rows() -> list[tuple[str, list[dict[str, Any]]]]:
    try:
        from tools.knowledge_tools import (
            _build_knowledge_corpus_index,
            _hydrate_raw_texts_for_indexing,
            _load_manifest,
            _read_text_file,
            _synthesis_path,
            list_knowledge_corpora,
        )
    except Exception:
        return []
    rows: list[tuple[str, list[dict[str, Any]]]] = []
    for corpus in list_knowledge_corpora():
        slug = str(corpus.get("slug") or "").strip()
        label = str(corpus.get("label") or slug).strip()
        if not slug:
            continue
        thread_id = f"knowledge:{slug}"
        try:
            manifest = _load_manifest(slug) or {}
            sources = manifest.get("sources") if isinstance(manifest.get("sources"), list) else []
            from tools.knowledge_tools import read_synthesis_text

            synthesis_text = read_synthesis_text(slug)
            raw_texts = _hydrate_raw_texts_for_indexing(slug, sources, None)
            msgs, _meta, _nodes = _build_knowledge_corpus_index(
                slug, label, synthesis_text, sources, raw_texts
            )
            if msgs:
                rows.append((thread_id, msgs))
        except Exception as exc:
            _log.debug("[search] knowledge row build failed for %s: %s", slug, str(exc)[:120])
            continue
    return rows


def _upsert_vector_chunks(
    conn: sqlite3.Connection,
    *,
    thread_id: str,
    role: str,
    created_at: str,
    message_row_id: int,
    text: str,
    dimensions: int,
    pre_chunked: bool = False,
) -> int:
    if (thread_id or "").startswith("knowledge:"):
        try:
            from utils.embedding_model import embedding_work_should_defer

            if embedding_work_should_defer():
                return 0
        except Exception:
            pass
        try:
            from chat_storage import _looks_secret

            if _looks_secret(text):
                return 0
        except Exception:
            pass
    if pre_chunked:
        t = (text or "").strip()
        chunks = [t] if t else []
    elif (thread_id or "").startswith("knowledge:"):
        try:
            from utils.embedding_chunking import chunk_text_for_embedding

            chunks = chunk_text_for_embedding(text)
        except Exception:
            chunks = _chunk_text(text)
    else:
        chunks = _chunk_text(text)
    if not chunks:
        return 0
    existing: dict[int, str] = {}
    for chunk_index, content_hash in conn.execute(
        "SELECT chunk_index, content_hash FROM vector_chunks WHERE thread_id = ? AND message_row_id = ?",
        (thread_id, message_row_id),
    ).fetchall():
        existing[int(chunk_index)] = str(content_hash or "")

    needed_chunks: list[tuple[int, str, str]] = []
    for idx, chunk in enumerate(chunks):
        h = sha256(chunk.encode("utf-8", errors="ignore")).hexdigest()
        if existing.get(idx) == h:
            continue
        needed_chunks.append((idx, chunk, h))
    if not needed_chunks:
        # Remove trailing chunks if source became shorter.
        if len(existing) > len(chunks):
            _purge_vector_chunks_range(
                conn, thread_id, message_row_id, len(chunks)
            )
        return 0
    # Cost cap per indexing call; remaining chunks are eventually caught by rebuild.
    needed_chunks = needed_chunks[: _embed_max_chunks_per_call()]

    inserted = 0
    batch_max = _embed_batch_max()
    for start in range(0, len(needed_chunks), batch_max):
        batch = needed_chunks[start : start + batch_max]
        embeds = _utility_embed_texts([c[1] for c in batch], dimensions=dimensions)
        if not embeds or len(embeds) != len(batch):
            if (thread_id or "").startswith("knowledge:"):
                _log.warning(
                    "[search] embedding batch incomplete for %s row %s (%d/%d)",
                    thread_id,
                    message_row_id,
                    len(embeds or []),
                    len(batch),
                )
            continue
        for (chunk_index, chunk, h), vec in zip(batch, embeds):
            if not vec:
                continue
            vec_dim = len(vec)
            if vec_dim <= 0 or vec_dim != dimensions:
                continue
            blob = _encode_embedding(vec)
            conn.execute(
                """
                INSERT INTO vector_chunks
                    (thread_id, message_row_id, chunk_index, role, created_at, content_hash, text_preview, embedding)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(thread_id, message_row_id, chunk_index) DO UPDATE SET
                    role = excluded.role,
                    created_at = excluded.created_at,
                    content_hash = excluded.content_hash,
                    text_preview = excluded.text_preview,
                    embedding = excluded.embedding
                """,
                (
                    thread_id,
                    message_row_id,
                    chunk_index,
                    role,
                    created_at,
                    h,
                    chunk[:240],
                    blob,
                ),
            )
            if _mirror_vector_chunk_writes(conn):
                try:
                    from utils import sqlite_vec_index as svi

                    rid = _vector_chunk_rowid(conn, thread_id, message_row_id, chunk_index)
                    if rid is not None:
                        svi.mirror_insert(
                            conn, rid, thread_id, blob, expected_dim=dimensions
                        )
                except Exception as exc:
                    _log.warning(
                        "[search] vec mirror_insert failed for %s row %s chunk %s: %s",
                        thread_id,
                        message_row_id,
                        chunk_index,
                        str(exc)[:120],
                    )
            inserted += 1
    if len(existing) > len(chunks):
        _purge_vector_chunks_range(conn, thread_id, message_row_id, len(chunks))
    return inserted


def index_messages(
    thread_id: str,
    messages: list[dict[str, Any]],
    *,
    force_semantic: bool = False,
    pre_chunked: bool = False,
) -> None:
    """Index new messages into the FTS5 search database.

    Called from chat_storage.append_messages() after the primary write.
    Silently fails if the search DB is unavailable (derived index, not critical path).

    When *force_semantic* is True, vector chunks are always generated regardless
    of the global ``semantic_search_enabled`` config (used for knowledge corpora).

    ``btw_`` prefixed threads are ephemeral side-questions (see utils/btw.py)
    and are intentionally excluded from cross-conversation search.
    """
    if isinstance(thread_id, str) and thread_id.startswith("btw_"):
        return
    if _rebuilding.is_set():
        return
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return
        try:
            dimensions = _vector_dim()
            _set_meta(conn, "vector_dimensions", str(dimensions))
            _stamp_chunker_version_conn(conn)
            _set_meta(conn, "utility_embed_identity", _utility_embed_identity())

            semantic_enabled = force_semantic or _semantic_enabled(None)
            row_pre_chunked = pre_chunked or (
                isinstance(thread_id, str) and thread_id.startswith("knowledge:")
            )
            inserted_vectors = 0
            for msg in messages:
                if (thread_id or "").startswith("knowledge:"):
                    try:
                        from utils.embedding_model import embedding_work_should_defer

                        if embedding_work_should_defer():
                            _log.info(
                                "[search] knowledge index_messages paused for chat (%s)",
                                thread_id,
                            )
                            break
                    except Exception:
                        pass
                role = str(msg.get("role", "unknown") or "unknown")
                content = _content_to_text(msg.get("content"))
                if not content or len(content) < 5:
                    continue
                if (thread_id or "").startswith("knowledge:"):
                    try:
                        from chat_storage import _looks_secret

                        if _looks_secret(content):
                            continue
                    except Exception:
                        pass
                created_at = str(msg.get("created_at") or "").strip() or datetime.now(UTC).isoformat().replace("+00:00", "Z")
                conn.execute(
                    "INSERT INTO messages_fts (thread_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                    (thread_id, role, content, created_at),
                )
                # Semantic vectors require stable row ids from chat_storage plumbing.
                row_id = msg.get("_row_id")
                if not semantic_enabled:
                    continue
                try:
                    row_id_int = int(row_id)
                except Exception:
                    continue
                inserted_vectors += _upsert_vector_chunks(
                    conn,
                    thread_id=thread_id,
                    role=role,
                    created_at=created_at,
                    message_row_id=row_id_int,
                    text=content,
                    dimensions=dimensions,
                    pre_chunked=row_pre_chunked,
                )
            if semantic_enabled:
                stamp_ok = False
                if (thread_id or "").startswith("knowledge:"):
                    stamp_ok = knowledge_semantic_index_complete(conn, thread_id)
                else:
                    row = conn.execute(
                        "SELECT COUNT(*) FROM vector_chunks WHERE thread_id = ?",
                        (thread_id,),
                    ).fetchone()
                    stamp_ok = int(row[0] or 0) > 0 if row else False
                if stamp_ok:
                    _stamp_thread_embed_identity_conn(conn, thread_id)
            conn.commit()
            if inserted_vectors:
                _log.debug("Indexed %d vector chunks for thread %s", inserted_vectors, thread_id)
        except Exception as e:
            if (thread_id or "").startswith("knowledge:"):
                _log.warning(
                    "[search] Failed to index knowledge messages for %s: %s",
                    thread_id,
                    str(e)[:160],
                )
            else:
                _log.debug("Failed to index messages: %s", e)
        finally:
            conn.close()


def _thread_id_is_reindex_staging(thread_id: str) -> bool:
    return (thread_id or "").endswith(_REINDEX_STAGING_SUFFIX)


def _thread_id_is_background(thread_id: str) -> bool:
    tid = thread_id or ""
    if _thread_id_is_reindex_staging(tid):
        return True
    return any(tid.startswith(p) for p in _BACKGROUND_THREAD_ID_PREFIXES)


def _exclude_background_threads_flag(explicit: bool | None) -> bool:
    if explicit is not None:
        return bool(explicit)
    try:
        from core.config import get_session_search_config

        return bool(get_session_search_config().get("exclude_background_threads", True))
    except Exception:
        return True


def _search_fts_rows(
    conn: sqlite3.Connection,
    query: str,
    *,
    exclude_thread_id: str,
    include_thread_id: str,
    max_results: int,
    exclude_background_threads: bool | None,
    fts_mode: str = "and",
) -> list[dict[str, Any]]:
    fts_query = _safe_fts_query(query, mode=fts_mode)
    sql = """
        SELECT thread_id, role, snippet(messages_fts, 2, '>>>', '<<<', '...', 64) as snippet,
               created_at, rank
        FROM messages_fts
        WHERE messages_fts MATCH ?
    """
    params: list[Any] = [fts_query]
    if include_thread_id:
        sql += " AND thread_id = ?"
        params.append(include_thread_id)
    elif exclude_thread_id:
        sql += " AND thread_id != ?"
        params.append(exclude_thread_id)
    exclude_bg = _exclude_background_threads_flag(exclude_background_threads)
    sql_limit = max(max_results * 8, max_results) if exclude_bg else max_results
    sql += " ORDER BY rank LIMIT ?"
    params.append(sql_limit)
    try:
        rows = conn.execute(sql, params).fetchall()
    except Exception:
        # Retry with raw stripped query before escalating.
        raw = (query or "").strip()
        if not raw or raw == fts_query:
            raise
        params[0] = raw
        rows = conn.execute(sql, params).fetchall()
    results: list[dict[str, Any]] = []
    for thread_id, role, snippet, created_at, rank in rows:
        if not _thread_id_is_search_visible(str(thread_id or "")):
            continue
        if exclude_bg and _thread_id_is_background(thread_id):
            continue
        results.append(
            {
                "thread_id": thread_id,
                "role": role,
                "snippet": snippet,
                "created_at": created_at,
                "rank": rank,
                "_source": "lexical",
            }
        )
        if len(results) >= max_results:
            break
    return results


def _semantic_finalize_scored(
    scored: list[tuple[float, dict[str, Any]]],
    *,
    conn: sqlite3.Connection,
    stale_threads: set[str],
    max_results: int,
) -> list[dict[str, Any]]:
    if stale_threads:
        try:
            from utils.embedding_model import maybe_schedule_embedding_reindex

            for tid in stale_threads:
                maybe_schedule_embedding_reindex(tid)
        except Exception:
            pass
    scored.sort(key=lambda x: x[0], reverse=True)
    hard_cap = min(max_results, _semantic_max_hits())
    return [row for _, row in scored[:hard_cap]]


def _semantic_row_postfilter_ok(
    tid: str,
    *,
    exclude_thread_id: str,
    include_thread_id: str,
    exclude_bg: bool,
) -> bool:
    if not _thread_id_is_search_visible(tid):
        return False
    if include_thread_id and tid != include_thread_id:
        return False
    if exclude_thread_id and tid == exclude_thread_id:
        return False
    if exclude_bg and _thread_id_is_background(tid):
        return False
    return True


def _thread_id_is_search_visible(thread_id: str) -> bool:
    """False for internal ephemeral indexes that must never surface in operator search."""
    return not _thread_id_is_reindex_staging(thread_id)


def _search_semantic_rows_bruteforce(
    conn: sqlite3.Connection,
    qv: list[float],
    *,
    exclude_thread_id: str,
    include_thread_id: str,
    max_results: int,
    exclude_background_threads: bool | None,
    dimensions: int,
) -> list[dict[str, Any]]:
    exclude_bg = _exclude_background_threads_flag(exclude_background_threads)
    sql = """
        SELECT thread_id, role, created_at, text_preview, embedding
        FROM vector_chunks
        WHERE 1=1
    """
    params: list[Any] = []
    if include_thread_id:
        sql += " AND thread_id = ?"
        params.append(include_thread_id)
    elif exclude_thread_id:
        sql += " AND thread_id != ?"
        params.append(exclude_thread_id)
    if exclude_bg:
        for prefix in _BACKGROUND_THREAD_ID_PREFIXES:
            sql += " AND thread_id NOT LIKE ?"
            params.append(f"{prefix}%")
    sql += " AND thread_id NOT LIKE ?"
    params.append(f"%{_REINDEX_STAGING_SUFFIX}")
    sql += " LIMIT ?"
    params.append(_semantic_candidate_limit())

    rows = conn.execute(sql, params).fetchall()
    scored: list[tuple[float, dict[str, Any]]] = []
    stale_threads: set[str] = set()
    for thread_id, role, created_at, text_preview, blob in rows:
        tid = str(thread_id or "")
        if tid and not _semantic_search_ok(conn, tid):
            stale_threads.add(tid)
            continue
        vec = _decode_embedding(blob, dimensions=dimensions)
        if not vec or len(vec) != dimensions:
            continue
        sim = _cosine_similarity(qv, vec)
        if sim <= 0.0:
            continue
        scored.append(
            (
                sim,
                {
                    "thread_id": thread_id,
                    "role": role,
                    "snippet": f">>>{(text_preview or '')}<<<",
                    "created_at": created_at,
                    "rank": -sim,
                    "semantic_score": sim,
                    "_source": "semantic",
                },
            )
        )
    return _semantic_finalize_scored(
        scored, conn=conn, stale_threads=stale_threads, max_results=max_results
    )


def _search_semantic_rows_vec(
    conn: sqlite3.Connection,
    qv: list[float],
    *,
    exclude_thread_id: str,
    include_thread_id: str,
    max_results: int,
    exclude_background_threads: bool | None,
    dimensions: int,
) -> list[dict[str, Any]]:
    from utils import sqlite_vec_index as svi

    svi.sync_distance_metric_from_ddl(conn)
    exclude_bg = _exclude_background_threads_flag(exclude_background_threads)
    scope = (include_thread_id or "").strip() or None
    k_base = _semantic_ann_k()
    k_fetch = min(
        _semantic_candidate_limit(),
        k_base * _semantic_ann_overfetch(),
    )
    qblob = _encode_embedding(qv)
    if len(qblob) != dimensions * 4:
        return []
    candidates = svi.knn(conn, qblob, k=k_fetch, thread_id=scope)
    if not candidates:
        return []

    rowid_dist = {int(rid): float(dist) for rid, dist in candidates}
    placeholders = ",".join("?" * len(rowid_dist))
    rows = conn.execute(
        f"""
        SELECT rowid, thread_id, role, created_at, text_preview
        FROM vector_chunks
        WHERE rowid IN ({placeholders})
        """,
        list(rowid_dist.keys()),
    ).fetchall()

    scored: list[tuple[float, dict[str, Any]]] = []
    stale_threads: set[str] = set()
    for rid, thread_id, role, created_at, text_preview in rows:
        tid = str(thread_id or "")
        if not _semantic_row_postfilter_ok(
            tid,
            exclude_thread_id=exclude_thread_id,
            include_thread_id=include_thread_id,
            exclude_bg=exclude_bg,
        ):
            continue
        if tid and not _semantic_search_ok(conn, tid):
            stale_threads.add(tid)
            continue
        dist = rowid_dist.get(int(rid))
        if dist is None:
            continue
        sim = svi.distance_to_semantic_score(dist)
        if sim <= 0.0:
            continue
        scored.append(
            (
                sim,
                {
                    "thread_id": thread_id,
                    "role": role,
                    "snippet": f">>>{(text_preview or '')}<<<",
                    "created_at": created_at,
                    "rank": -sim,
                    "semantic_score": sim,
                    "_source": "semantic",
                },
            )
        )
    return _semantic_finalize_scored(
        scored, conn=conn, stale_threads=stale_threads, max_results=max_results
    )


def _search_semantic_rows(
    conn: sqlite3.Connection,
    query: str,
    *,
    exclude_thread_id: str,
    include_thread_id: str,
    max_results: int,
    exclude_background_threads: bool | None,
) -> list[dict[str, Any]]:
    q = (query or "").strip()
    if len(q) < _semantic_query_min_chars():
        return []
    dimensions = _vector_dim()
    scope_tid = (include_thread_id or "").strip()
    if scope_tid and not _semantic_search_ok(conn, scope_tid):
        try:
            from utils.embedding_model import maybe_schedule_embedding_reindex

            maybe_schedule_embedding_reindex(scope_tid)
        except Exception:
            pass
        return []
    q_embed = _utility_embed_texts([q], dimensions=dimensions, kind="query")
    if not q_embed:
        return []
    qv = q_embed[0]
    if not qv or len(qv) != dimensions:
        return []

    if _vec_ann_active(conn, scope_thread_id=scope_tid):
        ann_rows = _search_semantic_rows_vec(
            conn,
            qv,
            exclude_thread_id=exclude_thread_id,
            include_thread_id=include_thread_id,
            max_results=max_results,
            exclude_background_threads=exclude_background_threads,
            dimensions=dimensions,
        )
        if ann_rows:
            return ann_rows
    return _search_semantic_rows_bruteforce(
        conn,
        qv,
        exclude_thread_id=exclude_thread_id,
        include_thread_id=include_thread_id,
        max_results=max_results,
        exclude_background_threads=exclude_background_threads,
        dimensions=dimensions,
    )


def _merge_rows_rrf(
    lexical_rows: list[dict[str, Any]],
    semantic_rows: list[dict[str, Any]],
    *,
    max_results: int,
) -> list[dict[str, Any]]:
    if not semantic_rows:
        return lexical_rows[:max_results]
    if not lexical_rows:
        return semantic_rows[:max_results]
    k = float(_rrf_k())
    scored: dict[tuple[str, str, str], dict[str, Any]] = {}
    for rank_idx, row in enumerate(lexical_rows, start=1):
        key = (str(row.get("thread_id", "")), str(row.get("created_at", "")), str(row.get("snippet", "")))
        item = scored.setdefault(key, {"row": dict(row), "score": 0.0, "source_set": set()})
        item["score"] += 1.0 / (k + rank_idx)
        item["source_set"].add("lexical")
    for rank_idx, row in enumerate(semantic_rows, start=1):
        key = (str(row.get("thread_id", "")), str(row.get("created_at", "")), str(row.get("snippet", "")))
        item = scored.setdefault(key, {"row": dict(row), "score": 0.0, "source_set": set()})
        item["score"] += 1.0 / (k + rank_idx)
        item["source_set"].add("semantic")
        if row.get("semantic_score") and not item["row"].get("semantic_score"):
            item["row"]["semantic_score"] = row.get("semantic_score")
    merged = sorted(scored.values(), key=lambda x: float(x["score"]), reverse=True)
    out: list[dict[str, Any]] = []
    for m in merged[:max_results]:
        row = dict(m["row"])
        src = m["source_set"]
        if src == {"lexical"}:
            row["_source"] = "lexical"
        elif src == {"semantic"}:
            row["_source"] = "semantic"
        else:
            row["_source"] = "both"
        row["rrf_score"] = float(m["score"])
        out.append(row)
    return out


def search(
    query: str,
    exclude_thread_id: str = "",
    max_results: int = 20,
    *,
    include_thread_id: str = "",
    exclude_background_threads: bool | None = None,
    semantic: bool | None = None,
    hybrid: bool | None = None,
    fts_mode: str = "and",
) -> list[dict[str, Any]]:
    """Search indexed conversations (FTS + optional semantic vectors).

    When ``include_thread_id`` is set, only rows for that thread are considered
    (``exclude_thread_id`` is ignored for the SQL filter).

    *fts_mode*: ``"and"`` (default) | ``"and_content"`` (stopword-stripped) |
    ``"or"`` (broadest recall).  Passed through to ``_safe_fts_query``.
    """
    if _rebuilding.is_set():
        return []
    if not query or not query.strip():
        return []
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return []
        try:
            lexical_rows = _search_fts_rows(
                conn,
                query,
                exclude_thread_id=exclude_thread_id,
                include_thread_id=include_thread_id,
                max_results=max(max_results * 5, max_results),
                exclude_background_threads=exclude_background_threads,
                fts_mode=fts_mode,
            )
            if not _semantic_enabled(semantic):
                return lexical_rows[:max_results]
            semantic_rows = _search_semantic_rows(
                conn,
                query,
                exclude_thread_id=exclude_thread_id,
                include_thread_id=include_thread_id,
                max_results=max(max_results * 5, max_results),
                exclude_background_threads=exclude_background_threads,
            )
            if _semantic_hybrid_enabled(hybrid):
                return _merge_rows_rrf(lexical_rows, semantic_rows, max_results=max_results)
            if semantic_rows:
                return semantic_rows[:max_results]
            return lexical_rows[:max_results]
        except Exception as e:
            _log.warning("Search failed (%s); triggering rebuild.", e)
            _trigger_rebuild_async()
            return []
        finally:
            conn.close()


def is_rebuilding() -> bool:
    """True if the search index is currently being rebuilt."""
    return _rebuilding.is_set()


def search_index_uses_plain_sqlite() -> bool:
    """True when the search DB is not SQLCipher-encrypted (vector previews are plaintext at rest)."""
    global _search_index_plain_sqlite
    if _search_index_plain_sqlite is None:
        conn = _safe_get_connection()
        if conn is not None:
            conn.close()  # probe only; _open_search_index already set _search_index_plain_sqlite
    return bool(_search_index_plain_sqlite)


def clear_thread_index(thread_id: str, *, clear_fts: bool = True, clear_vectors: bool = True) -> None:
    """Delete indexed rows for a single thread."""
    if not thread_id:
        return
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return
        try:
            if clear_fts:
                conn.execute("DELETE FROM messages_fts WHERE thread_id = ?", (thread_id,))
            if clear_vectors:
                _purge_vector_chunks_thread(conn, thread_id)
            if (thread_id or "").startswith("knowledge:") and (clear_fts or clear_vectors):
                _clear_thread_semantic_meta_conn(conn, thread_id)
            conn.commit()
        finally:
            conn.close()


def reindex_staging_thread_id(thread_id: str) -> str:
    """Ephemeral thread id used while rebuilding a knowledge corpus search index."""
    tid = (thread_id or "").strip()
    return f"{tid}{_REINDEX_STAGING_SUFFIX}"


def clear_orphan_reindex_staging_threads(*, preserve: set[str] | None = None) -> int:
    """Drop leftover staging indexes from interrupted corpus reindex runs."""
    preserve_ids = set(preserve or ())
    try:
        from utils.embedding_model import active_reindex_staging_thread_ids

        preserve_ids |= active_reindex_staging_thread_ids()
    except Exception:
        pass
    suffix_like = f"%{_REINDEX_STAGING_SUFFIX}"
    staging_ids: list[str] = []
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return 0
        try:
            for (tid,) in conn.execute(
                "SELECT DISTINCT thread_id FROM messages_fts WHERE thread_id LIKE ?",
                (suffix_like,),
            ).fetchall():
                if tid:
                    staging_ids.append(str(tid))
            for (tid,) in conn.execute(
                "SELECT DISTINCT thread_id FROM vector_chunks WHERE thread_id LIKE ?",
                (suffix_like,),
            ).fetchall():
                text = str(tid or "")
                if text and text not in staging_ids:
                    staging_ids.append(text)
        finally:
            conn.close()
    cleared = 0
    for tid in staging_ids:
        if tid in preserve_ids:
            continue
        clear_thread_index(tid, clear_fts=True, clear_vectors=True)
        cleared += 1
    if cleared:
        _log.info("[search] cleared %d orphan reindex staging thread(s)", cleared)
    return cleared


def _rebuilt_knowledge_index_viable(conn: sqlite3.Connection) -> bool:
    """False when swapping in a rebuild would leave knowledge corpora FTS-only."""
    try:
        from utils.embedding_model import embedding_enabled

        if not embedding_enabled():
            return True
    except Exception:
        return True
    suffix_like = f"%{_REINDEX_STAGING_SUFFIX}"
    try:
        rows = conn.execute(
            """
            SELECT DISTINCT thread_id FROM messages_fts
            WHERE thread_id LIKE 'knowledge:%' AND thread_id NOT LIKE ?
            """,
            (suffix_like,),
        ).fetchall()
        for (tid,) in rows:
            tid_s = str(tid or "").strip()
            if not tid_s or knowledge_semantic_index_complete(conn, tid_s):
                continue
            fts_row = conn.execute(
                "SELECT COUNT(*) FROM messages_fts WHERE thread_id = ?",
                (tid_s,),
            ).fetchone()
            if int(fts_row[0] or 0) if fts_row else 0 > 0:
                return False
        return True
    except Exception as exc:
        _log.warning(
            "[search] knowledge rebuild viability check failed: %s",
            str(exc)[:120],
        )
        return False


def _schedule_knowledge_reindex_for_incomplete(conn: sqlite3.Connection) -> None:
    suffix_like = f"%{_REINDEX_STAGING_SUFFIX}"
    try:
        from utils.embedding_model import maybe_schedule_embedding_reindex

        rows = conn.execute(
            """
            SELECT DISTINCT thread_id FROM messages_fts
            WHERE thread_id LIKE 'knowledge:%' AND thread_id NOT LIKE ?
            """,
            (suffix_like,),
        ).fetchall()
        for (tid,) in rows:
            tid_s = str(tid or "").strip()
            if tid_s and not knowledge_semantic_index_complete(conn, tid_s):
                maybe_schedule_embedding_reindex(tid_s)
    except Exception as exc:
        _log.debug("[search] schedule knowledge reindex after rebuild abort: %s", str(exc)[:120])


_FTS_RECOVERY_MIN_CHARS = 5


def _pad_fts_recovery_content(text: str) -> str:
    """Ensure recovered FTS rows satisfy index_messages minimum length."""
    t = (text or "").strip() or " "
    if len(t) < _FTS_RECOVERY_MIN_CHARS:
        t = t + (" " * (_FTS_RECOVERY_MIN_CHARS - len(t)))
    return t


def _thread_fts_as_index_messages(conn: sqlite3.Connection, thread_id: str) -> list[dict[str, Any]]:
    """Rebuild index_messages payloads from existing FTS rows (stable 1..N row ids)."""
    tid = (thread_id or "").strip()
    if not tid:
        return []
    rows = conn.execute(
        """
        SELECT role, content, created_at
        FROM messages_fts
        WHERE thread_id = ?
        ORDER BY rowid ASC
        """,
        (tid,),
    ).fetchall()
    messages: list[dict[str, Any]] = []
    for row_id, (role, content, created_at) in enumerate(rows, start=1):
        text = _pad_fts_recovery_content(_content_to_text(content))
        messages.append(
            {
                "role": str(role or "system"),
                "content": text,
                "created_at": str(created_at or "").strip()
                or datetime.now(UTC).isoformat().replace("+00:00", "Z"),
                "_row_id": row_id,
            }
        )
    return messages


def reindex_vectors_from_existing_fts(target_thread_id: str) -> bool:
    """Embed search vectors from production FTS rows (no re-digest or disk chunk rebuild).

    Uses the staging promote pattern so production FTS is only swapped after vectors
    are verified complete on staging.
    """
    target = (target_thread_id or "").strip()
    if not target.startswith("knowledge:"):
        return False
    try:
        from utils.embedding_model import (
            embedding_enabled,
            embedding_work_should_defer,
            load,
        )
    except Exception:
        return False
    if embedding_work_should_defer():
        _log.debug("[search] FTS vector recovery deferred — chat active (%s)", target)
        return False
    staging = reindex_staging_thread_id(target)
    if not embedding_enabled() or not load():
        _log.debug(
            "[search] FTS vector recovery skipped — embedder unavailable (%s)",
            target,
        )
        return False
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            _log.warning("[search] FTS vector recovery failed — search DB unavailable")
            return False
        try:
            if knowledge_semantic_index_complete(conn, target):
                return True
            fts_n = int(
                conn.execute(
                    "SELECT COUNT(*) FROM messages_fts WHERE thread_id = ?",
                    (target,),
                ).fetchone()[0]
                or 0
            )
            if fts_n <= 0:
                return False
            messages = _thread_fts_as_index_messages(conn, target)
            if len(messages) != fts_n:
                _log.warning(
                    "[search] FTS recovery for %s aborted: %d messages vs %d fts rows",
                    target,
                    len(messages),
                    fts_n,
                )
                return False
            if not messages:
                return False
        finally:
            conn.close()
    if embedding_work_should_defer():
        _log.debug("[search] FTS vector recovery deferred before staging (%s)", target)
        return False
    clear_thread_index(staging, clear_fts=True, clear_vectors=True)
    _FTS_RECOVER_BATCH = 64
    for start in range(0, len(messages), _FTS_RECOVER_BATCH):
        if embedding_work_should_defer():
            clear_thread_index(staging, clear_fts=True, clear_vectors=True)
            _log.info(
                "[search] FTS vector recovery paused for chat (%s, %d/%d rows staged)",
                target,
                start,
                len(messages),
            )
            return False
        index_messages(
            staging,
            messages[start : start + _FTS_RECOVER_BATCH],
            force_semantic=True,
            pre_chunked=True,
        )
    if embedding_work_should_defer():
        clear_thread_index(staging, clear_fts=True, clear_vectors=True)
        return False
    if not knowledge_semantic_index_complete_for_thread(staging):
        clear_thread_index(staging, clear_fts=True, clear_vectors=True)
        _log.warning(
            "[search] FTS vector recovery incomplete on staging for %s",
            target,
        )
        return False
    if not promote_staged_thread_index(staging, target):
        clear_thread_index(staging, clear_fts=True, clear_vectors=True)
        _log.warning("[search] FTS vector recovery promote failed for %s", target)
        return False
    _log.info(
        "[search] recovered %d search vectors for %s from existing FTS",
        fts_n,
        target,
    )
    return True


def promote_staged_thread_index(staging_thread_id: str, target_thread_id: str) -> bool:
    """Replace *target_thread_id* index with a verified staging build."""
    staging = (staging_thread_id or "").strip()
    target = (target_thread_id or "").strip()
    if not staging or not target or staging == target:
        return False
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return False
        try:
            if target.startswith("knowledge:") and not knowledge_semantic_index_complete(
                conn, staging
            ):
                return False
            conn.execute("DELETE FROM messages_fts WHERE thread_id = ?", (target,))
            conn.execute(
                """
                INSERT INTO messages_fts (thread_id, role, content, created_at)
                SELECT ?, role, content, created_at FROM messages_fts WHERE thread_id = ?
                """,
                (target, staging),
            )
            _purge_vector_chunks_thread(conn, target)
            conn.execute(
                """
                INSERT INTO vector_chunks
                    (thread_id, message_row_id, chunk_index, role, created_at,
                     content_hash, text_preview, embedding)
                SELECT ?, message_row_id, chunk_index, role, created_at,
                       content_hash, text_preview, embedding
                FROM vector_chunks WHERE thread_id = ?
                """,
                (target, staging),
            )
            for meta_key_fn in (_embed_identity_meta_key, _chunker_version_meta_key):
                staged_val = _get_meta(conn, meta_key_fn(staging))
                target_key = meta_key_fn(target)
                if staged_val:
                    _set_meta(conn, target_key, staged_val)
                else:
                    conn.execute("DELETE FROM search_meta WHERE key = ?", (target_key,))
            if _mirror_vector_chunk_writes(conn):
                try:
                    from utils import sqlite_vec_index as svi

                    dim = _vector_dim()
                    if svi.vec_available(conn) and svi.ensure_vec_table(conn, dim):
                        mirrored = svi.backfill(conn, expected_dim=dim)
                        if mirrored:
                            _log.debug(
                                "[search] promote staged mirrored %d vec rows for %s",
                                mirrored,
                                target,
                            )
                except Exception as exc:
                    _log.warning(
                        "[search] promote staged vec backfill failed for %s: %s",
                        target,
                        str(exc)[:120],
                    )
            conn.execute("DELETE FROM messages_fts WHERE thread_id = ?", (staging,))
            _purge_vector_chunks_thread(conn, staging)
            _clear_thread_semantic_meta_conn(conn, staging)
            conn.commit()
            return True
        except Exception as exc:
            _log.warning(
                "[search] promote staged index failed %s -> %s: %s",
                staging,
                target,
                str(exc)[:120],
            )
            try:
                conn.rollback()
            except Exception:
                pass
            return False
        finally:
            conn.close()


def rebuild_index() -> None:
    """Rebuild the entire FTS5 index from chat_storage. Blocks until complete."""
    global _vec_needs_rebuild, _vec_backfill_in_progress
    # Cross-process guard: ``_rebuilding`` only dedupes within one process. The backend, a
    # stale runner, MCP, and CLI all share ``~/.mugi/search.db``; without this, two of them
    # can run full rebuilds at once (wasted work + vec backfill racing an os.replace).
    from utils.process_lock import CrossProcessLock

    # Intra-process single-flight: atomically check-and-set ``_rebuilding`` so a second
    # same-process thread bails instead of racing a duplicate rebuild (the fcntl lock below
    # only guards across *processes*).
    with _rebuild_start_lock:
        if _rebuilding.is_set():
            _log.debug("[search] rebuild skipped — already rebuilding in this process")
            return
        _rebuilding.set()
    _xproc_lock = CrossProcessLock("search_index_rebuild")
    if not _xproc_lock.try_acquire():
        _log.info("[search] rebuild skipped — another process holds the rebuild lock")
        _rebuilding.clear()
        return
    _log.info("Rebuilding search index from chat history...")
    tmp_path: str | None = None
    try:
        base_dir = os.path.dirname(SEARCH_DB_PATH) or "."
        fd, tmp_path = tempfile.mkstemp(prefix="mugi_search_rebuild_", suffix=".db", dir=base_dir)
        os.close(fd)
        conn = _open_search_index(tmp_path, schedule_vec_backfill=False)
        try:
            from chat_storage import _get_connection as _get_chat_conn, _decrypt
            chat_conn = _get_chat_conn()
            try:
                dimensions = _vector_dim()
                _set_meta(conn, "vector_dimensions", str(dimensions))
                _stamp_chunker_version_conn(conn)
                _set_meta(conn, "utility_embed_identity", _utility_embed_identity())
                semantic_enabled = _semantic_enabled(None)
                thread_ids = [
                    row[0] for row in
                    chat_conn.execute("SELECT DISTINCT thread_id FROM messages").fetchall()
                ]
                total = 0
                total_vec = 0
                chat_threads_with_vectors: set[str] = set()
                for tid in thread_ids:
                    rows = chat_conn.execute(
                        "SELECT id, role, content, created_at FROM messages WHERE thread_id = ? ORDER BY id ASC",
                        (tid,),
                    ).fetchall()
                    for row_id, role, content_blob, created_at in rows:
                        try:
                            text = _decrypt(content_blob) if content_blob else ""
                        except Exception:
                            continue
                        if not text or not isinstance(text, str) or len(text.strip()) < 5:
                            continue
                        # Multipart JSON: extract text parts
                        if text.strip().startswith("["):
                            try:
                                import json
                                parts = json.loads(text)
                                if isinstance(parts, list):
                                    text_parts = []
                                    for p in parts:
                                        if isinstance(p, dict) and p.get("type") == "text":
                                            text_parts.append(p.get("text", ""))
                                        elif isinstance(p, str):
                                            text_parts.append(p)
                                    text = " ".join(text_parts)
                            except Exception:
                                pass
                        text = text.strip()
                        if not text or len(text) < 5:
                            continue
                        conn.execute(
                            "INSERT INTO messages_fts (thread_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                            (tid, role, text, created_at or ""),
                        )
                        total += 1
                        if semantic_enabled:
                            n_vec = _upsert_vector_chunks(
                                conn,
                                thread_id=tid,
                                role=str(role or "unknown"),
                                created_at=str(created_at or ""),
                                message_row_id=int(row_id),
                                text=text,
                                dimensions=dimensions,
                            )
                            total_vec += n_vec
                            if n_vec > 0 and not str(tid).startswith("knowledge:"):
                                chat_threads_with_vectors.add(str(tid))
                for tid in chat_threads_with_vectors:
                    _stamp_thread_embed_identity_conn(conn, tid)
                for kb_thread_id, kb_rows in _iter_knowledge_rows():
                    for msg in kb_rows:
                        role = str(msg.get("role") or "system")
                        text = _content_to_text(msg.get("content"))
                        created_at = str(msg.get("created_at") or "")
                        conn.execute(
                            "INSERT INTO messages_fts (thread_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                            (kb_thread_id, role, text, created_at),
                        )
                        total += 1
                        try:
                            row_id = int(msg.get("_row_id"))
                        except Exception:
                            continue
                        total_vec += _upsert_vector_chunks(
                            conn,
                            thread_id=kb_thread_id,
                            role=role,
                            created_at=created_at,
                            message_row_id=row_id,
                            text=text,
                            dimensions=dimensions,
                            pre_chunked=True,
                        )
                    if knowledge_semantic_index_complete(conn, kb_thread_id):
                        _stamp_thread_embed_identity_conn(conn, kb_thread_id)
                if not _rebuilt_knowledge_index_viable(conn):
                    _log.warning(
                        "Search index rebuild aborted: knowledge corpora would lose semantic "
                        "vectors (FTS without matching embeddings). Keeping existing search.db."
                    )
                    _schedule_knowledge_reindex_for_incomplete(conn)
                    return
                conn.commit()
                _log.info(
                    "Search index rebuilt: %d FTS rows, %d vector chunks across %d threads.",
                    total,
                    total_vec,
                    len(thread_ids),
                )
            finally:
                chat_conn.close()
        finally:
            conn.close()
        os.replace(tmp_path, SEARCH_DB_PATH)
        tmp_path = None
        _vec_needs_rebuild = False
        _vec_backfill_in_progress = False
    except Exception as e:
        _log.error("Search index rebuild failed: %s", e)
    finally:
        if tmp_path:
            try:
                os.remove(tmp_path)
            except OSError:
                pass
        _vec_backfill_in_progress = False
        _rebuilding.clear()
        _xproc_lock.release()


_deferred_rebuild_timer: threading.Timer | None = None
_deferred_rebuild_lock = threading.Lock()


def _main_chat_busy() -> bool:
    try:
        from persistence import read_main_chat_busy

        return read_main_chat_busy()
    except Exception:
        return False


def _schedule_deferred_rebuild(delay_s: float = 120.0) -> None:
    global _deferred_rebuild_timer

    def _fire() -> None:
        global _deferred_rebuild_timer
        with _deferred_rebuild_lock:
            _deferred_rebuild_timer = None
        _trigger_rebuild_async()

    with _deferred_rebuild_lock:
        if _deferred_rebuild_timer is not None:
            return
        _deferred_rebuild_timer = threading.Timer(delay_s, _fire)
        _deferred_rebuild_timer.daemon = True
        _deferred_rebuild_timer.start()


def _embedding_ready_for_full_rebuild() -> bool:
    """When MLX embeddings are required, defer full rebuild until the embedder can load."""
    try:
        from utils.embedding_model import embedding_enabled, load

        if not embedding_enabled():
            return True
        return load()
    except Exception:
        return False


def _trigger_rebuild_async() -> None:
    """Delete the corrupted DB and rebuild in a background thread."""
    if _rebuilding.is_set():
        return
    if _main_chat_busy():
        _log.info("Search index rebuild deferred — main chat active.")
        _schedule_deferred_rebuild()
        return
    if not _embedding_ready_for_full_rebuild():
        _log.info("Search index rebuild deferred — MLX embedder unavailable.")
        _schedule_deferred_rebuild(delay_s=180.0)
        return
    t = threading.Thread(target=rebuild_index, daemon=True, name="search-index-rebuild")
    t.start()


def ensure_index_exists() -> None:
    """If the search DB doesn't exist, build it in the background."""
    if not os.path.exists(SEARCH_DB_PATH):
        _trigger_rebuild_async()
