diff --git a/api/search.py b/api/search.py index 425d069..e16427f 100644 --- a/api/search.py +++ b/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, + ) + ) diff --git a/app.py b/app.py index 2786cc8..67b5b21 100644 --- a/app.py +++ b/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. diff --git a/benchmarks/README.md b/benchmarks/README.md index 5ef54c2..a02fa10 100644 --- a/benchmarks/README.md +++ b/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. diff --git a/benchmarks/baselines.json b/benchmarks/baselines.json index 99e5e50..626a537 100644 --- a/benchmarks/baselines.json +++ b/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 } } } diff --git a/scripts/check_benchmark_regression.py b/scripts/check_benchmark_regression.py index 3a27dab..6c356a5 100644 --- a/scripts/check_benchmark_regression.py +++ b/scripts/check_benchmark_regression.py @@ -8,12 +8,22 @@ 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"}) + + +def benchmark_regression_limit(name: str, threshold: float = THRESHOLD) -> float: + """Return the allowed regression ratio for a gated benchmark.""" + if name in RELAXED_THRESHOLD_TESTS: + return SEARCH_BENCHMARK_THRESHOLD + return threshold + # 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 +163,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 = benchmark_regression_limit(name, 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: @@ -169,7 +180,15 @@ def check_regression( print(f"WARN: {name!r} has no baseline yet; not gated") if failures: - print(f"\nREGRESSION: {len(failures)} benchmark(s) exceeded {threshold:.0%}") + limits = {benchmark_regression_limit(name, threshold) for name in failures} + if len(limits) == 1: + (limit,) = limits + print(f"\nREGRESSION: {len(failures)} benchmark(s) exceeded {limit:.0%}") + else: + details = ", ".join( + f"{name} ({benchmark_regression_limit(name, threshold):.0%})" for name in failures + ) + print(f"\nREGRESSION: {len(failures)} benchmark(s) exceeded limit: {details}") if missing: print(f"\nMISSING: {len(missing)} gated benchmark(s) absent from current results") if failures or missing: diff --git a/scripts/reduce_baselines.py b/scripts/reduce_baselines.py index a3b4114..a0c8c63 100644 --- a/scripts/reduce_baselines.py +++ b/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)." ), diff --git a/tests/test_api_integration.py b/tests/test_api_integration.py index f1124dd..6944efd 100644 --- a/tests/test_api_integration.py +++ b/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) diff --git a/tests/test_search.py b/tests/test_search.py index 3442b32..01c7816 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -5,7 +5,15 @@ from __future__ import annotations -from tests.conftest import assert_error_response +import json +import shutil +from datetime import UTC, datetime, timedelta +from pathlib import Path +from unittest.mock import patch + +from app import create_app +from tests.conftest import FIXTURES, assert_error_response +from utils.search_index import build_search_index, reset_background_for_tests _SEARCH_HIT_KEYS = frozenset( { @@ -76,3 +84,76 @@ def test_empty_query(client_single): resp = client_single.get("/api/search?q=") assert resp.status_code == 200 assert resp.get_json() == [] + + +def _index_patches(cache_root: Path): + return (patch("utils.search_index.cache_dir", return_value=cache_root),) + + +def _seed_indexed_client(tmp_path, monkeypatch, *, timestamp: str): + cache_root = tmp_path / "cache" + cache_root.mkdir() + project = tmp_path / "projects" / "demo-proj" + project.mkdir(parents=True) + session_path = project / "session_alpha.jsonl" + shutil.copy(FIXTURES / "session_minimal.jsonl", session_path) + lines = session_path.read_text(encoding="utf-8").splitlines() + entry = json.loads(lines[0]) + entry["timestamp"] = timestamp + lines[0] = json.dumps(entry, ensure_ascii=False) + session_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + monkeypatch.delenv("CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX", raising=False) + reset_background_for_tests() + + patches = _index_patches(cache_root) + with patches[0]: + assert build_search_index(str(tmp_path / "projects"), [], force=True) is True + + app = create_app(base_dir=str(tmp_path / "projects")) + app.config["TESTING"] = True + return app.test_client() + + +def test_default_window_excludes_old_session(tmp_path, monkeypatch): + old_ts = (datetime.now(UTC) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ") + client = _seed_indexed_client(tmp_path, monkeypatch, timestamp=old_ts) + resp = client.get("/api/search?q=Hello") + assert resp.status_code == 200 + assert resp.get_json() == [] + + +def test_all_history_includes_old_session(tmp_path, monkeypatch): + old_ts = (datetime.now(UTC) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ") + client = _seed_indexed_client(tmp_path, monkeypatch, timestamp=old_ts) + resp = client.get("/api/search?q=Hello&all_history=1") + assert resp.status_code == 200 + assert len(resp.get_json()) >= 1 + + +def test_search_uses_index_when_usable(tmp_path, monkeypatch): + recent_ts = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + client = _seed_indexed_client(tmp_path, monkeypatch, timestamp=recent_ts) + with patch("api.search.get_cached_session") as live_parse: + live_parse.side_effect = AssertionError("live-scan should not run when index is warm") + resp = client.get("/api/search?q=Hello") + assert resp.status_code == 200 + assert len(resp.get_json()) >= 1 + + +def test_search_falls_back_when_index_query_fails(tmp_path, monkeypatch): + recent_ts = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + client = _seed_indexed_client(tmp_path, monkeypatch, timestamp=recent_ts) + with patch( + "api.search.query_index_hits", + return_value={ + "hits": [], + "query_ok": False, + "sql_rows_fetched": 0, + "sql_exhausted": True, + }, + ): + resp = client.get("/api/search?q=Hello") + assert resp.status_code == 200 + assert len(resp.get_json()) >= 1 diff --git a/tests/test_search_index.py b/tests/test_search_index.py new file mode 100644 index 0000000..09af796 --- /dev/null +++ b/tests/test_search_index.py @@ -0,0 +1,392 @@ +"""Tests for utils/search_index.py (FTS search index).""" + +from __future__ import annotations + +import json +import sqlite3 +from contextlib import closing, contextmanager +from datetime import UTC, datetime, timedelta +from pathlib import Path +from unittest.mock import patch + +import pytest + +from api.search import _search_via_index +from utils.search_index import ( + build_search_index, + index_is_usable, + index_search_enabled, + query_index_hits, + reset_background_for_tests, + resolve_search_since_ms, + start_search_index_background, + timestamp_to_ms, + tool_result_searchable_text, +) + + +def _write_session(path: Path, lines: list[dict[str, object]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + for line in lines: + handle.write(json.dumps(line, ensure_ascii=False) + "\n") + + +def _index_patches(cache_root: Path): + return (patch("utils.search_index.cache_dir", return_value=cache_root),) + + +@pytest.fixture +def indexed_tree(tmp_path, monkeypatch): + """Temp projects dir + isolated search index cache.""" + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + project = projects / "demo-proj" + session_path = project / "session_alpha.jsonl" + _write_session( + session_path, + [ + { + "type": "user", + "timestamp": "2026-05-19T10:00:00Z", + "message": {"content": [{"type": "text", "text": "find indexed-unique-sentinel"}]}, + }, + { + "type": "assistant", + "timestamp": "2026-05-19T10:00:01Z", + "message": { + "model": "claude-test", + "content": [{"type": "text", "text": "acknowledged"}], + }, + }, + ], + ) + monkeypatch.delenv("CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX", raising=False) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + reset_background_for_tests() + + patches = _index_patches(cache_root) + with patches[0]: + built = build_search_index(str(projects), [], force=True) + assert built is True + pointer = cache_root / "search_index.active" + assert pointer.is_file() + index_name = pointer.read_text(encoding="utf-8").strip() + index_path = cache_root / index_name + assert index_path.is_file() + yield { + "projects": str(projects), + "cache_root": cache_root, + "index_path": index_path, + "term": "indexed-unique-sentinel", + } + + +class TestSearchIndexBuild: + def test_schema_tables_exist(self, indexed_tree): + with closing(sqlite3.connect(indexed_tree["index_path"])) as conn: + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type IN ('table', 'virtual table')" + ) + } + assert "index_meta" in tables + assert "sessions" in tables + assert "messages_fts" in tables + + def test_index_is_usable_after_build(self, indexed_tree): + patches = _index_patches(indexed_tree["cache_root"]) + with patches[0]: + assert index_is_usable(indexed_tree["projects"], []) is True + + def test_fts_finds_term(self, indexed_tree): + patches = _index_patches(indexed_tree["cache_root"]) + with patches[0]: + result = query_index_hits(indexed_tree["term"], since_ms=None, max_results=10) + assert result["query_ok"] is True + assert len(result["hits"]) == 1 + assert result["hits"][0]["session_id"] == "session_alpha" + assert indexed_tree["term"] in result["hits"][0]["text"] + + def test_pointer_swap_uses_uuid_file(self, indexed_tree): + index_path = indexed_tree["index_path"] + assert index_path.name.startswith("search_index.") + assert index_path.name.endswith(".sqlite") + assert index_path.name != "search_index.sqlite" + + def test_rebuild_when_manifest_changes(self, indexed_tree): + patches = _index_patches(indexed_tree["cache_root"]) + with patches[0]: + assert build_search_index(indexed_tree["projects"], [], force=False) is False + new_session = Path(indexed_tree["projects"]) / "demo-proj" / "session_beta.jsonl" + _write_session( + new_session, + [ + { + "type": "user", + "timestamp": "2026-06-01T10:00:00Z", + "message": {"content": [{"type": "text", "text": "beta only"}]}, + } + ], + ) + assert build_search_index(indexed_tree["projects"], [], force=False) is True + assert index_is_usable(indexed_tree["projects"], []) is True + + +class TestToolResultIndexing: + def test_tool_result_stdout_indexed(self, tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + session_path = projects / "tool-proj" / "session_tool.jsonl" + sentinel = "tool-result-sentinel-stdout" + _write_session( + session_path, + [ + { + "type": "user", + "timestamp": "2026-05-19T11:00:02Z", + "message": {"content": []}, + "toolUseResult": {"stdout": f"{sentinel}\n", "stderr": "", "exitCode": 0}, + } + ], + ) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + assert build_search_index(str(projects), [], force=True) is True + hits = query_index_hits(sentinel, since_ms=None, max_results=5) + assert hits["query_ok"] is True + assert any(sentinel in hit["text"] for hit in hits["hits"]) + + def test_tool_result_searchable_text_helper(self): + text = tool_result_searchable_text({"stdout": "hello", "stderr": "warn"}) + assert "hello" in text + assert "warn" in text + + +class TestSearchWindow: + def test_window_excludes_old_session(self, tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + old_ts = (datetime.now(UTC) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ") + _write_session( + projects / "win-proj" / "old_session.jsonl", + [ + { + "type": "user", + "timestamp": old_ts, + "message": {"content": [{"type": "text", "text": "window-old-sentinel"}]}, + } + ], + ) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + build_search_index(str(projects), [], force=True) + since_ms = resolve_search_since_ms(all_history=False) + hits = query_index_hits("window-old-sentinel", since_ms=since_ms, max_results=5) + assert hits["query_ok"] is True + assert hits["hits"] == [] + + def test_all_history_includes_old_session(self, tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + old_ts = (datetime.now(UTC) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ") + _write_session( + projects / "win-proj" / "old_session.jsonl", + [ + { + "type": "user", + "timestamp": old_ts, + "message": {"content": [{"type": "text", "text": "history-old-sentinel"}]}, + } + ], + ) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + build_search_index(str(projects), [], force=True) + hits = query_index_hits("history-old-sentinel", since_ms=None, max_results=5) + assert hits["query_ok"] is True + assert len(hits["hits"]) == 1 + + def test_undated_message_in_window(self, tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + _write_session( + projects / "win-proj" / "undated.jsonl", + [ + { + "type": "user", + "message": {"content": [{"type": "text", "text": "undated-window-sentinel"}]}, + } + ], + ) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + build_search_index(str(projects), [], force=True) + since_ms = resolve_search_since_ms(all_history=False) + hits = query_index_hits("undated-window-sentinel", since_ms=since_ms, max_results=5) + assert hits["query_ok"] is True + assert len(hits["hits"]) == 1 + + +class TestQueryIndexHits: + def test_results_ordered_newest_first(self, tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + _write_session( + projects / "order-proj" / "session.jsonl", + [ + { + "type": "user", + "timestamp": "2026-05-19T10:00:00Z", + "message": {"content": [{"type": "text", "text": "order alpha token"}]}, + }, + { + "type": "user", + "timestamp": "2026-06-19T10:00:00Z", + "message": {"content": [{"type": "text", "text": "order beta token"}]}, + }, + { + "type": "user", + "timestamp": "2026-07-19T10:00:00Z", + "message": {"content": [{"type": "text", "text": "order gamma token"}]}, + }, + ], + ) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + build_search_index(str(projects), [], force=True) + hits = query_index_hits("order", since_ms=None, max_results=10) + assert hits["query_ok"] is True + assert len(hits["hits"]) == 3 + timestamps = [hit["timestamp"] for hit in hits["hits"]] + assert timestamps == sorted(timestamps, reverse=True) + + def test_phrase_filter_paginates_past_decoy_token_matches(self, tmp_path, monkeypatch): + """Multi-word phrase matches beyond one FTS batch are not skipped.""" + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + lines: list[dict[str, object]] = [] + for i in range(250): + lines.append( + { + "type": "user", + "timestamp": f"2026-07-{(i % 28) + 1:02d}T10:00:00Z", + "message": { + "content": [ + { + "type": "text", + "text": f"phrase target decoy unique token only {i}", + } + ] + }, + } + ) + phrase = "decoy unique token only phrase target" + lines.append( + { + "type": "user", + "timestamp": "2025-12-01T10:00:00Z", + "message": { + "content": [ + { + "type": "text", + "text": phrase, + } + ] + }, + } + ) + _write_session(projects / "page-proj" / "session.jsonl", lines) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + patches = _index_patches(cache_root) + with patches[0]: + build_search_index(str(projects), [], force=True) + results = _search_via_index( + str(projects), + [], + phrase, + phrase.lower(), + since_ms=None, + max_results=1, + ) + assert results is not None + assert len(results) == 1 + assert phrase in results[0]["snippet"] or "phrase target" in results[0]["snippet"] + + def test_fts_failure_returns_query_not_ok(self, indexed_tree): + @contextmanager + def _broken_conn(*, readonly: bool = True): + class _Conn: + def execute(self, *args: object, **kwargs: object) -> None: + raise sqlite3.OperationalError("database is locked") + + def close(self) -> None: + pass + + yield _Conn() + + patches = _index_patches(indexed_tree["cache_root"]) + with ( + patches[0], + patch( + "utils.search_index._index_db_conn", + _broken_conn, + ), + ): + result = query_index_hits(indexed_tree["term"], since_ms=None, max_results=5) + assert result["query_ok"] is False + assert result["hits"] == [] + + +class TestBypassAndBackground: + def test_no_search_index_env_disables_index(self, indexed_tree, monkeypatch): + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX", "1") + patches = _index_patches(indexed_tree["cache_root"]) + with patches[0]: + assert index_search_enabled() is False + result = query_index_hits(indexed_tree["term"], since_ms=None, max_results=5) + assert result["query_ok"] is False + assert result["hits"] == [] + + def test_background_worker_starts_once(self, indexed_tree, monkeypatch): + monkeypatch.setenv( + "CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", + str(indexed_tree["cache_root"]), + ) + reset_background_for_tests() + patches = _index_patches(indexed_tree["cache_root"]) + with ( + patches[0], + patch("utils.search_index.threading.Thread") as mock_thread, + ): + start_search_index_background(indexed_tree["projects"], []) + start_search_index_background(indexed_tree["projects"], []) + assert mock_thread.call_count == 1 + assert mock_thread.call_args.kwargs.get("daemon") is True + assert mock_thread.return_value.start.call_count == 1 + + +class TestResolveSearchSinceMs: + def test_all_history_none(self): + assert resolve_search_since_ms(all_history=True) is None + + def test_default_thirty_day_window(self): + now = datetime(2026, 7, 7, 12, 0, tzinfo=UTC) + since = resolve_search_since_ms(all_history=False, now=now) + assert since == timestamp_to_ms("2026-06-07T12:00:00Z") + + def test_zero_days_searches_all(self): + assert resolve_search_since_ms(all_history=False, since_days=0) is None diff --git a/utils/search_index.py b/utils/search_index.py new file mode 100644 index 0000000..9ddf0b8 --- /dev/null +++ b/utils/search_index.py @@ -0,0 +1,689 @@ +"""Local FTS5 search index over Claude Code JSONL session files. + +Derived ``search_index..sqlite`` under ``~/.claude-code-chat-browser/``. +Session JSONL on disk remains source of truth; the index rebuilds when the +projects manifest or exclusion-rules fingerprint changes. + +Bypass: ``CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX=1``. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import sqlite3 +import threading +import time +import uuid +from collections.abc import Iterator +from contextlib import closing, contextmanager +from datetime import UTC, datetime, timedelta, timezone +from pathlib import Path +from typing import Any, TypedDict + +from models.session import RoleLiteral +from utils.jsonl_helpers import entry_message, extract_text, first_title_line +from utils.session_path import list_projects, list_sessions +from utils.session_summary_cache import rules_fingerprint + +__all__ = [ + "DEFAULT_SEARCH_WINDOW_DAYS", + "IndexMessageHitDict", + "IndexQueryResult", + "build_search_index", + "ensure_search_index", + "index_is_usable", + "index_search_enabled", + "query_index_hits", + "resolve_search_since_ms", + "search_snippet", + "start_search_index_background", + "timestamp_in_search_window_iso", + "timestamp_in_search_window_ms", + "tool_result_searchable_text", +] + +_logger = logging.getLogger(__name__) + +INDEX_VERSION = 1 +DEFAULT_SEARCH_WINDOW_DAYS = 30 +_INCLUDE_UNKNOWN_TIMESTAMPS_IN_WINDOW = True +_SKIP_ENTRY_TYPES = frozenset({"file-history-snapshot", "summary"}) + +_index_lock = threading.Lock() +_index_build_lock = threading.Lock() +_background_started = False +_usability_cache: dict[tuple[str, str], tuple[bool, float]] = {} +_usability_cache_lock = threading.Lock() +_USABILITY_CACHE_TTL_SECONDS = 30.0 +_FTS_BATCH_SIZE = 200 + + +class IndexMessageHitDict(TypedDict): + session_id: str + project_name: str + title: str + role: RoleLiteral + timestamp: str | None + text: str + file_path: str + mtime: float + + +class IndexQueryResult(TypedDict): + hits: list[IndexMessageHitDict] + query_ok: bool + sql_rows_fetched: int + sql_exhausted: bool + + +def cache_dir() -> Path: + """Return directory for search index files (overridable in tests).""" + override = os.environ.get("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", "").strip() + if override: + return Path(override) + return Path.home() / ".claude-code-chat-browser" + + +def index_search_enabled() -> bool: + return os.environ.get("CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX", "").strip().lower() not in ( + "1", + "true", + "yes", + ) + + +def resolve_search_since_ms( + *, + all_history: bool = False, + since_days: int | None = None, + now: datetime | None = None, +) -> int | None: + """Return epoch-ms cutoff for search, or ``None`` to search all history.""" + if all_history: + return None + days = since_days if since_days is not None else DEFAULT_SEARCH_WINDOW_DAYS + if days > 36_500: + days = 36_500 + if days <= 0: + return None + ref = now or datetime.now(timezone.utc) + cutoff = ref - timedelta(days=days) + return int(cutoff.timestamp() * 1000) + + +def timestamp_to_ms(ts: str | None) -> int: + if not ts: + return 0 + try: + dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) + return int(dt.timestamp() * 1000) + except (ValueError, AttributeError, OverflowError, OSError): + return 0 + + +def ms_to_timestamp(ms: int) -> str | None: + if ms <= 0: + return None + return datetime.fromtimestamp(ms / 1000, tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _clear_usability_cache() -> None: + with _usability_cache_lock: + _usability_cache.clear() + + +def _usability_cache_key(projects_dir: str, rules: list[Any]) -> tuple[str, str]: + return (str(Path(projects_dir).resolve()), rules_fingerprint(rules)) + + +def timestamp_in_search_window_ms(timestamp_ms: int, since_ms: int | None) -> bool: + if since_ms is None: + return True + if timestamp_ms <= 0: + return _INCLUDE_UNKNOWN_TIMESTAMPS_IN_WINDOW + return timestamp_ms >= since_ms + + +def timestamp_in_search_window_iso(timestamp: str | None, since_ms: int | None) -> bool: + return timestamp_in_search_window_ms(timestamp_to_ms(timestamp), since_ms) + + +def search_snippet(text: str, query: str, *, context: int = 80) -> str: + """Return a ±*context* character snippet around the first case-insensitive match.""" + needle = query.lower() + haystack = text.lower() + idx = haystack.find(needle) + if idx < 0: + return text[: context * 2] + start = max(0, idx - context) + end = min(len(text), idx + len(query) + context) + return text[start:end] + + +def _resolve_active_index_db_path() -> Path | None: + pointer = cache_dir() / "search_index.active" + legacy = cache_dir() / "search_index.sqlite" + if pointer.is_file(): + try: + name = pointer.read_text(encoding="utf-8").strip() + except OSError: + name = "" + if name: + candidate = cache_dir() / name + if candidate.is_file(): + return candidate + if legacy.is_file(): + return legacy + return None + + +def _publish_active_index(new_db_path: Path) -> None: + root = cache_dir() + root.mkdir(parents=True, exist_ok=True) + pointer = root / "search_index.active" + pointer_tmp = pointer.with_suffix(".active.tmp") + pointer_tmp.write_text(new_db_path.name, encoding="utf-8") + try: + pointer_tmp.replace(pointer) + except OSError: + pointer.write_text(new_db_path.name, encoding="utf-8") + try: + pointer_tmp.unlink() + except OSError: + pass + _prune_stale_index_files(keep=new_db_path) + + +def _prune_stale_index_files(*, keep: Path) -> None: + root = cache_dir() + for pattern in ("search_index.*.sqlite", "search_index.sqlite"): + for path in root.glob(pattern): + if path.resolve() == keep.resolve(): + continue + try: + path.unlink() + except OSError: + pass + for suffix in (".sqlite.tmp", ".active.tmp"): + for path in root.glob(f"search_index*{suffix}"): + try: + path.unlink() + except OSError: + pass + + +def _projects_fingerprint(projects_dir: str, rules: list[Any]) -> dict[str, Any]: + base = Path(projects_dir).resolve() + manifest: list[list[Any]] = [] + for project in list_projects(projects_dir): + for sess in list_sessions(project["path"]): + try: + rel = os.path.relpath(sess["path"], projects_dir) + stat = os.stat(sess["path"]) + except OSError: + continue + manifest.append([rel, stat.st_mtime_ns, stat.st_size]) + manifest.sort() + return { + "projects_dir": str(base), + "manifest": manifest, + "rules_fp": rules_fingerprint(rules), + } + + +def _fingerprints_match(a: dict[str, Any], b: dict[str, Any]) -> bool: + return ( + a.get("projects_dir") == b.get("projects_dir") + and a.get("rules_fp") == b.get("rules_fp") + and a.get("manifest") == b.get("manifest") + ) + + +def _open_index_db(*, readonly: bool = True) -> sqlite3.Connection | None: + db_path = _resolve_active_index_db_path() + if db_path is None: + return None + uri = db_path.resolve().as_uri() + if readonly: + uri += "?mode=ro" + try: + conn = sqlite3.connect(uri, uri=True, check_same_thread=False) + conn.row_factory = sqlite3.Row + return conn + except sqlite3.Error as exc: + _logger.debug("Failed to open search index: %s", exc) + return None + + +@contextmanager +def _index_db_conn(*, readonly: bool = True) -> Iterator[sqlite3.Connection | None]: + conn = _open_index_db(readonly=readonly) + try: + yield conn + finally: + if conn is not None: + conn.close() + + +def _read_stored_fingerprint(conn: sqlite3.Connection) -> dict[str, Any] | None: + try: + row = conn.execute("SELECT value FROM index_meta WHERE key = 'fingerprint'").fetchone() + if not row or not row[0]: + return None + data = json.loads(row[0]) + return data if isinstance(data, dict) else None + except (sqlite3.Error, json.JSONDecodeError): + return None + + +def _fts_match_query(query_lower: str) -> str | None: + tokens = [t for t in re.split(r"\W+", query_lower) if t] + if not tokens: + return None + parts: list[str] = [] + for token in tokens: + escaped = token.replace('"', '""') + parts.append(f'"{escaped}"*') + return " AND ".join(parts) + + +def _create_schema(conn: sqlite3.Connection) -> None: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS index_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS sessions ( + session_id TEXT NOT NULL, + project_name TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + first_ms INTEGER NOT NULL DEFAULT 0, + last_ms INTEGER NOT NULL DEFAULT 0, + file_path TEXT NOT NULL, + mtime REAL NOT NULL DEFAULT 0, + PRIMARY KEY (session_id, project_name) + ); + CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( + session_id UNINDEXED, + project_name UNINDEXED, + role UNINDEXED, + timestamp_ms UNINDEXED, + text, + tokenize='unicode61' + ); + """ + ) + + +def tool_result_searchable_text(raw: object) -> str: + if not isinstance(raw, dict): + return "" + parts: list[str] = [] + for key in ("stdout", "stderr", "content", "filePath", "query", "url", "plan"): + val = raw.get(key) + if isinstance(val, str) and val.strip(): + parts.append(val) + elif isinstance(val, list): + for item in val: + if isinstance(item, str) and item.strip(): + parts.append(item) + elif isinstance(item, dict): + nested = item.get("text") or item.get("content") + if isinstance(nested, str) and nested.strip(): + parts.append(nested) + return "\n".join(parts) + + +def _coerce_role(entry_type: str | None) -> RoleLiteral | None: + if entry_type == "user": + return "user" + if entry_type == "assistant": + return "assistant" + return None + + +def _indexable_texts_from_entry( + entry: dict[str, Any], +) -> list[tuple[str, str | None, RoleLiteral]]: + """Extract searchable (text, timestamp, role) tuples from one JSONL entry.""" + entry_type = entry.get("type") + if entry_type in _SKIP_ENTRY_TYPES: + return [] + + role = _coerce_role(entry_type if isinstance(entry_type, str) else None) + timestamp = entry.get("timestamp") + ts = timestamp if isinstance(timestamp, str) else None + texts: list[tuple[str, str | None, RoleLiteral]] = [] + + if role is not None: + msg = entry_message(entry) + content = msg.get("content", []) + text = extract_text(content) + if isinstance(text, str) and text.strip(): + texts.append((text, ts, role)) + + if entry_type == "user": + tool_text = tool_result_searchable_text(entry.get("toolUseResult")) + if tool_text: + texts.append((tool_text, ts, "user")) + + if not texts and role is not None: + raw_content = entry.get("content", "") + if isinstance(raw_content, str) and raw_content.strip(): + texts.append((raw_content, ts, role)) + + return texts + + +def _scan_session_file( + filepath: str, +) -> tuple[str, str, int, int, float, list[tuple[str, str | None, RoleLiteral]]]: + session_id = os.path.basename(filepath).replace(".jsonl", "") + title = "Untitled Session" + first_ms = 0 + last_ms = 0 + messages: list[tuple[str, str | None, RoleLiteral]] = [] + mtime = os.path.getmtime(filepath) + + with open(filepath, "r", encoding="utf-8", errors="replace") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + + ts = entry.get("timestamp") + if isinstance(ts, str): + ms = timestamp_to_ms(ts) + if ms > 0: + if first_ms == 0: + first_ms = ms + last_ms = ms + + if title == "Untitled Session" and entry.get("type") == "user": + msg = entry_message(entry) + text = extract_text(msg.get("content", [])) + if text: + candidate = first_title_line(text) + if candidate: + title = candidate + + messages.extend(_indexable_texts_from_entry(entry)) + + return session_id, title, first_ms, last_ms, mtime, messages + + +def build_search_index( + projects_dir: str, + rules: list[Any], + *, + force: bool = False, +) -> bool: + """Rebuild search index when fingerprint differs. Returns True if rebuilt.""" + if not index_search_enabled(): + return False + if not os.path.isdir(projects_dir): + return False + + fingerprint = _projects_fingerprint(projects_dir, rules) + + with _index_build_lock: + active_path = _resolve_active_index_db_path() + if not force and active_path is not None: + with _index_db_conn(readonly=True) as existing: + if existing is not None: + stored = _read_stored_fingerprint(existing) + if stored is not None and _fingerprints_match(stored, fingerprint): + return False + + root = cache_dir() + root.mkdir(parents=True, exist_ok=True) + new_path = root / f"search_index.{uuid.uuid4().hex[:12]}.sqlite" + session_count = 0 + message_count = 0 + + try: + with closing(sqlite3.connect(new_path)) as conn: + _create_schema(conn) + for project in list_projects(projects_dir): + for sess in list_sessions(project["path"]): + try: + session_id, title, first_ms, last_ms, mtime, texts = _scan_session_file( + sess["path"] + ) + except OSError as exc: + _logger.warning( + "Skipping session during index build: %s (%s)", + sess["path"], + exc, + ) + continue + if not texts: + continue + conn.execute( + "INSERT OR REPLACE INTO sessions" + " (session_id, project_name, title, first_ms, last_ms," + " file_path, mtime)" + " VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + session_id, + project["name"], + title, + first_ms, + last_ms, + sess["path"], + mtime, + ), + ) + session_count += 1 + for text, ts, role in texts: + conn.execute( + "INSERT INTO messages_fts" + " (session_id, project_name, role, timestamp_ms, text)" + " VALUES (?, ?, ?, ?, ?)", + ( + session_id, + project["name"], + role, + timestamp_to_ms(ts), + text, + ), + ) + message_count += 1 + + conn.execute( + "INSERT OR REPLACE INTO index_meta(key, value) VALUES (?, ?)", + ("fingerprint", json.dumps(fingerprint, ensure_ascii=False)), + ) + conn.execute( + "INSERT OR REPLACE INTO index_meta(key, value) VALUES (?, ?)", + ("version", str(INDEX_VERSION)), + ) + conn.execute( + "INSERT OR REPLACE INTO index_meta(key, value) VALUES (?, ?)", + ( + "stats", + json.dumps( + {"sessions": session_count, "messages": message_count}, + ensure_ascii=False, + ), + ), + ) + conn.commit() + + _publish_active_index(new_path) + _clear_usability_cache() + _logger.info( + "Search index rebuilt: %d sessions, %d messages -> %s", + session_count, + message_count, + new_path.name, + ) + return True + except Exception: + _logger.exception("Search index rebuild failed") + if new_path.is_file(): + try: + new_path.unlink() + except OSError: + pass + return False + + +def ensure_search_index(projects_dir: str, rules: list[Any]) -> None: + """Build index synchronously if missing or stale.""" + if not index_search_enabled(): + return + if _resolve_active_index_db_path() is None: + build_search_index(projects_dir, rules) + return + fingerprint = _projects_fingerprint(projects_dir, rules) + with _index_db_conn(readonly=True) as conn: + if conn is None: + build_search_index(projects_dir, rules) + return + stored = _read_stored_fingerprint(conn) + if stored is None or not _fingerprints_match(stored, fingerprint): + build_search_index(projects_dir, rules) + + +def start_search_index_background( + projects_dir: str, + rules: list[Any], + *, + poll_seconds: int = 60, +) -> None: + """Kick off initial + periodic index refresh in a daemon thread.""" + global _background_started + if not index_search_enabled(): + return + with _index_lock: + if _background_started: + return + _background_started = True + + def _worker() -> None: + while True: + try: + ensure_search_index(projects_dir, rules) + except Exception: + _logger.exception("Background search index refresh failed") + time.sleep(poll_seconds) + + thread = threading.Thread(target=_worker, name="search-index-refresh", daemon=True) + thread.start() + + +def index_is_usable(projects_dir: str, rules: list[Any]) -> bool: + """True when the on-disk index matches the current projects fingerprint.""" + if not index_search_enabled() or _resolve_active_index_db_path() is None: + return False + + key = _usability_cache_key(projects_dir, rules) + now = time.monotonic() + with _usability_cache_lock: + cached = _usability_cache.get(key) + if cached is not None and now - cached[1] < _USABILITY_CACHE_TTL_SECONDS: + return cached[0] + + fingerprint = _projects_fingerprint(projects_dir, rules) + with _index_db_conn(readonly=True) as conn: + if conn is None: + result = False + else: + stored = _read_stored_fingerprint(conn) + result = stored is not None and _fingerprints_match(stored, fingerprint) + + with _usability_cache_lock: + _usability_cache[key] = (result, now) + return result + + +def query_index_hits( + query_lower: str, + *, + since_ms: int | None, + max_results: int, + sql_offset: int = 0, +) -> IndexQueryResult: + """Return message hits from the FTS index (pre-exclusion, pre-snippet).""" + empty: IndexQueryResult = { + "hits": [], + "query_ok": True, + "sql_rows_fetched": 0, + "sql_exhausted": True, + } + if not query_lower or not index_search_enabled(): + return {"hits": [], "query_ok": False, "sql_rows_fetched": 0, "sql_exhausted": True} + + fts_q = _fts_match_query(query_lower) + if not fts_q: + return empty + + sql_limit = max(max_results, _FTS_BATCH_SIZE) + with _index_db_conn(readonly=True) as conn: + if conn is None: + return {"hits": [], "query_ok": False, "sql_rows_fetched": 0, "sql_exhausted": True} + try: + rows = conn.execute( + "SELECT m.session_id, m.project_name, m.role, m.timestamp_ms, m.text," + " s.title, s.file_path, s.mtime" + " FROM messages_fts m" + " JOIN sessions s" + " ON s.session_id = m.session_id AND s.project_name = m.project_name" + " WHERE messages_fts MATCH ?" + " ORDER BY m.timestamp_ms DESC" + " LIMIT ? OFFSET ?", + (fts_q, sql_limit, sql_offset), + ).fetchall() + except sqlite3.Error as exc: + _logger.debug("FTS query failed (%s); index may be rebuilding", exc) + return {"hits": [], "query_ok": False, "sql_rows_fetched": 0, "sql_exhausted": True} + + hits: list[IndexMessageHitDict] = [] + rows_scanned = 0 + for row in rows: + rows_scanned += 1 + text = row["text"] or "" + if query_lower not in text.lower(): + continue + ts_ms = int(row["timestamp_ms"] or 0) + if not timestamp_in_search_window_ms(ts_ms, since_ms): + continue + hits.append( + IndexMessageHitDict( + session_id=row["session_id"], + project_name=row["project_name"], + title=row["title"] or "Untitled Session", + role=row["role"], + timestamp=ms_to_timestamp(ts_ms), + text=text, + file_path=row["file_path"], + mtime=float(row["mtime"] or 0), + ) + ) + if len(hits) >= max_results: + break + return { + "hits": hits, + "query_ok": True, + "sql_rows_fetched": rows_scanned, + "sql_exhausted": len(rows) < sql_limit, + } + + +def reset_background_for_tests() -> None: + """Allow tests to restart the background worker.""" + global _background_started + with _index_lock: + _background_started = False + _clear_usability_cache()