fix: search index completeness (exclusion pagination, parity, live backfill)

Author: clean6378-max-it

api/search.py

@@ -12,17 +12,16 @@
 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,
+    message_searchable_text,
     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
@@ -73,6 +72,35 @@ def _rank_search_hits(results: list[SearchHitDict]) -> list[SearchHitDict]:
     )
 
 
+def _hit_dedup_key(hit: SearchHitDict) -> tuple[str, str, str | None, str]:
+    ts = hit.get("timestamp")
+    return (
+        hit["project"],
+        hit["session_id"],
+        ts if isinstance(ts, str) else None,
+        hit["role"],
+    )
+
+
+def _merge_search_hits(
+    primary: list[SearchHitDict],
+    extra: list[SearchHitDict],
+    *,
+    max_results: int,
+) -> list[SearchHitDict]:
+    seen = {_hit_dedup_key(hit) for hit in primary}
+    merged = list(primary)
+    for hit in extra:
+        key = _hit_dedup_key(hit)
+        if key in seen:
+            continue
+        merged.append(hit)
+        seen.add(key)
+        if len(merged) >= max_results:
+            break
+    return _rank_search_hits(merged)[:max_results]
+
+
 def _projects_dir_inaccessible(projects_dir: str) -> bool:
     """True when the projects path exists but cannot be listed (503 case)."""
     try:
@@ -84,18 +112,6 @@ def _projects_dir_inaccessible(projects_dir: str) -> bool:
         return 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,
@@ -129,13 +145,14 @@ def _search_via_index(
     *,
     since_ms: int | None,
     max_results: int,
-) -> list[SearchHitDict] | None:
+) -> tuple[list[SearchHitDict] | None, bool]:
     if not index_search_enabled() or not index_is_usable(projects_dir, rules):
-        return None
+        return None, False
 
     rules_fp = rules_fingerprint(rules)
     results: list[SearchHitDict] = []
     sql_offset = 0
