feat: FTS5 search index with background rebuild and 30-day default wi… (#120)

* feat: FTS5 search index with background rebuild and 30-day default window

* fix search index edge cases from review

Clamp oversized since_days instead of dropping them back to the 30-day default.
Keep paging FTS hits when exclusion rules filter out the first batch so limit
still fills. Skip session files that vanish during the fingerprint walk, drop
dead index path constants, and print the real regression limit in the benchmark
gate. Refresh test_search_full_corpus baseline for the indexed search path.

* style: ruff format check_benchmark_regression

* Fixed Timon's feedback on feat/search-fts-index. Summary:  utils/search_index.py Added ORDER BY m.timestamp_ms DESC so index hits are newest-first Replaced the 20× over-fetch with a fixed _FTS_BATCH_SIZE = 200; _search_via_index keeps paging until results are filled or FTS is exhausted Fixed offset: sql_rows_fetched is now rows actually scanned (not full batch size), so early break no longer skips unprocessed rows api/search.py Added _rank_search_hits() for timestamp-desc ordering Index and live-scan paths both rank before returning (live scan collects all matches, then sorts and slices — correct global newest-first on fallback) Tests test_results_ordered_newest_first — verifies DESC ordering test_phrase_filter_paginates_past_decoy_token_matches — 250 token-only decoys + real phrase at end; would fail with the old offset bug 30 tests passed (ruff + mypy clean).  Suggested commit message:  fix: FTS search ordering and pagination for phrase filtering Add ORDER BY timestamp_ms DESC, advance SQL offset by rows scanned, and page in fixed batches until enough hits or FTS exhausted. Rank live-scan results the same way for parity. Want me to commit and push to PR #120?  /humanizer commit message   fix: FTS search ordering and pagination

Author: clean6378-max-it

api/search.py

@@ -1,18 +1,38 @@
-"""Search endpoint. Brute-force substring match across all sessions."""
+"""Search endpoint — FTS index with live-scan fallback."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
 
 from flask import Blueprint, current_app, request
 
 from api._flask_types import FlaskReturn, json_response
 from api.error_codes import ErrorCode, error_response
 from models.search import SearchHitDict
+from models.session import MessageDict
 from utils.exclusion_rules import is_session_excluded
+from utils.search_index import (
+    index_is_usable,
+    index_search_enabled,
+    query_index_hits,
+    resolve_search_since_ms,
+    search_snippet,
+    timestamp_in_search_window_iso,
+    timestamp_to_ms,
+    tool_result_searchable_text,
+)
 from utils.session_cache import get_cached_session
 from utils.session_path import get_claude_projects_dir, list_projects, list_sessions
+from utils.session_summary_cache import get_summary, rules_fingerprint
 
 search_bp = Blueprint("search", __name__)
+_logger = logging.getLogger(__name__)
 
 _DEFAULT_LIMIT = 50
 _MAX_LIMIT = 500
+_MAX_SEARCH_SINCE_DAYS = 36_500
 
 
 def _parse_limit(raw: str | None, default: int = _DEFAULT_LIMIT) -> int:
@@ -28,60 +48,215 @@ def _parse_limit(raw: str | None, default: int = _DEFAULT_LIMIT) -> int:
     return min(value, _MAX_LIMIT)
 
 
-@search_bp.route("/api/search")
-def search() -> FlaskReturn:
-    query = request.args.get("q", "").strip().lower()
-    if not query:
-        return json_response([])
-
+def _parse_since_days(raw: str | None) -> int | None:
+    if raw is None or not str(raw).strip():
+        return None
     try:
-        max_results = _parse_limit(request.args.get("limit"))
+        days = int(str(raw).strip())
     except ValueError:
-        return error_response(
-            ErrorCode.SEARCH_INVALID_LIMIT,
-            "Invalid limit: must be a positive integer",
-            400,
+        return None
+    if days <= 0:
+        return None
+    return min(days, _MAX_SEARCH_SINCE_DAYS)
+
+
+def _rank_search_hits(results: list[SearchHitDict]) -> list[SearchHitDict]:
+    """Sort hits by timestamp descending (missing timestamps last)."""
+    return sorted(
+        results,
+        key=lambda hit: timestamp_to_ms(
+            hit["timestamp"] if isinstance(hit.get("timestamp"), str) else None
+        ),
+        reverse=True,
+    )
+
+
+def _message_searchable_text(msg: MessageDict) -> str:
+    text = msg.get("text", "") or msg.get("content", "")
+    if not isinstance(text, str):
+        text = ""
+    tool_result = msg.get("tool_result")
+    if isinstance(tool_result, dict):
+        tool_text = tool_result_searchable_text(tool_result)
+        if tool_text:
+            text = f"{text}\n{tool_text}" if text else tool_text
+    return text
+
+
+def _index_hit_excluded(
+    rules: list[Any],
+    rules_fp: str,
+    *,
+    project_name: str,
+    file_path: str,
+    mtime: float,
+) -> bool:
+    if not rules:
+        return False
+    cached = get_summary(file_path, mtime, rules_fp)
+    if cached is not None and cached["is_complete"]:
+        return cached["is_excluded"]
+    try:
+        session = get_cached_session(file_path)
+    except Exception:
+        _logger.warning(
+            "Could not load session for exclusion check during index search: %s",
+            file_path,
+            exc_info=True,
         )
+        return False
+    return is_session_excluded(rules, session, project_name)
 
-    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
-    projects = list_projects(base)
 
-    rules = current_app.config.get("EXCLUSION_RULES") or []
+def _search_via_index(
+    projects_dir: str,
+    rules: list[Any],
+    query: str,
+    query_lower: str,
+    *,
+    since_ms: int | None,
+    max_results: int,
+) -> list[SearchHitDict] | None:
+    if not index_search_enabled() or not index_is_usable(projects_dir, rules):
+        return None
+
+    rules_fp = rules_fingerprint(rules)
     results: list[SearchHitDict] = []
-    for project in projects:
-        if len(results) >= max_results:
+    sql_offset = 0
+    while len(results) < max_results:
+        need = max_results - len(results)
+        indexed = query_index_hits(
+            query_lower,
+            since_ms=since_ms,
+            max_results=need,
+            sql_offset=sql_offset,
+        )
+        if not indexed["query_ok"]:
+            return None
+        if indexed["sql_rows_fetched"] == 0:
             break
-        sessions = list_sessions(project["path"])
-        for sess_info in sessions:
+
+        for hit in indexed["hits"]:
             if len(results) >= max_results:
                 break
+            if _index_hit_excluded(
+                rules,
+                rules_fp,
+                project_name=hit["project_name"],
+                file_path=hit["file_path"],
+                mtime=hit["mtime"],
+            ):
+                continue
+            results.append(
+                {
+                    "project": hit["project_name"],
+                    "session_id": hit["session_id"],
+                    "title": hit["title"],
+                    "role": hit["role"],
+                    "timestamp": hit["timestamp"],
+                    "snippet": search_snippet(hit["text"], query),
+                }
+            )
+
+        sql_offset += indexed["sql_rows_fetched"]
+        if indexed["sql_exhausted"]:
+            break
+    return _rank_search_hits(results)[:max_results]
+
+
+def _search_live_scan(
+    base: str,
+    rules: list[Any],
+    query: str,
+    query_lower: str,
+    *,
+    since_ms: int | None,
+    max_results: int,
+) -> list[SearchHitDict]:
+    projects = list_projects(base)
+    results: list[SearchHitDict] = []
+    for project in projects:
+        sessions = list_sessions(project["path"])
+        for sess_info in sessions:
             try:
                 session = get_cached_session(sess_info["path"])
             except Exception:
+                _logger.warning(
+                    "Skipping session during live search: %s",
+                    sess_info["path"],
+                    exc_info=True,
+                )
                 continue
 
             if is_session_excluded(rules, session, project["name"]):
                 continue
 
             for msg in session["messages"]:
-                text = msg.get("text", "") or msg.get("content", "")
-                if query in text.lower():
-                    idx = text.lower().index(query)
-                    start = max(0, idx - 80)
-                    end = min(len(text), idx + len(query) + 80)
-                    snippet = text[start:end]
-
-                    results.append(
-                        {
-                            "project": project["name"],
-                            "session_id": session["session_id"],
-                            "title": session["title"],
-                            "role": msg["role"],
-                            "timestamp": msg.get("timestamp"),
-                            "snippet": snippet,
-                        }
-                    )
-                    if len(results) >= max_results:
-                        break
-
-    return json_response(results)
+                text = _message_searchable_text(msg)
+                if not text or query_lower not in text.lower():
+                    continue
+                if not timestamp_in_search_window_iso(
+                    msg.get("timestamp") if isinstance(msg.get("timestamp"), str) else None,
+                    since_ms,
+                ):
+                    continue
+                results.append(
+                    {
+                        "project": project["name"],
+                        "session_id": session["session_id"],
+                        "title": session["title"],
+                        "role": msg["role"],
+                        "timestamp": msg.get("timestamp"),
+                        "snippet": search_snippet(text, query),
+                    }
+                )
+    return _rank_search_hits(results)[:max_results]
+
+
+@search_bp.route("/api/search")
+def search() -> FlaskReturn:
+    query = request.args.get("q", "").strip()
+    if not query:
+        return json_response([])
+
+    try:
+        max_results = _parse_limit(request.args.get("limit"))
+    except ValueError:
+        return error_response(
+            ErrorCode.SEARCH_INVALID_LIMIT,
+            "Invalid limit: must be a positive integer",
+            400,
+        )
+
+    query_lower = query.lower()
+    all_history = request.args.get("all_history") in ("1", "true")
+    since_ms = resolve_search_since_ms(
+        all_history=all_history,
+        since_days=_parse_since_days(request.args.get("since_days")),
+        now=datetime.now(timezone.utc),
+    )
+
+    base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
+    rules = current_app.config.get("EXCLUSION_RULES") or []
+
+    indexed = _search_via_index(
+        base,
+        rules,
+        query,
+        query_lower,
+        since_ms=since_ms,
+        max_results=max_results,
+    )
+    if indexed is not None:
+        return json_response(indexed)
+
+    return json_response(
+        _search_live_scan(
+            base,
+            rules,
+            query,
+            query_lower,
+            since_ms=since_ms,
+            max_results=max_results,
+        )
+    )

app.py

@@ -3,6 +3,7 @@
 __version__ = "0.2.0.dev0"
 
 import argparse
+import logging
 import sys
 
 from flask import Flask
@@ -13,6 +14,7 @@
 from api.search import search_bp
 from api.sessions import sessions_bp
 from utils.exclusion_rules import load_rules, resolve_exclusion_rules_path
+from utils.session_path import get_claude_projects_dir
 
 # Content-Security-Policy for all Flask responses. 'unsafe-inline' in style-src is
 # required because highlight.js themes apply inline styles; can be tightened with
@@ -104,6 +106,15 @@ def create_app(
     app.register_blueprint(export_bp)
     app.register_blueprint(schema_report_bp)
 
+    if not app.config.get("TESTING"):
+        try:
+            from utils.search_index import start_search_index_background
+
+            projects_dir = base_dir or get_claude_projects_dir()
+            start_search_index_background(projects_dir, app.config["EXCLUSION_RULES"])
+        except Exception:
+            logging.getLogger(__name__).exception("Failed to start search index background worker")
+
     @app.after_request
     def set_security_headers(response):
         # Always set — do not use setdefault; a blueprint must not weaken CSP.

benchmarks/README.md

@@ -38,9 +38,9 @@ The memory ceiling test (`test_large_parse_peak_memory_under_ceiling`) runs in t
 
 The `benchmarks` job on **ubuntu-latest** runs pytest-benchmark (`--benchmark-json=benchmark-results.json`), then `scripts/check_benchmark_regression.py benchmark-results.json benchmarks/baselines.json`. CI fails when any **gated** benchmark mean exceeds its baseline by more than **20%**.
 
-**Gated:** parse medium/large + large peak memory; export 10/50/100 session latency + ZIP peak memory.
+**Gated:** parse medium/large + large peak memory; export 10/50/100 session latency + ZIP peak memory; `test_search_full_corpus` (relaxed **2.0×** threshold vs 1.2× for other gates — sub-ms search timings are noisy on ubuntu CI).
 
-**Not gated (informational only):** `test_parse_session_small`, `test_search_full_corpus` (sub-ms CI noise), and the `cache` group. These may appear in `baselines.json` for reference but are skipped by `check_benchmark_regression.py`. Benchmarks without a baseline entry print a warning and do not fail the gate.
+**Not gated (informational only):** `test_parse_session_small` and the `cache` group.
 
 Missing gated benchmarks (renamed or removed tests still listed in `baselines.json`) fail the gate.
 

benchmarks/baselines.json

@@ -1,6 +1,6 @@
 {
-  "_note": "Gated means from ubuntu-latest CI benchmark-results.json. PR #108 (schema drift): parse/export latency baselines raised for per-entry field-path fingerprinting. Excluded from gate (recorded for reference): test_parse_session_small, test_search_full_corpus (sub-ms CI noise). Memory benchmarks use extra_info.peak_bytes (bytes); latency uses stats.mean (seconds).",
-  "updated": "2026-07-03T00:00:00Z",
+  "_note": "Gated means from ubuntu-latest CI benchmark-results.json. PR #108 (schema drift): parse/export latency baselines raised for per-entry field-path fingerprinting. Excluded from gate (recorded for reference): test_parse_session_small (sub-ms CI noise). test_search_full_corpus baseline refreshed for FTS search path (PR #120). Memory benchmarks use extra_info.peak_bytes (bytes); latency uses stats.mean (seconds).",
+  "updated": "2026-07-07T00:00:00Z",
   "machine": "Linux",
   "groups": {
     "parse": {
@@ -18,7 +18,7 @@
       "test_bulk_export_zip_peak_memory[sessions-100]": 694088.0
     },
     "search": {
-      "test_search_full_corpus": 0.0011120838654706596
+      "test_search_full_corpus": 0.002595
     }
   }
 }