fix(search): filter session logs from retrieval

Author: r266-tech

openviking/retrieve/hierarchical_retriever.py

@@ -14,7 +14,9 @@
 import time
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import unquote
 
+from openviking.core.namespace import is_session_uri, uri_parts
 from openviking.core.retrieval_targets import default_target_directories
 from openviking.models.embedder.base import EmbedResult, embed_compat
 from openviking.models.rerank import RerankClient
@@ -36,12 +38,163 @@
 
 logger = get_logger(__name__)
 
+_INTERNAL_SESSION_LOG_FILENAMES = frozenset({"messages.jsonl"})
+_INTERNAL_SESSION_LOG_SIDECAR_FILENAMES = frozenset({".abstract.md", ".overview.md"})
+_SESSION_LOG_FILTER_OVERFETCH_FACTOR = 4
+_SESSION_LOG_FILTER_MAX_DECODE_PASSES = 3
+
 
 class RetrieverMode(str):
     THINKING = "thinking"
     QUICK = "quick"
 
 
+def _is_internal_session_log_uri(uri: str) -> bool:
+    """Return True for internal session transcript files that should not surface in recall."""
+    normalized = uri.split("?", 1)[0].split("#", 1)[0].rstrip("/")
+    scheme = "viking://"
+    if not normalized.casefold().startswith(scheme):
+        return False
+
+    path = normalized[len(scheme) :]
+    for _ in range(_SESSION_LOG_FILTER_MAX_DECODE_PASSES):
+        decoded = unquote(path)
+        if decoded == path:
+            break
+        path = decoded
+    path = path.split("?", 1)[0].split("#", 1)[0]
+    normalized_uri = f"{scheme}{path.rstrip('/').casefold()}"
+    if not is_session_uri(normalized_uri):
+        return False
+
+    parts = uri_parts(normalized_uri)
+    root_depth = _session_root_depth(parts)
+    if root_depth is None or len(parts) <= root_depth:
+        return False
+
+    tail = parts[root_depth:]
+    return _is_internal_session_log_tail(tail)
+
+
+def _session_root_depth(parts: List[str]) -> Optional[int]:
+    if len(parts) >= 2 and parts[0] == "session":
+        return 2
+    if len(parts) >= 3 and parts[:2] == ["user", "sessions"]:
+        return 3
+    if len(parts) >= 4 and parts[0] == "user" and parts[2] == "sessions":
+        return 4
+    return None
+
+
+def _is_internal_session_log_tail(tail: List[str]) -> bool:
+    if _is_session_transcript_tail(tail):
+        return True
+    if len(tail) >= 3 and tail[0] == "history" and tail[1].startswith("archive_"):
+        return _is_session_transcript_tail(tail[2:])
+    return False
+
+
+def _is_session_transcript_tail(tail: List[str]) -> bool:
+    if tail == ["messages.jsonl"] or tail in (
+        [".abstract.md"],
+        [".overview.md"],
+    ):
+        return True
+    if (
+        len(tail) == 2
+        and tail[0] in _INTERNAL_SESSION_LOG_FILENAMES
+        and tail[1] in _INTERNAL_SESSION_LOG_SIDECAR_FILENAMES
+    ):
+        return True
+    return False
+
+
+def _session_log_filtered_fetch_limit(limit: int, minimum: int) -> int:
+    return max(limit * _SESSION_LOG_FILTER_OVERFETCH_FACTOR, minimum)
+
+
+async def _search_in_tenant_excluding_session_logs(
+    vector_proxy: VikingDBManagerProxy,
+    *,
+    desired_limit: int,
+    page_limit: int,
+    query_vector: Optional[List[float]],
+    sparse_query_vector: Optional[Dict[str, float]],
+    context_type: Optional[str],
+    target_directories: List[str],
+    extra_filter: Optional[FilterExpr | Dict[str, Any]],
+    level: Optional[List[int]],
+    max_scan_pages: int,
+) -> tuple[List[Dict[str, Any]], int, int]:
+    results: List[Dict[str, Any]] = []
+    searches = 0
+    scanned = 0
+    offset = 0
+    for _ in range(max_scan_pages):
+        page = await vector_proxy.search_in_tenant(
+            query_vector=query_vector,
+            sparse_query_vector=sparse_query_vector,
+            context_type=context_type,
+            target_directories=target_directories,
+            extra_filter=extra_filter,
+            level=level,
+            limit=page_limit,
+            offset=offset,
+        )
+        searches += 1
+        scanned += len(page)
+        results.extend(
+            result
+            for result in page
+            if not _is_internal_session_log_uri(str(result.get("uri", "")))
+        )
+        if len(results) >= desired_limit or len(page) < page_limit:
+            break
+        offset += page_limit
+    return results[:desired_limit], searches, scanned
+
+
+async def _search_children_excluding_session_logs(
+    vector_proxy: VikingDBManagerProxy,
+    *,
+    parent_uri: str,
+    desired_limit: int,
+    page_limit: int,
+    query_vector: Optional[List[float]],
+    sparse_query_vector: Optional[Dict[str, float]],
+    context_type: Optional[str],
+    target_directories: Optional[List[str]],
+    extra_filter: Optional[FilterExpr | Dict[str, Any]],
+    max_scan_pages: int,
+) -> tuple[List[Dict[str, Any]], int, int]:
+    results: List[Dict[str, Any]] = []
+    searches = 0
+    scanned = 0
+    offset = 0
+    for _ in range(max_scan_pages):
+        page = await vector_proxy.search_children_in_tenant(
+            parent_uri=parent_uri,
+            query_vector=query_vector,
+            sparse_query_vector=sparse_query_vector,
+            context_type=context_type,
+            target_directories=target_directories,
+            extra_filter=extra_filter,
+            limit=page_limit,
+            offset=offset,
+        )
+        searches += 1
+        scanned += len(page)
+        results.extend(
+            result
+            for result in page
+            if not _is_internal_session_log_uri(str(result.get("uri", "")))
+        )
+        if len(results) >= desired_limit or len(page) < page_limit:
+            break
+        offset += page_limit
+    return results[:desired_limit], searches, scanned
+
+
 class HierarchicalRetriever:
     """Hierarchical retriever with dense and sparse vector support."""
 