+    fts_exhausted = False
     while len(results) < max_results:
         need = max_results - len(results)
         indexed = query_index_hits(
@@ -150,11 +167,12 @@ def _search_via_index(
                 len(results),
             )
             if results:
-                return _rank_search_hits(results)[:max_results]
-            return None
+                return _rank_search_hits(results)[:max_results], False
+            return None, False
         if not indexed["query_ok"]:
-            return None
+            return None, False
         if indexed["sql_rows_fetched"] == 0:
+            fts_exhausted = indexed["sql_exhausted"]
             break
 
         for hit in indexed["hits"]:
@@ -181,8 +199,50 @@ def _search_via_index(
 
         sql_offset += indexed["sql_rows_fetched"]
         if indexed["sql_exhausted"]:
+            fts_exhausted = True
             break
-    return _rank_search_hits(results)[:max_results]
+    return _rank_search_hits(results)[:max_results], fts_exhausted
+
+
+def _resolve_search_results(
+    base: str,
+    rules: list[Any],
+    query: str,
+    query_lower: str,
+    *,
+    since_ms: int | None,
+    max_results: int,
+) -> list[SearchHitDict]:
+    indexed, _fts_exhausted = _search_via_index(
+        base,
+        rules,
+        query,
+        query_lower,
+        since_ms=since_ms,
+        max_results=max_results,
+    )
+    if indexed is None:
+        return _search_live_scan(
+            base,
+            rules,
+            query,
+            query_lower,
+            since_ms=since_ms,
+            max_results=max_results,
+        )
+
+    if len(indexed) >= max_results:
+        return indexed
+
+    live = _search_live_scan(
+        base,
+        rules,
+        query,
+        query_lower,
+        since_ms=since_ms,
+        max_results=max_results,
+    )
+    return _merge_search_hits(indexed, live, max_results=max_results)
 
 
 def _search_live_scan(
@@ -213,7 +273,7 @@ def _search_live_scan(
                 continue
 
             for msg in session["messages"]:
-                text = _message_searchable_text(msg)
+                text = message_searchable_text(msg)
                 if not text or query_lower not in text.lower():
                     continue
                 if not timestamp_in_search_window_iso(
@@ -289,19 +349,8 @@ def search() -> FlaskReturn:
     rules = current_app.config.get("EXCLUSION_RULES") or []
 
     try:
-        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(
+            _resolve_search_results(
                 base,
                 rules,
                 query,

tests/test_error_propagation.py

@@ -212,7 +212,7 @@ def test_search_internal_error_does_not_leak(client_single, monkeypatch):
     def _boom(*_args, **_kwargs):
         raise RuntimeError("internal_secret_search_token")
 
-    monkeypatch.setattr("api.search._search_via_index", lambda *_a, **_kw: None)
+    monkeypatch.setattr("api.search._search_via_index", lambda *_a, **_kw: (None, False))
     monkeypatch.setattr("api.search._search_live_scan", _boom)
 
     resp = client_single.get("/api/search?q=Hello&all_history=1")

tests/test_search_index.py

@@ -11,7 +11,8 @@
 
 import pytest
 
-from api.search import _search_via_index
+from api.search import _resolve_search_results, _search_via_index
+from utils.exclusion_rules import load_rules
 from utils.search_index import (
     build_search_index,
     index_is_usable,
@@ -314,7 +315,7 @@ def test_phrase_filter_paginates_past_decoy_token_matches(self, tmp_path, monkey
         patches = _index_patches(cache_root)
         with patches[0]:
             build_search_index(str(projects), [], force=True)
-            results = _search_via_index(
+            results, _fts_exhausted = _search_via_index(
                 str(projects),
                 [],
                 phrase,
@@ -386,6 +387,130 @@ def test_background_worker_starts_once(self, indexed_tree, monkeypatch):
             assert mock_thread.return_value.start.call_count == 1
 
 
+class TestIndexSearchCompleteness:
+    def test_partial_batch_not_marked_exhausted(self, tmp_path, monkeypatch):
+        """sql_exhausted stays false until every row in the SQL batch is scanned."""
+        cache_root = tmp_path / "cache"
+        cache_root.mkdir()
+        projects = tmp_path / "projects"
+        lines: list[dict[str, object]] = []
+        for i in range(120):
+            lines.append(
+                {
+                    "type": "user",
+                    "timestamp": f"2026-07-{(i % 28) + 1:02d}T10:00:00Z",
+                    "message": {
+                        "content": [{"type": "text", "text": f"batch sentinel token {i}"}],
+                    },
+                }
+            )
+        _write_session(projects / "batch-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)
+            result = query_index_hits(
+                "batch sentinel token",
+                since_ms=None,
+                max_results=50,
+                sql_offset=0,
+            )
+            assert result["query_ok"] is True
+            assert len(result["hits"]) == 50
+            assert result["sql_rows_fetched"] < 120
+            assert result["sql_exhausted"] is False
+
+    def test_index_search_fills_limit_after_excluded_hits(self, tmp_path, monkeypatch):
+        cache_root = tmp_path / "cache"
+        cache_root.mkdir()
+        projects = tmp_path / "projects"
+        term = "quota-fill-sentinel"
+        excl_lines: list[dict[str, object]] = []
+        for i in range(80):
+            excl_lines.append(
+                {
+                    "type": "user",
+                    "timestamp": f"2026-08-{(i % 28) + 1:02d}T10:00:00Z",
+                    "message": {"content": [{"type": "text", "text": f"{term} excluded"}]},
+                }
+            )
+        good_lines: list[dict[str, object]] = []
+        for i in range(60):
+            good_lines.append(
+                {
+                    "type": "user",
+                    "timestamp": f"2026-05-{(i % 28) + 1:02d}T10:00:00Z",
+                    "message": {"content": [{"type": "text", "text": f"{term} good"}]},
+                }
+            )
+        _write_session(projects / "excl-proj" / "session.jsonl", excl_lines)
+        _write_session(projects / "good-proj" / "session.jsonl", good_lines)
+        rules_path = tmp_path / "rules.txt"
+        rules_path.write_text("excl-proj\n", encoding="utf-8")
+        rules = load_rules(str(rules_path))
+
+        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), rules, force=True)
+            results, _fts_exhausted = _search_via_index(
+                str(projects),
+                rules,
+                term,
+                term.lower(),
+                since_ms=None,
+                max_results=50,
+            )
+            assert results is not None
+            assert len(results) == 50
+            assert all(hit["project"] == "good-proj" for hit in results)
+
+    def test_system_content_index_and_live_scan_parity(self, tmp_path, monkeypatch):
+        cache_root = tmp_path / "cache"
+        cache_root.mkdir()
+        projects = tmp_path / "projects"
+        sentinel = "system-role-sentinel-xyz"
+        _write_session(
+            projects / "sys-proj" / "session.jsonl",
+            [
+                {
+                    "type": "user",
+                    "timestamp": "2026-06-01T10:00:00Z",
+                    "message": {"content": [{"type": "text", "text": "hello"}]},
+                },
+                {
+                    "type": "system",
+                    "timestamp": "2026-06-01T10:00:01Z",
+                    "content": f"compact boundary note {sentinel}",
+                },
+                {
+                    "type": "assistant",
+                    "timestamp": "2026-06-01T10:00:02Z",
+                    "message": {"content": [{"type": "text", "text": "ack"}]},
+                },
+            ],
+        )
+        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)
+            indexed = query_index_hits(sentinel, since_ms=None, max_results=5)
+            assert indexed["query_ok"] is True
+            assert len(indexed["hits"]) == 1
+            assert indexed["hits"][0]["role"] == "system"
+
+            resolved = _resolve_search_results(
+                str(projects),
+                [],
+                sentinel,
+                sentinel.lower(),
+                since_ms=None,
+                max_results=5,
+            )
+            assert len(resolved) == 1
+            assert resolved[0]["role"] == "system"
+
+
 class TestResolveSearchSinceMs:
     def test_all_history_none(self):
         assert resolve_search_since_ms(all_history=True) is None