From 14d9a31f9dfd2f16286cd8a4665cc10ed95dcf16 Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 15 Sep 2026 00:05:22 -0500 Subject: [PATCH 1/2] feat(core): add explicit multi-project API search Signed-off-by: phernandez --- src/basic_memory/api/app.py | 4 + .../v2/routers/multi_project_search_router.py | 184 ++++ .../multi_project_search_repository.py | 292 +++++++ .../repository/note_type_filters.py | 7 +- .../repository/postgres_search_query.py | 798 ++++++++++++++++++ .../repository/postgres_search_repository.py | 767 +---------------- src/basic_memory/repository/search_query.py | 25 + .../repository/sqlite_search_query.py | 649 ++++++++++++++ .../repository/sqlite_search_repository.py | 610 +------------ .../repository/temporal_filters.py | 10 +- .../schemas/multi_project_search.py | 28 + src/basic_memory/services/search_service.py | 29 +- test-int/test_multi_project_search.py | 531 ++++++++++++ 13 files changed, 2538 insertions(+), 1396 deletions(-) create mode 100644 src/basic_memory/api/v2/routers/multi_project_search_router.py create mode 100644 src/basic_memory/repository/multi_project_search_repository.py create mode 100644 src/basic_memory/repository/postgres_search_query.py create mode 100644 src/basic_memory/repository/sqlite_search_query.py create mode 100644 src/basic_memory/schemas/multi_project_search.py create mode 100644 test-int/test_multi_project_search.py diff --git a/src/basic_memory/api/app.py b/src/basic_memory/api/app.py index 5e2cb8cf6..b8cfd6481 100644 --- a/src/basic_memory/api/app.py +++ b/src/basic_memory/api/app.py @@ -9,6 +9,9 @@ from basic_memory import __version__ as version from basic_memory.api.container import ApiContainer, set_container +from basic_memory.api.v2.routers.multi_project_search_router import ( + router as multi_project_search_router, +) from basic_memory.api.v2.routers import ( accepted_content_router as v2_accepted_content, knowledge_router as v2_knowledge, @@ -136,6 +139,7 @@ async def workspace_permalink_context_middleware(request: Request, call_next): app.include_router(v2_schema, prefix="/v2/projects/{project_id}") app.include_router(v2_inspect, prefix="/v2/projects/{project_id}") app.include_router(v2_project, prefix="/v2") +app.include_router(multi_project_search_router, prefix="/v2") # Legacy web app proxy paths (compat with /proxy/projects/projects) app.include_router(v2_project, prefix="/proxy/projects") diff --git a/src/basic_memory/api/v2/routers/multi_project_search_router.py b/src/basic_memory/api/v2/routers/multi_project_search_router.py new file mode 100644 index 000000000..03b81999c --- /dev/null +++ b/src/basic_memory/api/v2/routers/multi_project_search_router.py @@ -0,0 +1,184 @@ +"""Database-scoped search for trusted API callers supplying effective project IDs. + +Authorization belongs to Cloud. This endpoint never discovers an implicit scope. +Caching is deliberately bypassed: the existing cache owns one project generation. +Milvus remains available through project search; this route supports semantic +retrieval only over shared sqlite-vec/pgvector storage, without project fan-out. +""" + +from dataclasses import replace + +from fastapi import APIRouter, HTTPException, Query, Response +from sqlalchemy import select, tuple_ + +from basic_memory import db +from basic_memory.api.v2.utils import _temporal_result_metadata +from basic_memory.config import DatabaseBackend +from basic_memory.deps.config import AppConfigDep +from basic_memory.deps.db import SessionMakerDep +from basic_memory.models import Entity, MemoryTimeIndex, Project +from basic_memory.repository.embedding_provider_factory import create_embedding_provider +from basic_memory.repository.multi_project_search_repository import ( + MultiProjectSearchRepository, + UnsupportedMultiProjectVectorIndexError, +) +from basic_memory.repository.semantic_errors import ( + SemanticDependenciesMissingError, + SemanticSearchDisabledError, +) +from basic_memory.schemas.base import normalize_note_type +from basic_memory.schemas.multi_project_search import ( + MultiProjectSearchQuery, + MultiProjectSearchResponse, + MultiProjectSearchResult, +) +from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode, TemporalResultMetadata +from basic_memory.services.search_service import SearchService + +router = APIRouter(tags=["search"]) + + +@router.api_route("/search/", methods=["QUERY"], response_model=MultiProjectSearchResponse) +async def search( + query: MultiProjectSearchQuery, + config: AppConfigDep, + session_maker: SessionMakerDep, + response: Response, + page: int = Query(1, ge=1), + page_size: int = Query(10, ge=1, le=1000), +) -> MultiProjectSearchResponse: + """Search one tenant database using an explicit, already-authorized project set.""" + response.headers["Accept-Query"] = "application/json" + response.headers["Cache-Control"] = "no-store" + exact = query.retrieval_mode == SearchRetrievalMode.FTS + result = MultiProjectSearchResponse( + results=[], + current_page=page, + page_size=page_size, + total_is_exact=exact, + temporal_applied=True if query.has_temporal_filter() else None, + ) + try: + prepared = SearchService.prepare_query(query) + if not query.project_ids or prepared is None: + return result + provider = None + compatible_adapter = ( + config.database_backend == DatabaseBackend.SQLITE + or config.semantic_vector_index == "pgvector" + ) + if not exact and compatible_adapter and config.semantic_search_enabled: + provider = create_embedding_provider(config) + repository = MultiProjectSearchRepository( + session_maker, + query.project_ids, + app_config=config, + embedding_provider=provider, + ) + if prepared.note_types: + async with db.scoped_session(session_maker) as session: + stored = await session.scalars( + select(Entity.note_type) + .where(Entity.project_id.in_(repository.project_ids)) + .distinct() + ) + canonical = set(prepared.note_types) + compatible = canonical | { + value for value in stored if value and normalize_note_type(value) in canonical + } + prepared = replace(prepared, note_types=sorted(compatible)) + offset = (page - 1) * page_size + rows = await repository.search( + prepared, limit=page_size if exact else page_size + 1, offset=offset + ) + if exact: + result.total = await repository.count(prepared) + result.has_more = offset + len(rows) < result.total + else: + result.has_more = len(rows) > page_size + rows = rows[:page_size] + except UnsupportedMultiProjectVectorIndexError as exc: + raise HTTPException( + status_code=400, + detail={"code": "unsupported_multi_project_vector_adapter", "message": str(exc)}, + ) from exc + except (SemanticDependenciesMissingError, SemanticSearchDisabledError, ValueError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + if not rows: + return result + # Batch hydration retains scope even for relation targets outside the visible set. + # No project service fan-out, and no hidden unrestricted entity lookup. + entity_ids = { + entity_id + for row in rows + for entity_id in (row.entity_id, row.from_id, row.to_id) + if entity_id is not None + } + temporal: dict[tuple[int, str, int], list[TemporalResultMetadata]] = {} + async with db.scoped_session(session_maker) as session: + entities = await session.scalars( + select(Entity).where( + Entity.project_id.in_(repository.project_ids), Entity.id.in_(entity_ids) + ) + ) + entities_by_key = {(entity.project_id, entity.id): entity for entity in entities} + projects = await session.scalars( + select(Project).where(Project.id.in_({row.project_id for row in rows})) + ) + projects_by_id = {project.id: project.external_id for project in projects} + if query.has_temporal_filter(): + assertions = await session.scalars( + select(MemoryTimeIndex) + .where( + tuple_( + MemoryTimeIndex.project_id, + MemoryTimeIndex.source_type, + MemoryTimeIndex.source_id, + ).in_([(row.project_id, row.type, row.id) for row in rows]) + ) + .order_by(MemoryTimeIndex.id) + ) + for assertion in assertions: + key = (assertion.project_id, assertion.source_type, assertion.source_id) + temporal.setdefault(key, []).append(_temporal_result_metadata(assertion)) + for row in rows: + owner = ( + entities_by_key.get((row.project_id, row.entity_id)) + if row.entity_id is not None + else None + ) + source = ( + entities_by_key.get((row.project_id, row.from_id)) if row.from_id is not None else None + ) + target = entities_by_key.get((row.project_id, row.to_id)) if row.to_id is not None else None + result.results.append( + MultiProjectSearchResult( + project_id=row.project_id, + project_external_id=projects_by_id[row.project_id], + title=row.title or "", + type=SearchItemType(row.type), + score=row.score or 0.0, + permalink=row.permalink, + entity=owner.permalink if owner else None, + external_id=owner.external_id if owner else None, + content=row.content, + content_length=row.content_length, + content_truncated=row.content_truncated, + matched_chunk=row.matched_chunk_text, + file_path=row.file_path, + updated_at=row.updated_at, + metadata=row.metadata, + entity_id=row.entity_id, + observation_id=row.id if row.type == SearchItemType.OBSERVATION else None, + relation_id=row.id if row.type == SearchItemType.RELATION else None, + category=row.category, + from_entity=source.permalink if source else None, + to_entity=target.permalink if target else None, + relation_type=row.relation_type, + temporal=temporal.get((row.project_id, row.type, row.id)) + if query.has_temporal_filter() + else None, + ) + ) + return result diff --git a/src/basic_memory/repository/multi_project_search_repository.py b/src/basic_memory/repository/multi_project_search_repository.py new file mode 100644 index 000000000..f7d26a006 --- /dev/null +++ b/src/basic_memory/repository/multi_project_search_repository.py @@ -0,0 +1,292 @@ +"""Read-only, database-scoped search over an explicit effective project set. + +Shared-storage sqlite-vec and pgvector support one vector query. Project-isolated +external indexes (including Milvus) remain supported by the single-project API; +this reader rejects them for vector/hybrid retrieval without opening an adapter. +""" + +import json +from collections.abc import Sequence +from dataclasses import replace +from typing import Any + +import logfire +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.config import BasicMemoryConfig, DatabaseBackend +from basic_memory.repository.embedding_provider import EmbeddingProvider +from basic_memory.repository.postgres_search_query import PostgresSearchQuery +from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.search_query import PreparedSearchQuery +from basic_memory.repository.search_repository_base import FUSION_BONUS, TOP_CHUNKS_PER_RESULT +from basic_memory.repository.semantic_errors import SemanticSearchDisabledError +from basic_memory.repository.semantic_vector_index_factory import semantic_embedding_identity +from basic_memory.repository.sqlite_search_query import SQLiteSearchQuery +from basic_memory.schemas.search import SearchRetrievalMode + + +# Exclude backend index columns and content_stems from page hydration. +_RESULT_COLUMNS = ( + "project_id", + "id", + "title", + "permalink", + "file_path", + "type", + "metadata", + "from_id", + "to_id", + "relation_type", + "entity_id", + "content_snippet", + "category", + "created_at", + "updated_at", +) + + +class UnsupportedMultiProjectVectorIndexError(ValueError): + """The configured adapter cannot retrieve an explicit project set in one query.""" + + +class MultiProjectSearchRepository: + """Retrieve and rank one database scope; never initialize or mutate its indexes.""" + + def __init__( + self, + session_maker: async_sessionmaker[AsyncSession], + project_ids: Sequence[int], + *, + app_config: BasicMemoryConfig, + embedding_provider: EmbeddingProvider | None = None, + ) -> None: + # tuple(None) raises instead of interpreting absence as unrestricted access. + ids = tuple(project_ids) + if any(type(project_id) is not int or project_id <= 0 for project_id in ids): + raise ValueError("Project IDs must be positive integers") + self.project_ids = tuple(sorted(set(ids))) + self.session_maker = session_maker + self.config = app_config + self.provider = embedding_provider + self.postgres = app_config.database_backend == DatabaseBackend.POSTGRES + self.compiler = ( + PostgresSearchQuery(session_maker, self.project_ids) + if self.postgres + else SQLiteSearchQuery(session_maker, self.project_ids) + ) + + async def _parts( + self, query: PreparedSearchQuery, *, lexical: bool + ) -> tuple[str, str, dict[str, Any], str, str]: + # Filters are shared with the project API. Vector candidates are restricted + # before ranking, so disallowed projects/rows never occupy a top-k window. + return await self.compiler._build_fts_query_parts( + search_text=query.search_text if lexical else None, + permalink=query.permalink, + permalink_match=query.permalink_match, + title=query.title, + note_types=query.note_types, + after_date=query.after_date, + search_item_types=query.search_item_types, + categories=query.categories, + metadata_filters=query.metadata_filters, + file_path_prefix=query.file_path_prefix, + temporal=query.temporal, + ) + + async def count(self, query: PreparedSearchQuery) -> int: + if query.retrieval_mode != SearchRetrievalMode.FTS: + raise ValueError("Exact counts are only supported for full-text search retrieval.") + if not self.project_ids: + return 0 + source, where, params, _, _ = await self._parts(query, lexical=True) + async with db.scoped_session(self.session_maker) as session: + result = await session.execute( + text(f"SELECT COUNT(*) FROM {source} WHERE {where}"), params + ) + return int(result.scalar_one()) + + async def search( + self, query: PreparedSearchQuery, *, limit: int = 10, offset: int = 0 + ) -> list[SearchIndexRow]: + if limit <= 0 or offset < 0: + raise ValueError("Search limit must be positive and offset must be nonnegative") + if not self.project_ids: + return [] + mode = query.retrieval_mode + result_columns = ", ".join(f"search_index.{name}" for name in _RESULT_COLUMNS) + if mode == SearchRetrievalMode.FTS: + source, where, params, order, score = await self._parts(query, lexical=True) + direction = "DESC" if self.postgres else "ASC" + sql = f""" + SELECT {result_columns}, {score} AS score + FROM {source} WHERE {where} + ORDER BY score {direction} {order}, search_index.project_id, + search_index.type, search_index.id + LIMIT :limit OFFSET :offset + """ + params.update(limit=limit, offset=offset) + async with db.scoped_session(self.session_maker) as session: + with logfire.span("search.fts", project_count=len(self.project_ids)): + result = await session.execute(text(sql), params) + return [SearchIndexRow.from_mapping(row._asdict()) for row in result] + + expected_index = "pgvector" if self.postgres else "sqlite-vec" + configured_index = self.config.semantic_vector_index if self.postgres else "sqlite-vec" + if configured_index != expected_index: + raise UnsupportedMultiProjectVectorIndexError( + f"Database-scoped vector/hybrid search does not support adapter '{configured_index}'. " + "Use pgvector or sqlite-vec, or request FTS. Single-project search remains supported." + ) + if not self.config.semantic_search_enabled or self.provider is None: + raise SemanticSearchDisabledError("Semantic search is disabled") + if not query.search_text or not query.search_text.strip(): + raise ValueError("Vector/hybrid search requires nonempty text") + + with logfire.span("search.embed_query", project_count=len(self.project_ids)): + embedding = await self.provider.embed_query(query.search_text.strip()) + if len(embedding) != self.provider.dimensions: + raise ValueError("Query dimensions do not match the configured embedding provider") + + source, where, params, _, _ = await self._parts(query, lexical=False) + params.update( + query_vector=json.dumps(embedding), + embedding_model=semantic_embedding_identity(self.provider), + vector_index=expected_index, + dimensions=self.provider.dimensions, + min_similarity=query.min_similarity + if query.min_similarity is not None + else self.config.semantic_min_similarity, + limit=limit, + offset=offset, + top_chunks=TOP_CHUNKS_PER_RESULT, + ) + distance = ( + "e.embedding <=> CAST(:query_vector AS vector)" + if self.postgres + else "vec_distance_L2(e.embedding, :query_vector)" + ) + similarity = ( + f"1 - ({distance})" if self.postgres else f"1 - ({distance}) * ({distance}) / 2.0" + ) + vector_join = ( + "e.chunk_id = c.id AND e.project_id = c.project_id" + if self.postgres + else "e.rowid = c.id" + ) + dimension_check = ( + "e.embedding_dims = :dimensions" + if self.postgres + else "vec_length(e.embedding) = :dimensions" + ) + # Rank the complete eligible set in SQL before paginating. Unlike a page-sized + # candidate pool, this keeps fusion membership/normalization stable on deep pages. + # Only page rows and their top chunks cross the database boundary. + ctes = f""" + eligible AS MATERIALIZED ( + SELECT search_index.project_id, search_index.type, search_index.id, + search_index.entity_id FROM {source} WHERE {where} + ), + vector_chunks AS MATERIALIZED ( + SELECT eligible.project_id, eligible.type, eligible.id, + c.chunk_key, c.chunk_text, {similarity} AS similarity + FROM eligible + JOIN search_vector_chunks c ON c.project_id = eligible.project_id + AND c.entity_id = eligible.entity_id + AND c.chunk_key LIKE eligible.type || ':' || CAST(eligible.id AS TEXT) || ':%' + JOIN search_vector_embeddings e ON {vector_join} + WHERE c.embedding_status = 'ready' AND c.vector_index = :vector_index + AND c.embedding_model = :embedding_model AND {dimension_check} + AND e.source_hash = c.source_hash + ), + vector_scores AS ( + SELECT project_id, type, id, + CASE WHEN MAX(similarity) > 1 THEN 1 ELSE MAX(similarity) END AS score + FROM vector_chunks WHERE similarity >= :min_similarity + GROUP BY project_id, type, id + ) + """ + ranking = "SELECT project_id, type, id, score FROM vector_scores" + if mode == SearchRetrievalMode.HYBRID: + fts_source, fts_where, fts_params, _, fts_score = await self._parts(query, lexical=True) + params.update(fts_params) + ctes += f""", + fts AS MATERIALIZED ( + SELECT search_index.project_id, search_index.type, search_index.id, + ABS({fts_score}) AS score FROM {fts_source} WHERE {fts_where} + ), + channels AS ( + SELECT project_id, type, id, score AS vector_score, 0.0 AS fts_score + FROM vector_scores + UNION ALL + SELECT project_id, type, id, 0.0, + COALESCE(score / NULLIF(MAX(score) OVER (), 0), 0) FROM fts + ) + """ + params["fusion_bonus"] = FUSION_BONUS + ranking = """ + SELECT project_id, type, id, + CASE WHEN MAX(vector_score) > MAX(fts_score) + THEN MAX(vector_score) + :fusion_bonus * MAX(fts_score) + ELSE MAX(fts_score) + :fusion_bonus * MAX(vector_score) END AS score + FROM channels GROUP BY project_id, type, id + """ + page_columns = ", ".join(f"s.{name}" for name in _RESULT_COLUMNS) + sql = f""" + WITH {ctes}, ranked AS ({ranking}), + page AS MATERIALIZED ( + SELECT * FROM ranked ORDER BY score DESC, project_id, type, id + LIMIT :limit OFFSET :offset + ), + page_chunks AS ( + SELECT c.*, ROW_NUMBER() OVER ( + PARTITION BY c.project_id, c.type, c.id + ORDER BY c.similarity DESC, c.chunk_key + ) AS chunk_rank + FROM vector_chunks c JOIN page p + ON (c.project_id, c.type, c.id) = (p.project_id, p.type, p.id) + ) + SELECT {page_columns}, p.score, c.chunk_text + FROM page p JOIN search_index s + ON (s.project_id, s.type, s.id) = (p.project_id, p.type, p.id) + LEFT JOIN page_chunks c + ON (c.project_id, c.type, c.id) = (p.project_id, p.type, p.id) + AND c.chunk_rank <= :top_chunks + ORDER BY p.score DESC, p.project_id, p.type, p.id, c.chunk_rank + """ + async with db.scoped_session(self.session_maker) as session: + if not self.postgres: + # Connection setup only: this reader never calls adapter.initialize(), + # which may recreate storage and invalidate manifests on schema mismatch. + import sqlite_vec + + connection = await session.connection() + raw = await connection.get_raw_connection() + driver = raw.driver_connection + assert driver is not None + await driver.enable_load_extension(True) + try: + await driver.load_extension(sqlite_vec.loadable_path()) + finally: + await driver.enable_load_extension(False) + with logfire.span( + "search.vector_query", + project_count=len(self.project_ids), + retrieval_mode=mode.value, + ): + result = await session.execute(text(sql), params) + rows: dict[tuple[int, str, int], SearchIndexRow] = {} + chunks: dict[tuple[int, str, int], list[str]] = {} + for record in result: + mapping = record._asdict() + row = SearchIndexRow.from_mapping(mapping) + key = (row.project_id, row.type, row.id) + rows.setdefault(key, row) + if mapping["chunk_text"] is not None: + chunks.setdefault(key, []).append(mapping["chunk_text"]) + return [ + replace(row, matched_chunk_text="\n---\n".join(chunks[key]) if key in chunks else None) + for key, row in rows.items() + ] diff --git a/src/basic_memory/repository/note_type_filters.py b/src/basic_memory/repository/note_type_filters.py index 9da86df09..55386bfb7 100644 --- a/src/basic_memory/repository/note_type_filters.py +++ b/src/basic_memory/repository/note_type_filters.py @@ -50,6 +50,7 @@ def build_note_type_predicate( params: dict[str, Any], *, note_type_value: str, + project_scope_sql: str = "= :project_id", ) -> str: """Build the WHERE-clause fragment restricting rows to notes of the given types. @@ -66,10 +67,10 @@ def build_note_type_predicate( placeholders.append(f":{name}") return ( - f"{SEARCH_TABLE}.entity_id IN (\n" - f" SELECT {_OWNER}.id\n" + f"({SEARCH_TABLE}.project_id, {SEARCH_TABLE}.entity_id) IN (\n" + f" SELECT {_OWNER}.project_id, {_OWNER}.id\n" f" FROM {SEARCH_TABLE} AS {_OWNER}\n" f" WHERE {_OWNER}.type = '{SearchItemType.ENTITY.value}'\n" - f" AND {_OWNER}.project_id = :project_id\n" + f" AND {_OWNER}.project_id {project_scope_sql}\n" f" AND LOWER({note_type_value}) IN ({', '.join(placeholders)}))" ) diff --git a/src/basic_memory/repository/postgres_search_query.py b/src/basic_memory/repository/postgres_search_query.py new file mode 100644 index 000000000..5fe10f33f --- /dev/null +++ b/src/basic_memory/repository/postgres_search_query.py @@ -0,0 +1,798 @@ +"""Read-only FTS query compilation for explicit project scopes.""" + +import json +import re +from collections.abc import Sequence +from datetime import datetime +from typing import Any, List, Optional + +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory.repository.search_query import relaxed_query_words, relaxation_word_tokens +from basic_memory.repository.script_ngrams import analyze_script_query +from basic_memory.repository.search_repository_base import ( + SearchIndexKey, + candidate_key_restriction_condition, + file_path_prefix_condition, + metadata_contains_like_condition, + metadata_filter_content_type_condition, +) +from basic_memory.repository.metadata_filters import parse_metadata_filters +from basic_memory.repository.note_type_filters import ( + POSTGRES_NOTE_TYPE_VALUE, + build_note_type_predicate, +) +from basic_memory.repository.temporal_filters import build_temporal_predicate +from basic_memory.schemas.search import SearchItemType +from basic_memory.temporal import TemporalFilter + + +_TSQUERY_OPERAND_PATTERN = re.compile(r"'(?:''|[^'])*'(?::\*)?|[^\s&|!()]+") +_TSQUERY_WORD_PATTERN = re.compile(r"[^\W_]+(?:'[^\W_]+)?", re.UNICODE) +_QUOTED_QUERY_PATTERN = re.compile(r'"([^"]*)"') +_BOOLEAN_WORDS = frozenset({"AND", "OR", "NOT"}) +_TSQUERY_METACHARACTERS = frozenset("&|!:<>") + + +def _tsquery_operands(processed_text: str) -> list[tuple[str, str]]: + """Return unique (query operand, representative text) pairs in source order.""" + operands: dict[str, str] = {} + for operand in _TSQUERY_OPERAND_PATTERN.findall(processed_text): + representative = operand.removesuffix(":*") + if representative.startswith("'") and representative.endswith("'"): + representative = representative[1:-1].replace("''", "'") + operands.setdefault(operand, representative) + continue + + # An unquoted apostrophe is invalid tsquery syntax. Keep the literal + # word for the synthetic document, but quote and escape its probe so a + # strict syntax failure can proceed to the relaxed retry. + if "'" in representative: + escaped = "'{}'".format(representative.replace("'", "''")) + safe_operand = f"{escaped}:*" if operand.endswith(":*") else escaped + operands.setdefault(safe_operand, representative) + continue + + # PostgreSQL legitimately parses punctuation inside operands such as + # ``v0.13.0b2:*`` and ``auth-service:*``. Preserve those bytes so the + # synthetic document is tokenized the same way as the original note. + if "<" not in representative and ">" not in representative: + operands.setdefault(operand, representative) + continue + + # A malformed strict operand (for example ``foo str: + """Render user text as complete, individually escaped tsquery operands.""" + words = relaxation_word_tokens(text_value) + if drop_boolean_words: + words = [word for word in words if word.upper() not in _BOOLEAN_WORDS] + if not words: + return "NOSPECIALCHARS:*" + + operands = [] + for word in words: + escaped_word = "'{}'".format(word.replace("'", "''")) if "'" in word else word + operands.append(f"{escaped_word}:*" if is_prefix else escaped_word) + return operator.join(operands) + + +def _render_boolean_operand(operand: str) -> str: + """Preserve safe structured text while escaping tsquery syntax bytes.""" + if "'" not in operand and not any(char in _TSQUERY_METACHARACTERS for char in operand): + return operand + return _render_tsquery_words( + operand, + operator=" & ", + is_prefix=False, + ) + + +def _has_valid_boolean_shape(expression: str) -> bool: + """Reject incomplete operator structure before it reaches strict ``to_tsquery``.""" + depth = 0 + for char in expression: + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth < 0: + return False + if depth: + return False + + stripped = expression.strip() + if not stripped or stripped[0] in "&|" or stripped[-1] in "&|!": + return False + return not any( + re.search(pattern, stripped) + for pattern in ( + r"[&|]\s*[&|]", + r"!\s*[&|)]", + r"\(\s*[&|)]", + r"[&|!(]\s*\)", + ) + ) + + +class PostgresSearchQuery: + """Compile FTS and filter predicates without owning index mutations.""" + + def __init__( + self, + session_maker: async_sessionmaker[AsyncSession], + project_ids: Sequence[int], + ) -> None: + self.session_maker = session_maker + self._entity_columns: set[str] | None = None + # A missing scope is never unrestricted. Empty scopes compile to no rows. + ids = tuple(project_ids) + if any(type(project_id) is not int or project_id <= 0 for project_id in ids): + raise ValueError("Project IDs must be positive integers") + ids = tuple(sorted(set(ids))) + self._scope_params = {f"scope_{index}": value for index, value in enumerate(ids)} + self._scope_sql = ( + "IN (" + ", ".join(f":{key}" for key in self._scope_params) + ")" + if ids + else "IN (NULL)" + ) + + def _prepare_search_term(self, term: str, is_prefix: bool = True) -> str: + """Prepare a search term for tsquery format. + + Args: + term: The search term to prepare + is_prefix: Whether to add prefix search capability (:* operator) + + Returns: + Formatted search term for tsquery + + For Postgres: + - Boolean operators are converted to tsquery format (&, |, !) + - Prefix matching uses the :* operator + - Terms are sanitized to prevent tsquery syntax errors + """ + # Check for explicit boolean operators + boolean_operators = [" AND ", " OR ", " NOT "] + if '"' in term or any(op in f" {term} " for op in boolean_operators): + return self._prepare_boolean_query(term) + + # For non-Boolean queries, prepare single term + return self._prepare_single_term(term, is_prefix) + + @staticmethod + def _relaxed_tsquery_term(word: str) -> str: + """Render one relaxed word as a tsquery-safe prefix expression. + + Mirrors the SQLite renderer: a word token can contain an apostrophe, and + tsquery reads that as lexeme-quoting syntax rather than text. Quoting the + lexeme and doubling any interior quote keeps it literal. + """ + if "'" in word: + return "'{}':*".format(word.replace("'", "''")) + return f"{word}:*" + + @staticmethod + def _relaxed_tsquery_text(search_text: Optional[str]) -> Optional[str]: + """OR-relaxed tsquery expression for a failed strict query, or None.""" + words = relaxed_query_words(search_text) + if not words: + return None + return " | ".join(PostgresSearchQuery._relaxed_tsquery_term(word) for word in words) + + def _prepare_boolean_query(self, query: str) -> str: + """Convert Boolean query to tsquery format. + + Args: + query: A Boolean query like "coffee AND brewing" or "(pour OR french) AND press" + + Returns: + tsquery-formatted string with & (AND), | (OR), ! (NOT) operators + + Examples: + "coffee AND brewing" -> "coffee & brewing" + "(pour OR french) AND press" -> "(pour | french) & press" + "coffee NOT decaf" -> "coffee & !decaf" + """ + # PostgreSQL's strict to_tsquery grammar does not accept web-style double + # quotes. Convert complete quoted groups first so operator-looking words + # inside them remain text and every word becomes a complete operand. + quoted_phrases: dict[str, str] = {} + + def replace_quoted_phrase(match: re.Match[str]) -> str: + phrase = _render_tsquery_words( + match.group(1), + operator=" & ", + is_prefix=False, + ) + placeholder = f"BMQUOTEDPHRASE{len(quoted_phrases)}" + while placeholder in query: + placeholder += "X" + quoted_phrases[placeholder] = f"({phrase})" + # Surround the placeholder so quotes adjacent to plain text become + # explicit operands instead of restoring into ``word(group)``. + return f" {placeholder} " + + result = _QUOTED_QUERY_PATTERN.sub(replace_quoted_phrase, query) + if '"' in result: + # An unmatched quote is user text, not a reason to abort the database + # transaction. Boolean-looking words lose their operator role here. + return _render_tsquery_words( + query, + operator=" & ", + is_prefix=True, + drop_boolean_words=True, + ) + + # Boolean syntax is the only structure retained from user input. A + # whitespace-delimited operand still needs explicit conjunctions, but + # PostgreSQL must tokenize structured single operands such as + # ``auth-service`` and ``config.json`` exactly as it did before quoted + # query normalization. + normalized_parts: list[str] = [] + operator_pattern = r"((? str: + """Prepare a single search term for tsquery. + + Args: + term: A single search term + is_prefix: Whether to add prefix search capability (:* suffix) + + Returns: + A properly formatted single term for tsquery + + For Postgres tsquery: + - Multi-word queries become "word1 & word2" + - Prefix matching uses ":*" suffix (e.g., "coff:*") + - Special characters that need escaping: & | ! ( ) : + """ + if not term or not term.strip(): + return term + + term = term.strip() + + # Check if term is already a wildcard pattern + if "*" in term: + # Replace * with :* for Postgres prefix matching + return term.replace("*", ":*") + + # Remove tsquery special characters from the search term + # These characters have special meaning in tsquery and cause syntax errors + # if not used as operators + special_chars = ["&", "|", "!", "(", ")", ":"] + cleaned_term = term + for char in special_chars: + cleaned_term = cleaned_term.replace(char, " ") + + # Handle multi-word queries + if " " in cleaned_term: + # Strip sentence punctuation from word edges so question-form + # queries produce clean lexemes (parity with SQLite FTS5 prep). + # The tsquery tokenizer ignores this punctuation anyway; leaving it + # in only risks tsquery syntax errors. Interior characters are kept. + words = [w.strip("?!.,;") for w in cleaned_term.split()] + words = [w for w in words if w] + if not words: + # All characters were special chars, search won't match anything + # Return a safe search term that won't cause syntax errors + return "NOSPECIALCHARS:*" + if is_prefix: + # Add prefix matching to each word + prepared_words = [f"{word}:*" for word in words] + else: + prepared_words = words + # Join with AND operator + return " & ".join(prepared_words) + + # Single word: strip edge punctuation; guard the now-empty case so a + # bare ":*"/"" never reaches tsquery. + cleaned_term = cleaned_term.strip().strip("?!.,;") + if not cleaned_term: + return "NOSPECIALCHARS:*" + if is_prefix: + return f"{cleaned_term}:*" + else: + return cleaned_term + + async def _build_fts_query_parts( + self, + search_text: Optional[str] = None, + permalink: Optional[str] = None, + permalink_match: Optional[str] = None, + title: Optional[str] = None, + note_types: Optional[List[str]] = None, + after_date: Optional[datetime] = None, + search_item_types: Optional[List[SearchItemType]] = None, + categories: Optional[List[str]] = None, + metadata_filters: Optional[dict[str, Any]] = None, + file_path_prefix: Optional[str] = None, + temporal: Optional[TemporalFilter] = None, + allow_relaxed: bool = False, + candidate_keys: Sequence[SearchIndexKey] | None = None, + ) -> tuple[str, str, dict[str, Any], str, str]: + """Build Postgres FTS FROM/WHERE params shared by search and count.""" + conditions = [] + params: dict[str, Any] = dict(self._scope_params) + order_by_clause = "" + from_clause = "search_index" + document_vector_sql: str | None = None + script_tsqueries: list[str] = [] + + # Handle text search for title and content using tsvector + if search_text: + if search_text.strip() == "*" or search_text.strip() == "": + # For wildcard searches, don't add any text conditions + pass + else: + script_query = analyze_script_query(search_text.strip()) + if script_query.word_text: + processed_text = self._prepare_search_term(script_query.word_text) + params["text"] = processed_text + probe_texts = [processed_text] + if allow_relaxed: + relaxed_text = self._relaxed_tsquery_text(script_query.word_text) + if relaxed_text: + probe_texts.append(relaxed_text) + + candidate_operands: dict[str, None] = {} + for probe_text in probe_texts: + for operand, _representative in _tsquery_operands(probe_text): + candidate_operands.setdefault(operand, None) + if candidate_operands: + params["text_candidate"] = " | ".join(candidate_operands) + + # Trigger: PostgreSQL can extract a required-positive query tree. + # Why: OR-ing its operands is a safe indexed superset even when + # terms live in different chunks. Pure/optional negation returns + # ``T`` and must retain all project rows for correct semantics. + # Outcome: ordinary and required-positive NOT queries use both + # GIN indexes; only genuinely unindexable negation scans the project. + from_clause = f""" + search_index JOIN ( + SELECT + candidate_parent.project_id, + candidate_parent.id, + candidate_parent.type + FROM search_index AS candidate_parent + WHERE candidate_parent.project_id {self._scope_sql} + AND querytree(to_tsquery('english', :text)) <> 'T' + AND candidate_parent.textsearchable_index_col + @@ to_tsquery('english', :text_candidate) + UNION + SELECT + candidate_chunk.project_id, + candidate_chunk.search_index_id AS id, + candidate_chunk.search_index_type AS type + FROM search_index_fts_chunks AS candidate_chunk + WHERE candidate_chunk.project_id {self._scope_sql} + AND querytree(to_tsquery('english', :text)) <> 'T' + AND candidate_chunk.textsearchable_index_col + @@ to_tsquery('english', :text_candidate) + UNION + SELECT + candidate_all.project_id, + candidate_all.id, + candidate_all.type + FROM search_index AS candidate_all + WHERE candidate_all.project_id {self._scope_sql} + AND querytree(to_tsquery('english', :text)) = 'T' + ) AS fts_candidate + ON fts_candidate.project_id = search_index.project_id + AND fts_candidate.id = search_index.id + AND fts_candidate.type = search_index.type + """ + document_vector_sql = self._document_fts_vector_sql(probe_texts, params) + word_condition = f"{document_vector_sql} @@ to_tsquery('english', :text)" + if script_query.gram_phrases: + # Trigger: PostgreSQL's English dictionary removes every word term. + # Why: an empty word query must not suppress a required script match. + # Outcome: only mixed queries treat the empty word channel as neutral; + # word-only stopword queries retain their established empty result. + word_condition = ( + f"(numnode(to_tsquery('english', :text)) = 0 OR {word_condition})" + ) + conditions.append(word_condition) + + if script_query.gram_phrases: + script_tsqueries = [ + " <-> ".join(f"'{gram}'" for gram in phrase) + for phrase in script_query.gram_phrases + ] + for index, script_tsquery in enumerate(script_tsqueries): + params[f"script_text_{index}"] = script_tsquery + # Trigger: a query contains script grams, with or without word terms. + # Why: every script phrase is required, while an English word clause can + # reduce to an empty tsquery after dictionary processing. + # Outcome: start from the parent and child script GIN indexes, then apply + # every word and script predicate below. + params["script_candidate_text"] = " | ".join( + f"({script_tsquery})" for script_tsquery in script_tsqueries + ) + from_clause = f""" + search_index JOIN ( + SELECT + script_parent.project_id, + script_parent.id, + script_parent.type + FROM search_index AS script_parent + WHERE script_parent.project_id {self._scope_sql} + AND script_parent.script_ngrams_index_col + @@ to_tsquery('simple', :script_candidate_text) + UNION + SELECT + script_candidate.project_id, + script_candidate.search_index_id AS id, + script_candidate.search_index_type AS type + FROM search_index_fts_chunks AS script_candidate + WHERE script_candidate.project_id {self._scope_sql} + AND script_candidate.script_ngrams_index_col + @@ to_tsquery('simple', :script_candidate_text) + ) AS fts_candidate + ON fts_candidate.project_id = search_index.project_id + AND fts_candidate.id = search_index.id + AND fts_candidate.type = search_index.type + """ + conditions.extend( + "(search_index.script_ngrams_index_col " + f"@@ to_tsquery('simple', :script_text_{index}) OR EXISTS (" + "SELECT 1 FROM search_index_fts_chunks AS script_chunk " + "WHERE script_chunk.project_id = search_index.project_id " + "AND script_chunk.search_index_id = search_index.id " + "AND script_chunk.search_index_type = search_index.type " + "AND script_chunk.script_ngrams_index_col " + f"@@ to_tsquery('simple', :script_text_{index})))" + for index in range(len(script_tsqueries)) + ) + + # Handle title search + if title: + title_text = self._prepare_search_term(title.strip(), is_prefix=False) + params["title_text"] = title_text + conditions.append( + "to_tsvector('english', search_index.title) @@ to_tsquery('english', :title_text)" + ) + + # Handle permalink exact search + if permalink: + params["permalink"] = permalink + conditions.append("search_index.permalink = :permalink") + + # Handle permalink pattern match + if permalink_match: + permalink_text = permalink_match.lower().strip() + params["permalink"] = permalink_text + if "*" in permalink_match: + # Use LIKE for pattern matching in Postgres + # Convert * to % for SQL LIKE + permalink_pattern = permalink_text.replace("*", "%") + params["permalink"] = permalink_pattern + conditions.append("search_index.permalink LIKE :permalink") + else: + conditions.append("search_index.permalink = :permalink") + + # Handle directory subtree scope. The predicate is built by the shared + # helper so Postgres and SQLite scope by the identical rule; see + # file_path_prefix_condition for the boundary and escaping reasoning. + subtree_condition = file_path_prefix_condition(file_path_prefix, params) + if subtree_condition is not None: + conditions.append(subtree_condition) + + # Handle an explicit candidate-row restriction. Built by the shared helper so + # both backends restrict by the identical rule; see + # candidate_key_restriction_condition for why the vector filter pass asks about + # its candidates rather than paging the filter's whole match set (#1431). + if candidate_keys is not None: + conditions.append(candidate_key_restriction_condition(candidate_keys, params)) + + # Handle search item type filter (parameterized for defense-in-depth) + if search_item_types: + type_placeholders = [] + for idx, t in enumerate(search_item_types): + param_name = f"search_type_{idx}" + params[param_name] = t.value + type_placeholders.append(f":{param_name}") + conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})") + + # Handle observation category filter (parameterized for defense-in-depth). + # Trigger: caller passed `categories` to scope observation results. + # Why: `entity_types=["observation"]` only narrows to the observation row type; + # callers expect exact-category matching, not incidental text matches. + # Outcome: only rows whose indexed category exactly equals a requested value + # survive (entities/relations have NULL category and are excluded). + if categories: + category_placeholders = [] + for idx, category in enumerate(categories): + param_name = f"category_{idx}" + params[param_name] = category + category_placeholders.append(f":{param_name}") + conditions.append(f"search_index.category IN ({', '.join(category_placeholders)})") + + # Handle note type filter (frontmatter type field, parameterized). + # Trigger: caller passed `note_types` to scope by the frontmatter `type` field. + # Why: the type belongs to the note, but only its entity row carries the + # frontmatter; observation and relation rows do not. Reading it off each row + # silently excluded every non-entity row, which made `note_types` combined + # with a valid-time filter unsatisfiable. + # Outcome: resolved through the owning note in one shared builder, so both + # backends ask the same question and observation rows of a matching note + # are admitted. + if note_types: + conditions.append( + build_note_type_predicate( + note_types, + params, + project_scope_sql=self._scope_sql, + note_type_value=POSTGRES_NOTE_TYPE_VALUE, + ) + ) + + # Handle date filter + if after_date: + params["after_date"] = after_date + # Filter on updated_at so recently-edited notes are included even when created_at is old + conditions.append("search_index.updated_at > :after_date") + # order by most recent first + order_by_clause = ", search_index.updated_at DESC" + + # Handle authored valid time (SPEC-82). + # Trigger: caller asked when a statement was true of the world. + # Why: `after_date` above filters `updated_at`, which records when the note was + # last edited. That is bookkeeping, never a semantic claim; a decision + # effective through July says nothing about when its file was touched. + # Outcome: an independent predicate over the temporal projection, textually + # identical to the SQLite one because canonical bounds compare + # lexicographically on both backends. Undated sources carry no row and + # are therefore excluded whenever a valid-time filter is present. + if temporal is not None: + conditions.append( + build_temporal_predicate(temporal, params, project_scope_sql=self._scope_sql) + ) + + # Handle structured metadata filters (frontmatter) + # Uses jsonb_extract_path_text() / jsonb_extract_path() with parameterized + # path parts instead of #>> / #> with interpolated paths. + if metadata_filters: + parsed_filters = parse_metadata_filters(metadata_filters) + from_clause = f"{from_clause} JOIN entity ON search_index.entity_id = entity.id" + # Frontmatter filters answer for notes only; see + # metadata_filter_content_type_condition for why every regular file + # would otherwise satisfy a null predicate. + conditions.append(metadata_filter_content_type_condition(params)) + metadata_expr = "entity.entity_metadata::jsonb" + + for idx, filt in enumerate(parsed_filters): + # Parameterize each JSON path part individually + path_param_names = [] + for j, part in enumerate(filt.path_parts): + path_param = f"meta_path_{idx}_{j}" + params[path_param] = part + path_param_names.append(f":{path_param}") + path_args = ", ".join(path_param_names) + text_expr = f"jsonb_extract_path_text({metadata_expr}, {path_args})" + json_expr = f"jsonb_extract_path({metadata_expr}, {path_args})" + + # jsonb_extract_path_text returns SQL NULL both for a missing key + # and for an explicit JSON null — the same two cases SQLite's + # json_extract collapses — so the dialects answer + # `{"owner": None}` row for row. `= NULL` is never true, so + # equality here would report a confident zero. + if filt.op == "is_null": + conditions.append(f"{text_expr} IS NULL") + continue + + if filt.op == "eq": + value_param = f"meta_val_{idx}" + params[value_param] = filt.value + conditions.append(f"{text_expr} = :{value_param}") + continue + + if filt.op == "in": + placeholders = [] + for j, val in enumerate(filt.value): + value_param = f"meta_val_{idx}_{j}" + params[value_param] = val + placeholders.append(f":{value_param}") + conditions.append(f"{text_expr} IN ({', '.join(placeholders)})") + continue + + if filt.op == "contains": + base_param = f"meta_val_{idx}" + tag_conditions = [] + # Require all values to be present + for j, val in enumerate(filt.value): + tag_param = f"{base_param}_{j}" + params[tag_param] = json.dumps([val]) + # The exact JSONB containment test is the primary path; the + # substring patterns only reach values stored as array text. + like_condition = metadata_contains_like_condition( + text_expr, + val, + param_prefix=tag_param, + params=params, + ) + tag_conditions.append( + f"({json_expr} @> CAST(:{tag_param} AS jsonb) OR {like_condition})" + ) + conditions.append(" AND ".join(tag_conditions)) + continue + + if filt.op in {"gt", "gte", "lt", "lte", "between"}: + compare_expr = ( + f"{text_expr}::double precision" + if filt.comparison == "numeric" + else text_expr + ) + + if filt.op == "between": + min_param = f"meta_val_{idx}_min" + max_param = f"meta_val_{idx}_max" + params[min_param] = filt.value[0] + params[max_param] = filt.value[1] + conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}") + else: + value_param = f"meta_val_{idx}" + params[value_param] = filt.value + operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op] + conditions.append(f"{compare_expr} {operator} :{value_param}") + continue + + conditions.append(f"search_index.project_id {self._scope_sql}") + + # Build WHERE clause + where_clause = " AND ".join(conditions) if conditions else "1=1" + + # Build SQL with ts_rank() for scoring + # Note: If no text search, score will be NULL, so we use COALESCE to default to 0 + score_parts: list[str] = [] + if document_vector_sql is not None: + score_parts.append( + "GREATEST(" + f"ts_rank({document_vector_sql}, to_tsquery('english', :text)), " + "ts_rank(search_index.textsearchable_index_col, to_tsquery('english', :text)), " + "COALESCE((SELECT MAX(ts_rank(" + "fts_chunk.textsearchable_index_col, to_tsquery('english', :text))) " + "FROM search_index_fts_chunks AS fts_chunk " + "WHERE fts_chunk.project_id = search_index.project_id " + "AND fts_chunk.search_index_id = search_index.id " + "AND fts_chunk.search_index_type = search_index.type " + "AND fts_chunk.textsearchable_index_col " + "@@ to_tsquery('english', :text)), 0))" + ) + score_parts.extend( + "GREATEST(" + "ts_rank(search_index.script_ngrams_index_col, " + f"to_tsquery('simple', :script_text_{index})), " + "COALESCE((SELECT MAX(ts_rank(script_rank.script_ngrams_index_col, " + f"to_tsquery('simple', :script_text_{index}))) " + "FROM search_index_fts_chunks AS script_rank " + "WHERE script_rank.project_id = search_index.project_id " + "AND script_rank.search_index_id = search_index.id " + "AND script_rank.search_index_type = search_index.type " + "AND script_rank.script_ngrams_index_col " + f"@@ to_tsquery('simple', :script_text_{index})), 0))" + for index in range(len(script_tsqueries)) + ) + # Each condition above is required, so every query component should contribute to + # relevance. Taking only the strongest rank makes additional script runs invisible. + score_expr = " + ".join(score_parts) if score_parts else "0" + + return from_clause, where_clause, params, order_by_clause, score_expr + + @staticmethod + def _document_fts_vector_sql(processed_texts: Sequence[str], params: dict[str, Any]) -> str: + """Build a query-sized vector representing lexemes found anywhere in one item.""" + operands: dict[str, str] = {} + for processed_text in processed_texts: + for operand, representative in _tsquery_operands(processed_text): + operands.setdefault(operand, representative) + + present_lexemes: list[str] = [] + for index, (operand, representative) in enumerate(operands.items()): + operand_param = f"text_operand_{index}" + representative_param = f"text_representative_{index}" + params[operand_param] = operand + params[representative_param] = representative + present_lexemes.append( + "CASE WHEN (search_index.textsearchable_index_col " + f"@@ to_tsquery('english', :{operand_param}) OR EXISTS (" + "SELECT 1 FROM search_index_fts_chunks AS operand_chunk " + "WHERE operand_chunk.project_id = search_index.project_id " + "AND operand_chunk.search_index_id = search_index.id " + "AND operand_chunk.search_index_type = search_index.type " + "AND operand_chunk.textsearchable_index_col " + f"@@ to_tsquery('english', :{operand_param}))) " + f"THEN :{representative_param} ELSE '' END" + ) + + if not present_lexemes: + return "search_index.textsearchable_index_col" + + # The synthesized text contains at most the query operands, never the note body. + # This preserves document-wide Boolean semantics without recreating an unbounded vector. + lexeme_array = f"ARRAY[{', '.join(present_lexemes)}]" + return f"to_tsvector('english', array_to_string({lexeme_array}, ' '))" diff --git a/src/basic_memory/repository/postgres_search_repository.py b/src/basic_memory/repository/postgres_search_repository.py index a6d7e3088..429fd1655 100644 --- a/src/basic_memory/repository/postgres_search_repository.py +++ b/src/basic_memory/repository/postgres_search_repository.py @@ -2,7 +2,6 @@ import asyncio import json -import re import time from collections.abc import Sequence from datetime import datetime @@ -14,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from basic_memory import db +from basic_memory.repository.postgres_search_query import PostgresSearchQuery from basic_memory.config import BasicMemoryConfig, ConfigManager, DatabaseBackend from basic_memory.models.search import SEARCH_INDEX_ROW_KEY_COLUMNS from basic_memory.repository.embedding_provider import EmbeddingProvider @@ -21,28 +21,18 @@ from basic_memory.repository.rerank_provider import RerankProvider from basic_memory.repository.rerank_provider_factory import create_rerank_provider from basic_memory.repository.search_index_row import SearchIndexRow -from basic_memory.repository.search_query import relaxed_query_words, relaxation_word_tokens -from basic_memory.repository.script_ngrams import analyze_script_query, build_script_ngrams +from basic_memory.repository.search_query import relaxed_query_words +from basic_memory.repository.script_ngrams import build_script_ngrams from basic_memory.repository.semantic_chunking import VectorChunkRecord from basic_memory.repository.search_repository_base import ( SearchIndexKey, SearchRepositoryBase, VectorChunkState, - candidate_key_restriction_condition, - file_path_prefix_condition, - metadata_contains_like_condition, - metadata_filter_content_type_condition, ) from basic_memory.repository.search_trace import ( SearchTraceCollector, build_fts_page_stage, ) -from basic_memory.repository.metadata_filters import parse_metadata_filters -from basic_memory.repository.note_type_filters import ( - POSTGRES_NOTE_TYPE_VALUE, - build_note_type_predicate, -) -from basic_memory.repository.temporal_filters import build_temporal_predicate from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError from basic_memory.repository.semantic_vector_index import SemanticVectorIndex from basic_memory.repository.semantic_vector_sync import ( @@ -59,111 +49,6 @@ from basic_memory.temporal import TemporalFilter -_TSQUERY_OPERAND_PATTERN = re.compile(r"'(?:''|[^'])*'(?::\*)?|[^\s&|!()]+") -_TSQUERY_WORD_PATTERN = re.compile(r"[^\W_]+(?:'[^\W_]+)?", re.UNICODE) -_QUOTED_QUERY_PATTERN = re.compile(r'"([^"]*)"') -_BOOLEAN_WORDS = frozenset({"AND", "OR", "NOT"}) -_TSQUERY_METACHARACTERS = frozenset("&|!:<>") - - -def _tsquery_operands(processed_text: str) -> list[tuple[str, str]]: - """Return unique (query operand, representative text) pairs in source order.""" - operands: dict[str, str] = {} - for operand in _TSQUERY_OPERAND_PATTERN.findall(processed_text): - representative = operand.removesuffix(":*") - if representative.startswith("'") and representative.endswith("'"): - representative = representative[1:-1].replace("''", "'") - operands.setdefault(operand, representative) - continue - - # An unquoted apostrophe is invalid tsquery syntax. Keep the literal - # word for the synthetic document, but quote and escape its probe so a - # strict syntax failure can proceed to the relaxed retry. - if "'" in representative: - escaped = "'{}'".format(representative.replace("'", "''")) - safe_operand = f"{escaped}:*" if operand.endswith(":*") else escaped - operands.setdefault(safe_operand, representative) - continue - - # PostgreSQL legitimately parses punctuation inside operands such as - # ``v0.13.0b2:*`` and ``auth-service:*``. Preserve those bytes so the - # synthetic document is tokenized the same way as the original note. - if "<" not in representative and ">" not in representative: - operands.setdefault(operand, representative) - continue - - # A malformed strict operand (for example ``foo str: - """Render user text as complete, individually escaped tsquery operands.""" - words = relaxation_word_tokens(text_value) - if drop_boolean_words: - words = [word for word in words if word.upper() not in _BOOLEAN_WORDS] - if not words: - return "NOSPECIALCHARS:*" - - operands = [] - for word in words: - escaped_word = "'{}'".format(word.replace("'", "''")) if "'" in word else word - operands.append(f"{escaped_word}:*" if is_prefix else escaped_word) - return operator.join(operands) - - -def _render_boolean_operand(operand: str) -> str: - """Preserve safe structured text while escaping tsquery syntax bytes.""" - if "'" not in operand and not any(char in _TSQUERY_METACHARACTERS for char in operand): - return operand - return _render_tsquery_words( - operand, - operator=" & ", - is_prefix=False, - ) - - -def _has_valid_boolean_shape(expression: str) -> bool: - """Reject incomplete operator structure before it reaches strict ``to_tsquery``.""" - depth = 0 - for char in expression: - if char == "(": - depth += 1 - elif char == ")": - depth -= 1 - if depth < 0: - return False - if depth: - return False - - stripped = expression.strip() - if not stripped or stripped[0] in "&|" or stripped[-1] in "&|!": - return False - return not any( - re.search(pattern, stripped) - for pattern in ( - r"[&|]\s*[&|]", - r"!\s*[&|)]", - r"\(\s*[&|)]", - r"[&|!(]\s*\)", - ) - ) - - def _strip_nul_from_row(row_data: dict[str, Any]) -> dict[str, Any]: """Strip NUL bytes from all string values in a row dict. @@ -173,7 +58,7 @@ def _strip_nul_from_row(row_data: dict[str, Any]) -> dict[str, Any]: return {k: v.replace("\x00", "") if isinstance(v, str) else v for k, v in row_data.items()} -class PostgresSearchRepository(SearchRepositoryBase): +class PostgresSearchRepository(PostgresSearchQuery, SearchRepositoryBase): """PostgreSQL tsvector implementation of search repository. Uses PostgreSQL's full-text search capabilities with: @@ -199,7 +84,8 @@ def __init__( vector_index: SemanticVectorIndex | None = None, rerank_provider: RerankProvider | None = None, ): - super().__init__(session_maker, project_id) + SearchRepositoryBase.__init__(self, session_maker, project_id) + PostgresSearchQuery.__init__(self, session_maker, (project_id,)) self._app_config = app_config or ConfigManager().config self._semantic_enabled = self._app_config.semantic_search_enabled self._semantic_vector_k = self._app_config.semantic_vector_k @@ -400,232 +286,6 @@ async def _replace_fts_chunks( # tsquery preparation (backend-specific) # ------------------------------------------------------------------ - @override - def _prepare_search_term(self, term: str, is_prefix: bool = True) -> str: - """Prepare a search term for tsquery format. - - Args: - term: The search term to prepare - is_prefix: Whether to add prefix search capability (:* operator) - - Returns: - Formatted search term for tsquery - - For Postgres: - - Boolean operators are converted to tsquery format (&, |, !) - - Prefix matching uses the :* operator - - Terms are sanitized to prevent tsquery syntax errors - """ - # Check for explicit boolean operators - boolean_operators = [" AND ", " OR ", " NOT "] - if '"' in term or any(op in f" {term} " for op in boolean_operators): - return self._prepare_boolean_query(term) - - # For non-Boolean queries, prepare single term - return self._prepare_single_term(term, is_prefix) - - @staticmethod - def _relaxed_tsquery_term(word: str) -> str: - """Render one relaxed word as a tsquery-safe prefix expression. - - Mirrors the SQLite renderer: a word token can contain an apostrophe, and - tsquery reads that as lexeme-quoting syntax rather than text. Quoting the - lexeme and doubling any interior quote keeps it literal. - """ - if "'" in word: - return "'{}':*".format(word.replace("'", "''")) - return f"{word}:*" - - @staticmethod - def _relaxed_tsquery_text(search_text: Optional[str]) -> Optional[str]: - """OR-relaxed tsquery expression for a failed strict query, or None.""" - words = relaxed_query_words(search_text) - if not words: - return None - return " | ".join(PostgresSearchRepository._relaxed_tsquery_term(word) for word in words) - - def _prepare_boolean_query(self, query: str) -> str: - """Convert Boolean query to tsquery format. - - Args: - query: A Boolean query like "coffee AND brewing" or "(pour OR french) AND press" - - Returns: - tsquery-formatted string with & (AND), | (OR), ! (NOT) operators - - Examples: - "coffee AND brewing" -> "coffee & brewing" - "(pour OR french) AND press" -> "(pour | french) & press" - "coffee NOT decaf" -> "coffee & !decaf" - """ - # PostgreSQL's strict to_tsquery grammar does not accept web-style double - # quotes. Convert complete quoted groups first so operator-looking words - # inside them remain text and every word becomes a complete operand. - quoted_phrases: dict[str, str] = {} - - def replace_quoted_phrase(match: re.Match[str]) -> str: - phrase = _render_tsquery_words( - match.group(1), - operator=" & ", - is_prefix=False, - ) - placeholder = f"BMQUOTEDPHRASE{len(quoted_phrases)}" - while placeholder in query: - placeholder += "X" - quoted_phrases[placeholder] = f"({phrase})" - # Surround the placeholder so quotes adjacent to plain text become - # explicit operands instead of restoring into ``word(group)``. - return f" {placeholder} " - - result = _QUOTED_QUERY_PATTERN.sub(replace_quoted_phrase, query) - if '"' in result: - # An unmatched quote is user text, not a reason to abort the database - # transaction. Boolean-looking words lose their operator role here. - return _render_tsquery_words( - query, - operator=" & ", - is_prefix=True, - drop_boolean_words=True, - ) - - # Boolean syntax is the only structure retained from user input. A - # whitespace-delimited operand still needs explicit conjunctions, but - # PostgreSQL must tokenize structured single operands such as - # ``auth-service`` and ``config.json`` exactly as it did before quoted - # query normalization. - normalized_parts: list[str] = [] - operator_pattern = r"((? str: - """Prepare a single search term for tsquery. - - Args: - term: A single search term - is_prefix: Whether to add prefix search capability (:* suffix) - - Returns: - A properly formatted single term for tsquery - - For Postgres tsquery: - - Multi-word queries become "word1 & word2" - - Prefix matching uses ":*" suffix (e.g., "coff:*") - - Special characters that need escaping: & | ! ( ) : - """ - if not term or not term.strip(): - return term - - term = term.strip() - - # Check if term is already a wildcard pattern - if "*" in term: - # Replace * with :* for Postgres prefix matching - return term.replace("*", ":*") - - # Remove tsquery special characters from the search term - # These characters have special meaning in tsquery and cause syntax errors - # if not used as operators - special_chars = ["&", "|", "!", "(", ")", ":"] - cleaned_term = term - for char in special_chars: - cleaned_term = cleaned_term.replace(char, " ") - - # Handle multi-word queries - if " " in cleaned_term: - # Strip sentence punctuation from word edges so question-form - # queries produce clean lexemes (parity with SQLite FTS5 prep). - # The tsquery tokenizer ignores this punctuation anyway; leaving it - # in only risks tsquery syntax errors. Interior characters are kept. - words = [w.strip("?!.,;") for w in cleaned_term.split()] - words = [w for w in words if w] - if not words: - # All characters were special chars, search won't match anything - # Return a safe search term that won't cause syntax errors - return "NOSPECIALCHARS:*" - if is_prefix: - # Add prefix matching to each word - prepared_words = [f"{word}:*" for word in words] - else: - prepared_words = words - # Join with AND operator - return " & ".join(prepared_words) - - # Single word: strip edge punctuation; guard the now-empty case so a - # bare ":*"/"" never reaches tsquery. - cleaned_term = cleaned_term.strip().strip("?!.,;") - if not cleaned_term: - return "NOSPECIALCHARS:*" - if is_prefix: - return f"{cleaned_term}:*" - else: - return cleaned_term - # ------------------------------------------------------------------ # Abstract hook implementations (vector/semantic, Postgres-specific) # ------------------------------------------------------------------ @@ -967,421 +627,6 @@ def _is_tsquery_syntax_error(exc: Exception) -> bool: or "no operator in tsquery" in msg ) - async def _build_fts_query_parts( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - allow_relaxed: bool = False, - candidate_keys: Sequence[SearchIndexKey] | None = None, - ) -> tuple[str, str, dict[str, Any], str, str]: - """Build Postgres FTS FROM/WHERE params shared by search and count.""" - conditions = [] - params = {} - order_by_clause = "" - from_clause = "search_index" - document_vector_sql: str | None = None - script_tsqueries: list[str] = [] - - # Handle text search for title and content using tsvector - if search_text: - if search_text.strip() == "*" or search_text.strip() == "": - # For wildcard searches, don't add any text conditions - pass - else: - script_query = analyze_script_query(search_text.strip()) - if script_query.word_text: - processed_text = self._prepare_search_term(script_query.word_text) - params["text"] = processed_text - probe_texts = [processed_text] - if allow_relaxed: - relaxed_text = self._relaxed_tsquery_text(script_query.word_text) - if relaxed_text: - probe_texts.append(relaxed_text) - - candidate_operands: dict[str, None] = {} - for probe_text in probe_texts: - for operand, _representative in _tsquery_operands(probe_text): - candidate_operands.setdefault(operand, None) - if candidate_operands: - params["text_candidate"] = " | ".join(candidate_operands) - - # Trigger: PostgreSQL can extract a required-positive query tree. - # Why: OR-ing its operands is a safe indexed superset even when - # terms live in different chunks. Pure/optional negation returns - # ``T`` and must retain all project rows for correct semantics. - # Outcome: ordinary and required-positive NOT queries use both - # GIN indexes; only genuinely unindexable negation scans the project. - from_clause = """ - search_index JOIN ( - SELECT - candidate_parent.project_id, - candidate_parent.id, - candidate_parent.type - FROM search_index AS candidate_parent - WHERE candidate_parent.project_id = :project_id - AND querytree(to_tsquery('english', :text)) <> 'T' - AND candidate_parent.textsearchable_index_col - @@ to_tsquery('english', :text_candidate) - UNION - SELECT - candidate_chunk.project_id, - candidate_chunk.search_index_id AS id, - candidate_chunk.search_index_type AS type - FROM search_index_fts_chunks AS candidate_chunk - WHERE candidate_chunk.project_id = :project_id - AND querytree(to_tsquery('english', :text)) <> 'T' - AND candidate_chunk.textsearchable_index_col - @@ to_tsquery('english', :text_candidate) - UNION - SELECT - candidate_all.project_id, - candidate_all.id, - candidate_all.type - FROM search_index AS candidate_all - WHERE candidate_all.project_id = :project_id - AND querytree(to_tsquery('english', :text)) = 'T' - ) AS fts_candidate - ON fts_candidate.project_id = search_index.project_id - AND fts_candidate.id = search_index.id - AND fts_candidate.type = search_index.type - """ - document_vector_sql = self._document_fts_vector_sql(probe_texts, params) - word_condition = f"{document_vector_sql} @@ to_tsquery('english', :text)" - if script_query.gram_phrases: - # Trigger: PostgreSQL's English dictionary removes every word term. - # Why: an empty word query must not suppress a required script match. - # Outcome: only mixed queries treat the empty word channel as neutral; - # word-only stopword queries retain their established empty result. - word_condition = ( - f"(numnode(to_tsquery('english', :text)) = 0 OR {word_condition})" - ) - conditions.append(word_condition) - - if script_query.gram_phrases: - script_tsqueries = [ - " <-> ".join(f"'{gram}'" for gram in phrase) - for phrase in script_query.gram_phrases - ] - for index, script_tsquery in enumerate(script_tsqueries): - params[f"script_text_{index}"] = script_tsquery - # Trigger: a query contains script grams, with or without word terms. - # Why: every script phrase is required, while an English word clause can - # reduce to an empty tsquery after dictionary processing. - # Outcome: start from the parent and child script GIN indexes, then apply - # every word and script predicate below. - params["script_candidate_text"] = " | ".join( - f"({script_tsquery})" for script_tsquery in script_tsqueries - ) - from_clause = """ - search_index JOIN ( - SELECT - script_parent.project_id, - script_parent.id, - script_parent.type - FROM search_index AS script_parent - WHERE script_parent.project_id = :project_id - AND script_parent.script_ngrams_index_col - @@ to_tsquery('simple', :script_candidate_text) - UNION - SELECT - script_candidate.project_id, - script_candidate.search_index_id AS id, - script_candidate.search_index_type AS type - FROM search_index_fts_chunks AS script_candidate - WHERE script_candidate.project_id = :project_id - AND script_candidate.script_ngrams_index_col - @@ to_tsquery('simple', :script_candidate_text) - ) AS fts_candidate - ON fts_candidate.project_id = search_index.project_id - AND fts_candidate.id = search_index.id - AND fts_candidate.type = search_index.type - """ - conditions.extend( - "(search_index.script_ngrams_index_col " - f"@@ to_tsquery('simple', :script_text_{index}) OR EXISTS (" - "SELECT 1 FROM search_index_fts_chunks AS script_chunk " - "WHERE script_chunk.project_id = search_index.project_id " - "AND script_chunk.search_index_id = search_index.id " - "AND script_chunk.search_index_type = search_index.type " - "AND script_chunk.script_ngrams_index_col " - f"@@ to_tsquery('simple', :script_text_{index})))" - for index in range(len(script_tsqueries)) - ) - - # Handle title search - if title: - title_text = self._prepare_search_term(title.strip(), is_prefix=False) - params["title_text"] = title_text - conditions.append( - "to_tsvector('english', search_index.title) @@ to_tsquery('english', :title_text)" - ) - - # Handle permalink exact search - if permalink: - params["permalink"] = permalink - conditions.append("search_index.permalink = :permalink") - - # Handle permalink pattern match - if permalink_match: - permalink_text = permalink_match.lower().strip() - params["permalink"] = permalink_text - if "*" in permalink_match: - # Use LIKE for pattern matching in Postgres - # Convert * to % for SQL LIKE - permalink_pattern = permalink_text.replace("*", "%") - params["permalink"] = permalink_pattern - conditions.append("search_index.permalink LIKE :permalink") - else: - conditions.append("search_index.permalink = :permalink") - - # Handle directory subtree scope. The predicate is built by the shared - # helper so Postgres and SQLite scope by the identical rule; see - # file_path_prefix_condition for the boundary and escaping reasoning. - subtree_condition = file_path_prefix_condition(file_path_prefix, params) - if subtree_condition is not None: - conditions.append(subtree_condition) - - # Handle an explicit candidate-row restriction. Built by the shared helper so - # both backends restrict by the identical rule; see - # candidate_key_restriction_condition for why the vector filter pass asks about - # its candidates rather than paging the filter's whole match set (#1431). - if candidate_keys is not None: - conditions.append(candidate_key_restriction_condition(candidate_keys, params)) - - # Handle search item type filter (parameterized for defense-in-depth) - if search_item_types: - type_placeholders = [] - for idx, t in enumerate(search_item_types): - param_name = f"search_type_{idx}" - params[param_name] = t.value - type_placeholders.append(f":{param_name}") - conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})") - - # Handle observation category filter (parameterized for defense-in-depth). - # Trigger: caller passed `categories` to scope observation results. - # Why: `entity_types=["observation"]` only narrows to the observation row type; - # callers expect exact-category matching, not incidental text matches. - # Outcome: only rows whose indexed category exactly equals a requested value - # survive (entities/relations have NULL category and are excluded). - if categories: - category_placeholders = [] - for idx, category in enumerate(categories): - param_name = f"category_{idx}" - params[param_name] = category - category_placeholders.append(f":{param_name}") - conditions.append(f"search_index.category IN ({', '.join(category_placeholders)})") - - # Handle note type filter (frontmatter type field, parameterized). - # Trigger: caller passed `note_types` to scope by the frontmatter `type` field. - # Why: the type belongs to the note, but only its entity row carries the - # frontmatter; observation and relation rows do not. Reading it off each row - # silently excluded every non-entity row, which made `note_types` combined - # with a valid-time filter unsatisfiable. - # Outcome: resolved through the owning note in one shared builder, so both - # backends ask the same question and observation rows of a matching note - # are admitted. - if note_types: - conditions.append( - build_note_type_predicate( - note_types, params, note_type_value=POSTGRES_NOTE_TYPE_VALUE - ) - ) - - # Handle date filter - if after_date: - params["after_date"] = after_date - # Filter on updated_at so recently-edited notes are included even when created_at is old - conditions.append("search_index.updated_at > :after_date") - # order by most recent first - order_by_clause = ", search_index.updated_at DESC" - - # Handle authored valid time (SPEC-82). - # Trigger: caller asked when a statement was true of the world. - # Why: `after_date` above filters `updated_at`, which records when the note was - # last edited. That is bookkeeping, never a semantic claim; a decision - # effective through July says nothing about when its file was touched. - # Outcome: an independent predicate over the temporal projection, textually - # identical to the SQLite one because canonical bounds compare - # lexicographically on both backends. Undated sources carry no row and - # are therefore excluded whenever a valid-time filter is present. - if temporal is not None: - conditions.append(build_temporal_predicate(temporal, params)) - - # Handle structured metadata filters (frontmatter) - # Uses jsonb_extract_path_text() / jsonb_extract_path() with parameterized - # path parts instead of #>> / #> with interpolated paths. - if metadata_filters: - parsed_filters = parse_metadata_filters(metadata_filters) - from_clause = f"{from_clause} JOIN entity ON search_index.entity_id = entity.id" - # Frontmatter filters answer for notes only; see - # metadata_filter_content_type_condition for why every regular file - # would otherwise satisfy a null predicate. - conditions.append(metadata_filter_content_type_condition(params)) - metadata_expr = "entity.entity_metadata::jsonb" - - for idx, filt in enumerate(parsed_filters): - # Parameterize each JSON path part individually - path_param_names = [] - for j, part in enumerate(filt.path_parts): - path_param = f"meta_path_{idx}_{j}" - params[path_param] = part - path_param_names.append(f":{path_param}") - path_args = ", ".join(path_param_names) - text_expr = f"jsonb_extract_path_text({metadata_expr}, {path_args})" - json_expr = f"jsonb_extract_path({metadata_expr}, {path_args})" - - # jsonb_extract_path_text returns SQL NULL both for a missing key - # and for an explicit JSON null — the same two cases SQLite's - # json_extract collapses — so the dialects answer - # `{"owner": None}` row for row. `= NULL` is never true, so - # equality here would report a confident zero. - if filt.op == "is_null": - conditions.append(f"{text_expr} IS NULL") - continue - - if filt.op == "eq": - value_param = f"meta_val_{idx}" - params[value_param] = filt.value - conditions.append(f"{text_expr} = :{value_param}") - continue - - if filt.op == "in": - placeholders = [] - for j, val in enumerate(filt.value): - value_param = f"meta_val_{idx}_{j}" - params[value_param] = val - placeholders.append(f":{value_param}") - conditions.append(f"{text_expr} IN ({', '.join(placeholders)})") - continue - - if filt.op == "contains": - base_param = f"meta_val_{idx}" - tag_conditions = [] - # Require all values to be present - for j, val in enumerate(filt.value): - tag_param = f"{base_param}_{j}" - params[tag_param] = json.dumps([val]) - # The exact JSONB containment test is the primary path; the - # substring patterns only reach values stored as array text. - like_condition = metadata_contains_like_condition( - text_expr, - val, - param_prefix=tag_param, - params=params, - ) - tag_conditions.append( - f"({json_expr} @> CAST(:{tag_param} AS jsonb) OR {like_condition})" - ) - conditions.append(" AND ".join(tag_conditions)) - continue - - if filt.op in {"gt", "gte", "lt", "lte", "between"}: - compare_expr = ( - f"{text_expr}::double precision" - if filt.comparison == "numeric" - else text_expr - ) - - if filt.op == "between": - min_param = f"meta_val_{idx}_min" - max_param = f"meta_val_{idx}_max" - params[min_param] = filt.value[0] - params[max_param] = filt.value[1] - conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}") - else: - value_param = f"meta_val_{idx}" - params[value_param] = filt.value - operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op] - conditions.append(f"{compare_expr} {operator} :{value_param}") - continue - - # Always filter by project_id - params["project_id"] = self.project_id - conditions.append("search_index.project_id = :project_id") - - # Build WHERE clause - where_clause = " AND ".join(conditions) if conditions else "1=1" - - # Build SQL with ts_rank() for scoring - # Note: If no text search, score will be NULL, so we use COALESCE to default to 0 - score_parts: list[str] = [] - if document_vector_sql is not None: - score_parts.append( - "GREATEST(" - f"ts_rank({document_vector_sql}, to_tsquery('english', :text)), " - "ts_rank(search_index.textsearchable_index_col, to_tsquery('english', :text)), " - "COALESCE((SELECT MAX(ts_rank(" - "fts_chunk.textsearchable_index_col, to_tsquery('english', :text))) " - "FROM search_index_fts_chunks AS fts_chunk " - "WHERE fts_chunk.project_id = search_index.project_id " - "AND fts_chunk.search_index_id = search_index.id " - "AND fts_chunk.search_index_type = search_index.type " - "AND fts_chunk.textsearchable_index_col " - "@@ to_tsquery('english', :text)), 0))" - ) - score_parts.extend( - "GREATEST(" - "ts_rank(search_index.script_ngrams_index_col, " - f"to_tsquery('simple', :script_text_{index})), " - "COALESCE((SELECT MAX(ts_rank(script_rank.script_ngrams_index_col, " - f"to_tsquery('simple', :script_text_{index}))) " - "FROM search_index_fts_chunks AS script_rank " - "WHERE script_rank.project_id = search_index.project_id " - "AND script_rank.search_index_id = search_index.id " - "AND script_rank.search_index_type = search_index.type " - "AND script_rank.script_ngrams_index_col " - f"@@ to_tsquery('simple', :script_text_{index})), 0))" - for index in range(len(script_tsqueries)) - ) - # Each condition above is required, so every query component should contribute to - # relevance. Taking only the strongest rank makes additional script runs invisible. - score_expr = " + ".join(score_parts) if score_parts else "0" - - return from_clause, where_clause, params, order_by_clause, score_expr - - @staticmethod - def _document_fts_vector_sql(processed_texts: Sequence[str], params: dict[str, Any]) -> str: - """Build a query-sized vector representing lexemes found anywhere in one item.""" - operands: dict[str, str] = {} - for processed_text in processed_texts: - for operand, representative in _tsquery_operands(processed_text): - operands.setdefault(operand, representative) - - present_lexemes: list[str] = [] - for index, (operand, representative) in enumerate(operands.items()): - operand_param = f"text_operand_{index}" - representative_param = f"text_representative_{index}" - params[operand_param] = operand - params[representative_param] = representative - present_lexemes.append( - "CASE WHEN (search_index.textsearchable_index_col " - f"@@ to_tsquery('english', :{operand_param}) OR EXISTS (" - "SELECT 1 FROM search_index_fts_chunks AS operand_chunk " - "WHERE operand_chunk.project_id = search_index.project_id " - "AND operand_chunk.search_index_id = search_index.id " - "AND operand_chunk.search_index_type = search_index.type " - "AND operand_chunk.textsearchable_index_col " - f"@@ to_tsquery('english', :{operand_param}))) " - f"THEN :{representative_param} ELSE '' END" - ) - - if not present_lexemes: - return "search_index.textsearchable_index_col" - - # The synthesized text contains at most the query operands, never the note body. - # This preserves document-wide Boolean semantics without recreating an unbounded vector. - lexeme_array = f"ARRAY[{', '.join(present_lexemes)}]" - return f"to_tsvector('english', array_to_string({lexeme_array}, ' '))" - @override async def search( self, diff --git a/src/basic_memory/repository/search_query.py b/src/basic_memory/repository/search_query.py index d73d70c0c..412d81741 100644 --- a/src/basic_memory/repository/search_query.py +++ b/src/basic_memory/repository/search_query.py @@ -1,5 +1,11 @@ """Shared full-text query preparation rules.""" +from dataclasses import dataclass +from datetime import datetime +from typing import Any +from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode +from basic_memory.temporal import TemporalFilter + import re import unicodedata @@ -300,3 +306,22 @@ def relaxed_query_words(search_text: str | None) -> list[str] | None: return None pruned_words = [token for token in tokens if token not in RELAXATION_STOPWORDS] return _emit_relaxation_terms(pruned_words or tokens) or None + + +@dataclass(frozen=True) +class PreparedSearchQuery: + """Normalized query inputs shared by search and count.""" + + search_text: str | None + permalink: str | None + permalink_match: str | None + title: str | None + note_types: list[str] | None + search_item_types: list[SearchItemType] | None + categories: list[str] | None + after_date: datetime | None + metadata_filters: dict[str, Any] | None + file_path_prefix: str | None + temporal: TemporalFilter | None + retrieval_mode: SearchRetrievalMode + min_similarity: float | None diff --git a/src/basic_memory/repository/sqlite_search_query.py b/src/basic_memory/repository/sqlite_search_query.py new file mode 100644 index 000000000..f1f81de13 --- /dev/null +++ b/src/basic_memory/repository/sqlite_search_query.py @@ -0,0 +1,649 @@ +"""Read-only FTS query compilation for explicit project scopes.""" + +import re +from collections.abc import Sequence +from datetime import datetime +from typing import Any, List, Optional + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.repository.search_query import relaxed_query_words +from basic_memory.repository.search_repository_base import ( + SearchIndexKey, + candidate_key_restriction_condition, + file_path_prefix_condition, + metadata_contains_like_condition, + metadata_filter_content_type_condition, +) +from basic_memory.repository.script_ngrams import analyze_script_query +from basic_memory.repository.metadata_filters import parse_metadata_filters, build_sqlite_json_path +from basic_memory.repository.note_type_filters import ( + SQLITE_NOTE_TYPE_VALUE, + build_note_type_predicate, +) +from basic_memory.repository.temporal_filters import build_temporal_predicate +from basic_memory.schemas.search import SearchItemType +from basic_memory.temporal import TemporalFilter + + +SQLITE_WORD_COLUMNS = "{title content_stems content_snippet}" + + +class SQLiteSearchQuery: + """Compile FTS and filter predicates without owning index mutations.""" + + def __init__( + self, + session_maker: async_sessionmaker[AsyncSession], + project_ids: Sequence[int], + ) -> None: + self.session_maker = session_maker + self._entity_columns: set[str] | None = None + # A missing scope is never unrestricted. Empty scopes compile to no rows. + ids = tuple(project_ids) + if any(type(project_id) is not int or project_id <= 0 for project_id in ids): + raise ValueError("Project IDs must be positive integers") + ids = tuple(sorted(set(ids))) + self._scope_params = {f"scope_{index}": value for index, value in enumerate(ids)} + self._scope_sql = ( + "IN (" + ", ".join(f":{key}" for key in self._scope_params) + ")" + if ids + else "IN (NULL)" + ) + + async def _get_entity_columns(self) -> set[str]: + if self._entity_columns is None: + async with db.scoped_session(self.session_maker) as session: + result = await session.execute(text("PRAGMA table_info(entity)")) + self._entity_columns = {row[1] for row in result.fetchall()} + return self._entity_columns + + def _prepare_boolean_query(self, query: str) -> str: + """Prepare a Boolean query by quoting individual terms while preserving operators. + + Args: + query: A Boolean query like "tier1-test AND unicode" or "(hello OR world) NOT test" + + Returns: + A properly formatted Boolean query with quoted terms that need quoting + """ + # Define Boolean operators and their boundaries + boolean_pattern = r"(\bAND\b|\bOR\b|\bNOT\b)" + + # Split the query by Boolean operators, keeping the operators + parts = re.split(boolean_pattern, query) + + processed_parts = [] + for part in parts: + part = part.strip() + if not part: + continue + + # If it's a Boolean operator, keep it as is + if part in ["AND", "OR", "NOT"]: + processed_parts.append(part) + else: + # Handle parentheses specially - they should be preserved for grouping + if "(" in part or ")" in part: + # Parse parenthetical expressions carefully + processed_part = self._prepare_parenthetical_term(part) + processed_parts.append(processed_part) + else: + # This is a search term - for Boolean queries, don't add prefix wildcards + prepared_term = self._prepare_single_term(part, is_prefix=False) + processed_parts.append(prepared_term) + + return " ".join(processed_parts) + + def _prepare_parenthetical_term(self, term: str) -> str: + """Prepare a term that contains parentheses, preserving the parentheses for grouping. + + Args: + term: A term that may contain parentheses like "(hello" or "world)" or "(hello OR world)" + + Returns: + A properly formatted term with parentheses preserved + """ + # Handle terms that start/end with parentheses but may contain quotable content + result = "" + i = 0 + while i < len(term): + if term[i] in "()": + # Preserve parentheses as-is + result += term[i] + i += 1 + else: + # Find the next parenthesis or end of string + start = i + while i < len(term) and term[i] not in "()": + i += 1 + + # Extract the content between parentheses + content = term[start:i].strip() + if content: + # Only quote if it actually needs quoting (has hyphens, special chars, etc) + # but don't quote if it's just simple words + if self._needs_quoting(content): + escaped_content = content.replace('"', '""') + result += f'"{escaped_content}"' + else: + result += content + + return result + + def _needs_quoting(self, term: str) -> bool: + """Check if a term needs to be quoted for FTS5 safety. + + Args: + term: The term to check + + Returns: + True if the term should be quoted + """ + if not term or not term.strip(): + return False + + # Characters that indicate we should quote (excluding parentheses which are valid syntax) + needs_quoting_chars = [ + " ", + ".", + ":", + ";", + ",", + "<", + ">", + "?", + "/", + "-", + "'", + '"', + "[", + "]", + "{", + "}", + "+", + "!", + "@", + "#", + "$", + "%", + "^", + "&", + "=", + "|", + "\\", + "~", + "`", + ] + + return any(c in term for c in needs_quoting_chars) + + def _prepare_single_term(self, term: str, is_prefix: bool = True) -> str: + """Prepare a single search term (no Boolean operators). + + Args: + term: A single search term + is_prefix: Whether to add prefix search capability (* suffix) + + Returns: + A properly formatted single term + """ + if not term or not term.strip(): + return term + + term = term.strip() + + # Check if term is already a proper wildcard pattern (alphanumeric + *) + # e.g., "hello*", "test*world" - these should be left alone + if "*" in term and all(c.isalnum() or c in "*_-" for c in term): + return term + + # Natural-language queries arrive with sentence punctuation that FTS5 + # treats as syntax ("When did Melanie paint a sunrise?"). The tokenizer + # ignores this punctuation in the INDEX, so stripping it from word + # edges loses nothing — but leaving it forces the whole question into + # an exact-phrase match that returns zero rows, silently disabling the + # FTS half of hybrid search. Interior characters (hyphens, slashes — + # permalinks and paths) are untouched. + if " " in term: + words = [word.strip("?!.,;:") for word in term.split()] + term = " ".join(word for word in words if word) + if not term: + return "" + + # Characters that can cause FTS5 syntax errors when used as operators + # We're more conservative here - only quote when we detect problematic patterns + problematic_chars = [ + '"', + "'", + "(", + ")", + "[", + "]", + "{", + "}", + "+", + "!", + "@", + "#", + "$", + "%", + "^", + "&", + "=", + "|", + "\\", + "~", + "`", + ] + + # Characters that indicate we should quote (spaces, dots, colons, etc.) + # Adding hyphens here because FTS5 can have issues with hyphens followed by wildcards + needs_quoting_chars = [" ", ".", ":", ";", ",", "<", ">", "?", "/", "-"] + + # Check if term needs quoting + has_problematic = any(c in term for c in problematic_chars) + has_spaces_or_special = any(c in term for c in needs_quoting_chars) + + if has_problematic or has_spaces_or_special: + # Handle multi-word queries differently from special character queries + if " " in term and not any(c in term for c in problematic_chars): + # Check if any individual word contains special characters that need quoting + words = term.strip().split() + has_special_in_words = any( + any(c in word for c in needs_quoting_chars if c != " ") for word in words + ) + + if not has_special_in_words: + # For multi-word queries with simple words (like "emoji unicode"), + # use boolean AND to handle word order variations + if is_prefix: + # Add prefix wildcard to each word for better matching + prepared_words = [f"{word}*" for word in words if word] + else: + prepared_words = words + term = " AND ".join(prepared_words) + else: + # If any word has special characters, quote the entire phrase + escaped_term = term.replace('"', '""') + if is_prefix and not ("/" in term and term.endswith(".md")): + term = f'"{escaped_term}"*' + else: + term = f'"{escaped_term}"' # pragma: no cover + else: + # For terms with problematic characters or file paths, use exact phrase matching + # Escape any existing quotes by doubling them + escaped_term = term.replace('"', '""') + # Quote the entire term to handle special characters safely + if is_prefix and not ("/" in term and term.endswith(".md")): + # For search terms (not file paths), add prefix matching + term = f'"{escaped_term}"*' + else: + # For file paths, use exact matching + term = f'"{escaped_term}"' + elif is_prefix: + # Only add wildcard for simple terms without special characters + term = f"{term}*" + + return term + + def _prepare_search_term(self, term: str, is_prefix: bool = True) -> str: + """Prepare a search term for FTS5 query. + + Args: + term: The search term to prepare + is_prefix: Whether to add prefix search capability (* suffix) + + For FTS5: + - Boolean operators (AND, OR, NOT) are preserved for complex queries + - Terms with FTS5 special characters are quoted to prevent syntax errors + - Simple terms get prefix wildcards for better matching + """ + # Check for explicit boolean operators - if present, process as Boolean query + boolean_operators = [" AND ", " OR ", " NOT "] + if any(op in f" {term} " for op in boolean_operators): + return self._prepare_boolean_query(term) + + # For non-Boolean queries, use the single term preparation logic + return self._prepare_single_term(term, is_prefix) + + @staticmethod + def _relaxed_fts_term(word: str) -> str: + """Render one relaxed word as an FTS5-safe prefix expression. + + A word token can contain an apostrophe ("об'єкт", "don't"). Interpolated + bare it is FTS5 syntax, not text: the whole expression fails to parse, the + caller swallows the syntax error, and the relaxed retry returns nothing — + the exact silent-empty-FTS failure this fallback exists to prevent. + """ + if "'" in word or '"' in word: + return '"{}"*'.format(word.replace('"', '""')) + return f"{word}*" + + @staticmethod + def _relaxed_fts_text(search_text: Optional[str]) -> Optional[str]: + """OR-relaxed FTS5 expression for a failed strict query, or None.""" + words = relaxed_query_words(search_text) + if not words: + return None + return " OR ".join(SQLiteSearchQuery._relaxed_fts_term(word) for word in words) + + async def _build_fts_query_parts( + self, + search_text: Optional[str] = None, + permalink: Optional[str] = None, + permalink_match: Optional[str] = None, + title: Optional[str] = None, + note_types: Optional[List[str]] = None, + after_date: Optional[datetime] = None, + search_item_types: Optional[List[SearchItemType]] = None, + categories: Optional[List[str]] = None, + metadata_filters: Optional[dict[str, Any]] = None, + file_path_prefix: Optional[str] = None, + temporal: Optional[TemporalFilter] = None, + candidate_keys: Sequence[SearchIndexKey] | None = None, + ) -> tuple[str, str, dict[str, Any], str, str]: + """Build SQLite FTS FROM/WHERE params shared by search and count.""" + conditions = [] + match_conditions = [] + params: dict[str, Any] = dict(self._scope_params) + order_by_clause = "" + from_clause = "search_index" + score_expression = "bm25(search_index)" + preserve_match_score = False + + # Handle text search for title and content + if search_text: + # Skip FTS for wildcard-only queries that would cause "unknown special query" errors + if search_text.strip() == "*" or search_text.strip() == "": + # For wildcard searches, don't add any text conditions - return all results + pass + else: + script_query = analyze_script_query(search_text.strip()) + # Trigger: the query contains text from an unsegmented script. + # Why: the script channel needs one table-level MATCH alongside word fields. + # Outcome: mixed queries rank all terms together; word-only queries retain their + # established per-column matching and ranking behavior. + if script_query.gram_phrases: + preserve_match_score = True + params["text"] = "" + params["script_text"] = "" + if script_query.word_text: + prepared_text = self._prepare_search_term(script_query.word_text) + params["text"] = ( + f"(title: ({prepared_text}) OR " + f"content_stems: ({prepared_text}) OR " + f"content_snippet: ({prepared_text}))" + ) + script_phrases = " AND ".join( + f'"{" ".join(phrase)}"' for phrase in script_query.gram_phrases + ) + script_clause = f"script_ngrams: ({script_phrases})" + params["script_text"] = ( + f" AND ({script_clause})" if script_query.word_text else script_clause + ) + match_conditions.append("search_index MATCH (:text || :script_text)") + else: + word_text = ( + script_query.word_text + if script_query.word_text is not None + else search_text.strip() + ) + processed_text = self._prepare_search_term(word_text) + params["text"] = processed_text + # content_stems is capped for Postgres index-row compatibility, while + # SQLite stores the complete note body in its FTS5 content_snippet column. + match_conditions.append( + "(search_index.title MATCH :text OR " + "search_index.content_stems MATCH :text OR " + "search_index.content_snippet MATCH :text)" + ) + + # Handle title match search + if title: + title_text = self._prepare_search_term(title.strip(), is_prefix=False) + params["title_text"] = title_text + match_conditions.append("search_index.title MATCH :title_text") + + # Handle permalink exact search + if permalink: + params["permalink"] = permalink + conditions.append("search_index.permalink = :permalink") + + # Handle permalink match search, supports * + if permalink_match: + # For GLOB patterns, don't use _prepare_search_term as it will quote slashes + # GLOB patterns need to preserve their syntax + permalink_text = permalink_match.lower().strip() + params["permalink"] = permalink_text + if "*" in permalink_match: + conditions.append("search_index.permalink GLOB :permalink") + else: + # For exact matches without *, we can use FTS5 MATCH + # but only prepare the term if it doesn't look like a path + if "/" in permalink_text: + conditions.append("search_index.permalink = :permalink") + else: + permalink_text = self._prepare_search_term(permalink_text, is_prefix=False) + params["permalink"] = permalink_text + match_conditions.append("search_index.permalink MATCH :permalink") + + # Handle directory subtree scope. The predicate is built by the shared + # helper so SQLite and Postgres scope by the identical rule; see + # file_path_prefix_condition for the boundary and escaping reasoning. + subtree_condition = file_path_prefix_condition(file_path_prefix, params) + if subtree_condition is not None: + conditions.append(subtree_condition) + + # Handle an explicit candidate-row restriction. Built by the shared helper so + # both backends restrict by the identical rule; see + # candidate_key_restriction_condition for why the vector filter pass asks about + # its candidates rather than paging the filter's whole match set (#1431). + if candidate_keys is not None: + conditions.append(candidate_key_restriction_condition(candidate_keys, params)) + + # Handle entity type filter (parameterized for defense-in-depth) + if search_item_types: + type_placeholders = [] + for idx, t in enumerate(search_item_types): + param_name = f"search_type_{idx}" + params[param_name] = t.value + type_placeholders.append(f":{param_name}") + conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})") + + # Handle observation category filter (parameterized for defense-in-depth). + # Trigger: caller passed `categories` to scope observation results. + # Why: `entity_types=["observation"]` only narrows to the observation row type; + # callers expect exact-category matching, not incidental text matches. + # Outcome: only rows whose indexed category exactly equals a requested value + # survive (entities/relations have NULL category and are excluded). + if categories: + category_placeholders = [] + for idx, category in enumerate(categories): + param_name = f"category_{idx}" + params[param_name] = category + category_placeholders.append(f":{param_name}") + conditions.append(f"search_index.category IN ({', '.join(category_placeholders)})") + + # Handle note type filter (frontmatter type field, parameterized). + # Trigger: caller passed `note_types` to scope by the frontmatter `type` field. + # Why: the type belongs to the note, but only its entity row carries the + # frontmatter; observation and relation rows do not. Reading it off each row + # silently excluded every non-entity row, which made `note_types` combined + # with a valid-time filter unsatisfiable. + # Outcome: resolved through the owning note in one shared builder, so both + # backends ask the same question and observation rows of a matching note + # are admitted. + if note_types: + conditions.append( + build_note_type_predicate( + note_types, + params, + project_scope_sql=self._scope_sql, + note_type_value=SQLITE_NOTE_TYPE_VALUE, + ) + ) + + # Handle date filter using datetime() for proper comparison + if after_date: + params["after_date"] = after_date + # Filter on updated_at so recently-edited notes are included even when created_at is old + conditions.append("datetime(search_index.updated_at) > datetime(:after_date)") + + # order by most recent first + order_by_clause = ", search_index.updated_at DESC" + + # Handle authored valid time (SPEC-82). + # Trigger: caller asked when a statement was true of the world. + # Why: `after_date` above filters `updated_at`, which records when the note was + # last edited. That is bookkeeping, never a semantic claim; a decision + # effective through July says nothing about when its file was touched. + # Outcome: an independent predicate over the temporal projection. It matches + # only sources carrying a structured qualifier, so undated sources are + # excluded whenever a valid-time filter is present, and no ordering + # changes -- relevance still decides the ranking. + if temporal is not None: + conditions.append( + build_temporal_predicate(temporal, params, project_scope_sql=self._scope_sql) + ) + + # Handle structured metadata filters (frontmatter) + if metadata_filters: + parsed_filters = parse_metadata_filters(metadata_filters) + from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id" + # Frontmatter filters answer for notes only; see + # metadata_filter_content_type_condition for why every regular file + # would otherwise satisfy a null predicate. + conditions.append(metadata_filter_content_type_condition(params)) + entity_columns = await self._get_entity_columns() + + for idx, filt in enumerate(parsed_filters): + path_param = f"meta_path_{idx}" + extract_expr = None + use_tags_column = False + + if filt.path_parts == ["status"] and "frontmatter_status" in entity_columns: + extract_expr = "entity.frontmatter_status" + elif filt.path_parts == ["type"] and "frontmatter_type" in entity_columns: + extract_expr = "entity.frontmatter_type" + elif filt.path_parts == ["tags"] and "tags_json" in entity_columns: + extract_expr = "entity.tags_json" + use_tags_column = True + + if extract_expr is None: + params[path_param] = build_sqlite_json_path(filt.path_parts) + extract_expr = f"json_extract(entity.entity_metadata, :{path_param})" + + # json_extract returns SQL NULL both for a missing key and for an + # explicit JSON null, and the generated frontmatter_* columns are + # that same json_extract — so IS NULL means "the note carries no + # value here", the question `{"owner": None}` asks. `= NULL` is + # never true, so equality here would report a confident zero. + if filt.op == "is_null": + conditions.append(f"{extract_expr} IS NULL") + continue + + if filt.op == "eq": + value_param = f"meta_val_{idx}" + params[value_param] = filt.value + conditions.append(f"{extract_expr} = :{value_param}") + continue + + if filt.op == "in": + placeholders = [] + for j, val in enumerate(filt.value): + value_param = f"meta_val_{idx}_{j}" + params[value_param] = val + placeholders.append(f":{value_param}") + conditions.append(f"{extract_expr} IN ({', '.join(placeholders)})") + continue + + if filt.op == "contains": + tag_conditions = [] + for j, val in enumerate(filt.value): + value_param = f"meta_val_{idx}_{j}" + params[value_param] = val + # The exact JSON-membership test is the primary path; the + # substring patterns only reach values stored as array text. + like_condition = metadata_contains_like_condition( + extract_expr, + val, + param_prefix=value_param, + params=params, + ) + json_each_expr = ( + "json_each(entity.tags_json)" + if use_tags_column + else f"json_each(entity.entity_metadata, :{path_param})" + ) + tag_conditions.append( + "(" + f"EXISTS (SELECT 1 FROM {json_each_expr} WHERE value = :{value_param}) " + f"OR {like_condition}" + ")" + ) + conditions.append(" AND ".join(tag_conditions)) + continue + + if filt.op in {"gt", "gte", "lt", "lte", "between"}: + compare_expr = ( + f"CAST({extract_expr} AS REAL)" + if filt.comparison == "numeric" + else extract_expr + ) + + if filt.op == "between": + min_param = f"meta_val_{idx}_min" + max_param = f"meta_val_{idx}_max" + params[min_param] = filt.value[0] + params[max_param] = filt.value[1] + conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}") + else: + value_param = f"meta_val_{idx}" + params[value_param] = filt.value + operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op] + conditions.append(f"{compare_expr} {operator} :{value_param}") + continue + + # Trigger: SQLite rejects some Boolean combinations of MATCH predicates, + # including a word-field OR expression combined with the script channel. + # Why: each MATCH must be evaluated in an FTS-valid query context. + # Outcome: keep one outer MATCH for bm25 ranking and intersect the rest by rowid. + if len(match_conditions) > 1: + ranked_match, *additional_matches = match_conditions + conditions.extend( + f"search_index.rowid IN (SELECT rowid FROM search_index WHERE {match_condition})" + for match_condition in additional_matches + ) + match_conditions = [ranked_match] + + # Trigger: SQLite FTS MATCH predicates combined with JOINs can fail with + # "unable to use function MATCH in the requested context". + # Why: script queries need MATCH and bm25 together for ranking, while legacy + # word-column OR predicates cannot evaluate bm25 in the same derived query. + # Outcome: rank script matches before joining metadata; retain the established + # rowid-filter path for word-only searches. + if metadata_filters and match_conditions: + match_where = " AND ".join(match_conditions) + if preserve_match_score: + from_clause = ( + "(SELECT search_index.rowid AS rowid, search_index.*, " + "bm25(search_index) AS fts_score " + f"FROM search_index WHERE {match_where}) AS search_index " + "JOIN entity ON search_index.entity_id = entity.id" + ) + score_expression = "search_index.fts_score" + else: + conditions.append( + f"search_index.rowid IN (SELECT rowid FROM search_index WHERE {match_where})" + ) + else: + conditions.extend(match_conditions) + + conditions.append(f"search_index.project_id {self._scope_sql}") + + # Build WHERE clause + where_clause = " AND ".join(conditions) if conditions else "1=1" + return from_clause, where_clause, params, order_by_clause, score_expression diff --git a/src/basic_memory/repository/sqlite_search_repository.py b/src/basic_memory/repository/sqlite_search_repository.py index 4160d7932..cac93d965 100644 --- a/src/basic_memory/repository/sqlite_search_repository.py +++ b/src/basic_memory/repository/sqlite_search_repository.py @@ -1,7 +1,6 @@ """SQLite FTS5-based search repository implementation.""" import asyncio -import re import time from collections.abc import Sequence from contextlib import asynccontextmanager @@ -15,6 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from basic_memory import db +from basic_memory.repository.sqlite_search_query import SQLiteSearchQuery from basic_memory.config import BasicMemoryConfig, ConfigManager from basic_memory.models.search import ( CREATE_SEARCH_INDEX, @@ -31,22 +31,12 @@ from basic_memory.repository.search_repository_base import ( SearchIndexKey, SearchRepositoryBase, - candidate_key_restriction_condition, - file_path_prefix_condition, - metadata_contains_like_condition, - metadata_filter_content_type_condition, ) from basic_memory.repository.script_ngrams import analyze_script_query from basic_memory.repository.search_trace import ( SearchTraceCollector, build_fts_page_stage, ) -from basic_memory.repository.metadata_filters import parse_metadata_filters, build_sqlite_json_path -from basic_memory.repository.note_type_filters import ( - SQLITE_NOTE_TYPE_VALUE, - build_note_type_predicate, -) -from basic_memory.repository.temporal_filters import build_temporal_predicate from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError from basic_memory.repository.semantic_vector_index import SemanticVectorIndex from basic_memory.repository.semantic_vector_sync import StagedVectorDeletion @@ -59,7 +49,7 @@ SQLITE_WORD_COLUMNS = "{title content_stems content_snippet}" -class SQLiteSearchRepository(SearchRepositoryBase): +class SQLiteSearchRepository(SQLiteSearchQuery, SearchRepositoryBase): """SQLite FTS5 implementation of search repository. Uses SQLite's FTS5 virtual tables for full-text search with: @@ -79,7 +69,8 @@ def __init__( vector_index: SemanticVectorIndex | None = None, rerank_provider: RerankProvider | None = None, ): - super().__init__(session_maker, project_id) + SearchRepositoryBase.__init__(self, session_maker, project_id) + SQLiteSearchQuery.__init__(self, session_maker, (project_id,)) self._entity_columns: set[str] | None = None self._app_config = app_config or ConfigManager().config self._semantic_enabled = self._app_config.semantic_search_enabled @@ -117,13 +108,6 @@ def __init__( ), ) - async def _get_entity_columns(self) -> set[str]: - if self._entity_columns is None: - async with db.scoped_session(self.session_maker) as session: - result = await session.execute(text("PRAGMA table_info(entity)")) - self._entity_columns = {row[1] for row in result.fetchall()} - return self._entity_columns - @override async def init_search_index(self): """Create FTS5 virtual table for search if it doesn't exist. @@ -164,277 +148,6 @@ async def init_search_index(self): # FTS5 query preparation (backend-specific) # ------------------------------------------------------------------ - def _prepare_boolean_query(self, query: str) -> str: - """Prepare a Boolean query by quoting individual terms while preserving operators. - - Args: - query: A Boolean query like "tier1-test AND unicode" or "(hello OR world) NOT test" - - Returns: - A properly formatted Boolean query with quoted terms that need quoting - """ - # Define Boolean operators and their boundaries - boolean_pattern = r"(\bAND\b|\bOR\b|\bNOT\b)" - - # Split the query by Boolean operators, keeping the operators - parts = re.split(boolean_pattern, query) - - processed_parts = [] - for part in parts: - part = part.strip() - if not part: - continue - - # If it's a Boolean operator, keep it as is - if part in ["AND", "OR", "NOT"]: - processed_parts.append(part) - else: - # Handle parentheses specially - they should be preserved for grouping - if "(" in part or ")" in part: - # Parse parenthetical expressions carefully - processed_part = self._prepare_parenthetical_term(part) - processed_parts.append(processed_part) - else: - # This is a search term - for Boolean queries, don't add prefix wildcards - prepared_term = self._prepare_single_term(part, is_prefix=False) - processed_parts.append(prepared_term) - - return " ".join(processed_parts) - - def _prepare_parenthetical_term(self, term: str) -> str: - """Prepare a term that contains parentheses, preserving the parentheses for grouping. - - Args: - term: A term that may contain parentheses like "(hello" or "world)" or "(hello OR world)" - - Returns: - A properly formatted term with parentheses preserved - """ - # Handle terms that start/end with parentheses but may contain quotable content - result = "" - i = 0 - while i < len(term): - if term[i] in "()": - # Preserve parentheses as-is - result += term[i] - i += 1 - else: - # Find the next parenthesis or end of string - start = i - while i < len(term) and term[i] not in "()": - i += 1 - - # Extract the content between parentheses - content = term[start:i].strip() - if content: - # Only quote if it actually needs quoting (has hyphens, special chars, etc) - # but don't quote if it's just simple words - if self._needs_quoting(content): - escaped_content = content.replace('"', '""') - result += f'"{escaped_content}"' - else: - result += content - - return result - - def _needs_quoting(self, term: str) -> bool: - """Check if a term needs to be quoted for FTS5 safety. - - Args: - term: The term to check - - Returns: - True if the term should be quoted - """ - if not term or not term.strip(): - return False - - # Characters that indicate we should quote (excluding parentheses which are valid syntax) - needs_quoting_chars = [ - " ", - ".", - ":", - ";", - ",", - "<", - ">", - "?", - "/", - "-", - "'", - '"', - "[", - "]", - "{", - "}", - "+", - "!", - "@", - "#", - "$", - "%", - "^", - "&", - "=", - "|", - "\\", - "~", - "`", - ] - - return any(c in term for c in needs_quoting_chars) - - def _prepare_single_term(self, term: str, is_prefix: bool = True) -> str: - """Prepare a single search term (no Boolean operators). - - Args: - term: A single search term - is_prefix: Whether to add prefix search capability (* suffix) - - Returns: - A properly formatted single term - """ - if not term or not term.strip(): - return term - - term = term.strip() - - # Check if term is already a proper wildcard pattern (alphanumeric + *) - # e.g., "hello*", "test*world" - these should be left alone - if "*" in term and all(c.isalnum() or c in "*_-" for c in term): - return term - - # Natural-language queries arrive with sentence punctuation that FTS5 - # treats as syntax ("When did Melanie paint a sunrise?"). The tokenizer - # ignores this punctuation in the INDEX, so stripping it from word - # edges loses nothing — but leaving it forces the whole question into - # an exact-phrase match that returns zero rows, silently disabling the - # FTS half of hybrid search. Interior characters (hyphens, slashes — - # permalinks and paths) are untouched. - if " " in term: - words = [word.strip("?!.,;:") for word in term.split()] - term = " ".join(word for word in words if word) - if not term: - return "" - - # Characters that can cause FTS5 syntax errors when used as operators - # We're more conservative here - only quote when we detect problematic patterns - problematic_chars = [ - '"', - "'", - "(", - ")", - "[", - "]", - "{", - "}", - "+", - "!", - "@", - "#", - "$", - "%", - "^", - "&", - "=", - "|", - "\\", - "~", - "`", - ] - - # Characters that indicate we should quote (spaces, dots, colons, etc.) - # Adding hyphens here because FTS5 can have issues with hyphens followed by wildcards - needs_quoting_chars = [" ", ".", ":", ";", ",", "<", ">", "?", "/", "-"] - - # Check if term needs quoting - has_problematic = any(c in term for c in problematic_chars) - has_spaces_or_special = any(c in term for c in needs_quoting_chars) - - if has_problematic or has_spaces_or_special: - # Handle multi-word queries differently from special character queries - if " " in term and not any(c in term for c in problematic_chars): - # Check if any individual word contains special characters that need quoting - words = term.strip().split() - has_special_in_words = any( - any(c in word for c in needs_quoting_chars if c != " ") for word in words - ) - - if not has_special_in_words: - # For multi-word queries with simple words (like "emoji unicode"), - # use boolean AND to handle word order variations - if is_prefix: - # Add prefix wildcard to each word for better matching - prepared_words = [f"{word}*" for word in words if word] - else: - prepared_words = words - term = " AND ".join(prepared_words) - else: - # If any word has special characters, quote the entire phrase - escaped_term = term.replace('"', '""') - if is_prefix and not ("/" in term and term.endswith(".md")): - term = f'"{escaped_term}"*' - else: - term = f'"{escaped_term}"' # pragma: no cover - else: - # For terms with problematic characters or file paths, use exact phrase matching - # Escape any existing quotes by doubling them - escaped_term = term.replace('"', '""') - # Quote the entire term to handle special characters safely - if is_prefix and not ("/" in term and term.endswith(".md")): - # For search terms (not file paths), add prefix matching - term = f'"{escaped_term}"*' - else: - # For file paths, use exact matching - term = f'"{escaped_term}"' - elif is_prefix: - # Only add wildcard for simple terms without special characters - term = f"{term}*" - - return term - - @override - def _prepare_search_term(self, term: str, is_prefix: bool = True) -> str: - """Prepare a search term for FTS5 query. - - Args: - term: The search term to prepare - is_prefix: Whether to add prefix search capability (* suffix) - - For FTS5: - - Boolean operators (AND, OR, NOT) are preserved for complex queries - - Terms with FTS5 special characters are quoted to prevent syntax errors - - Simple terms get prefix wildcards for better matching - """ - # Check for explicit boolean operators - if present, process as Boolean query - boolean_operators = [" AND ", " OR ", " NOT "] - if any(op in f" {term} " for op in boolean_operators): - return self._prepare_boolean_query(term) - - # For non-Boolean queries, use the single term preparation logic - return self._prepare_single_term(term, is_prefix) - - @staticmethod - def _relaxed_fts_term(word: str) -> str: - """Render one relaxed word as an FTS5-safe prefix expression. - - A word token can contain an apostrophe ("об'єкт", "don't"). Interpolated - bare it is FTS5 syntax, not text: the whole expression fails to parse, the - caller swallows the syntax error, and the relaxed retry returns nothing — - the exact silent-empty-FTS failure this fallback exists to prevent. - """ - if "'" in word or '"' in word: - return '"{}"*'.format(word.replace('"', '""')) - return f"{word}*" - - @staticmethod - def _relaxed_fts_text(search_text: Optional[str]) -> Optional[str]: - """OR-relaxed FTS5 expression for a failed strict query, or None.""" - words = relaxed_query_words(search_text) - if not words: - return None - return " OR ".join(SQLiteSearchRepository._relaxed_fts_term(word) for word in words) - @override async def semantic_effectively_enabled(self) -> bool: """Probe the sqlite-vec runtime instead of trusting still-enabled config. @@ -784,321 +497,6 @@ async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> Non def _is_fts5_syntax_error(exc: Exception) -> bool: return "fts5: syntax error" in str(exc).lower() - async def _build_fts_query_parts( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - candidate_keys: Sequence[SearchIndexKey] | None = None, - ) -> tuple[str, str, dict[str, Any], str, str]: - """Build SQLite FTS FROM/WHERE params shared by search and count.""" - conditions = [] - match_conditions = [] - params = {} - order_by_clause = "" - from_clause = "search_index" - score_expression = "bm25(search_index)" - preserve_match_score = False - - # Handle text search for title and content - if search_text: - # Skip FTS for wildcard-only queries that would cause "unknown special query" errors - if search_text.strip() == "*" or search_text.strip() == "": - # For wildcard searches, don't add any text conditions - return all results - pass - else: - script_query = analyze_script_query(search_text.strip()) - # Trigger: the query contains text from an unsegmented script. - # Why: the script channel needs one table-level MATCH alongside word fields. - # Outcome: mixed queries rank all terms together; word-only queries retain their - # established per-column matching and ranking behavior. - if script_query.gram_phrases: - preserve_match_score = True - params["text"] = "" - params["script_text"] = "" - if script_query.word_text: - prepared_text = self._prepare_search_term(script_query.word_text) - params["text"] = ( - f"(title: ({prepared_text}) OR " - f"content_stems: ({prepared_text}) OR " - f"content_snippet: ({prepared_text}))" - ) - script_phrases = " AND ".join( - f'"{" ".join(phrase)}"' for phrase in script_query.gram_phrases - ) - script_clause = f"script_ngrams: ({script_phrases})" - params["script_text"] = ( - f" AND ({script_clause})" if script_query.word_text else script_clause - ) - match_conditions.append("search_index MATCH (:text || :script_text)") - else: - word_text = ( - script_query.word_text - if script_query.word_text is not None - else search_text.strip() - ) - processed_text = self._prepare_search_term(word_text) - params["text"] = processed_text - # content_stems is capped for Postgres index-row compatibility, while - # SQLite stores the complete note body in its FTS5 content_snippet column. - match_conditions.append( - "(search_index.title MATCH :text OR " - "search_index.content_stems MATCH :text OR " - "search_index.content_snippet MATCH :text)" - ) - - # Handle title match search - if title: - title_text = self._prepare_search_term(title.strip(), is_prefix=False) - params["title_text"] = title_text - match_conditions.append("search_index.title MATCH :title_text") - - # Handle permalink exact search - if permalink: - params["permalink"] = permalink - conditions.append("search_index.permalink = :permalink") - - # Handle permalink match search, supports * - if permalink_match: - # For GLOB patterns, don't use _prepare_search_term as it will quote slashes - # GLOB patterns need to preserve their syntax - permalink_text = permalink_match.lower().strip() - params["permalink"] = permalink_text - if "*" in permalink_match: - conditions.append("search_index.permalink GLOB :permalink") - else: - # For exact matches without *, we can use FTS5 MATCH - # but only prepare the term if it doesn't look like a path - if "/" in permalink_text: - conditions.append("search_index.permalink = :permalink") - else: - permalink_text = self._prepare_search_term(permalink_text, is_prefix=False) - params["permalink"] = permalink_text - match_conditions.append("search_index.permalink MATCH :permalink") - - # Handle directory subtree scope. The predicate is built by the shared - # helper so SQLite and Postgres scope by the identical rule; see - # file_path_prefix_condition for the boundary and escaping reasoning. - subtree_condition = file_path_prefix_condition(file_path_prefix, params) - if subtree_condition is not None: - conditions.append(subtree_condition) - - # Handle an explicit candidate-row restriction. Built by the shared helper so - # both backends restrict by the identical rule; see - # candidate_key_restriction_condition for why the vector filter pass asks about - # its candidates rather than paging the filter's whole match set (#1431). - if candidate_keys is not None: - conditions.append(candidate_key_restriction_condition(candidate_keys, params)) - - # Handle entity type filter (parameterized for defense-in-depth) - if search_item_types: - type_placeholders = [] - for idx, t in enumerate(search_item_types): - param_name = f"search_type_{idx}" - params[param_name] = t.value - type_placeholders.append(f":{param_name}") - conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})") - - # Handle observation category filter (parameterized for defense-in-depth). - # Trigger: caller passed `categories` to scope observation results. - # Why: `entity_types=["observation"]` only narrows to the observation row type; - # callers expect exact-category matching, not incidental text matches. - # Outcome: only rows whose indexed category exactly equals a requested value - # survive (entities/relations have NULL category and are excluded). - if categories: - category_placeholders = [] - for idx, category in enumerate(categories): - param_name = f"category_{idx}" - params[param_name] = category - category_placeholders.append(f":{param_name}") - conditions.append(f"search_index.category IN ({', '.join(category_placeholders)})") - - # Handle note type filter (frontmatter type field, parameterized). - # Trigger: caller passed `note_types` to scope by the frontmatter `type` field. - # Why: the type belongs to the note, but only its entity row carries the - # frontmatter; observation and relation rows do not. Reading it off each row - # silently excluded every non-entity row, which made `note_types` combined - # with a valid-time filter unsatisfiable. - # Outcome: resolved through the owning note in one shared builder, so both - # backends ask the same question and observation rows of a matching note - # are admitted. - if note_types: - conditions.append( - build_note_type_predicate( - note_types, params, note_type_value=SQLITE_NOTE_TYPE_VALUE - ) - ) - - # Handle date filter using datetime() for proper comparison - if after_date: - params["after_date"] = after_date - # Filter on updated_at so recently-edited notes are included even when created_at is old - conditions.append("datetime(search_index.updated_at) > datetime(:after_date)") - - # order by most recent first - order_by_clause = ", search_index.updated_at DESC" - - # Handle authored valid time (SPEC-82). - # Trigger: caller asked when a statement was true of the world. - # Why: `after_date` above filters `updated_at`, which records when the note was - # last edited. That is bookkeeping, never a semantic claim; a decision - # effective through July says nothing about when its file was touched. - # Outcome: an independent predicate over the temporal projection. It matches - # only sources carrying a structured qualifier, so undated sources are - # excluded whenever a valid-time filter is present, and no ordering - # changes -- relevance still decides the ranking. - if temporal is not None: - conditions.append(build_temporal_predicate(temporal, params)) - - # Handle structured metadata filters (frontmatter) - if metadata_filters: - parsed_filters = parse_metadata_filters(metadata_filters) - from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id" - # Frontmatter filters answer for notes only; see - # metadata_filter_content_type_condition for why every regular file - # would otherwise satisfy a null predicate. - conditions.append(metadata_filter_content_type_condition(params)) - entity_columns = await self._get_entity_columns() - - for idx, filt in enumerate(parsed_filters): - path_param = f"meta_path_{idx}" - extract_expr = None - use_tags_column = False - - if filt.path_parts == ["status"] and "frontmatter_status" in entity_columns: - extract_expr = "entity.frontmatter_status" - elif filt.path_parts == ["type"] and "frontmatter_type" in entity_columns: - extract_expr = "entity.frontmatter_type" - elif filt.path_parts == ["tags"] and "tags_json" in entity_columns: - extract_expr = "entity.tags_json" - use_tags_column = True - - if extract_expr is None: - params[path_param] = build_sqlite_json_path(filt.path_parts) - extract_expr = f"json_extract(entity.entity_metadata, :{path_param})" - - # json_extract returns SQL NULL both for a missing key and for an - # explicit JSON null, and the generated frontmatter_* columns are - # that same json_extract — so IS NULL means "the note carries no - # value here", the question `{"owner": None}` asks. `= NULL` is - # never true, so equality here would report a confident zero. - if filt.op == "is_null": - conditions.append(f"{extract_expr} IS NULL") - continue - - if filt.op == "eq": - value_param = f"meta_val_{idx}" - params[value_param] = filt.value - conditions.append(f"{extract_expr} = :{value_param}") - continue - - if filt.op == "in": - placeholders = [] - for j, val in enumerate(filt.value): - value_param = f"meta_val_{idx}_{j}" - params[value_param] = val - placeholders.append(f":{value_param}") - conditions.append(f"{extract_expr} IN ({', '.join(placeholders)})") - continue - - if filt.op == "contains": - tag_conditions = [] - for j, val in enumerate(filt.value): - value_param = f"meta_val_{idx}_{j}" - params[value_param] = val - # The exact JSON-membership test is the primary path; the - # substring patterns only reach values stored as array text. - like_condition = metadata_contains_like_condition( - extract_expr, - val, - param_prefix=value_param, - params=params, - ) - json_each_expr = ( - "json_each(entity.tags_json)" - if use_tags_column - else f"json_each(entity.entity_metadata, :{path_param})" - ) - tag_conditions.append( - "(" - f"EXISTS (SELECT 1 FROM {json_each_expr} WHERE value = :{value_param}) " - f"OR {like_condition}" - ")" - ) - conditions.append(" AND ".join(tag_conditions)) - continue - - if filt.op in {"gt", "gte", "lt", "lte", "between"}: - compare_expr = ( - f"CAST({extract_expr} AS REAL)" - if filt.comparison == "numeric" - else extract_expr - ) - - if filt.op == "between": - min_param = f"meta_val_{idx}_min" - max_param = f"meta_val_{idx}_max" - params[min_param] = filt.value[0] - params[max_param] = filt.value[1] - conditions.append(f"{compare_expr} BETWEEN :{min_param} AND :{max_param}") - else: - value_param = f"meta_val_{idx}" - params[value_param] = filt.value - operator = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="}[filt.op] - conditions.append(f"{compare_expr} {operator} :{value_param}") - continue - - # Trigger: SQLite rejects some Boolean combinations of MATCH predicates, - # including a word-field OR expression combined with the script channel. - # Why: each MATCH must be evaluated in an FTS-valid query context. - # Outcome: keep one outer MATCH for bm25 ranking and intersect the rest by rowid. - if len(match_conditions) > 1: - ranked_match, *additional_matches = match_conditions - conditions.extend( - f"search_index.rowid IN (SELECT rowid FROM search_index WHERE {match_condition})" - for match_condition in additional_matches - ) - match_conditions = [ranked_match] - - # Trigger: SQLite FTS MATCH predicates combined with JOINs can fail with - # "unable to use function MATCH in the requested context". - # Why: script queries need MATCH and bm25 together for ranking, while legacy - # word-column OR predicates cannot evaluate bm25 in the same derived query. - # Outcome: rank script matches before joining metadata; retain the established - # rowid-filter path for word-only searches. - if metadata_filters and match_conditions: - match_where = " AND ".join(match_conditions) - if preserve_match_score: - from_clause = ( - "(SELECT search_index.rowid AS rowid, search_index.*, " - "bm25(search_index) AS fts_score " - f"FROM search_index WHERE {match_where}) AS search_index " - "JOIN entity ON search_index.entity_id = entity.id" - ) - score_expression = "search_index.fts_score" - else: - conditions.append( - f"search_index.rowid IN (SELECT rowid FROM search_index WHERE {match_where})" - ) - else: - conditions.extend(match_conditions) - - # Always filter by project_id - params["project_id"] = self.project_id - conditions.append("search_index.project_id = :project_id") - - # Build WHERE clause - where_clause = " AND ".join(conditions) if conditions else "1=1" - return from_clause, where_clause, params, order_by_clause, score_expression - @override async def search( self, diff --git a/src/basic_memory/repository/temporal_filters.py b/src/basic_memory/repository/temporal_filters.py index 81c3c666c..8e507c529 100644 --- a/src/basic_memory/repository/temporal_filters.py +++ b/src/basic_memory/repository/temporal_filters.py @@ -84,7 +84,9 @@ def _not_window_ends_before_source(window: TemporalRange) -> str | None: return f"({' OR '.join(clauses)})" -def build_temporal_predicate(temporal: TemporalFilter, params: dict[str, Any]) -> str: +def build_temporal_predicate( + temporal: TemporalFilter, params: dict[str, Any], *, project_scope_sql: str = "= :project_id" +) -> str: """Build the WHERE-clause fragment restricting search rows by authored valid time. Two intervals overlap exactly when neither lies entirely before the other, which @@ -105,7 +107,7 @@ def build_temporal_predicate(temporal: TemporalFilter, params: dict[str, Any]) - # false constant is both correct and cheaper than running the subquery. return _MATCHES_NOTHING - conditions = [f"{TEMPORAL_INDEX_TABLE}.project_id = :project_id"] + conditions = [f"{TEMPORAL_INDEX_TABLE}.project_id {project_scope_sql}"] if temporal.kind is not None: params["tq_kind"] = temporal.kind.value @@ -138,8 +140,8 @@ def build_temporal_predicate(temporal: TemporalFilter, params: dict[str, Any]) - # (type, id) is the search row's own identity and the address this projection # stores, so the pair joins the two without a correlated reference. return ( - "(search_index.type, search_index.id) IN (\n" - f" SELECT {TEMPORAL_INDEX_TABLE}.source_type, {TEMPORAL_INDEX_TABLE}.source_id\n" + "(search_index.project_id, search_index.type, search_index.id) IN (\n" + f" SELECT {TEMPORAL_INDEX_TABLE}.project_id, {TEMPORAL_INDEX_TABLE}.source_type, {TEMPORAL_INDEX_TABLE}.source_id\n" f" FROM {TEMPORAL_INDEX_TABLE}\n" f" WHERE {where_clause})" ) diff --git a/src/basic_memory/schemas/multi_project_search.py b/src/basic_memory/schemas/multi_project_search.py new file mode 100644 index 000000000..50af5c2cd --- /dev/null +++ b/src/basic_memory/schemas/multi_project_search.py @@ -0,0 +1,28 @@ +"""API-only database search contract; effective authorization scope is required.""" + +from typing import Annotated + +from pydantic import BaseModel, Field + +from basic_memory.schemas.search import SearchQuery, SearchResult + + +class MultiProjectSearchQuery(SearchQuery): + """Cloud resolves authorization before passing internal database project IDs.""" + + project_ids: list[Annotated[int, Field(strict=True, gt=0)]] + + +class MultiProjectSearchResult(SearchResult): + project_id: int + project_external_id: str + + +class MultiProjectSearchResponse(BaseModel): + results: list[MultiProjectSearchResult] + current_page: int + page_size: int + total: int = 0 + total_is_exact: bool = True + has_more: bool = False + temporal_applied: bool | None = None diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 301e643f2..9c6396dde 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -4,7 +4,7 @@ import ast import re from collections.abc import Mapping, Sequence -from dataclasses import dataclass, replace +from dataclasses import replace from datetime import datetime from typing import Any, List, Optional, Set, Dict @@ -23,7 +23,10 @@ SearchIndexRow, SearchRepository, ) -from basic_memory.repository.search_query import relaxed_query_words +from basic_memory.repository.search_query import ( + PreparedSearchQuery as PreparedSearchQuery, + relaxed_query_words, +) from basic_memory.repository.search_trace import SearchTraceCollector from basic_memory.schemas.base import normalize_note_type from basic_memory.schemas.search import SearchQuery, SearchItemType, SearchRetrievalMode @@ -42,25 +45,6 @@ MAX_CONTENT_STEMS_SIZE = 6000 -@dataclass(frozen=True) -class PreparedSearchQuery: - """Normalized query inputs shared by search and count.""" - - search_text: str | None - permalink: str | None - permalink_match: str | None - title: str | None - note_types: list[str] | None - search_item_types: list[SearchItemType] | None - categories: list[str] | None - after_date: datetime | None - metadata_filters: dict[str, Any] | None - file_path_prefix: str | None - temporal: TemporalFilter | None - retrieval_mode: SearchRetrievalMode - min_similarity: float | None - - def entity_embeddings_enabled(entity: Entity) -> bool: """Return whether semantic embeddings should be generated for this entity. @@ -213,7 +197,8 @@ async def reindex_all(self, background_tasks: Optional[BackgroundTasks] = None) logger.info("Reindex complete") - def prepare_query(self, query: SearchQuery) -> PreparedSearchQuery | None: + @staticmethod + def prepare_query(query: SearchQuery) -> PreparedSearchQuery | None: """Normalize a SearchQuery into repository arguments.""" search_text = query.text tags = query.tags diff --git a/test-int/test_multi_project_search.py b/test-int/test_multi_project_search.py new file mode 100644 index 000000000..214ae2a91 --- /dev/null +++ b/test-int/test_multi_project_search.py @@ -0,0 +1,531 @@ +"""One database pipeline over real SQLite/sqlite-vec or Postgres/pgvector storage. + +Run with BASIC_MEMORY_TEST_POSTGRES=1 for Postgres. Only the embedding provider +is deterministic; indexing, vector persistence, retrieval, and API hydration are real. +""" + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +import pytest +from fastapi import FastAPI +from httpx import AsyncClient +from sqlalchemy import event, text +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.config import BasicMemoryConfig, DatabaseBackend +from basic_memory.deps.read_cache import get_read_cache +from basic_memory.models import Entity, Project +from basic_memory.repository.multi_project_search_repository import MultiProjectSearchRepository +from basic_memory.repository.postgres_search_repository import PostgresSearchRepository +from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository +from basic_memory.schemas.search import SearchQuery, SearchRetrievalMode +from basic_memory.services.search_service import SearchService + + +class CountingEmbeddingProvider: + model_name = "scope-test" + dimensions = 4 + + def __init__(self) -> None: + self.query_calls = 0 + + async def embed_query(self, text: str) -> list[float]: + self.query_calls += 1 + return [1.0, 0.0, 0.0, 0.0] + + async def embed_documents(self, texts: list[str]) -> list[list[float]]: + return [[1.0, 0.0, 0.0, 0.0] for _ in texts] + + def runtime_log_attrs(self) -> dict[str, Any]: + return {} + + +@dataclass +class Corpus: + session_maker: async_sessionmaker[AsyncSession] + engine: AsyncEngine + config: BasicMemoryConfig + projects: list[Project] + provider: CountingEmbeddingProvider + + def repository(self, project_ids: Sequence[int]) -> MultiProjectSearchRepository: + return MultiProjectSearchRepository( + self.session_maker, + project_ids, + app_config=self.config, + embedding_provider=self.provider, + ) + + +@pytest.fixture +async def corpus( + engine_factory, app_config: BasicMemoryConfig, monkeypatch: pytest.MonkeyPatch +) -> Corpus: + engine, session_maker = engine_factory + app_config.semantic_search_enabled = True + app_config.reranker_enabled = False + app_config.semantic_min_similarity = 0.0 + provider = CountingEmbeddingProvider() + from basic_memory.api.v2.routers import multi_project_search_router + + monkeypatch.setattr( + multi_project_search_router, "create_embedding_provider", lambda _config: provider + ) + projects = [] + now = datetime(2026, 9, 1, tzinfo=timezone.utc) + for index in range(3): + async with db.scoped_session(session_maker) as session: + project = Project( + name=f"scope-{index}", permalink=f"scope-{index}", path=f"/scope-{index}" + ) + session.add(project) + await session.flush() + entity = Entity( + project_id=project.id, + title="shared nebula", + note_type="note", + content_type="text/markdown", + permalink="notes/shared", + file_path="notes/shared.md", + entity_metadata={"status": "open" if index == 0 else "closed", "tags": ["scope"]}, + created_at=now, + updated_at=now, + ) + session.add(entity) + await session.commit() + projects.append(project) + repo_type = ( + PostgresSearchRepository + if app_config.database_backend == DatabaseBackend.POSTGRES + else SQLiteSearchRepository + ) + writer = repo_type( + session_maker, project.id, app_config=app_config, embedding_provider=provider + ) + await writer.init_search_index() + # Search rows have a composite (project, type, id) identity. Deliberately + # collide the observation/relation IDs and paths across all three projects. + rows = [ + SearchIndexRow( + project_id=project.id, + id=entity.id, + entity_id=entity.id, + type="entity", + file_path=entity.file_path, + permalink=entity.permalink, + title=entity.title, + content_stems="shared nebula", + content_snippet="shared nebula", + metadata={"note_type": "note"}, + created_at=now, + updated_at=now, + ) + ] + for row_type in ("observation", "relation"): + rows.append( + SearchIndexRow( + project_id=project.id, + id=500, + entity_id=entity.id, + type=row_type, + file_path=entity.file_path, + permalink=f"notes/shared/{row_type}", + title="shared nebula", + content_stems="shared nebula", + content_snippet=f"shared nebula project {index} {row_type}", + category="fact" if row_type == "observation" else None, + from_id=entity.id if row_type == "relation" else None, + relation_type="relates_to" if row_type == "relation" else None, + created_at=now, + updated_at=now, + ) + ) + await writer.bulk_index_items(rows) + await writer.sync_entity_vectors(entity.id) + return Corpus(session_maker, engine, app_config, projects, provider) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(SearchRetrievalMode)) +@pytest.mark.parametrize("scope_size", [0, 1, 2]) +async def test_explicit_scope_one_pipeline_and_read_only( + corpus: Corpus, mode: SearchRetrievalMode, scope_size: int +) -> None: + ids = [project.id for project in corpus.projects[:scope_size]] + repo = corpus.repository(ids) + prepared = SearchService.prepare_query(SearchQuery(text="nebula", retrieval_mode=mode)) + assert prepared is not None + statements: list[str] = [] + + def record(_conn, _cursor, statement, _params, _context, _many): + statements.append(statement) + + event.listen(corpus.engine.sync_engine, "before_cursor_execute", record) + try: + rows = await repo.search(prepared, limit=100) + finally: + event.remove(corpus.engine.sync_engine, "before_cursor_execute", record) + assert len(rows) == scope_size * 3 + assert {row.project_id for row in rows} == set(ids) + assert len({(row.project_id, row.type, row.id) for row in rows}) == len(rows) + assert corpus.provider.query_calls == int(scope_size > 0 and mode != SearchRetrievalMode.FTS) + retrievals = [ + sql for sql in statements if "search_index" in sql and not sql.startswith("PRAGMA") + ] + assert len(retrievals) == int(scope_size > 0) + assert all( + not sql.lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE", "CREATE", "DROP")) + for sql in statements + ) + assert not hasattr(repo, "index_item") and not hasattr(repo, "delete_by_entity_id") + if mode == SearchRetrievalMode.FTS: + assert await repo.count(prepared) == scope_size * 3 + else: + assert all(row.matched_chunk_text for row in rows) + with pytest.raises(ValueError, match="Exact counts"): + await repo.count(prepared) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(SearchRetrievalMode)) +async def test_global_pagination_is_independent_of_page_size( + corpus: Corpus, mode: SearchRetrievalMode +) -> None: + repo = corpus.repository([project.id for project in corpus.projects[:2]]) + query = SearchService.prepare_query(SearchQuery(text="nebula", retrieval_mode=mode)) + assert query is not None + complete = await repo.search(query, limit=100) + pages = [ + row + for offset in range(0, len(complete), 2) + for row in await repo.search(query, limit=2, offset=offset) + ] + assert [(r.project_id, r.type, r.id, r.score) for r in pages] == [ + (r.project_id, r.type, r.id, r.score) for r in complete + ] + assert await repo.search(query, offset=100) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(SearchRetrievalMode)) +async def test_filters_scope_counts_and_api_hydration( + corpus: Corpus, client: AsyncClient, mode: SearchRetrievalMode +) -> None: + ids = [project.id for project in corpus.projects[:2]] + response = await client.request( + "QUERY", + "/v2/search/", + json={ + "project_ids": ids, + "text": "nebula", + "retrieval_mode": mode.value, + "metadata_filters": {"status": "open"}, + "note_types": ["note"], + "file_path_prefix": "notes", + "entity_types": ["observation"], + "categories": ["fact"], + }, + ) + assert response.status_code == 200, response.text + data = response.json() + assert len(data["results"]) == 1 + row = data["results"][0] + assert row["project_id"] == ids[0] + assert row["project_external_id"] == corpus.projects[0].external_id + assert row["external_id"] and row["observation_id"] == 500 + assert data["total_is_exact"] == (mode == SearchRetrievalMode.FTS) + assert data["total"] == int(mode == SearchRetrievalMode.FTS) + assert not data["has_more"] + assert response.headers["cache-control"] == "no-store" + + +@pytest.mark.asyncio +async def test_route_bypasses_cache_on_scope_change( + corpus: Corpus, client: AsyncClient, app: FastAPI +) -> None: + def forbidden_cache(): + pytest.fail("The database-scoped route must not acquire the project cache") + + app.dependency_overrides[get_read_cache] = forbidden_cache + for ids, count in [([p.id for p in corpus.projects], 9), ([corpus.projects[1].id], 3), ([], 0)]: + response = await client.request( + "QUERY", "/v2/search/", json={"project_ids": ids, "text": "nebula"} + ) + assert response.status_code == 200, response.text + assert response.json()["total"] == count + assert {r["project_id"] for r in response.json()["results"]} <= set(ids) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scope", [None, [0], [True], ["1"]]) +async def test_route_rejects_invalid_scope(client: AsyncClient, scope: object) -> None: + response = await client.request( + "QUERY", "/v2/search/", json={"text": "nebula", "project_ids": scope} + ) + assert response.status_code == 422 + response = await client.request("QUERY", "/v2/search/", json={"text": "nebula"}) + assert response.status_code == 422 + + +@pytest.mark.asyncio +async def test_unsupported_adapter_is_explicit_and_fts_still_works( + corpus: Corpus, client: AsyncClient +) -> None: + # The capability decision follows the configured DB adapter, before embedding. + original_backend = corpus.config.database_backend + corpus.config.database_backend = DatabaseBackend.POSTGRES + corpus.config.semantic_vector_index = "milvus" + for mode in ("vector", "hybrid"): + response = await client.request( + "QUERY", + "/v2/search/", + json={"project_ids": [corpus.projects[0].id], "text": "nebula", "retrieval_mode": mode}, + ) + assert response.status_code == 400 + assert response.json()["detail"]["code"] == "unsupported_multi_project_vector_adapter" + assert corpus.provider.query_calls == 0 + corpus.config.database_backend = original_backend + response = await client.request( + "QUERY", + "/v2/search/", + json={"project_ids": [corpus.projects[0].id], "text": "nebula", "retrieval_mode": "fts"}, + ) + assert response.status_code == 200, response.text + assert response.json()["total"] == 3 + + +@pytest.mark.asyncio +async def test_stale_manifest_model_and_source_hash_are_excluded(corpus: Corpus) -> None: + ids = [project.id for project in corpus.projects] + async with db.scoped_session(corpus.session_maker) as session: + await session.execute( + text( + "UPDATE search_vector_chunks SET embedding_status = 'pending' WHERE project_id = :p" + ), + {"p": ids[0]}, + ) + await session.execute( + text( + "UPDATE search_vector_chunks SET embedding_model = 'other-model' WHERE project_id = :p" + ), + {"p": ids[1]}, + ) + await session.execute( + text("UPDATE search_vector_chunks SET source_hash = 'stale' WHERE project_id = :p"), + {"p": ids[2]}, + ) + await session.commit() + prepared = SearchService.prepare_query( + SearchQuery(text="nebula", retrieval_mode=SearchRetrievalMode.VECTOR) + ) + assert prepared is not None + assert await corpus.repository(ids).search(prepared) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(SearchRetrievalMode)) +async def test_temporal_collision_filter_and_hydration( + corpus: Corpus, client: AsyncClient, mode: SearchRetrievalMode +) -> None: + from basic_memory.models import MemoryTimeIndex + from sqlalchemy import select + + ids = [project.id for project in corpus.projects[:2]] + async with db.scoped_session(corpus.session_maker) as session: + entities = (await session.scalars(select(Entity).where(Entity.project_id.in_(ids)))).all() + for entity in entities: + current = entity.project_id == ids[0] + session.add( + MemoryTimeIndex( + project_id=entity.project_id, + entity_id=entity.id, + source_type="observation", + source_id=500, + time_kind="effective", + range_axis="date", + lower_value="2026-09-01" if current else "2025-01-01", + upper_value=None if current else "2025-02-01", + lower_inclusive=True, + upper_inclusive=False, + is_empty=False, + extractor="test", + source_text="@effective[2026-09-01,)" + if current + else "@effective[2025-01-01,2025-02-01)", + ) + ) + await session.commit() + response = await client.request( + "QUERY", + "/v2/search/", + json={ + "project_ids": ids, + "text": "nebula", + "retrieval_mode": mode.value, + "valid_at": "2026-09-14", + "note_types": ["note"], + }, + ) + assert response.status_code == 200, response.text + payload = response.json() + assert payload["temporal_applied"] is True + assert len(payload["results"]) == 1 + assert payload["results"][0]["project_id"] == ids[0] + assert payload["results"][0]["temporal"][0]["source_text"] == "@effective[2026-09-01,)" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(SearchRetrievalMode)) +async def test_api_pagination_and_empty_later_page( + corpus: Corpus, client: AsyncClient, mode: SearchRetrievalMode +) -> None: + query = { + "project_ids": [project.id for project in corpus.projects[:2]], + "text": "nebula", + "retrieval_mode": mode.value, + } + for page, count, has_more in [(1, 2, True), (3, 2, False), (4, 0, False)]: + response = await client.request( + "QUERY", "/v2/search/", params={"page": page, "page_size": 2}, json=query + ) + assert response.status_code == 200, response.text + assert len(response.json()["results"]) == count + assert response.json()["has_more"] == has_more + + +@pytest.mark.asyncio +async def test_repository_guards_and_disabled_route(corpus: Corpus, client: AsyncClient) -> None: + from basic_memory.repository.postgres_search_query import PostgresSearchQuery + from basic_memory.repository.sqlite_search_query import SQLiteSearchQuery + + for bad_scope in ([0], [-1], [True]): + with pytest.raises(ValueError, match="positive integers"): + corpus.repository(bad_scope) + for compiler in (PostgresSearchQuery, SQLiteSearchQuery): + with pytest.raises(ValueError, match="positive integers"): + compiler(corpus.session_maker, bad_scope) + repo = corpus.repository([corpus.projects[0].id]) + prepared = SearchService.prepare_query( + SearchQuery(text="nebula", retrieval_mode=SearchRetrievalMode.VECTOR) + ) + assert prepared is not None + for limit, offset in [(0, 0), (1, -1)]: + with pytest.raises(ValueError, match="limit"): + await repo.search(prepared, limit=limit, offset=offset) + no_text = SearchService.prepare_query( + SearchQuery(title="nebula", retrieval_mode=SearchRetrievalMode.VECTOR) + ) + assert no_text is not None + with pytest.raises(ValueError, match="nonempty text"): + await repo.search(no_text) + corpus.provider.dimensions = 5 + with pytest.raises(ValueError, match="dimensions"): + await repo.search(prepared) + corpus.config.semantic_search_enabled = False + response = await client.request( + "QUERY", + "/v2/search/", + json={"project_ids": [corpus.projects[0].id], "text": "nebula", "retrieval_mode": "vector"}, + ) + assert response.status_code == 400 + assert "disabled" in response.json()["detail"] + response = await client.request( + "QUERY", "/v2/search/", json={"project_ids": [corpus.projects[0].id]} + ) + assert response.status_code == 200 and not response.json()["results"] + + +@pytest.mark.asyncio +async def test_hybrid_fts_only_rows_and_threshold(corpus: Corpus) -> None: + ids = [project.id for project in corpus.projects[:2]] + async with db.scoped_session(corpus.session_maker) as session: + await session.execute( + text( + "UPDATE search_vector_chunks SET embedding_status = 'pending' WHERE project_id = :p" + ), + {"p": ids[1]}, + ) + await session.commit() + prepared = SearchService.prepare_query( + SearchQuery(text="nebula", retrieval_mode=SearchRetrievalMode.HYBRID, min_similarity=1) + ) + assert prepared is not None + rows = await corpus.repository(ids).search(prepared, limit=100) + assert len(rows) == 6 + assert all(row.matched_chunk_text is None for row in rows if row.project_id == ids[1]) + assert max(row.score or 0.0 for row in rows) == pytest.approx(1.3) + assert all(1.0 <= (row.score or 0.0) <= 1.3 + 1e-6 for row in rows if row.project_id == ids[0]) + assert all(0.0 <= (row.score or 0.0) <= 1.0 for row in rows if row.project_id == ids[1]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", [SearchRetrievalMode.VECTOR, SearchRetrievalMode.HYBRID]) +async def test_compare_project_pipelines_on_same_corpus( + corpus: Corpus, mode: SearchRetrievalMode +) -> None: + """Record bounded corpus equivalence and actual cold/warm retrieval costs.""" + import json + import time + from dataclasses import asdict + + repo_type = ( + PostgresSearchRepository + if corpus.config.database_backend == DatabaseBackend.POSTGRES + else SQLiteSearchRepository + ) + writers = [ + repo_type( + corpus.session_maker, + project.id, + app_config=corpus.config, + embedding_provider=corpus.provider, + ) + for project in corpus.projects + ] + reader = corpus.repository([project.id for project in corpus.projects]) + prepared = SearchService.prepare_query(SearchQuery(text="nebula", retrieval_mode=mode)) + assert prepared is not None + for temperature in ("cold", "warm"): + corpus.provider.query_calls = 0 + started = time.perf_counter() + baseline = [ + row + for writer in writers + for row in await writer.search(search_text="nebula", retrieval_mode=mode, limit=100) + ] + baseline_ms = (time.perf_counter() - started) * 1000 + assert corpus.provider.query_calls == len(writers) + corpus.provider.query_calls = 0 + started = time.perf_counter() + combined = await reader.search(prepared, limit=100) + combined_ms = (time.perf_counter() - started) * 1000 + assert corpus.provider.query_calls == 1 + assert {(r.project_id, r.type, r.id) for r in combined} == { + (r.project_id, r.type, r.id) for r in baseline + } + print( + json.dumps( + { + "backend": corpus.config.database_backend.value, + "mode": mode.value, + "temperature": temperature, + "projects": len(writers), + "results": len(combined), + "baseline_embeddings": len(writers), + "combined_embeddings": 1, + "baseline_ms": round(baseline_ms, 2), + "combined_ms": round(combined_ms, 2), + "baseline_payload_bytes": len( + json.dumps([asdict(row) for row in baseline], default=str) + ), + "combined_payload_bytes": len( + json.dumps([asdict(row) for row in combined], default=str) + ), + } + ) + ) From 0159119c7328729b7ac69be342031e072c073a6d Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 15 Sep 2026 00:22:48 -0500 Subject: [PATCH 2/2] fix(core): preserve guarded multi-project lexical retrieval Signed-off-by: phernandez --- .../multi_project_search_repository.py | 115 ++++++++++++++---- test-int/test_multi_project_search.py | 110 +++++++++++++++++ 2 files changed, 202 insertions(+), 23 deletions(-) diff --git a/src/basic_memory/repository/multi_project_search_repository.py b/src/basic_memory/repository/multi_project_search_repository.py index f7d26a006..a524821d2 100644 --- a/src/basic_memory/repository/multi_project_search_repository.py +++ b/src/basic_memory/repository/multi_project_search_repository.py @@ -6,24 +6,32 @@ """ import json +import re from collections.abc import Sequence from dataclasses import replace from typing import Any import logfire -from sqlalchemy import text +from sqlalchemy import Result, text +from sqlalchemy.exc import DBAPIError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from basic_memory import db from basic_memory.config import BasicMemoryConfig, DatabaseBackend from basic_memory.repository.embedding_provider import EmbeddingProvider from basic_memory.repository.postgres_search_query import PostgresSearchQuery +from basic_memory.repository.postgres_search_repository import PostgresSearchRepository +from basic_memory.repository.script_ngrams import analyze_script_query from basic_memory.repository.search_index_row import SearchIndexRow from basic_memory.repository.search_query import PreparedSearchQuery from basic_memory.repository.search_repository_base import FUSION_BONUS, TOP_CHUNKS_PER_RESULT -from basic_memory.repository.semantic_errors import SemanticSearchDisabledError +from basic_memory.repository.semantic_errors import ( + SemanticDependenciesMissingError, + SemanticSearchDisabledError, +) from basic_memory.repository.semantic_vector_index_factory import semantic_embedding_identity -from basic_memory.repository.sqlite_search_query import SQLiteSearchQuery +from basic_memory.repository.sqlite_search_query import SQLITE_WORD_COLUMNS, SQLiteSearchQuery +from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository from basic_memory.schemas.search import SearchRetrievalMode @@ -96,16 +104,72 @@ async def _parts( temporal=query.temporal, ) + async def _fts_ctes(self, query: PreparedSearchQuery) -> tuple[str, dict[str, Any], str]: + source, where, params, order, score = await self._parts(query, lexical=True) + columns = ", ".join(f"search_index.{name}" for name in _RESULT_COLUMNS) + selection = f"SELECT {columns}, {score} AS score FROM {source} WHERE {where}" + params["fts_enabled"] = True + selection += " AND :fts_enabled" + ctes = f"fts_strict AS MATERIALIZED ({selection})" + word_text = ( + analyze_script_query(query.search_text.strip()).word_text if query.search_text else None + ) + if isinstance(self.compiler, PostgresSearchQuery): + relaxed = self.compiler._relaxed_tsquery_text(word_text) + else: + relaxed = self.compiler._relaxed_fts_text(word_text) + if relaxed and "script_text" in params: + relaxed = f"{SQLITE_WORD_COLUMNS}: ({relaxed})" + if relaxed and params.get("text"): + # Decide relaxation over the complete scope, before pagination. A deep + # empty page must never switch to a different candidate set. + params["relaxed_text"] = relaxed + relaxed_selection = re.sub(r":text\b", ":relaxed_text", selection) + ctes += f""", fts AS MATERIALIZED ( + SELECT * FROM fts_strict UNION ALL + {relaxed_selection} AND NOT EXISTS (SELECT 1 FROM fts_strict) + )""" + else: + ctes += ", fts AS MATERIALIZED (SELECT * FROM fts_strict)" + return ctes, params, order.replace("search_index.", "") + + async def _execute( + self, session: AsyncSession, sql: str, params: dict[str, Any] + ) -> Result[Any]: + try: + if not self.postgres or "fts_enabled" not in params: + return await session.execute(text(sql), params) + # A PostgreSQL syntax failure aborts its transaction. The savepoint + # keeps the same connection usable for the guarded lexical retry. + async with session.begin_nested(): + return await session.execute(text(sql), params) + except DBAPIError as exc: + syntax_error = ( + PostgresSearchRepository._is_tsquery_syntax_error(exc) + if self.postgres + else SQLiteSearchRepository._is_fts5_syntax_error(exc) + ) + if not syntax_error: + raise + retry = dict(params) + if self.postgres and params.get("relaxed_text"): + retry["text"] = params["relaxed_text"] + else: + # Preserve the established empty lexical channel for invalid + # explicit syntax; a hybrid request still retains vector matches. + retry.update(fts_enabled=False, text="" if self.postgres else '""') + if "title_text" in retry: + retry["title_text"] = retry["text"] + return await session.execute(text(sql), retry) + async def count(self, query: PreparedSearchQuery) -> int: if query.retrieval_mode != SearchRetrievalMode.FTS: raise ValueError("Exact counts are only supported for full-text search retrieval.") if not self.project_ids: return 0 - source, where, params, _, _ = await self._parts(query, lexical=True) + ctes, params, _ = await self._fts_ctes(query) async with db.scoped_session(self.session_maker) as session: - result = await session.execute( - text(f"SELECT COUNT(*) FROM {source} WHERE {where}"), params - ) + result = await self._execute(session, f"WITH {ctes} SELECT COUNT(*) FROM fts", params) return int(result.scalar_one()) async def search( @@ -116,21 +180,18 @@ async def search( if not self.project_ids: return [] mode = query.retrieval_mode - result_columns = ", ".join(f"search_index.{name}" for name in _RESULT_COLUMNS) if mode == SearchRetrievalMode.FTS: - source, where, params, order, score = await self._parts(query, lexical=True) + ctes, params, order = await self._fts_ctes(query) direction = "DESC" if self.postgres else "ASC" sql = f""" - SELECT {result_columns}, {score} AS score - FROM {source} WHERE {where} - ORDER BY score {direction} {order}, search_index.project_id, - search_index.type, search_index.id + WITH {ctes} SELECT * FROM fts + ORDER BY score {direction} {order}, project_id, type, id LIMIT :limit OFFSET :offset """ params.update(limit=limit, offset=offset) async with db.scoped_session(self.session_maker) as session: with logfire.span("search.fts", project_count=len(self.project_ids)): - result = await session.execute(text(sql), params) + result = await self._execute(session, sql, params) return [SearchIndexRow.from_mapping(row._asdict()) for row in result] expected_index = "pgvector" if self.postgres else "sqlite-vec" @@ -210,19 +271,16 @@ async def search( """ ranking = "SELECT project_id, type, id, score FROM vector_scores" if mode == SearchRetrievalMode.HYBRID: - fts_source, fts_where, fts_params, _, fts_score = await self._parts(query, lexical=True) + fts_ctes, fts_params, _ = await self._fts_ctes(query) params.update(fts_params) ctes += f""", - fts AS MATERIALIZED ( - SELECT search_index.project_id, search_index.type, search_index.id, - ABS({fts_score}) AS score FROM {fts_source} WHERE {fts_where} - ), + {fts_ctes}, channels AS ( SELECT project_id, type, id, score AS vector_score, 0.0 AS fts_score FROM vector_scores UNION ALL SELECT project_id, type, id, 0.0, - COALESCE(score / NULLIF(MAX(score) OVER (), 0), 0) FROM fts + COALESCE(ABS(score) / NULLIF(MAX(ABS(score)) OVER (), 0), 0) FROM fts ) """ params["fusion_bonus"] = FUSION_BONUS @@ -260,13 +318,24 @@ async def search( if not self.postgres: # Connection setup only: this reader never calls adapter.initialize(), # which may recreate storage and invalidate manifests on schema mismatch. - import sqlite_vec + try: + import sqlite_vec + except ImportError as exc: + raise SemanticDependenciesMissingError( + "sqlite-vec package is missing. Install/update basic-memory." + ) from exc connection = await session.connection() raw = await connection.get_raw_connection() driver = raw.driver_connection assert driver is not None - await driver.enable_load_extension(True) + try: + await driver.enable_load_extension(True) + except AttributeError as exc: + raise SemanticDependenciesMissingError( + "This Python build does not support SQLite extension loading. " + "Use a Python build with extension support or request FTS." + ) from exc try: await driver.load_extension(sqlite_vec.loadable_path()) finally: @@ -276,7 +345,7 @@ async def search( project_count=len(self.project_ids), retrieval_mode=mode.value, ): - result = await session.execute(text(sql), params) + result = await self._execute(session, sql, params) rows: dict[tuple[int, str, int], SearchIndexRow] = {} chunks: dict[tuple[int, str, int], list[str]] = {} for record in result: diff --git a/test-int/test_multi_project_search.py b/test-int/test_multi_project_search.py index 214ae2a91..a3c8d487d 100644 --- a/test-int/test_multi_project_search.py +++ b/test-int/test_multi_project_search.py @@ -529,3 +529,113 @@ async def test_compare_project_pipelines_on_same_corpus( } ) ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["fts", "hybrid"]) +@pytest.mark.parametrize("query", ["Did nebula go hiking at sunrise?", "foo None: + ids = [project.id for project in corpus.projects[:2]] + # Remove the vector channel so a hybrid success proves lexical recovery. + async with db.scoped_session(corpus.session_maker) as session: + await session.execute(text("UPDATE search_vector_chunks SET embedding_status = 'pending'")) + await session.commit() + body = {"project_ids": ids, "text": query, "retrieval_mode": mode} + complete = await client.request("QUERY", "/v2/search/", json=body) + assert complete.status_code == 200, complete.text + expected = complete.json()["results"] + assert len(expected) == 6 + assert {row["project_id"] for row in expected} == set(ids) + if mode == "fts": + assert complete.json()["total"] == 6 + pages = [] + for page in range(1, 5): + response = await client.request( + "QUERY", "/v2/search/", json=body, params={"page": page, "page_size": 2} + ) + assert response.status_code == 200, response.text + pages.extend(response.json()["results"]) + assert response.json()["has_more"] == (page < 3) + assert pages == expected + assert corpus.provider.query_calls == (5 if mode == "hybrid" else 0) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["fts", "hybrid"]) +async def test_relaxation_keeps_strict_matches_on_deep_pages( + corpus: Corpus, client: AsyncClient, mode: str +) -> None: + # Three terms permit relaxation, but strict matches anywhere in the selected + # scope must suppress it even after the final page. + body = { + "project_ids": [project.id for project in corpus.projects[:2]], + "text": "shared nebula observation", + "retrieval_mode": mode, + "min_similarity": 1, + } + async with db.scoped_session(corpus.session_maker) as session: + await session.execute(text("UPDATE search_vector_chunks SET embedding_status = 'pending'")) + await session.commit() + first = await client.request("QUERY", "/v2/search/", json=body) + assert first.status_code == 200, first.text + assert len(first.json()["results"]) == 2 + later = await client.request("QUERY", "/v2/search/", json=body, params={"page": 2}) + assert later.status_code == 200, later.text + assert later.json()["results"] == [] + if mode == "fts": + assert later.json()["total"] == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["package", "extension"]) +async def test_sqlite_semantic_dependency_errors_are_actionable( + corpus: Corpus, client: AsyncClient, monkeypatch: pytest.MonkeyPatch, failure: str +) -> None: + if corpus.config.database_backend == DatabaseBackend.POSTGRES: + pytest.skip("SQLite connection capability boundary") + import sys + import aiosqlite + + if failure == "package": + monkeypatch.setitem(sys.modules, "sqlite_vec", None) + else: + + async def unavailable(_self, _enabled): + raise AttributeError("enable_load_extension") + + monkeypatch.setattr(aiosqlite.Connection, "enable_load_extension", unavailable) + for mode in ("vector", "hybrid"): + response = await client.request( + "QUERY", + "/v2/search/", + json={"project_ids": [corpus.projects[0].id], "text": "nebula", "retrieval_mode": mode}, + ) + assert response.status_code == 400, response.text + assert ("sqlite-vec" if failure == "package" else "extension loading") in response.json()[ + "detail" + ] + response = await client.request( + "QUERY", "/v2/search/", json={"project_ids": [corpus.projects[0].id], "text": "nebula"} + ) + assert response.status_code == 200, response.text + assert response.json()["total"] == 3 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("query", ["foo None: + response = await client.request( + "QUERY", + "/v2/search/", + json={"project_ids": [corpus.projects[0].id], "text": query, "retrieval_mode": mode}, + ) + assert response.status_code == 200, response.text + assert len(response.json()["results"]) == (3 if mode == "hybrid" else 0) + if mode == "fts": + assert response.json()["total"] == 0 + assert corpus.provider.query_calls == int(mode == "hybrid")