@@ -73,6 +226,9 @@ def __init__(
         self.retrieval_config = retrieval_config or RetrievalConfig()
         self.hotness_alpha = self.retrieval_config.hotness_alpha
         self.score_propagation_alpha = self.retrieval_config.score_propagation_alpha
+        self.session_log_filter_max_scan_pages = (
+            self.retrieval_config.session_log_filter_max_scan_pages
+        )
 
         # Use rerank threshold if available, otherwise use a default
         self.threshold = rerank_config.threshold if rerank_config else 0
@@ -151,19 +307,27 @@ async def retrieve(
         context_type = query.context_type.value if query.context_type else None
 
         if mode == RetrieverMode.QUICK:
+            quick_fetch_limit = _session_log_filtered_fetch_limit(limit, self.GLOBAL_SEARCH_TOPK)
             with telemetry.measure("search.vector_retrieval"):
-                quick_results = await vector_proxy.search_in_tenant(
+                (
+                    quick_results,
+                    search_count,
+                    scanned_count,
+                ) = await _search_in_tenant_excluding_session_logs(
+                    vector_proxy,
+                    desired_limit=max(limit, self.GLOBAL_SEARCH_TOPK),
+                    page_limit=quick_fetch_limit,
                     query_vector=query_vector,
                     sparse_query_vector=sparse_query_vector,
                     context_type=context_type,
                     target_directories=target_dirs,
                     extra_filter=scope_dsl,
                     level=level,
-                    limit=max(limit, self.GLOBAL_SEARCH_TOPK),
+                    max_scan_pages=self.session_log_filter_max_scan_pages,
                 )
-            telemetry.count("vector.searches", 1)
-            telemetry.count("vector.scored", len(quick_results))
-            telemetry.count("vector.scanned", len(quick_results))
+            telemetry.count("vector.searches", search_count)
+            telemetry.count("vector.scored", scanned_count)
+            telemetry.count("vector.scanned", scanned_count)
 
             collected_by_uri: Dict[str, Dict[str, Any]] = {}
             for result in quick_results:
@@ -192,19 +356,27 @@ async def retrieve(
             rerank_used = False
         else:
             # Step 2: Global vector search to supplement starting points
+            global_fetch_limit = max(limit, self.GLOBAL_SEARCH_TOPK)
             with telemetry.measure("search.vector_retrieval"):
-                global_results = await vector_proxy.search_in_tenant(
+                (
+                    global_results,
+                    search_count,
+                    scanned_count,
+                ) = await _search_in_tenant_excluding_session_logs(
+                    vector_proxy,
+                    desired_limit=global_fetch_limit,
+                    page_limit=global_fetch_limit,
                     query_vector=query_vector,
                     sparse_query_vector=sparse_query_vector,
                     context_type=context_type,
                     target_directories=target_dirs,
                     extra_filter=scope_dsl,
                     level=[0, 1],
-                    limit=max(limit, self.GLOBAL_SEARCH_TOPK),
+                    max_scan_pages=self.session_log_filter_max_scan_pages,
                 )
-            telemetry.count("vector.searches", 1)
-            telemetry.count("vector.scored", len(global_results))
-            telemetry.count("vector.scanned", len(global_results))
+            telemetry.count("vector.searches", search_count)
+            telemetry.count("vector.scored", scanned_count)
+            telemetry.count("vector.scanned", scanned_count)
 
             # Debug: Print all URIs in global_results
             if logger.isEnabledFor(logging.DEBUG):
@@ -389,7 +561,7 @@ async def _recursive_search(
         if initial_candidates:
             for r in initial_candidates:
                 uri = r.get("uri", "")
-                if not uri:
+                if not uri or _is_internal_session_log_uri(uri):
                     continue
                 if level is None or r.get("level", 2) in level:
                     score = self._finite_score(r.get("_score", 0.0))
@@ -408,17 +580,25 @@ async def _recursive_search(
 
         # Initialize: process starting points
         for uri, score in starting_points:
+            if _is_internal_session_log_uri(uri):
+                continue
             heapq.heappush(dir_queue, (-score, uri))
 
-        async def search_children(current_uri: str) -> List[Dict[str, Any]]:
-            return await vector_proxy.search_children_in_tenant(
+        child_desired_limit = max(limit * 2, 20)
+        child_fetch_limit = _session_log_filtered_fetch_limit(limit, 20)
+
+        async def search_children(current_uri: str) -> tuple[List[Dict[str, Any]], int, int]:
+            return await _search_children_excluding_session_logs(
+                vector_proxy,
                 parent_uri=current_uri,
+                desired_limit=child_desired_limit,
+                page_limit=child_fetch_limit,
                 query_vector=query_vector,
-                sparse_query_vector=sparse_query_vector,  # Pass sparse vector
+                sparse_query_vector=sparse_query_vector,
                 context_type=context_type,
                 target_directories=target_dirs,
                 extra_filter=scope_dsl,
-                limit=max(limit * 2, 20),
+                max_scan_pages=self.session_log_filter_max_scan_pages,
             )
 
         parallelism = max(1, self.MAX_PARALLEL_CHILD_SEARCHES)
@@ -442,10 +622,12 @@ async def search_children(current_uri: str) -> List[Dict[str, Any]]:
             )
 
             telemetry = get_current_telemetry()
-            for (_, current_score), results in zip(batch, batch_results, strict=True):
-                telemetry.count("vector.searches", 1)
-                telemetry.count("vector.scored", len(results))
-                telemetry.count("vector.scanned", len(results))
+            for (_, current_score), (results, search_count, scanned_count) in zip(
+                batch, batch_results, strict=True
+            ):
+                telemetry.count("vector.searches", search_count)
+                telemetry.count("vector.scored", scanned_count)
+                telemetry.count("vector.scanned", scanned_count)
 
                 if not results:
                     continue
@@ -457,6 +639,8 @@ async def search_children(current_uri: str) -> List[Dict[str, Any]]:
 
                 for r, score in zip(results, query_scores, strict=True):
                     uri = r.get("uri", "")
+                    if not uri or _is_internal_session_log_uri(uri):
+                        continue
                     final_score = (
                         alpha * score + (1 - alpha) * current_score if current_score else score
                     )

openviking/storage/viking_vector_index_backend.py

@@ -1125,6 +1125,7 @@ async def search_children_in_tenant(
         target_directories: Optional[List[str]] = None,
         extra_filter: Optional[FilterExpr | Dict[str, Any]] = None,
         limit: int = 10,
+        offset: int = 0,
     ) -> List[Dict[str, Any]]:
         # TODO：Better Alternative to Current Temporary Fix
 
@@ -1156,6 +1157,7 @@ async def search_children_in_tenant(
             sparse_query_vector=sparse_query_vector,
             filter=merged_filter,
             limit=limit,
+            offset=offset,
             output_fields=RETRIEVAL_OUTPUT_FIELDS,
             ctx=ctx,
         )