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

Author: clean6378-max-it

api/search.py

@@ -1,18 +1,37 @@
-"""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,
+    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,25 +47,107 @@ 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 or days > _MAX_SEARCH_SINCE_DAYS:
+        return None
+    return days
+
+
+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)
+    indexed = query_index_hits(query_lower, since_ms=since_ms, max_results=max_results)
+    if not indexed["query_ok"]:
+        return None
+
+    results: list[SearchHitDict] = []
+    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),
+            }
+        )
+    return 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:
         if len(results) >= max_results:
@@ -58,30 +159,84 @@ def search() -> FlaskReturn:
             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),
+                    }
+                )
+                if len(results) >= max_results:
+                    break
+    return 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,17 @@ 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,5 +1,5 @@
 {
-  "_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).",
+  "_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). Memory benchmarks use extra_info.peak_bytes (bytes); latency uses stats.mean (seconds).",
   "updated": "2026-07-03T00:00:00Z",
   "machine": "Linux",
   "groups": {

scripts/check_benchmark_regression.py

@@ -8,12 +8,14 @@
 from pathlib import Path
 
 THRESHOLD = 1.20
+# Sub-ms search timings are noisy on ubuntu CI; allow a wider band than parse/export.
+SEARCH_BENCHMARK_THRESHOLD = 2.0
+RELAXED_THRESHOLD_TESTS = frozenset({"test_search_full_corpus"})
 
 # Sub-ms timings are too noisy for a fixed 20% gate on ubuntu CI.
 EXCLUDED_FROM_GATE = frozenset(
     {
         "test_parse_session_small",
-        "test_search_full_corpus",
     }
 )
 
@@ -153,13 +155,14 @@ def check_regression(
             print(f"WARN: baseline for {name!r} is zero; skipping ratio check")
             continue
         ratio = cur / base
-        tag = "FAIL" if ratio > threshold else "ok"
+        limit = SEARCH_BENCHMARK_THRESHOLD if name in RELAXED_THRESHOLD_TESTS else threshold
+        tag = "FAIL" if ratio > limit else "ok"
         entry = entries_by_name.get(name)
         if metric_is_bytes(name, entry):
             print(f"[{tag}] {name}: {cur:.0f} bytes vs {base:.0f} bytes ({ratio:.2f}x)")
         else:
             print(f"[{tag}] {name}: {cur:.6f}s vs {base:.6f}s ({ratio:.2f}x)")
-        if ratio > threshold:
+        if ratio > limit:
             failures.append(name)
 
     for name in flat:

scripts/reduce_baselines.py

@@ -70,8 +70,8 @@ def reduce_baselines(
         "_note": (
             "Gated means from ubuntu-latest CI benchmark-results.json."
             f"{slack_note} "
-            "Excluded from gate (recorded for reference): test_parse_session_small, "
-            "test_search_full_corpus (sub-ms CI noise). "
+            "Excluded from gate (recorded for reference): test_parse_session_small "
+            "(sub-ms CI noise). "
             "Memory benchmarks use extra_info.peak_bytes (bytes); "
             "latency uses stats.mean (seconds)."
         ),

tests/test_api_integration.py

@@ -100,7 +100,7 @@ def test_session_detail_includes_thinking_blocks(client_thinking):
 
 
 def test_search_returns_results(client):
-    resp = client.get("/api/search?q=Hello")
+    resp = client.get("/api/search?q=Hello&all_history=1")
     assert resp.status_code == 200
     results = resp.get_json()
     assert isinstance(results, list)
@@ -121,7 +121,7 @@ def test_search_invalid_limit(client):
 
 
 def test_search_valid_limit(client):
-    resp = client.get("/api/search?q=Hello&limit=5")
+    resp = client.get("/api/search?q=Hello&limit=5&all_history=1")
     assert resp.status_code == 200
     results = resp.get_json()
     assert isinstance(results, list)