iris_x_hermes/gateway-plugin/search.py

"""FTS5 session search bridge (M3).

Bridges the ``search {query, scope, chat_id?, thread_id?}`` frame to the hermes
session store (SQLite + FTS5, ``hermes_state.py`` / ``hermes_state_search.py``)
and returns ``search.results`` hits.

The session DB (``get_hermes_home()/"state.db"``) is opened **read-only** --
search never writes to the store. Two query paths:

* **FTS5** (primary): ``messages_fts MATCH <sanitized>`` with BM25 ranking.
* **LIKE** (fallback): when the FTS5 table is absent (FTS disabled / fresh DB)
  or the MATCH raises, a substring scan over ``messages.content``.

Scope:
* ``"all"``  -- every channel/thread/session.
* ``"chat"`` -- restrict to the given ``chat_id`` (and optional ``thread_id``).

Each hit: ``{message_id, chat_id, thread_id, role, snippet, ts}`` where
``message_id`` is the session-store row id (string) and ``ts`` is epoch
milliseconds. The app navigates to the hit's channel/thread and matches the
message by timestamp to scroll + highlight.

Privacy: search is local to the user's own hermes home; no data leaves the
machine.

Milestone M3.
"""

import contextlib
import logging
import re
import sqlite3
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

MAX_QUERY_CHARS = 200
DEFAULT_LIMIT = 20
MAX_LIMIT = 100

# FTS5 special chars (mirror of hermes_state_search._FTS5_SPECIAL_CHARS) for the
# fallback sanitizer when the real one can't be imported.
_FTS5_SPECIAL_CHARS = "+{}():\"^@/#&|~[]<>,;!?$=\\'"
_FTS5_SPECIAL_RE = re.compile(f"[{re.escape(_FTS5_SPECIAL_CHARS)}]")


def _sanitize(query: str) -> str:
    """Sanitize user input for a safe FTS5 MATCH.

    Prefers the gateway's own sanitizer (exact parity with session_search);
    falls back to a simplified strip-and-quote pass when it can't be imported.
    """
    q = (query or "").strip()
    if not q:
        return ""
    try:
        from hermes_state_search import SessionSearchMixin

        return SessionSearchMixin._sanitize_fts5_query(q)
    except Exception:
        return _sanitize_fallback(q)


def _sanitize_fallback(query: str) -> str:
    q = query[:MAX_QUERY_CHARS]
    q = _FTS5_SPECIAL_RE.sub(" ", q)
    if "%" in q:
        q = q.replace("%", " ")
    q = re.sub(r"\*+", "*", q)
    q = re.sub(r"(^|\s)\*", r"\1", q)
    q = re.sub(r"(?i)^(AND|OR|NOT)\b\s*", "", q.strip())
    q = re.sub(r"(?i)\s+(AND|OR|NOT)\s*$", "", q.strip())
    q = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', q)
    return q.strip()


def _fts_available(conn: sqlite3.Connection) -> bool:
    try:
        row = conn.execute(
            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages_fts' LIMIT 1"
        ).fetchone()
        return row is not None
    except sqlite3.Error:
        return False


def _scope_clauses(
    scope: str, chat_id: str | None, thread_id: str | None
) -> tuple[list[str], list[Any]]:
    """Build the scope WHERE clauses + params (empty for scope='all')."""
    clauses: list[str] = []
    params: list[Any] = []
    if scope == "chat" and chat_id:
        clauses.append("s.chat_id = ?")
        params.append(chat_id)
        if thread_id:
            clauses.append("s.thread_id = ?")
            params.append(thread_id)
    return clauses, params


def _row_to_hit(row: sqlite3.Row) -> dict[str, Any]:
    ts = row["timestamp"]
    try:
        ts_ms = int(float(ts) * 1000)
    except (TypeError, ValueError):
        ts_ms = 0
    return {
        "message_id": str(row["id"]),
        "chat_id": row["chat_id"],
        "thread_id": row["thread_id"],
        "role": row["role"],
        "snippet": row["snippet"] or "",
        "ts": ts_ms,
    }


