diff --git a/src/basic_memory/alembic/versions/0044_entity_embedded_checksum.py b/src/basic_memory/alembic/versions/0044_entity_embedded_checksum.py new file mode 100644 index 000000000..7a4b52355 --- /dev/null +++ b/src/basic_memory/alembic/versions/0044_entity_embedded_checksum.py @@ -0,0 +1,66 @@ +"""Record which checksum each entity's embeddings were built from. + +Search responses now say how current semantic results are: the oldest last-modified +time among notes whose embeddings lag their content. Basic Memory Cloud defers a +workspace's embeddings past a daily fair-use budget while notes stay saved and +full-text indexed, so a note can sit with new content and old vectors for a day. +Nothing in the schema showed that: an edited note keeps its earlier `ready` chunks, which +look the same as current ones. + +``entity.embedded_checksum`` holds the checksum a vector-sync pass finished embedding. +A note is behind when it is NULL, differs from ``checksum``, or the last pass left the +note part-way (``vector_sync_deferred_at``). A partial index on exactly those rows makes +the per-search lookup read only the notes that are behind. + +The backfill treats every entity the last pass did not leave part-way as current. That +is not provable from the manifest, but the alternative, NULL everywhere, would report +notes that owe nothing (opted out of embeddings, or with no text to embed) as behind +until something rewrites them. A note that was in fact behind at upgrade time reads as +current until its pending embedding pass runs and stamps the real value. + +A plain ALTER TABLE, not batch mode: ``entity`` carries generated columns, and SQLite's +batch recreation emits them twice (see x7d8e9f0a1b2). + +Revision ID: 0044 +Revises: 0043 +Create Date: 2026-11-04 20:00:00.000000 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0044" +down_revision: Union[str, None] = "0043" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +# Must match ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE in models/knowledge.py. Copied, not +# imported, so this revision keeps creating the index it was written with. +EMBEDDING_NOT_CURRENT_PREDICATE = ( + "content_type = 'text/markdown' AND (" + "embedded_checksum IS NULL " + "OR embedded_checksum <> checksum " + "OR vector_sync_deferred_at IS NOT NULL)" +) + + +def upgrade() -> None: + """Add entity.embedded_checksum, backfill it, and index the entities that are behind.""" + op.add_column("entity", sa.Column("embedded_checksum", sa.String(), nullable=True)) + op.execute( + "UPDATE entity SET embedded_checksum = checksum WHERE vector_sync_deferred_at IS NULL" + ) + op.execute( + "CREATE INDEX IF NOT EXISTS ix_entity_project_embedding_not_current " + "ON entity (project_id, updated_at) " + f"WHERE {EMBEDDING_NOT_CURRENT_PREDICATE}" + ) + + +def downgrade() -> None: + """Drop the index and the embedded checksum.""" + op.execute("DROP INDEX IF EXISTS ix_entity_project_embedding_not_current") + op.drop_column("entity", "embedded_checksum") diff --git a/src/basic_memory/api/v2/routers/scoped_search_router.py b/src/basic_memory/api/v2/routers/scoped_search_router.py index dbb34f195..7b9fbc381 100644 --- a/src/basic_memory/api/v2/routers/scoped_search_router.py +++ b/src/basic_memory/api/v2/routers/scoped_search_router.py @@ -21,6 +21,7 @@ from basic_memory.deps import AppConfigDep, SessionMakerDep from basic_memory.repository.search_repository import create_search_reader from basic_memory.repository.search_scope import ProjectScope +from basic_memory.repository.semantic_index_freshness import semantic_index_current_as_of from basic_memory.schemas.search import ( MAX_SCOPED_SEARCH_PAGE_SIZE, ScopedSearchQuery, @@ -119,4 +120,7 @@ async def search_scope( # None, not False, when nothing was asked: an ordinary search payload stays # exactly what it was before valid time existed. temporal_applied=True if temporal_requested else None, + semantic_index_current_as_of=await semantic_index_current_as_of( + session_maker, app_config, scope + ), ) diff --git a/src/basic_memory/api/v2/routers/search_router.py b/src/basic_memory/api/v2/routers/search_router.py index 120c0a86a..27d422c53 100644 --- a/src/basic_memory/api/v2/routers/search_router.py +++ b/src/basic_memory/api/v2/routers/search_router.py @@ -19,6 +19,7 @@ from basic_memory.read_cache.policy import SEARCH_READ_CACHE_TTL_SECONDS from basic_memory.read_cache.routes import ReadCacheRoute, cached_read, content_group from basic_memory.deps import ( + AppConfigDep, EntityServiceV2ExternalDep, MemoryTimeIndexRepositoryV2ExternalDep, ProjectExternalIdPathDep, @@ -28,6 +29,8 @@ ) from basic_memory.repository.metadata_filters import parse_metadata_filters from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.search_scope import ProjectScope +from basic_memory.repository.semantic_index_freshness import semantic_index_current_as_of from basic_memory.schemas.search import SearchQuery, SearchResponse, SearchRetrievalMode from basic_memory.services.search_service import SearchService from basic_memory.services.search_guidance import unspaced_script_query_hint @@ -87,6 +90,7 @@ async def search( entity_service: EntityServiceV2ExternalDep, temporal_repository: MemoryTimeIndexRepositoryV2ExternalDep, session_maker: SessionMakerDep, + app_config: AppConfigDep, response: Response, internal_project_id: ProjectExternalIdPathDep, project_id: str = Path(..., description="Project external UUID"), @@ -103,7 +107,8 @@ async def search( search_service: Search service scoped to project entity_service: Entity service scoped to project temporal_repository: Valid-time projection, read to explain temporal matches - session_maker: Session factory for the temporal hydration read + session_maker: Session factory for the temporal hydration and freshness reads + app_config: Configuration that says whether semantic search can run here page: Page number for pagination page_size: Number of results per page @@ -216,6 +221,9 @@ async def search( phase="build_response", result_count=len(search_results), ): + current_as_of = await semantic_index_current_as_of( + session_maker, app_config, ProjectScope.single(internal_project_id) + ) result = SearchResponse( results=search_results, current_page=page, @@ -231,6 +239,7 @@ async def search( # None, not False, when nothing was asked: an ordinary search # payload stays exactly what it was before valid time existed. temporal_applied=True if temporal_requested else None, + semantic_index_current_as_of=current_as_of, ) return result diff --git a/src/basic_memory/cli/auth.py b/src/basic_memory/cli/auth.py index fe8a3103d..9f725756d 100644 --- a/src/basic_memory/cli/auth.py +++ b/src/basic_memory/cli/auth.py @@ -16,6 +16,7 @@ from basic_memory.config import ConfigManager from basic_memory.cli.markup import literal +from basic_memory.user_agent import BASIC_MEMORY_USER_AGENT console = Console() @@ -50,7 +51,7 @@ async def _get_http_client(self) -> AsyncIterator[httpx.AsyncClient]: async with self._http_client_factory() as client: yield client else: - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(headers={"User-Agent": BASIC_MEMORY_USER_AGENT}) as client: yield client def generate_pkce_pair(self) -> tuple[str, str]: diff --git a/src/basic_memory/cli/commands/cloud/api_client.py b/src/basic_memory/cli/commands/cloud/api_client.py index 74f60c8de..b2c08b0f9 100644 --- a/src/basic_memory/cli/commands/cloud/api_client.py +++ b/src/basic_memory/cli/commands/cloud/api_client.py @@ -11,6 +11,7 @@ from basic_memory.cli.auth import CLIAuth from basic_memory.config import ConfigManager +from basic_memory.user_agent import BASIC_MEMORY_USER_AGENT console = Console() @@ -75,7 +76,9 @@ async def get_authenticated_headers(auth: CLIAuth | None = None) -> dict[str, st @asynccontextmanager async def _default_http_client(timeout: float) -> AsyncIterator[httpx.AsyncClient]: - async with httpx.AsyncClient(timeout=timeout) as client: + async with httpx.AsyncClient( + timeout=timeout, headers={"User-Agent": BASIC_MEMORY_USER_AGENT} + ) as client: yield client diff --git a/src/basic_memory/man/man3/bm-search(3).md b/src/basic_memory/man/man3/bm-search(3).md index 493d425da..1686411a7 100644 --- a/src/basic_memory/man/man3/bm-search(3).md +++ b/src/basic_memory/man/man3/bm-search(3).md @@ -113,6 +113,10 @@ bm_search(query="release", mode="text", page_size=1, output_format="json", proje a next-page call; follow it to check the remaining candidates. - Line results report `total_is_exact: false`; `total` counts ranked candidates, so page with `has_more` instead of using it as a result count. +- Embeddings can lag a note's text. When some notes are not embedded yet, the + response carries `semantic_index_current_as_of` and the text output says + "Semantic results current as of ...". Notes changed after that time are found + by `mode="text"` but may be missing from semantic and hybrid matches. ## AUTHORED VALID TIME diff --git a/src/basic_memory/mcp/async_client.py b/src/basic_memory/mcp/async_client.py index 5bf92b06c..262541883 100644 --- a/src/basic_memory/mcp/async_client.py +++ b/src/basic_memory/mcp/async_client.py @@ -11,6 +11,7 @@ from loguru import logger import logfire +from basic_memory.user_agent import BASIC_MEMORY_USER_AGENT from basic_memory.config import ( BasicMemoryConfig, ConfigManager, @@ -349,7 +350,7 @@ async def _cloud_client( token = await _resolve_cloud_token(config) proxy_base_url = f"{config.cloud_host}/proxy" - headers = {"Authorization": f"Bearer {token}"} + headers = {"Authorization": f"Bearer {token}", "User-Agent": BASIC_MEMORY_USER_AGENT} headers.update(workspace_permalink_headers()) if workspace: headers["X-Workspace-ID"] = workspace @@ -370,7 +371,7 @@ async def get_cloud_control_plane_client( config = ConfigManager().config timeout = _build_timeout() token = await _resolve_cloud_token(config) - headers = {"Authorization": f"Bearer {token}"} + headers = {"Authorization": f"Bearer {token}", "User-Agent": BASIC_MEMORY_USER_AGENT} if workspace: headers["X-Workspace-ID"] = workspace logger.info(f"Creating HTTP client for cloud control plane at: {config.cloud_host}") diff --git a/src/basic_memory/mcp/tools/ranked_search.py b/src/basic_memory/mcp/tools/ranked_search.py index 2707170f9..da71badd6 100644 --- a/src/basic_memory/mcp/tools/ranked_search.py +++ b/src/basic_memory/mcp/tools/ranked_search.py @@ -13,6 +13,7 @@ import re from dataclasses import asdict +from datetime import datetime, timezone from typing import Annotated, Any, Literal from fastmcp import Context @@ -678,12 +679,14 @@ def _format_results( if next_call is not None else "" ) + freshness = _semantic_freshness_note(payload) if not results: page_hint = " on this page" if next_call is not None else "" return ( (f"No results for '{query}'" if query else "No results") + f"{page_hint} in {scope}." + (f"\n\n{payload['query_hint']}" if payload.get("query_hint") else "") + + freshness + continuation ) @@ -713,4 +716,20 @@ def _format_results( lines.append(f" (more matches from line {row['next_match_line']})") elif row.get("matched_chunk"): lines.append(f" {row['matched_chunk'][:200]}") - return "\n".join(lines) + continuation + return "\n".join(lines) + freshness + continuation + + +def _semantic_freshness_note(payload: dict[str, Any]) -> str: + """Say how current semantic results are, when embeddings lag some notes. + + The field is absent when every note is embedded, so ordinary output is unchanged. + """ + current_as_of = payload.get("semantic_index_current_as_of") + if current_as_of is None: + return "" + as_of_utc = datetime.fromisoformat(current_as_of).astimezone(timezone.utc) + return ( + f"\n\nSemantic results current as of {as_of_utc:%Y-%m-%d %H:%M} UTC. " + "Notes changed since then are found by text search but may be missing " + "from semantic matches." + ) diff --git a/src/basic_memory/mcp/tools/search.py b/src/basic_memory/mcp/tools/search.py index db1887f28..496f480c2 100644 --- a/src/basic_memory/mcp/tools/search.py +++ b/src/basic_memory/mcp/tools/search.py @@ -7,6 +7,7 @@ import re from collections.abc import Sequence from dataclasses import dataclass +from datetime import datetime from typing import Any, cast from uuid import UUID @@ -589,6 +590,8 @@ async def _search_all_projects( databases_answered = 0 failures: list[str] = [] query_hint: str | None = None + # The merged page is only as current as its least current database. + semantic_current_as_of: list[datetime] = [] for tenant_id, refs in databases.items(): label = _database_label(tenant_id, refs) @@ -619,6 +622,8 @@ async def _search_all_projects( response = _compact_search_response(response) if response.query_hint: query_hint = response.query_hint + if response.semantic_index_current_as_of is not None: + semantic_current_as_of.append(response.semantic_index_current_as_of) total += ( response.total if response.total > 0 @@ -671,6 +676,9 @@ async def _search_all_projects( # Confirmed only because a database answered: `databases_answered` is # non-zero here for any temporal query, guarded immediately above. "temporal_applied": True if temporal_requested else None, + "semantic_index_current_as_of": min(semantic_current_as_of) + if semantic_current_as_of + else None, } ) diff --git a/src/basic_memory/models/knowledge.py b/src/basic_memory/models/knowledge.py index 58f9e524a..8c4fceebb 100644 --- a/src/basic_memory/models/knowledge.py +++ b/src/basic_memory/models/knowledge.py @@ -29,6 +29,18 @@ from basic_memory.utils import file_path_conflict_keys, generate_permalink +# A markdown entity whose semantic embeddings do not reflect its current content: never +# embedded at this checksum, or a pass left part of it unembedded. Shared verbatim by the +# partial index below and by the search-freshness query, because both engines use a +# partial index only when the query repeats the index's predicate. +ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE = ( + "content_type = 'text/markdown' AND (" + "embedded_checksum IS NULL " + "OR embedded_checksum <> checksum " + "OR vector_sync_deferred_at IS NOT NULL)" +) + + def file_path_conflict_columns(file_path: str) -> dict[str, str]: """Return the derived entity columns that index ``file_path`` for conflict checks. @@ -76,6 +88,15 @@ class Entity(Base): # File path conflict checks look up one project's rows by either conflict key. Index("ix_entity_project_file_path_key", "project_id", "file_path_key"), Index("ix_entity_project_file_path_permalink", "project_id", "file_path_permalink"), + # Search reports the oldest note whose embeddings are behind. Only those notes are + # in this index, so the lookup reads a handful of rows however large the project. + Index( + "ix_entity_project_embedding_not_current", + "project_id", + "updated_at", + sqlite_where=text(ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE), + postgresql_where=text(ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE), + ), ) # Core identity @@ -123,6 +144,12 @@ class Entity(Base): vector_sync_deferred_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True, default=None ) + # The entity checksum whose embeddings a vector-sync pass finished, or which owes no + # embeddings (opted out). It differs from `checksum` once the note changes and stays + # behind until a pass embeds the new content, which is how search can say how current + # semantic results are. Written only by vector sync, with the checksum read before + # that pass read the note's search rows. + embedded_checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True, default=None) # Metadata and tracking created_at: Mapped[datetime] = mapped_column( diff --git a/src/basic_memory/repository/search_repository.py b/src/basic_memory/repository/search_repository.py index 8cfe692a8..af6000665 100644 --- a/src/basic_memory/repository/search_repository.py +++ b/src/basic_memory/repository/search_repository.py @@ -6,7 +6,7 @@ - PostgresSearchRepository: Uses tsvector/tsquery with GIN indexes """ -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any, Callable, List, Optional, Protocol @@ -88,6 +88,10 @@ async def record_entity_vector_deferrals( """Record which entities a vector sync pass left unfinished.""" ... + async def record_entity_embedded_checksums(self, checksums: Mapping[int, str | None]) -> None: + """Stamp the checksum each listed entity's embeddings now reflect.""" + ... + async def semantic_effectively_enabled(self) -> bool: """Return whether semantic retrieval can actually run right now.""" ... diff --git a/src/basic_memory/repository/search_repository_base.py b/src/basic_memory/repository/search_repository_base.py index 1e3b301cf..223674997 100644 --- a/src/basic_memory/repository/search_repository_base.py +++ b/src/basic_memory/repository/search_repository_base.py @@ -540,6 +540,55 @@ async def record_entity_vector_deferrals( ) await session.commit() + async def read_entity_checksums(self, entity_ids: Iterable[int]) -> dict[int, str | None]: + """Read each listed entity's current checksum; deleted entities are absent.""" + ordered_ids = sorted(set(entity_ids)) + checksums: dict[int, str | None] = {} + if not ordered_ids: + return checksums + async with db.scoped_session(self.session_maker) as session: + # Chunked for the same bind-parameter cap as the deferral markers above. + for start in range(0, len(ordered_ids), VECTOR_HYDRATION_BATCH_SIZE): + chunk = ordered_ids[start : start + VECTOR_HYDRATION_BATCH_SIZE] + placeholders = ", ".join(f":entity_id_{index}" for index in range(len(chunk))) + params: dict[str, object] = {"project_id": self.project_id} + params.update( + {f"entity_id_{index}": entity_id for index, entity_id in enumerate(chunk)} + ) + result = await session.execute( + text( + "SELECT id, checksum FROM entity " + f"WHERE project_id = :project_id AND id IN ({placeholders})" + ), + params, + ) + checksums.update({int(row[0]): row[1] for row in result.all()}) + return checksums + + async def record_entity_embedded_checksums(self, checksums: Mapping[int, str | None]) -> None: + """Stamp the checksum each listed entity's embeddings now reflect. + + The caller passes the checksum it read *before* the pass read the note's + search rows. A write that lands during the pass then leaves the stamp behind + the new checksum, so the note reads as not current until the next pass embeds + it -- the safe direction. Stamping `checksum` as it is at the end of the pass + would mark content the pass never saw as embedded. + """ + if not checksums: + return + async with db.scoped_session(self.session_maker) as session: + # One executemany: a single bind set per row, so no parameter cap applies. + await session.execute( + text( + "UPDATE entity SET embedded_checksum = :checksum " + "WHERE project_id = :project_id AND id = :entity_id" + ), + [ + {"project_id": self.project_id, "entity_id": entity_id, "checksum": checksum} + for entity_id, checksum in sorted(checksums.items()) + ], + ) + async def _write_embeddings( self, session: AsyncSession, @@ -1690,6 +1739,9 @@ class owns a database session, and the helper is also driven by test doubles that have none. """ pass_started_at = datetime.now(timezone.utc) + # Read before the pass reads any search rows, so a completed entity is stamped + # with a checksum no newer than the content it embedded. + checksums_at_start = await self.read_entity_checksums(entity_ids) try: result = await semantic_vector_sync.sync_entity_vectors_internal( self, @@ -1726,6 +1778,13 @@ class owns a database session, and the helper is also driven by test completed_entity_ids=set(result.synced_entity_ids), pass_started_at=pass_started_at, ) + await self.record_entity_embedded_checksums( + { + entity_id: checksums_at_start[entity_id] + for entity_id in result.synced_entity_ids + if entity_id in checksums_at_start + } + ) return result def _vector_prepare_window_size(self) -> int: diff --git a/src/basic_memory/repository/semantic_index_freshness.py b/src/basic_memory/repository/semantic_index_freshness.py new file mode 100644 index 000000000..ea3109b5d --- /dev/null +++ b/src/basic_memory/repository/semantic_index_freshness.py @@ -0,0 +1,61 @@ +"""How current semantic search is for a set of projects. + +A note is saved and full-text indexed at once, but its embeddings come from a later +vector-sync pass. That pass can lag: a large note embeds a shard at a time, a failed +pass leaves work owed, and Basic Memory Cloud defers a legacy workspace's embeddings to +the next UTC day once it passes its fair-use budget. Search results say how far behind +the vectors are, so a client can tell "no semantic match" from "not embedded yet". + +The answer is the oldest last-modified time among notes whose embeddings are not +current. Every note changed before it is reflected in vector search. Derived state, +read without locking, per the consistency model in CLAUDE.md. +""" + +from datetime import datetime + +from sqlalchemy import Select, func, select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.config import BasicMemoryConfig, DatabaseBackend +from basic_memory.models.knowledge import ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE, Entity +from basic_memory.repository.search_scope import ProjectScope +from basic_memory.repository.semantic_runtime import semantic_runtime_available +from basic_memory.utils import ensure_timezone_aware + + +def oldest_behind_statement(scope: ProjectScope) -> Select[tuple[datetime]]: + """The oldest last-modified time among ``scope``'s notes whose embeddings lag. + + The predicate is the partial index's own text, so both engines can answer from + `ix_entity_project_embedding_not_current`, which holds only the notes behind. + """ + return select(func.min(Entity.updated_at)).where( + Entity.project_id.in_(scope.project_ids), + text(ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE), + ) + + +async def semantic_index_current_as_of( + session_maker: async_sessionmaker[AsyncSession], + app_config: BasicMemoryConfig, + scope: ProjectScope, +) -> datetime | None: + """Return when semantic search stopped being current for ``scope``, or None. + + None means every note's embeddings are current, or there is no semantic search to + be current: a host that cannot load sqlite-vec runs keyword-only (#711), and an + empty scope searched nothing. + """ + if scope.is_empty or not semantic_runtime_available(app_config): + return None + + async with db.scoped_session(session_maker) as session: + oldest_behind = (await session.execute(oldest_behind_statement(scope))).scalar_one() + if oldest_behind is None: + return None + # SQLite hands back naive local times. An aware value is what lets a client compare + # answers from a local and a cloud database, as the all-projects search does. + return ensure_timezone_aware( + oldest_behind, cloud_mode=app_config.database_backend == DatabaseBackend.POSTGRES + ) diff --git a/src/basic_memory/schemas/search.py b/src/basic_memory/schemas/search.py index 14f356a83..8afea720b 100644 --- a/src/basic_memory/schemas/search.py +++ b/src/basic_memory/schemas/search.py @@ -375,3 +375,16 @@ class SearchResponse(BaseModel): description="True when the server executed a requested valid-time filter; " "absent when the request carried none", ) + # Semantic (vector) results come from embeddings that a later pass builds, and that + # pass can lag the note itself -- Basic Memory Cloud defers a legacy workspace's + # embeddings to the next UTC day past its fair-use budget. Full-text search is not + # affected: it is indexed with the note. + semantic_index_current_as_of: datetime | None = Field( + default=None, + description="Semantic (vector) search reflects every note in the searched " + "projects changed before this time. It is the oldest last-modified time among " + "notes whose embeddings are not current (pending, deferred, or missing); notes " + "changed at or after it may be missing from semantic and hybrid results but are " + "still found by full-text search. Null when every note's embeddings are current, " + "or when semantic search is unavailable.", + ) diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 2f283df83..d481def61 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -6,7 +6,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from itertools import batched -from datetime import datetime +from datetime import datetime, timezone from typing import Any, List, Optional, Set, Dict from dateparser import parse @@ -813,6 +813,7 @@ async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: if not entity_embeddings_enabled(entity): await self._clear_entity_vectors(entity_id) + await self._settle_opted_out_entities({entity.id: entity.checksum}) # Opted-out entities count as skipped, matching the batch path. return VectorSyncBatchResult( entities_total=1, entities_synced=0, entities_failed=0, entities_skipped=1 @@ -864,6 +865,9 @@ async def sync_entity_vectors_batch( await asyncio.gather( *(self._clear_entity_vectors(entity_id) for entity_id in opted_out_ids) ) + await self._settle_opted_out_entities( + {entity_id: entities_by_id[entity_id].checksum for entity_id in opted_out_ids} + ) eligible_entity_ids = [ entity_id @@ -1073,6 +1077,21 @@ async def _purge_stale_search_rows(self) -> None: "Purged stale search rows", project_id=self.repository.project_id, purged=purged ) + async def _settle_opted_out_entities(self, checksums: dict[int, str | None]) -> None: + """Record that opted-out notes owe no embeddings. + + An opted-out note's content is as current as semantic search will ever make it, + so it is stamped embedded at its checksum. Opt-out bypasses the repository sync + boundary that clears deferral markers, so the marker a failed or sharded pass left + is cleared here too; otherwise it would pin the search freshness forever. + """ + await self.repository.record_entity_vector_deferrals( + unfinished_entity_ids=set(), + completed_entity_ids=set(checksums), + pass_started_at=datetime.now(timezone.utc), + ) + await self.repository.record_entity_embedded_checksums(checksums) + async def _clear_entity_vectors(self, entity_id: int) -> None: """Delete derived vector rows for one entity.""" from basic_memory.repository.search_repository_base import SearchRepositoryBase diff --git a/src/basic_memory/user_agent.py b/src/basic_memory/user_agent.py new file mode 100644 index 000000000..2d8e71ceb --- /dev/null +++ b/src/basic_memory/user_agent.py @@ -0,0 +1,15 @@ +"""The User-Agent Basic Memory sends to Basic Memory Cloud. + +httpx's default (`python-httpx/x`) looks the same as any script, so Cloud could not +tell our own CLI and MCP server from unknown clients when applying fair-use and rate +limits. Every HTTP client that calls Cloud sets this as a client default; a header a +caller passes on one request still replaces it, as httpx merges request headers over +client headers. + +The MCP server and the CLI share one process model and no surface flag, so both send +`(cli)`. +""" + +from basic_memory import __version__ + +BASIC_MEMORY_USER_AGENT = f"basic-memory/{__version__} (cli)" diff --git a/tests/api/v2/test_search_router_telemetry.py b/tests/api/v2/test_search_router_telemetry.py index b128f51e2..385fb0106 100644 --- a/tests/api/v2/test_search_router_telemetry.py +++ b/tests/api/v2/test_search_router_telemetry.py @@ -38,7 +38,12 @@ async def fake_to_search_results( return [] monkeypatch.setattr(logfire, "span", fake_span) + + async def fake_semantic_index_current_as_of(session_maker, app_config, scope): + return None + monkeypatch.setattr(router, "to_search_results", fake_to_search_results) + monkeypatch.setattr(router, "semantic_index_current_as_of", fake_semantic_index_current_as_of) http_response = Response() search_response = await router.search( @@ -48,6 +53,7 @@ async def fake_to_search_results( # This query carries no valid-time filter, so neither is touched. temporal_repository=object(), session_maker=object(), + app_config=object(), response=http_response, internal_project_id=1, project_id="11111111-1111-1111-1111-111111111111", diff --git a/tests/api/v2/test_search_semantic_freshness.py b/tests/api/v2/test_search_semantic_freshness.py new file mode 100644 index 000000000..78e501ecb --- /dev/null +++ b/tests/api/v2/test_search_semantic_freshness.py @@ -0,0 +1,119 @@ +"""Search responses say how current semantic results are, on both search routes. + +The note is written, full-text indexed and embedded through the real services; the +lagging note is made the way a held vector pass leaves one, with its checksum moved on +and no new embedding. +""" + +from datetime import datetime, timedelta +from pathlib import Path + +import pytest +from httpx import AsyncClient +from sqlalchemy import text + +from basic_memory import db +from basic_memory.models import Entity, Project +from basic_memory.repository import semantic_index_freshness as freshness_module + + +async def _embedded_note(test_project, entity_repository, search_service, file_service) -> Entity: + file_path = Path(test_project.path) / "notes" / "Ledger.md" + file_path.parent.mkdir(parents=True, exist_ok=True) + await file_service.write_file(file_path, "# Ledger\n\nQuarterly ledger notes") + async with db.scoped_session(search_service.session_maker) as session: + entity = await entity_repository.create( + session, + { + "title": "Ledger", + "note_type": "note", + "content_type": "text/markdown", + "file_path": "notes/Ledger.md", + "checksum": "ledger-v1", + }, + ) + await search_service.index_entity(entity) + assert (await search_service.sync_entity_vectors(entity.id)).entities_synced == 1 + return entity + + +async def _hold_embedding(session_maker, entity: Entity) -> datetime: + """Edit the note's content with its vector pass held back; return the edit time.""" + edited_at = (datetime.now() - timedelta(hours=2)).astimezone().replace(microsecond=0) + async with db.scoped_session(session_maker) as session: + await session.execute( + text("UPDATE entity SET checksum = 'ledger-v2', updated_at = :at WHERE id = :id"), + {"at": edited_at, "id": entity.id}, + ) + return edited_at + + +@pytest.mark.asyncio +async def test_project_search_reports_semantic_freshness( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_repository, + search_service, + file_service, + session_maker, +): + entity = await _embedded_note(test_project, entity_repository, search_service, file_service) + + current = await client.request("QUERY", f"{v2_project_url}/search/", json={"text": "ledger"}) + assert current.status_code == 200 + assert current.json()["semantic_index_current_as_of"] is None + + edited_at = await _hold_embedding(session_maker, entity) + behind = await client.request("QUERY", f"{v2_project_url}/search/", json={"text": "ledger"}) + + assert behind.status_code == 200 + assert datetime.fromisoformat(behind.json()["semantic_index_current_as_of"]) == edited_at + # Full-text search still finds the held note; only its vectors lag. + assert [row["title"] for row in behind.json()["results"]] == ["Ledger"] + + +@pytest.mark.asyncio +async def test_project_search_without_semantic_search_reports_nothing( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_repository, + search_service, + file_service, + session_maker, + monkeypatch, +): + entity = await _embedded_note(test_project, entity_repository, search_service, file_service) + await _hold_embedding(session_maker, entity) + monkeypatch.setattr(freshness_module, "semantic_runtime_available", lambda _config: False) + + response = await client.request( + "QUERY", f"{v2_project_url}/search/", json={"text": "ledger", "retrieval_mode": "fts"} + ) + + assert response.status_code == 200 + assert response.json()["semantic_index_current_as_of"] is None + + +@pytest.mark.asyncio +async def test_scoped_search_reports_semantic_freshness( + client: AsyncClient, + test_project: Project, + entity_repository, + search_service, + file_service, + session_maker, +): + entity = await _embedded_note(test_project, entity_repository, search_service, file_service) + body = {"project_ids": [test_project.id], "text": "ledger"} + + current = await client.request("QUERY", "/v2/search/", json=body) + assert current.status_code == 200 + assert current.json()["semantic_index_current_as_of"] is None + + edited_at = await _hold_embedding(session_maker, entity) + behind = await client.request("QUERY", "/v2/search/", json=body) + + assert behind.status_code == 200 + assert datetime.fromisoformat(behind.json()["semantic_index_current_as_of"]) == edited_at diff --git a/tests/mcp/test_tool_bm_search.py b/tests/mcp/test_tool_bm_search.py index 3d7878047..c3d1d0173 100644 --- a/tests/mcp/test_tool_bm_search.py +++ b/tests/mcp/test_tool_bm_search.py @@ -4,10 +4,14 @@ mode="text" so ranking does not depend on the embedding model. """ +from datetime import datetime, timezone from typing import Any import pytest from fastmcp.exceptions import ToolError +from sqlalchemy import text as sql_text + +from basic_memory import db from basic_memory.index.local_project import LocalProjectIndexRunner from basic_memory.mcp.tools import bm_read, bm_search, bm_write @@ -468,3 +472,56 @@ async def test_lines_skip_hits_without_markdown( assert ranked["results"], "the image must be a ranked hit for this test to mean anything" assert with_lines["results"] == [] + + +# --- Semantic freshness --- + + +async def _hold_deploy_plan_embedding(session_maker) -> datetime: + """Edit Deploy Plan's content the way a held vector pass leaves it; return the time.""" + edited_at = datetime(2026, 3, 4, 5, 6, tzinfo=timezone.utc) + async with db.scoped_session(session_maker) as session: + await session.execute( + sql_text( + "UPDATE entity SET checksum = 'held', updated_at = :at WHERE title = 'Deploy Plan'" + ), + {"at": edited_at.astimezone()}, + ) + return edited_at + + +@pytest.mark.asyncio +async def test_current_embeddings_add_no_freshness_note(client, test_project): + await _seed(test_project.name) + + payload = await search_json(query="deploy", project=test_project.name) + text = await search_text(query="deploy", project=test_project.name) + + assert "semantic_index_current_as_of" not in payload + assert "Semantic results current as of" not in text + + +@pytest.mark.asyncio +async def test_held_embeddings_say_how_current_semantic_results_are( + client, test_project, session_maker +): + await _seed(test_project.name) + edited_at = await _hold_deploy_plan_embedding(session_maker) + + payload = await search_json(query="deploy", project=test_project.name) + text = await search_text(query="deploy", project=test_project.name) + + assert datetime.fromisoformat(payload["semantic_index_current_as_of"]) == edited_at + assert "Semantic results current as of 2026-03-04 05:06 UTC." in text + + +@pytest.mark.asyncio +async def test_all_projects_search_carries_the_freshness(client, test_project, session_maker): + await _seed(test_project.name) + edited_at = await _hold_deploy_plan_embedding(session_maker) + + payload = await search_json(query="deploy", mode="text", all_projects=True) + text = await search_text(query="deploy", mode="text", all_projects=True) + + assert datetime.fromisoformat(payload["semantic_index_current_as_of"]) == edited_at + assert "Semantic results current as of 2026-03-04 05:06 UTC." in text diff --git a/tests/repository/test_postgres_search_repository_unit.py b/tests/repository/test_postgres_search_repository_unit.py index ee7123b19..e06e65325 100644 --- a/tests/repository/test_postgres_search_repository_unit.py +++ b/tests/repository/test_postgres_search_repository_unit.py @@ -32,16 +32,22 @@ def _skip_vector_deferral_writes(monkeypatch): """These tests mock session_maker, so the deferral write has nothing to talk to. - `sync_entity_vectors_batch` records which entities it deferred, which needs a - real session; the doubles here patch `scoped_session` only for the paths they - exercise. The write itself is covered against a database in - tests/services/test_project_readiness.py. + `sync_entity_vectors_batch` records which entities it deferred and which checksum + each finished entity embedded, which needs a real session; the doubles here patch + `scoped_session` only for the paths they exercise. Those writes are covered against + a database in tests/services/test_project_readiness.py and + tests/repository/test_semantic_index_freshness.py. """ - async def _noop(self, **_kwargs) -> None: + async def _noop(self, *_args, **_kwargs) -> None: return None + async def _no_checksums(self, _entity_ids) -> dict[int, str | None]: + return {} + monkeypatch.setattr(PostgresSearchRepository, "record_entity_vector_deferrals", _noop) + monkeypatch.setattr(PostgresSearchRepository, "record_entity_embedded_checksums", _noop) + monkeypatch.setattr(PostgresSearchRepository, "read_entity_checksums", _no_checksums) # --- Helpers --------------------------------------------------------------- diff --git a/tests/repository/test_semantic_index_freshness.py b/tests/repository/test_semantic_index_freshness.py new file mode 100644 index 000000000..6491a33f0 --- /dev/null +++ b/tests/repository/test_semantic_index_freshness.py @@ -0,0 +1,345 @@ +"""How current semantic search is, against a real database and real vector sync. + +Notes are created, full-text indexed and embedded through the real search service +(with the deterministic test embedder). A note whose embeddings lag its content is +then made the way production makes one: its checksum moves on and no vector pass runs, +which is what a Cloud fair-use deferral leaves behind. +""" + +from datetime import datetime, timedelta +from pathlib import Path + +import pytest +from sqlalchemy import text +from sqlalchemy.dialects import sqlite + +from basic_memory import db +from basic_memory.config import DatabaseBackend +from basic_memory.models import Entity, Project +from basic_memory.repository import semantic_index_freshness as freshness_module +from basic_memory.repository.search_scope import ProjectScope +from basic_memory.repository.semantic_index_freshness import ( + oldest_behind_statement, + semantic_index_current_as_of, +) + + +def _local_time(days_ago: int) -> datetime: + # Local-aware, because SQLite stores naive local times and reads them back as such. + return (datetime.now() - timedelta(days=days_ago)).astimezone().replace(microsecond=0) + + +async def _write_file(project: Project, title: str) -> None: + """Search indexing reads the note's file, so each note gets one.""" + path = Path(project.path) / "notes" / f"{title}.md" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"# {title}\n\nNotes about {title}.", encoding="utf-8") + + +async def _embedded_note( + entity_repository, + search_service, + session_maker, + project: Project, + *, + title: str, + updated_at: datetime, +) -> Entity: + """Create a note, index its text, and run the real vector sync over it.""" + await _write_file(project, title) + async with db.scoped_session(session_maker) as session: + entity = await entity_repository.create( + session, + { + "project_id": entity_repository.project_id, + "title": title, + "note_type": "note", + "permalink": f"notes/{title.lower().replace(' ', '-')}", + "file_path": f"notes/{title}.md", + "content_type": "text/markdown", + "checksum": f"{title}-v1", + "created_at": updated_at, + "updated_at": updated_at, + }, + ) + await search_service.index_entity(entity) + result = await search_service.sync_entity_vectors(entity.id) + assert result.entities_synced == 1 + return entity + + +async def _edit_without_embedding(session_maker, entity: Entity, updated_at: datetime) -> None: + """Change a note's content the way a write does, with its vector pass held back.""" + async with db.scoped_session(session_maker) as session: + await session.execute( + text("UPDATE entity SET checksum = :checksum, updated_at = :updated_at WHERE id = :id"), + {"checksum": f"{entity.title}-v2", "updated_at": updated_at, "id": entity.id}, + ) + + +async def _current_as_of(session_maker, app_config, test_project) -> datetime | None: + return await semantic_index_current_as_of( + session_maker, app_config, ProjectScope.single(test_project.id) + ) + + +@pytest.mark.asyncio +async def test_all_embeddings_current_is_none( + entity_repository, search_service, session_maker, app_config, test_project +): + entity = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Current", + updated_at=_local_time(3), + ) + + async with db.scoped_session(session_maker) as session: + stored = await session.get(Entity, entity.id) + assert stored is not None + assert stored.embedded_checksum == stored.checksum + assert await _current_as_of(session_maker, app_config, test_project) is None + + +@pytest.mark.asyncio +async def test_a_note_changed_after_its_embedding_reports_its_modified_time( + entity_repository, search_service, session_maker, app_config, test_project +): + """The oldest note behind is the answer; current notes, even older ones, do not count.""" + await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Old Current", + updated_at=_local_time(10), + ) + held = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Held", + updated_at=_local_time(5), + ) + newer_held = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Newer Held", + updated_at=_local_time(4), + ) + held_edit_time = _local_time(2) + await _edit_without_embedding(session_maker, held, held_edit_time) + await _edit_without_embedding(session_maker, newer_held, _local_time(1)) + + assert await _current_as_of(session_maker, app_config, test_project) == held_edit_time + + # The held pass runs: those notes catch up and search is current again. + await search_service.sync_entity_vectors(held.id) + await search_service.sync_entity_vectors(newer_held.id) + assert await _current_as_of(session_maker, app_config, test_project) is None + + +@pytest.mark.asyncio +async def test_a_note_never_embedded_is_behind( + entity_repository, search_service, session_maker, app_config, test_project +): + """A new note whose first vector pass has not run is not in semantic search yet.""" + created_at = _local_time(1) + await _write_file(test_project, "Fresh") + async with db.scoped_session(session_maker) as session: + entity = await entity_repository.create( + session, + { + "project_id": entity_repository.project_id, + "title": "Fresh", + "note_type": "note", + "permalink": "notes/fresh", + "file_path": "notes/Fresh.md", + "content_type": "text/markdown", + "checksum": "fresh-v1", + "created_at": created_at, + "updated_at": created_at, + }, + ) + await search_service.index_entity(entity) + + assert await _current_as_of(session_maker, app_config, test_project) == created_at + + +@pytest.mark.asyncio +async def test_a_note_left_part_way_is_behind( + entity_repository, search_service, session_maker, app_config, test_project +): + """A pass that embedded one shard of a large note leaves it behind (#1440).""" + updated_at = _local_time(2) + entity = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Sharded", + updated_at=updated_at, + ) + await search_service.repository.record_entity_vector_deferrals( + unfinished_entity_ids={entity.id}, + completed_entity_ids=set(), + pass_started_at=datetime.now().astimezone(), + ) + + assert await _current_as_of(session_maker, app_config, test_project) == updated_at + + +@pytest.mark.asyncio +async def test_an_opted_out_note_owes_nothing( + entity_repository, search_service, session_maker, app_config, test_project +): + """`embed: false` notes are never embedded, so they never hold search back.""" + updated_at = _local_time(2) + await _write_file(test_project, "Private") + async with db.scoped_session(session_maker) as session: + entity = await entity_repository.create( + session, + { + "project_id": entity_repository.project_id, + "title": "Private", + "note_type": "note", + "permalink": "notes/private", + "file_path": "notes/Private.md", + "content_type": "text/markdown", + "checksum": "private-v1", + "entity_metadata": {"embed": False}, + "created_at": updated_at, + "updated_at": updated_at, + }, + ) + await search_service.index_entity(entity) + assert await _current_as_of(session_maker, app_config, test_project) == updated_at + + await search_service.sync_entity_vectors(entity.id) + + assert await _current_as_of(session_maker, app_config, test_project) is None + + +@pytest.mark.asyncio +async def test_an_embedded_checksum_older_than_the_note_stays_behind( + entity_repository, search_service, session_maker, app_config, test_project +): + """A write during a pass leaves the stamp on the checksum the pass started from.""" + entity = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Racing", + updated_at=_local_time(3), + ) + edit_time = _local_time(1) + await _edit_without_embedding(session_maker, entity, edit_time) + # What the pass that started before the edit stamps when it finishes. + await search_service.repository.record_entity_embedded_checksums({entity.id: "Racing-v1"}) + + assert await _current_as_of(session_maker, app_config, test_project) == edit_time + + +@pytest.mark.asyncio +async def test_non_markdown_files_are_ignored( + entity_repository, session_maker, app_config, test_project +): + """Attachments are never embedded, so they never hold search back.""" + async with db.scoped_session(session_maker) as session: + await entity_repository.create( + session, + { + "project_id": entity_repository.project_id, + "title": "image.png", + "note_type": "file", + "file_path": "files/image.png", + "content_type": "image/png", + "checksum": "png", + "created_at": _local_time(1), + "updated_at": _local_time(1), + }, + ) + + assert await _current_as_of(session_maker, app_config, test_project) is None + + +@pytest.mark.asyncio +async def test_semantic_search_unavailable_is_none( + entity_repository, search_service, session_maker, app_config, test_project, monkeypatch +): + """A keyword-only host (#711) has no semantic results to be behind.""" + entity = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Keyword", + updated_at=_local_time(3), + ) + await _edit_without_embedding(session_maker, entity, _local_time(1)) + monkeypatch.setattr(freshness_module, "semantic_runtime_available", lambda _config: False) + + assert await _current_as_of(session_maker, app_config, test_project) is None + + +@pytest.mark.asyncio +async def test_an_empty_scope_is_none(session_maker, app_config): + assert ( + await semantic_index_current_as_of(session_maker, app_config, ProjectScope.of([])) is None + ) + + +@pytest.mark.asyncio +async def test_the_lookup_reads_the_partial_index(session_maker, app_config, test_project): + """Each search pays for this lookup, so it must not scan the project's entities.""" + if app_config.database_backend != DatabaseBackend.SQLITE: + pytest.skip("Postgres picks partial indexes by cost; the planner is not asserted.") + compiled = oldest_behind_statement(ProjectScope.single(test_project.id)).compile( + dialect=sqlite.dialect(), compile_kwargs={"literal_binds": True} + ) + async with db.scoped_session(session_maker) as session: + rows = (await session.execute(text(f"EXPLAIN QUERY PLAN {compiled}"))).all() + plan = " ".join(str(row[-1]) for row in rows) + + assert "ix_entity_project_embedding_not_current" in plan + + +@pytest.mark.asyncio +async def test_opting_out_a_part_way_note_settles_it( + entity_repository, search_service, session_maker, app_config, test_project +): + """A note left part-way, then marked `embed: false`, no longer holds search back. + + Opt-out skips the repository sync boundary that clears deferral markers, so the + marker the earlier pass left must be cleared by the opt-out itself. + """ + updated_at = _local_time(2) + entity = await _embedded_note( + entity_repository, + search_service, + session_maker, + test_project, + title="Abandoned", + updated_at=updated_at, + ) + await search_service.repository.record_entity_vector_deferrals( + unfinished_entity_ids={entity.id}, + completed_entity_ids=set(), + pass_started_at=datetime.now().astimezone(), + ) + assert await _current_as_of(session_maker, app_config, test_project) == updated_at + + async with db.scoped_session(session_maker) as session: + await session.execute( + text("UPDATE entity SET entity_metadata = :metadata WHERE id = :id"), + {"metadata": '{"embed": false}', "id": entity.id}, + ) + await search_service.sync_entity_vectors(entity.id) + + assert await _current_as_of(session_maker, app_config, test_project) is None diff --git a/tests/repository/test_semantic_search_base.py b/tests/repository/test_semantic_search_base.py index 76e8ef3db..90a1e7f09 100644 --- a/tests/repository/test_semantic_search_base.py +++ b/tests/repository/test_semantic_search_base.py @@ -4,7 +4,7 @@ from basic_memory.repository.search_scope import ProjectScope import asyncio import hashlib -from collections.abc import Sequence +from collections.abc import Iterable, Mapping, Sequence from contextlib import asynccontextmanager from datetime import datetime from types import SimpleNamespace @@ -95,6 +95,14 @@ async def record_entity_vector_deferrals( return None # no session_maker in this double; the real write is covered # in tests/services/test_project_readiness.py + @override + async def read_entity_checksums(self, entity_ids: Iterable[int]) -> dict[int, str | None]: + return {} # no session_maker in this double; covered in test_semantic_index_freshness.py + + @override + async def record_entity_embedded_checksums(self, checksums: Mapping[int, str | None]) -> None: + return None # no session_maker in this double; covered in test_semantic_index_freshness.py + @override async def search( self, diff --git a/tests/services/test_semantic_search.py b/tests/services/test_semantic_search.py index 44247d248..da64d65e0 100644 --- a/tests/services/test_semantic_search.py +++ b/tests/services/test_semantic_search.py @@ -111,7 +111,9 @@ async def test_semantic_vector_sync_skips_embed_opt_out_and_clears_vectors( monkeypatch.setattr( search_service.entity_repository, "find_by_id", - AsyncMock(return_value=SimpleNamespace(id=42, entity_metadata={"embed": False})), + AsyncMock( + return_value=SimpleNamespace(id=42, checksum="c42", entity_metadata={"embed": False}) + ), ) sync_vectors = AsyncMock() delete_entity_vectors = AsyncMock() @@ -133,7 +135,7 @@ async def test_semantic_vector_sync_resumes_when_embed_opt_out_removed(search_se monkeypatch.setattr( search_service.entity_repository, "find_by_id", - AsyncMock(return_value=SimpleNamespace(id=42, entity_metadata={})), + AsyncMock(return_value=SimpleNamespace(id=42, checksum="c42", entity_metadata={})), ) sync_vectors = AsyncMock() execute_query = AsyncMock() @@ -159,8 +161,8 @@ async def test_semantic_vector_sync_batch_skips_embed_opt_out_and_reports_skips( "find_by_ids", AsyncMock( return_value=[ - SimpleNamespace(id=41, entity_metadata={"embed": False}), - SimpleNamespace(id=42, entity_metadata={}), + SimpleNamespace(id=41, checksum="c41", entity_metadata={"embed": False}), + SimpleNamespace(id=42, checksum="c42", entity_metadata={}), ] ), ) @@ -254,8 +256,8 @@ async def test_reindex_vectors_respects_embed_opt_out(search_service, monkeypatc "find_all", AsyncMock( return_value=[ - SimpleNamespace(id=41, entity_metadata={"embed": False}), - SimpleNamespace(id=42, entity_metadata={}), + SimpleNamespace(id=41, checksum="c41", entity_metadata={"embed": False}), + SimpleNamespace(id=42, checksum="c42", entity_metadata={}), ] ), ) @@ -304,7 +306,7 @@ async def test_reindex_vectors_purges_sqlite_vectors_before_sync(search_service, monkeypatch.setattr( search_service.entity_repository, "find_all", - AsyncMock(return_value=[SimpleNamespace(id=42, entity_metadata={})]), + AsyncMock(return_value=[SimpleNamespace(id=42, checksum="c42", entity_metadata={})]), ) async def delete_stale_vector_rows(): @@ -413,8 +415,8 @@ async def test_reindex_vectors_force_full_clears_project_vectors_before_resync( "find_all", AsyncMock( return_value=[ - SimpleNamespace(id=41, entity_metadata={}), - SimpleNamespace(id=42, entity_metadata={}), + SimpleNamespace(id=41, checksum="c41", entity_metadata={}), + SimpleNamespace(id=42, checksum="c42", entity_metadata={}), ] ), ) @@ -463,7 +465,7 @@ async def test_semantic_vector_sync_batch_cleans_up_unknown_ids(search_service, monkeypatch.setattr( search_service.entity_repository, "find_by_ids", - AsyncMock(return_value=[SimpleNamespace(id=42, entity_metadata={})]), + AsyncMock(return_value=[SimpleNamespace(id=42, checksum="c42", entity_metadata={})]), ) sync_batch = AsyncMock( side_effect=[ diff --git a/tests/test_cloud_user_agent.py b/tests/test_cloud_user_agent.py new file mode 100644 index 000000000..ec639aca1 --- /dev/null +++ b/tests/test_cloud_user_agent.py @@ -0,0 +1,82 @@ +"""Every HTTP client Basic Memory opens toward Cloud identifies itself. + +httpx's default (`python-httpx/x`) looks like any script. These assert the header on +the wire, through a MockTransport swapped into the real clients the CLI and MCP open. +""" + +import httpx +import pytest + +from basic_memory import __version__ +from basic_memory.cli.auth import CLIAuth +from basic_memory.cli.commands.cloud.api_client import _default_http_client +from basic_memory.config import ProjectMode +from basic_memory.mcp import async_client as async_client_module +from basic_memory.mcp.async_client import get_client, get_cloud_control_plane_client +from basic_memory.user_agent import BASIC_MEMORY_USER_AGENT + + +@pytest.fixture(autouse=True) +def _no_client_factory(): + async_client_module._client_factory = None + yield + async_client_module._client_factory = None + + +@pytest.fixture +def cloud_config(config_manager): + cfg = config_manager.load_config() + cfg.cloud_host = "https://cloud.example.test" + cfg.cloud_api_key = "bmc_test_key_123" + cfg.set_project_mode("research", ProjectMode.CLOUD) + config_manager.save_config(cfg) + return cfg + + +async def _sent_user_agent(client: httpx.AsyncClient, **request_kwargs) -> str: + """Send one request through the client's own headers and return its User-Agent.""" + seen: dict[str, str] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["user_agent"] = request.headers["User-Agent"] + return httpx.Response(200, json={}) + + client._transport = httpx.MockTransport(handler) # pyright: ignore[reportPrivateUsage] + await client.get("https://cloud.example.test/health", **request_kwargs) + return seen["user_agent"] + + +def test_user_agent_names_the_package_version(): + assert BASIC_MEMORY_USER_AGENT == f"basic-memory/{__version__} (cli)" + + +@pytest.mark.asyncio +async def test_cloud_proxy_client_sends_the_user_agent(cloud_config): + async with get_client(project_name="research") as client: + assert await _sent_user_agent(client) == BASIC_MEMORY_USER_AGENT + + +@pytest.mark.asyncio +async def test_control_plane_client_sends_the_user_agent(cloud_config): + async with get_cloud_control_plane_client(workspace="tenant-123") as client: + assert await _sent_user_agent(client) == BASIC_MEMORY_USER_AGENT + + +@pytest.mark.asyncio +async def test_a_callers_user_agent_still_wins(cloud_config): + async with get_cloud_control_plane_client() as client: + sent = await _sent_user_agent(client, headers={"User-Agent": "custom/1.0"}) + assert sent == "custom/1.0" + + +@pytest.mark.asyncio +async def test_cloud_api_request_client_sends_the_user_agent(): + async with _default_http_client(timeout=5.0) as client: + assert await _sent_user_agent(client) == BASIC_MEMORY_USER_AGENT + + +@pytest.mark.asyncio +async def test_cli_auth_client_sends_the_user_agent(config_manager): + auth = CLIAuth(client_id="cid", authkit_domain="https://auth.example.test") + async with auth._get_http_client() as client: # pyright: ignore[reportPrivateUsage] + assert await _sent_user_agent(client) == BASIC_MEMORY_USER_AGENT diff --git a/tests/test_entity_embedded_checksum_migration.py b/tests/test_entity_embedded_checksum_migration.py new file mode 100644 index 000000000..dd6345109 --- /dev/null +++ b/tests/test_entity_embedded_checksum_migration.py @@ -0,0 +1,99 @@ +"""Migration 0044 adds entity.embedded_checksum, backfills it, and indexes lagging notes.""" + +from importlib import import_module + +from alembic import command +from sqlalchemy import inspect, text +from sqlalchemy.engine import Engine + +from basic_memory.models.knowledge import ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE +from tests.test_entity_file_path_conflict_keys_migration import ( # noqa: F401 (fixture) + _alembic_config, + migration_database, +) + +migration = import_module("basic_memory.alembic.versions.0044_entity_embedded_checksum") +PREVIOUS_REVISION = str(migration.down_revision) +INDEX_NAME = "ix_entity_project_embedding_not_current" + + +def _seed(engine: Engine) -> None: + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO project (id, external_id, name, permalink, path, is_active, " + "created_at, updated_at) VALUES (1, 'project-1', 'main', 'main', '/tmp/main', " + "true, '2026-01-01 00:00:00', '2026-01-01 00:00:00')" + ) + ) + connection.execute( + text( + "INSERT INTO entity (id, external_id, title, note_type, content_type, " + "project_id, file_path, file_path_key, file_path_permalink, checksum, " + "vector_sync_deferred_at, created_at, updated_at) VALUES (:id, :external_id, " + "'Note', 'note', 'text/markdown', 1, :file_path, :file_path, :file_path, " + ":checksum, :deferred_at, '2026-01-01 00:00:00', '2026-01-01 00:00:00')" + ), + [ + { + "id": 1, + "external_id": "settled", + "file_path": "settled.md", + "checksum": "settled-checksum", + "deferred_at": None, + }, + { + "id": 2, + "external_id": "part-way", + "file_path": "part-way.md", + "checksum": "part-way-checksum", + "deferred_at": "2026-01-02 00:00:00", + }, + ], + ) + + +def test_upgrade_backfills_and_indexes_lagging_notes( + migration_database: tuple[str, Engine], # noqa: F811 (fixture) +) -> None: + database_url, engine = migration_database + config = _alembic_config(database_url) + command.upgrade(config, PREVIOUS_REVISION) + _seed(engine) + + command.upgrade(config, migration.revision) + + with engine.connect() as connection: + stamps = { + row[0]: row[1] + for row in connection.execute(text("SELECT id, embedded_checksum FROM entity")) + } + behind = connection.execute( + text(f"SELECT id FROM entity WHERE {ENTITY_EMBEDDING_NOT_CURRENT_PREDICATE}") + ).all() + indexes = { + index["name"]: index["column_names"] + for index in inspect(connection).get_indexes("entity") + } + + # A settled note is taken as current; one a pass left part-way still owes work. + assert stamps == {1: "settled-checksum", 2: None} + assert [row[0] for row in behind] == [2] + assert indexes[INDEX_NAME] == ["project_id", "updated_at"] + + +def test_downgrade_drops_the_column_and_index( + migration_database: tuple[str, Engine], # noqa: F811 (fixture) +) -> None: + database_url, engine = migration_database + config = _alembic_config(database_url) + command.upgrade(config, migration.revision) + command.downgrade(config, PREVIOUS_REVISION) + + with engine.connect() as connection: + columns = [column["name"] for column in inspect(connection).get_columns("entity")] + indexes = [index["name"] for index in inspect(connection).get_indexes("entity")] + + assert "embedded_checksum" not in columns + assert len(columns) == len(set(columns)), f"entity has duplicated columns: {columns}" + assert INDEX_NAME not in indexes