feat: FTS5 search index with background rebuild and 30-day default wi…

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
     }
   }
 }