def _fts_query(
    conn: sqlite3.Connection,
    query: str,
    scope: str,
    chat_id: str | None,
    thread_id: str | None,
    limit: int,
) -> list[dict[str, Any]]:
    where = ["messages_fts MATCH ?", "(m.active = 1 OR m.compacted = 1)"]
    params: list[Any] = [query]
    scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id)
    where.extend(scope_clauses)
    params.extend(scope_params)
    params.extend([limit])
    # The f-string only splices a fixed set of static WHERE fragments; every
    # user value is bound via ``?`` placeholders (see execute below).
    sql = f"""
        SELECT
            m.id,
            m.role,
            snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet,
            m.timestamp,
            s.chat_id,
            s.thread_id
        FROM messages_fts
        JOIN messages m ON m.id = messages_fts.rowid
        JOIN sessions s ON s.id = m.session_id
        WHERE {" AND ".join(where)}
        ORDER BY rank
        LIMIT ?
    """
    # Safe: every value is bound via ``?`` placeholders (no user data in SQL).
    # pi-lens-ignore: python-sql-injection
    rows = conn.execute(sql, params).fetchall()
    return [_row_to_hit(r) for r in rows]


def _like_query(
    conn: sqlite3.Connection,
    query: str,
    scope: str,
    chat_id: str | None,
    thread_id: str | None,
    limit: int,
) -> list[dict[str, Any]]:
    """Substring fallback when FTS5 is unavailable."""
    # First plain word of the query is the LIKE needle (best-effort).
    needle = re.split(r"\s+", query.strip(), maxsplit=1)[0].strip('"')
    if not needle:
        return []
    like = f"%{needle}%"
    where = ["(m.active = 1 OR m.compacted = 1)", "m.content LIKE ?"]
    params: list[Any] = [like]
    scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id)
    where.extend(scope_clauses)
    params.extend(scope_params)
    params.extend([limit])
    # The f-string only splices a fixed set of static WHERE fragments; every
    # user value is bound via ``?`` placeholders (see execute below).
    sql = f"""
        SELECT
            m.id,
            m.role,
            substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet,
            m.timestamp,
            s.chat_id,
            s.thread_id
        FROM messages m
        JOIN sessions s ON s.id = m.session_id
        WHERE {" AND ".join(where)}
        ORDER BY m.timestamp DESC
        LIMIT ?
    """
    # The needle appears twice (LIKE + instr); params order: like, scope..., needle, limit
    full_params = [like, *scope_params, needle, limit]
    # Safe: every value is bound via ``?`` placeholders (no user data in SQL).
    # pi-lens-ignore: python-sql-injection
    rows = conn.execute(sql, full_params).fetchall()
    return [_row_to_hit(r) for r in rows]


def search(
    db_path: Path,
    query: str,
    scope: str = "all",
    chat_id: str | None = None,
    thread_id: str | None = None,
    limit: int = DEFAULT_LIMIT,
) -> list[dict[str, Any]]:
    """Run a scoped search over the session store. Returns a list of hits.

    Never raises: any DB/FTS error yields an empty result (the caller sends an
    empty ``search.results``).
    """
    sanitized = _sanitize(query)
    if not sanitized:
        return []
    db_path = Path(db_path)
    if not db_path.exists():
        return []
    limit = max(1, min(int(limit or DEFAULT_LIMIT), MAX_LIMIT))
    scope = (scope or "all").strip().lower()
    if scope not in ("all", "chat"):
        scope = "all"

    try:
        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    except sqlite3.Error as e:
        logger.warning("android search: open failed: %s", e)
        return []
    conn.row_factory = sqlite3.Row
    try:
        if _fts_available(conn):
            try:
                return _fts_query(conn, sanitized, scope, chat_id, thread_id, limit)
            except sqlite3.Error as e:
                logger.debug("android search: FTS5 failed, using LIKE: %s", e)
        return _like_query(conn, sanitized, scope, chat_id, thread_id, limit)
    except sqlite3.Error as e:
        logger.warning("android search: query failed: %s", e)
        return []
    finally:
        # Best-effort: a close failure on a read-only connection is not
        # actionable (nothing to roll back).
        with contextlib.suppress(Exception):
            conn.close()