diff --git a/src/basic_memory/api/app.py b/src/basic_memory/api/app.py index 3bf92b804..4d8aa4748 100644 --- a/src/basic_memory/api/app.py +++ b/src/basic_memory/api/app.py @@ -19,6 +19,7 @@ prompt_router as v2_prompt, importer_router as v2_importer, schema_router as v2_schema, + inspect_router as v2_inspect, ) import logfire from basic_memory.index.note_content_materialization import drain_pending_materializations @@ -131,6 +132,7 @@ async def workspace_permalink_context_middleware(request: Request, call_next): app.include_router(v2_prompt, prefix="/v2/projects/{project_id}") app.include_router(v2_importer, prefix="/v2/projects/{project_id}") 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") # Legacy web app proxy paths (compat with /proxy/projects/projects) diff --git a/src/basic_memory/api/v2/routers/__init__.py b/src/basic_memory/api/v2/routers/__init__.py index 21fb9b43d..9b68ef53f 100644 --- a/src/basic_memory/api/v2/routers/__init__.py +++ b/src/basic_memory/api/v2/routers/__init__.py @@ -9,6 +9,7 @@ from basic_memory.api.v2.routers.prompt_router import router as prompt_router from basic_memory.api.v2.routers.importer_router import router as importer_router from basic_memory.api.v2.routers.schema_router import router as schema_router +from basic_memory.api.v2.routers.inspect_router import router as inspect_router __all__ = [ "knowledge_router", @@ -20,4 +21,5 @@ "prompt_router", "importer_router", "schema_router", + "inspect_router", ] diff --git a/src/basic_memory/api/v2/routers/inspect_router.py b/src/basic_memory/api/v2/routers/inspect_router.py new file mode 100644 index 000000000..d2c711e2f --- /dev/null +++ b/src/basic_memory/api/v2/routers/inspect_router.py @@ -0,0 +1,150 @@ +"""V2 router for read-only retrieval inspection.""" + +from typing import assert_never + +from fastapi import APIRouter, HTTPException + +from basic_memory.deps import ( + FileServiceV2ExternalDep, + LinkResolverV2ExternalDep, + ProjectExternalIdPathDep, + SearchRepositoryV2ExternalDep, +) +from basic_memory.schemas.inspect import ( + InspectChunk, + InspectChunkReadiness, + InspectChunksRequest, + InspectChunksResponse, + InspectDetachedSearchRow, + InspectIndexBehindRowsDetail, + InspectRowsBehindFileDetail, + InspectSearchRow, +) +from basic_memory.services.retrieval_inspect import ( + ChunkFresh, + ChunkFreshnessUnknown, + ChunkNotIndexed, + ChunkIndexBehindRows, + ChunkRowsBehindFile, + inspect_entity_chunks, +) + +router = APIRouter(prefix="/inspect", tags=["inspect"]) + + +@router.post("/chunks", response_model=InspectChunksResponse) +async def inspect_chunks( + data: InspectChunksRequest, + project_id: ProjectExternalIdPathDep, + link_resolver: LinkResolverV2ExternalDep, + search_repository: SearchRepositoryV2ExternalDep, + file_service: FileServiceV2ExternalDep, +) -> InspectChunksResponse: + """Show how one note is represented by search rows and vector chunks.""" + entity = await link_resolver.resolve_entity( + data.identifier, + load_relations=False, + ) + if entity is None or entity.project_id != project_id: + raise HTTPException(status_code=404, detail=f"Entity not found: '{data.identifier}'") + + inspection = await inspect_entity_chunks(search_repository, entity, file_service) + freshness_detail: InspectIndexBehindRowsDetail | InspectRowsBehindFileDetail | None + match inspection.freshness: + case ChunkFresh() | ChunkNotIndexed(): + freshness_detail = None + case ChunkIndexBehindRows(): + freshness_detail = InspectIndexBehindRowsDetail( + entity_fingerprint_indexed=( + list(inspection.freshness.entity_fingerprint_indexed) + if isinstance( + inspection.freshness.entity_fingerprint_indexed, + tuple, + ) + else inspection.freshness.entity_fingerprint_indexed + ), + entity_fingerprint_current=(inspection.freshness.entity_fingerprint_current), + missing_chunk_count=inspection.freshness.missing_chunk_count, + ) + case ChunkRowsBehindFile() | ChunkFreshnessUnknown(): + evidence = inspection.freshness.evidence + freshness_detail = InspectRowsBehindFileDetail( + entity_checksum=evidence.entity_checksum, + current_file_checksum=evidence.current_file_checksum, + db_checksum=evidence.db_checksum, + file_checksum=evidence.file_checksum, + file_write_status=evidence.file_write_status, + ) + case unexpected: # pragma: no cover - EntityChunkFreshness is exhaustive + assert_never(unexpected) + return InspectChunksResponse( + entity_id=entity.id, + external_id=entity.external_id, + permalink=entity.permalink, + file_path=entity.file_path, + title=entity.title, + entity_checksum=entity.checksum, + configured_embedding_model=inspection.configured_identity.embedding_model, + configured_vector_index=inspection.configured_identity.vector_index, + readiness=InspectChunkReadiness( + total=inspection.readiness.total, + ready=inspection.readiness.ready, + pending=inspection.readiness.pending, + stale=inspection.readiness.stale, + orphaned=inspection.readiness.orphaned, + missing=inspection.readiness.missing, + ), + entity_fingerprint_indexed=( + list(inspection.entity_fingerprint_indexed) + if isinstance(inspection.entity_fingerprint_indexed, tuple) + else inspection.entity_fingerprint_indexed + ), + entity_fingerprint_current=inspection.entity_fingerprint_current, + stale=inspection.stale, + freshness=inspection.freshness.value, + freshness_detail=freshness_detail, + rows=[ + InspectSearchRow( + type=inspected_row.search_row.type, + id=inspected_row.search_row.id, + title=inspected_row.search_row.title, + category=inspected_row.search_row.category, + relation_type=inspected_row.search_row.relation_type, + content_preview=inspected_row.search_row.content_snippet, + chunks=[ + InspectChunk( + chunk_key=chunk.stored_row.chunk_key, + ordinal=chunk.ordinal, + text=chunk.stored_row.chunk_text, + source_hash=chunk.stored_row.source_hash, + embedding_model=chunk.stored_row.embedding_model, + vector_index=chunk.stored_row.vector_index, + status=chunk.status, + updated_at=chunk.stored_row.updated_at, + ) + for chunk in inspected_row.chunks + ], + ) + for inspected_row in inspection.rows + ], + detached=[ + InspectDetachedSearchRow( + type=detached_row.row_type, + id=detached_row.row_id, + chunks=[ + InspectChunk( + chunk_key=chunk.stored_row.chunk_key, + ordinal=chunk.ordinal, + text=chunk.stored_row.chunk_text, + source_hash=chunk.stored_row.source_hash, + embedding_model=chunk.stored_row.embedding_model, + vector_index=chunk.stored_row.vector_index, + status=chunk.status, + updated_at=chunk.stored_row.updated_at, + ) + for chunk in detached_row.chunks + ], + ) + for detached_row in inspection.detached + ], + ) diff --git a/src/basic_memory/cli/app.py b/src/basic_memory/cli/app.py index bf0ea489e..7cd2686e9 100644 --- a/src/basic_memory/cli/app.py +++ b/src/basic_memory/cli/app.py @@ -127,6 +127,7 @@ def _post_command_messages() -> None: # ('hook' returns above, before this point.) skip_init_commands = { "doctor", + "inspect", "man", "mcp", "status", diff --git a/src/basic_memory/cli/commands/__init__.py b/src/basic_memory/cli/commands/__init__.py index 0f0225ac1..78ad09a46 100644 --- a/src/basic_memory/cli/commands/__init__.py +++ b/src/basic_memory/cli/commands/__init__.py @@ -6,6 +6,7 @@ import_chatgpt, man, tool, + inspect, project, config, format, @@ -26,6 +27,7 @@ "import_claude_projects", "import_chatgpt", "tool", + "inspect", "project", "config", "format", diff --git a/src/basic_memory/cli/commands/inspect.py b/src/basic_memory/cli/commands/inspect.py new file mode 100644 index 000000000..7638053c6 --- /dev/null +++ b/src/basic_memory/cli/commands/inspect.py @@ -0,0 +1,320 @@ +"""Read-only retrieval inspection commands.""" + +from typing import Annotated, Optional, assert_never + +import typer +from loguru import logger +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +from rich.text import Text + +from basic_memory.cli.app import app +from basic_memory.cli.commands.routing import force_routing, validate_routing_flags +from basic_memory.cli.commands.tool import _resolve_output_mode, _validate_output_flags +from basic_memory.mcp.clients.inspect import InspectClient +from basic_memory.mcp.project_context import get_project_client +from basic_memory.schemas.inspect import ( + ChunkStatus, + InspectFreshness, + InspectChunksResponse, + InspectDetachedSearchRow, + InspectIndexBehindRowsDetail, + InspectRowsBehindFileDetail, + InspectSearchRow, +) + +inspect_app = typer.Typer() +app.add_typer(inspect_app, name="inspect", help="Inspect retrieval projections") + +console = Console() + + +async def run_inspect_chunks( + identifier: str, + *, + project: str | None, + project_id: str | None, +) -> InspectChunksResponse: + """Resolve the project route and fetch one chunk inspection.""" + async with get_project_client(project=project, project_id=project_id) as ( + http_client, + active_project, + ): + return await InspectClient(http_client, active_project.external_id).inspect_chunks( + identifier + ) + + +def _text_preview(text: str, limit: int = 120) -> str: + """Return a compact single-line chunk preview for human output.""" + single_line = " ".join(text.split()) + if len(single_line) <= limit: + return single_line + return f"{single_line[: limit - 3].rstrip()}..." + + +def _row_label(row: InspectSearchRow) -> str: + """Build a compact search-row identity without hiding type-specific metadata.""" + details = [f"{row.type}:{row.id}"] + if row.title: + details.append(row.title) + if row.category: + details.append(f"category={row.category}") + if row.relation_type: + details.append(f"relation={row.relation_type}") + return " · ".join(details) + + +def _detached_row_label(row: InspectDetachedSearchRow) -> str: + """Build the stored identity for chunks whose source row disappeared.""" + return f"{row.type}:{row.id} · source row gone" + + +def _rich_status(status: ChunkStatus) -> Text: + """Render every closed chunk status with a distinct terminal style.""" + match status: + case "ready": + return Text("ready", style="green") + case "pending": + return Text("pending", style="yellow") + case "stale": + return Text("stale", style="bold red") + case "orphaned": + return Text("orphaned", style="magenta") + case unexpected: # pragma: no cover - schema validation is exhaustive + assert_never(unexpected) + + +def _rich_freshness(freshness: InspectFreshness) -> Text: + """Render every closed freshness state with its diagnostic severity.""" + match freshness: + case "fresh": + return Text("fresh", style="green") + case "not_indexed": + return Text("not_indexed", style="yellow") + case "index_behind_rows": + return Text("index_behind_rows", style="yellow") + case "rows_behind_file": + return Text("rows_behind_file", style="red") + case "unknown": + return Text("unknown", style="dim") + case unexpected: # pragma: no cover - schema validation is exhaustive + assert_never(unexpected) + + +def _display_value(value: str | list[str] | None) -> str: + """Render optional and multi-valued diagnostic evidence compactly.""" + if value is None: + return "-" + if isinstance(value, list): + return ", ".join(value) + return value + + +def _freshness_detail_lines(response: InspectChunksResponse) -> tuple[str, ...]: + """Render the evidence required by each non-fresh state.""" + match response.freshness: + case "fresh" | "not_indexed": + return () + case "index_behind_rows": + detail = response.freshness_detail + if not isinstance(detail, InspectIndexBehindRowsDetail): # pragma: no cover + raise ValueError("index_behind_rows requires fingerprint detail") + return ( + f"Indexed fingerprint: {_display_value(detail.entity_fingerprint_indexed)}", + f"Current fingerprint: {detail.entity_fingerprint_current}", + f"Current chunks missing from manifest: {detail.missing_chunk_count}", + ) + case "rows_behind_file" | "unknown": + detail = response.freshness_detail + if not isinstance(detail, InspectRowsBehindFileDetail): # pragma: no cover + raise ValueError(f"{response.freshness} requires file lineage detail") + return ( + f"Entity checksum: {_display_value(detail.entity_checksum)}", + f"Current file checksum: {_display_value(detail.current_file_checksum)}", + f"DB checksum: {_display_value(detail.db_checksum)}", + f"Lineage file checksum: {_display_value(detail.file_checksum)}", + f"File write status: {_display_value(detail.file_write_status)}", + ) + case unexpected: # pragma: no cover - InspectFreshness is exhaustive + assert_never(unexpected) + + +def _display_chunks(response: InspectChunksResponse) -> None: + """Render a Rich identity header and one chunk table per search row.""" + readiness = response.readiness + if response.entity_fingerprint_indexed is None: + fingerprint_match = "not indexed" + elif response.stale: + fingerprint_match = "no" + else: + fingerprint_match = "yes" + + header = Text() + header.append(f"{response.title}\n", style="bold cyan") + header.append(f"{response.file_path}") + if response.permalink: + header.append(f" · {response.permalink}", style="green") + header.append(f"\nEntity: {response.entity_id} · {response.external_id}") + header.append( + f"\nEngine: {response.configured_vector_index} · {response.configured_embedding_model}" + ) + header.append( + "\nReadiness: " + f"{readiness.ready} ready, {readiness.pending} pending, " + f"{readiness.stale} stale, {readiness.orphaned} orphaned, " + f"{readiness.missing} missing ({readiness.total} total)" + ) + header.append(f"\nFingerprint match: {fingerprint_match}") + header.append("\nFreshness: ") + header.append_text(_rich_freshness(response.freshness)) + for line in _freshness_detail_lines(response): + header.append(f"\n{line}", style="dim") + console.print(Panel(header, title="Retrieval chunks", expand=False)) + + if readiness.total == 0: + console.print( + "[yellow]No vector chunks are stored; showing search rows only. " + "Semantic search may be disabled.[/yellow]" + ) + + for row in response.rows: + table = Table(title=Text(_row_label(row)), show_header=True, header_style="bold") + table.add_column("Ordinal", justify="right") + table.add_column("Status") + table.add_column("Chars", justify="right") + table.add_column("Text preview", max_width=80) + for chunk in row.chunks: + table.add_row( + str(chunk.ordinal), + _rich_status(chunk.status), + str(len(chunk.text)), + Text(_text_preview(chunk.text)), + ) + console.print(table) + + for row in response.detached: + table = Table( + title=Text(_detached_row_label(row), style="bold red"), + show_header=True, + header_style="bold", + ) + table.add_column("Ordinal", justify="right") + table.add_column("Status") + table.add_column("Chars", justify="right") + table.add_column("Text preview", max_width=80) + for chunk in row.chunks: + table.add_row( + str(chunk.ordinal), + _rich_status(chunk.status), + str(len(chunk.text)), + Text(_text_preview(chunk.text)), + ) + console.print(table) + + +def _plain_chunks(response: InspectChunksResponse) -> None: + """Render the same inspection as undecorated, greppable text.""" + readiness = response.readiness + if response.entity_fingerprint_indexed is None: + fingerprint_match = "not indexed" + else: + fingerprint_match = "no" if response.stale else "yes" + + typer.echo(f"Retrieval chunks: {response.title}") + typer.echo(f"Path: {response.file_path}") + typer.echo(f"Permalink: {response.permalink or '-'}") + typer.echo(f"Entity: {response.entity_id} ({response.external_id})") + typer.echo( + f"Engine: {response.configured_vector_index} / {response.configured_embedding_model}" + ) + typer.echo( + "Readiness: " + f"ready={readiness.ready} pending={readiness.pending} stale={readiness.stale} " + f"orphaned={readiness.orphaned} missing={readiness.missing} total={readiness.total}" + ) + typer.echo(f"Fingerprint match: {fingerprint_match}") + typer.echo(f"Freshness: {response.freshness}") + for line in _freshness_detail_lines(response): + typer.echo(line) + + if readiness.total == 0: + typer.echo( + "Note: No vector chunks are stored; showing search rows only. " + "Semantic search may be disabled." + ) + + for row in response.rows: + typer.echo(f"\n{_row_label(row)}") + if not row.chunks: + typer.echo(" (no chunks)") + continue + for chunk in row.chunks: + typer.echo( + f" {chunk.ordinal} {chunk.status} {len(chunk.text)} chars " + f"{_text_preview(chunk.text)}" + ) + + for row in response.detached: + typer.echo(f"\n{_detached_row_label(row)}") + for chunk in row.chunks: + typer.echo( + f" {chunk.ordinal} {chunk.status} {len(chunk.text)} chars " + f"{_text_preview(chunk.text)}" + ) + + +@inspect_app.command("chunks") +def inspect_chunks( + identifier: Annotated[str, typer.Argument(help="Note identifier to inspect")], + json_output: bool = typer.Option(False, "--json", help="Output raw JSON"), + plain: bool = typer.Option(False, "--plain", help="Output undecorated plain text"), + project: Annotated[ + Optional[str], + typer.Option(help="The project to use; defaults to the configured project."), + ] = None, + project_id: Annotated[ + Optional[str], + typer.Option( + "--project-id", + help="Project external_id (UUID); takes precedence over --project.", + ), + ] = None, + local: bool = typer.Option( + False, "--local", help="Force local API routing (ignore cloud mode)" + ), + cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"), +) -> None: + """Show how the retrieval index decomposes one note into rows and chunks.""" + from basic_memory.cli.commands.command_utils import run_with_cleanup + from fastmcp.exceptions import ToolError + + try: + validate_routing_flags(local, cloud) + _validate_output_flags(json_output, plain) + with force_routing(local=local, cloud=cloud): + response = run_with_cleanup( + run_inspect_chunks( + identifier, + project=project, + project_id=project_id, + ) + ) + + mode = _resolve_output_mode(json_output, plain) + if mode == "json": + print(response.model_dump_json(indent=2)) + elif mode == "plain": + _plain_chunks(response) + else: + _display_chunks(response) + except typer.Exit: + raise + except (ToolError, ValueError) as exc: + typer.echo(f"Error: {exc}", err=True) + raise typer.Exit(1) + except Exception as exc: # pragma: no cover + logger.error(f"Error inspecting retrieval chunks: {exc}") + typer.echo(f"Error: {exc}", err=True) + raise typer.Exit(1) diff --git a/src/basic_memory/cli/main.py b/src/basic_memory/cli/main.py index 631318ad4..31d379c27 100644 --- a/src/basic_memory/cli/main.py +++ b/src/basic_memory/cli/main.py @@ -25,6 +25,7 @@ def _version_only_invocation(argv: list[str]) -> bool: import_claude_conversations, import_claude_projects, import_memory_json, + inspect, man, mcp, orphans, diff --git a/src/basic_memory/mcp/clients/__init__.py b/src/basic_memory/mcp/clients/__init__.py index a9f16f4de..303ff9f67 100644 --- a/src/basic_memory/mcp/clients/__init__.py +++ b/src/basic_memory/mcp/clients/__init__.py @@ -18,6 +18,7 @@ from basic_memory.mcp.clients.resource import ResourceClient from basic_memory.mcp.clients.project import ProjectClient from basic_memory.mcp.clients.schema import SchemaClient +from basic_memory.mcp.clients.inspect import InspectClient __all__ = [ "KnowledgeClient", @@ -27,4 +28,5 @@ "ResourceClient", "ProjectClient", "SchemaClient", + "InspectClient", ] diff --git a/src/basic_memory/mcp/clients/inspect.py b/src/basic_memory/mcp/clients/inspect.py new file mode 100644 index 000000000..b9e28ed88 --- /dev/null +++ b/src/basic_memory/mcp/clients/inspect.py @@ -0,0 +1,34 @@ +"""Typed client for read-only retrieval inspection endpoints.""" + +from httpx import AsyncClient + +import logfire +from basic_memory.schemas.inspect import InspectChunksRequest, InspectChunksResponse + + +class InspectClient: + """Typed client for project-scoped retrieval inspection.""" + + def __init__(self, http_client: AsyncClient, project_id: str): + self.http_client = http_client + self._base_path = f"/v2/projects/{project_id}/inspect" + + async def inspect_chunks(self, identifier: str) -> InspectChunksResponse: + """Inspect one note's search rows and vector chunk manifest.""" + from basic_memory.mcp.tools.utils import call_post + + request = InspectChunksRequest(identifier=identifier) + with logfire.span( + "mcp.client.inspect.chunks", + client_name="inspect", + operation="chunks", + ): + response = await call_post( + self.http_client, + f"{self._base_path}/chunks", + json=request.model_dump(mode="json"), + client_name="inspect", + operation="chunks", + path_template="/v2/projects/{project_id}/inspect/chunks", + ) + return InspectChunksResponse.model_validate(response.json()) diff --git a/src/basic_memory/repository/postgres_search_repository.py b/src/basic_memory/repository/postgres_search_repository.py index 0d6a2c867..6c1e7b810 100644 --- a/src/basic_memory/repository/postgres_search_repository.py +++ b/src/basic_memory/repository/postgres_search_repository.py @@ -325,6 +325,36 @@ def _prepare_single_term(self, term: str, is_prefix: bool = True) -> str: # Abstract hook implementations (vector/semantic, Postgres-specific) # ------------------------------------------------------------------ + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + """Return chunk keys whose pgvector physical row is live for this entity.""" + # Trigger: semantic search is off, or the configured index is external. + # Why: a disabled config expects no physical rows, and external adapters expose + # no portable storage-inspection contract (same rule as get_embedding_status()). + # Outcome: None — chunk status stays manifest-only. + if not self._semantic_enabled or self._semantic_vector_index_name != "pgvector": + return None + async with db.scoped_session(self.session_maker) as session: + tables_result = await session.execute( + text( + "SELECT table_name FROM information_schema.tables " + "WHERE table_name IN ('search_vector_chunks', 'search_vector_embeddings')" + ) + ) + table_names = {str(name) for name in tables_result.scalars().all()} + if not {"search_vector_chunks", "search_vector_embeddings"} <= table_names: + return set() + result = await session.execute( + text( + "SELECT c.chunk_key FROM search_vector_chunks c " + "JOIN search_vector_embeddings e " + "ON e.chunk_id = c.id AND e.source_hash = c.source_hash " + "WHERE c.project_id = :project_id AND c.entity_id = :entity_id" + ), + {"project_id": self.project_id, "entity_id": entity_id}, + ) + return {str(chunk_key) for chunk_key in result.scalars().all()} + @override async def _ensure_vector_tables(self) -> None: self._assert_semantic_available() diff --git a/src/basic_memory/repository/search_repository.py b/src/basic_memory/repository/search_repository.py index e6914e21d..a25d0b538 100644 --- a/src/basic_memory/repository/search_repository.py +++ b/src/basic_memory/repository/search_repository.py @@ -18,6 +18,7 @@ from basic_memory.repository.rerank_provider_factory import create_rerank_provider from basic_memory.repository.postgres_search_repository import PostgresSearchRepository from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.search_repository_base import ChunkManifestRow from basic_memory.repository.semantic_vector_index_factory import ( create_semantic_vector_index, resolve_semantic_vector_index_name, @@ -33,13 +34,30 @@ class SearchRepository(Protocol): Both SQLite and Postgres implementations must satisfy this protocol. """ + session_maker: async_sessionmaker[AsyncSession] + @property def project_id(self) -> int: ... + @property + def configured_embedding_model(self) -> str: ... + + @property + def configured_vector_index(self) -> str: ... + async def init_search_index(self) -> None: """Initialize the search index schema.""" ... + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + """Return chunk keys with a live physical vector row, or None when + physical storage is not inspectable.""" + ... + + async def semantic_effectively_enabled(self) -> bool: + """Return whether semantic retrieval can actually run right now.""" + ... + async def search( self, search_text: Optional[str] = None, @@ -87,6 +105,14 @@ async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> Non """Index multiple items in a batch.""" ... + async def get_entity_search_rows(self, entity_id: int) -> list[SearchIndexRow]: + """Return every search projection owned by one entity.""" + ... + + async def get_entity_chunk_manifest(self, entity_id: int) -> list[ChunkManifestRow]: + """Return the stored vector-chunk manifest for one entity.""" + ... + async def delete_by_permalink(self, permalink: str) -> None: """Delete item by permalink.""" ... diff --git a/src/basic_memory/repository/search_repository_base.py b/src/basic_memory/repository/search_repository_base.py index f67585a45..9e834c7be 100644 --- a/src/basic_memory/repository/search_repository_base.py +++ b/src/basic_memory/repository/search_repository_base.py @@ -3,11 +3,11 @@ import hashlib import time from abc import ABC, abstractmethod -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Mapping, Sequence from contextlib import asynccontextmanager -from dataclasses import replace +from dataclasses import dataclass, replace from datetime import datetime -from typing import Any, Callable, Dict, List, Optional, cast +from typing import Any, Callable, Dict, List, Literal, Optional, cast import logfire as logfire from loguru import logger @@ -16,11 +16,15 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from basic_memory import db +from basic_memory.config import BasicMemoryConfig from basic_memory.repository import semantic_vector_sync from basic_memory.repository.embedding_provider import ( EmbeddingProvider, embedding_provider_identity, ) +from basic_memory.repository.embedding_provider_factory import ( + configured_embedding_provider_identity, +) from basic_memory.repository.rerank_provider import ( RerankProvider, build_rerank_document, @@ -59,6 +63,7 @@ ) from basic_memory.runtime.vector_sync import VectorSyncBatchResult from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode +from basic_memory.utils import ensure_timezone_aware # --- Semantic search constants --- @@ -80,6 +85,53 @@ # auto-increment sequences, so a bare id is ambiguous across row types. Every map in # the vector/hybrid retrieval path must key rows by (type, id) to avoid collisions. type SearchIndexKey = tuple[str, int] +type StoredEmbeddingStatus = Literal["pending", "ready"] + + +@dataclass(frozen=True, slots=True) +class ChunkManifestRow: + """One persisted vector-chunk manifest row.""" + + entity_id: int + chunk_key: str + chunk_text: str + source_hash: str + entity_fingerprint: str + embedding_model: str + vector_index: str + embedding_status: StoredEmbeddingStatus + updated_at: datetime + + @classmethod + def from_mapping(cls, row: Mapping[str, Any]) -> "ChunkManifestRow": + """Hydrate one portable manifest row from SQLite or PostgreSQL.""" + raw_status = str(row["embedding_status"]) + match raw_status: + case "pending": + embedding_status: StoredEmbeddingStatus = "pending" + case "ready": + embedding_status = "ready" + case _: + raise ValueError(f"Unknown vector chunk embedding status: {raw_status!r}") + + return cls( + entity_id=int(row["entity_id"]), + chunk_key=str(row["chunk_key"]), + chunk_text=str(row["chunk_text"]), + source_hash=str(row["source_hash"]), + entity_fingerprint=str(row["entity_fingerprint"]), + embedding_model=str(row["embedding_model"]), + vector_index=str(row["vector_index"]), + embedding_status=embedding_status, + updated_at=row["updated_at"], + ) + + def __post_init__(self) -> None: + """Restore a timezone-aware datetime from either backend's raw value.""" + updated_at = self.updated_at + if isinstance(updated_at, str): + updated_at = datetime.fromisoformat(updated_at) + object.__setattr__(self, "updated_at", ensure_timezone_aware(updated_at)) async def purge_stale_search_index_rows( @@ -131,6 +183,7 @@ class SearchRepositoryBase(ABC): # --- Subclass-populated attributes --- _semantic_enabled: bool + _app_config: BasicMemoryConfig _semantic_vector_k: int _semantic_min_similarity: float _embedding_provider: Optional[EmbeddingProvider] @@ -161,6 +214,26 @@ def __init__(self, session_maker: async_sessionmaker[AsyncSession], project_id: self.session_maker = session_maker self.project_id = project_id + async def semantic_effectively_enabled(self) -> bool: + """Return whether semantic retrieval can actually run for this repository. + + Configuration is the default signal. Backends with a startup runtime + fallback (SQLite degrades to keyword-only search when sqlite-vec cannot + load, #711) override this with a runtime probe so per-request instances + honor the degraded state instead of trusting still-enabled config. + """ + return self._semantic_enabled + + @property + def configured_embedding_model(self) -> str: + """Return the configured persisted embedding identity without loading a model.""" + return configured_embedding_provider_identity(self._app_config) + + @property + def configured_vector_index(self) -> str: + """Return the configured vector-index identity.""" + return self._semantic_vector_index_name + # ------------------------------------------------------------------ # Abstract methods — FTS and schema (backend-specific) # ------------------------------------------------------------------ @@ -859,6 +932,58 @@ async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> Non logger.debug(f"Bulk indexed {len(search_index_rows)} rows") await session.commit() + async def get_entity_search_rows(self, entity_id: int) -> list[SearchIndexRow]: + """Return every search projection owned by one entity.""" + async with db.scoped_session(self.session_maker) as session: + result = await session.execute( + text( + "SELECT project_id, id, title, content_stems, content_snippet, " + "permalink, file_path, type, metadata, from_id, to_id, relation_type, " + "entity_id, category, created_at, updated_at " + "FROM search_index " + "WHERE project_id = :project_id AND (" + "(type = 'entity' AND id = :entity_id) " + "OR entity_id = :entity_id " + "OR (type = 'relation' AND from_id = :entity_id)" + ") ORDER BY type, id" + ), + {"project_id": self.project_id, "entity_id": entity_id}, + ) + return [SearchIndexRow.from_mapping(dict(row)) for row in result.mappings().all()] + + async def get_entity_chunk_manifest(self, entity_id: int) -> list[ChunkManifestRow]: + """Return the stored vector-chunk manifest for one project-scoped entity.""" + async with db.scoped_session(self.session_maker) as session: + connection = await session.connection() + manifest_exists = await connection.run_sync( + lambda sync_connection: inspect(sync_connection).has_table("search_vector_chunks") + ) + if not manifest_exists: + return [] + + result = await session.execute( + text( + "SELECT entity_id, chunk_key, chunk_text, source_hash, " + "entity_fingerprint, embedding_model, vector_index, " + "embedding_status, updated_at " + "FROM search_vector_chunks " + "WHERE project_id = :project_id AND entity_id = :entity_id " + "ORDER BY chunk_key" + ), + {"project_id": self.project_id, "entity_id": entity_id}, + ) + return [ChunkManifestRow.from_mapping(dict(row)) for row in result.mappings().all()] + + @abstractmethod + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + """Return manifest chunk keys whose built-in physical vector row is live. + + ``None`` means physical storage is not inspectable here: semantic search is + disabled, or the configured index is external and exposes no portable + storage-inspection contract (mirroring get_embedding_status()). + """ + ... + async def delete_by_entity_id(self, entity_id: int) -> None: """Delete all search index entries for an entity. diff --git a/src/basic_memory/repository/sqlite_search_repository.py b/src/basic_memory/repository/sqlite_search_repository.py index 459d16c4c..62dee4899 100644 --- a/src/basic_memory/repository/sqlite_search_repository.py +++ b/src/basic_memory/repository/sqlite_search_repository.py @@ -400,6 +400,59 @@ def _relaxed_fts_text(search_text: Optional[str]) -> Optional[str]: return None return " OR ".join(f"{word}*" for word in words) + @override + async def semantic_effectively_enabled(self) -> bool: + """Probe the sqlite-vec runtime instead of trusting still-enabled config. + + init_search_index() disables semantics only on the startup repository + instance; per-request instances are rebuilt from config, so they must + re-check the runtime to honor the keyword-only fallback (#711). + """ + if not self._semantic_enabled: + return False + async with db.scoped_session(self.session_maker) as session: + try: + await self._ensure_sqlite_vec_loaded(session) + except SemanticDependenciesMissingError: + return False + return True + + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + """Return chunk keys whose sqlite-vec physical row is live for this entity.""" + # Trigger: semantic search is off, or the configured index is external. + # Why: a disabled config expects no physical rows, and external adapters expose + # no portable storage-inspection contract (same rule as get_embedding_status()). + # Outcome: None — chunk status stays manifest-only. + if not self._semantic_enabled or self._semantic_vector_index_name != "sqlite-vec": + return None + async with db.scoped_session(self.session_maker) as session: + tables_result = await session.execute( + text( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name IN ('search_vector_chunks', 'search_vector_embeddings')" + ) + ) + table_names = {str(name) for name in tables_result.scalars().all()} + if not {"search_vector_chunks", "search_vector_embeddings"} <= table_names: + return set() + try: + await self._ensure_sqlite_vec_loaded(session) + except SemanticDependenciesMissingError: + # Runtime fallback: tables remain from a working install but this host + # cannot load sqlite-vec, so physical storage is not inspectable here. + return None + result = await session.execute( + text( + "SELECT c.chunk_key FROM search_vector_chunks c " + "JOIN search_vector_embeddings e " + "ON e.rowid = c.id AND e.source_hash = c.source_hash " + "WHERE c.project_id = :project_id AND c.entity_id = :entity_id" + ), + {"project_id": self.project_id, "entity_id": entity_id}, + ) + return {str(chunk_key) for chunk_key in result.scalars().all()} + # ------------------------------------------------------------------ # sqlite-vec extension loading (SQLite-specific) # ------------------------------------------------------------------ diff --git a/src/basic_memory/schemas/inspect.py b/src/basic_memory/schemas/inspect.py new file mode 100644 index 000000000..d024e6ffb --- /dev/null +++ b/src/basic_memory/schemas/inspect.py @@ -0,0 +1,146 @@ +"""API schemas for note-level retrieval inspection.""" + +from datetime import datetime +from typing import Literal, Self, assert_never + +from pydantic import BaseModel, ConfigDict, model_validator + +type ChunkStatus = Literal["ready", "pending", "stale", "orphaned"] +type InspectFreshness = Literal[ + "fresh", + "not_indexed", + "index_behind_rows", + "rows_behind_file", + "unknown", +] +type InspectFileWriteStatus = Literal[ + "pending", + "writing", + "synced", + "failed", + "external_change_detected", +] + + +class InspectChunksRequest(BaseModel): + """Request to inspect the retrieval projections for one note identifier.""" + + identifier: str + + +class InspectChunkReadiness(BaseModel): + """Mutually exclusive vector-chunk readiness counts. + + ``missing`` counts current chunks with no stored manifest row; ``total`` covers + stored rows only. + """ + + total: int + ready: int + pending: int + stale: int + orphaned: int + missing: int + + +class InspectChunk(BaseModel): + """One vector chunk stored for a search row.""" + + chunk_key: str + ordinal: int + text: str + source_hash: str + embedding_model: str + vector_index: str + status: ChunkStatus + updated_at: datetime + + +class InspectSearchRow(BaseModel): + """One search row and its current stored vector chunks.""" + + type: str + id: int + title: str | None + category: str | None + relation_type: str | None + content_preview: str | None + chunks: list[InspectChunk] + + +class InspectDetachedSearchRow(BaseModel): + """Stored chunks grouped by a source search row that no longer exists.""" + + type: str + id: int + source_row_gone: Literal[True] = True + chunks: list[InspectChunk] + + +class InspectIndexBehindRowsDetail(BaseModel): + """Fingerprint mismatch or uncovered chunks proving the index trails current rows.""" + + model_config = ConfigDict(extra="forbid") + + entity_fingerprint_indexed: str | list[str] | None + entity_fingerprint_current: str + missing_chunk_count: int + + +class InspectRowsBehindFileDetail(BaseModel): + """File and note-content evidence used for file-to-row freshness.""" + + model_config = ConfigDict(extra="forbid") + + entity_checksum: str | None + current_file_checksum: str | None + db_checksum: str | None + file_checksum: str | None + file_write_status: InspectFileWriteStatus | None + + +type InspectFreshnessDetail = InspectIndexBehindRowsDetail | InspectRowsBehindFileDetail + + +class InspectChunksResponse(BaseModel): + """Note identity, chunk readiness, and search-row decomposition.""" + + entity_id: int + external_id: str + permalink: str | None + file_path: str + title: str + entity_checksum: str | None + configured_embedding_model: str + configured_vector_index: str + readiness: InspectChunkReadiness + entity_fingerprint_indexed: str | list[str] | None + entity_fingerprint_current: str + stale: bool + freshness: InspectFreshness + freshness_detail: InspectFreshnessDetail | None + rows: list[InspectSearchRow] + detached: list[InspectDetachedSearchRow] + + @model_validator(mode="after") + def validate_freshness_detail(self) -> Self: + """Keep the derived freshness value and its evidence in a valid combination.""" + match self.freshness: + case "fresh" | "not_indexed": + detail_is_valid = self.freshness_detail is None + case "index_behind_rows": + detail_is_valid = isinstance( + self.freshness_detail, + InspectIndexBehindRowsDetail, + ) + case "rows_behind_file" | "unknown": + detail_is_valid = isinstance( + self.freshness_detail, + InspectRowsBehindFileDetail, + ) + case unexpected: # pragma: no cover - InspectFreshness is exhaustive + assert_never(unexpected) + + if not detail_is_valid: + raise ValueError(f"Invalid detail for freshness={self.freshness}") + return self diff --git a/src/basic_memory/services/retrieval_inspect.py b/src/basic_memory/services/retrieval_inspect.py new file mode 100644 index 000000000..2318127c5 --- /dev/null +++ b/src/basic_memory/services/retrieval_inspect.py @@ -0,0 +1,468 @@ +"""Read-only inspection of one entity's retrieval projections.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Literal, assert_never + +from basic_memory import db +from basic_memory.indexing.note_content_reconciler import note_content_state_from_model +from basic_memory.indexing.note_content_reconciliation import ( + NoteContentState, + NoteContentWriteStatus, +) +from basic_memory.models import Entity +from basic_memory.repository.note_content_repository import NoteContentRepository +from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.search_repository import SearchRepository +from basic_memory.repository.search_repository_base import ( + ChunkManifestRow, + SearchRepositoryBase, +) +from basic_memory.repository.semantic_chunking import ( + build_entity_fingerprint, + build_vector_chunk_records, +) +from basic_memory.schemas.inspect import ChunkStatus +from basic_memory.services.exceptions import FileOperationError +from basic_memory.services.file_service import FileService +from basic_memory.services.search_service import entity_embeddings_enabled + + +@dataclass(frozen=True, slots=True) +class ConfiguredVectorIdentity: + """The vector configuration whose rows are visible to retrieval.""" + + embedding_model: str + vector_index: str + + +@dataclass(frozen=True, slots=True) +class CurrentSourceHashes: + """Current chunk and entity hashes derived from the search rows.""" + + by_chunk_key: Mapping[str, str] + entity_fingerprint: str + + +@dataclass(frozen=True, slots=True) +class InspectedChunk: + """One stored manifest chunk with its derived inspection status.""" + + stored_row: ChunkManifestRow + ordinal: int + status: ChunkStatus + + +@dataclass(frozen=True, slots=True) +class InspectedSearchRow: + """One current search row and the stored chunks derived from it.""" + + search_row: SearchIndexRow + chunks: tuple[InspectedChunk, ...] + + +@dataclass(frozen=True, slots=True) +class InspectedDetachedSearchRow: + """Stored chunks whose source search row no longer exists.""" + + row_type: str + row_id: int + chunks: tuple[InspectedChunk, ...] + + +@dataclass(frozen=True, slots=True) +class ChunkReadiness: + """Mutually exclusive readiness counts for one entity manifest. + + ``missing`` counts current chunks with no manifest row at all — they never enter the + stored-row comparison, so ``total`` (stored rows) cannot see them. + """ + + total: int + ready: int + pending: int + stale: int + orphaned: int + missing: int + + +@dataclass(frozen=True, slots=True) +class ChunkFresh: + """The file, search rows, and stored chunks agree.""" + + value: Literal["fresh"] = "fresh" + + +@dataclass(frozen=True, slots=True) +class ChunkIndexBehindRows: + """Stored chunks were built from older search rows, or do not cover them all.""" + + entity_fingerprint_indexed: str | tuple[str, ...] | None + entity_fingerprint_current: str + missing_chunk_count: int = 0 + value: Literal["index_behind_rows"] = "index_behind_rows" + + +@dataclass(frozen=True, slots=True) +class ChunkNotIndexed: + """No search projection exists for this entity yet (pre-index or cleared for reindex).""" + + value: Literal["not_indexed"] = "not_indexed" + + +@dataclass(frozen=True, slots=True) +class FileFreshnessEvidence: + """File and note-content checksums used to diagnose row freshness.""" + + entity_checksum: str | None + current_file_checksum: str | None + db_checksum: str | None + file_checksum: str | None + file_write_status: NoteContentWriteStatus | None + + +@dataclass(frozen=True, slots=True) +class ChunkRowsBehindFile: + """Search rows were derived from bytes older than the canonical file.""" + + evidence: FileFreshnessEvidence + value: Literal["rows_behind_file"] = "rows_behind_file" + + +@dataclass(frozen=True, slots=True) +class ChunkFreshnessUnknown: + """The file could not be read and lineage cannot prove its relation to the rows.""" + + evidence: FileFreshnessEvidence + value: Literal["unknown"] = "unknown" + + +type EntityChunkFreshness = ( + ChunkFresh + | ChunkNotIndexed + | ChunkIndexBehindRows + | ChunkRowsBehindFile + | ChunkFreshnessUnknown +) + + +@dataclass(frozen=True, slots=True) +class EntityChunkInspection: + """Complete note-level retrieval inspection result.""" + + entity: Entity + configured_identity: ConfiguredVectorIdentity + readiness: ChunkReadiness + entity_fingerprint_indexed: str | tuple[str, ...] | None + entity_fingerprint_current: str + stale: bool + freshness: EntityChunkFreshness + rows: tuple[InspectedSearchRow, ...] + detached: tuple[InspectedDetachedSearchRow, ...] + + +def classify_chunk_status( + stored_row: ChunkManifestRow, + current_source_hashes: CurrentSourceHashes, + configured_identity: ConfiguredVectorIdentity, + physical_chunk_keys: set[str] | None, +) -> ChunkStatus: + """Classify one stored chunk against current sources and configured retrieval identity. + + ``physical_chunk_keys`` names the manifest chunks whose built-in physical vector row + is live; ``None`` means physical storage is not inspectable (semantic disabled or an + external index) and status stays manifest-only. + """ + if ( + stored_row.embedding_model != configured_identity.embedding_model + or stored_row.vector_index != configured_identity.vector_index + ): + return "orphaned" + + current_source_hash = current_source_hashes.by_chunk_key.get(stored_row.chunk_key) + if ( + current_source_hash != stored_row.source_hash + or stored_row.entity_fingerprint != current_source_hashes.entity_fingerprint + ): + return "stale" + + match stored_row.embedding_status: + case "ready": + # Trigger: the manifest claims ready but the built-in physical vector row is + # gone or carries a different source_hash. + # Why: retrieval joins the manifest to physical storage, so a missing join + # partner can never be served — get_embedding_status() counts the same + # state as orphaned at the project level. + # Outcome: report the chunk as orphaned instead of ready. + if physical_chunk_keys is not None and stored_row.chunk_key not in physical_chunk_keys: + return "orphaned" + return "ready" + case "pending": + return "pending" + case unexpected: # pragma: no cover - repository hydration rejects this value + assert_never(unexpected) + + +def _summarize_readiness( + chunks: tuple[InspectedChunk, ...], + *, + missing_current_chunks: int, +) -> ChunkReadiness: + ready = 0 + pending = 0 + stale = 0 + orphaned = 0 + for chunk in chunks: + match chunk.status: + case "ready": + ready += 1 + case "pending": + pending += 1 + case "stale": + stale += 1 + case "orphaned": + orphaned += 1 + case unexpected: # pragma: no cover - ChunkStatus is exhaustive + assert_never(unexpected) + + return ChunkReadiness( + total=len(chunks), + ready=ready, + pending=pending, + stale=stale, + orphaned=orphaned, + missing=missing_current_chunks, + ) + + +def _file_freshness_evidence( + *, + entity_checksum: str | None, + current_file_checksum: str | None, + note_content: NoteContentState | None, +) -> FileFreshnessEvidence: + """Collect the portable checksum evidence exposed by the inspection response.""" + return FileFreshnessEvidence( + entity_checksum=entity_checksum, + current_file_checksum=current_file_checksum, + db_checksum=note_content.db_checksum if note_content is not None else None, + file_checksum=note_content.file_checksum if note_content is not None else None, + file_write_status=(note_content.file_write_status if note_content is not None else None), + ) + + +def lineage_shows_rows_behind_file( + *, + entity_checksum: str, + note_content: NoteContentState | None, +) -> bool: + """Return whether note-content lineage proves that indexed rows trail the file. + + ``synced`` proves that ``file_checksum`` names accepted bytes only when it agrees with + ``db_checksum``. ``external_change_detected`` instead records the actual unexpected file + checksum protected by the materialization conflict guard. Pending, writing, and failed + states retain historical bookkeeping but do not prove which bytes are currently in storage. + """ + if note_content is None or note_content.file_checksum is None: + return False + + match note_content.file_write_status: + case "synced": + file_checksum_is_observed = note_content.file_checksum == note_content.db_checksum + case "external_change_detected": + file_checksum_is_observed = note_content.file_checksum != note_content.db_checksum + case "pending" | "writing" | "failed": + file_checksum_is_observed = False + case unexpected: # pragma: no cover - NoteContentState validates the persisted status + assert_never(unexpected) + + return file_checksum_is_observed and note_content.file_checksum != entity_checksum + + +def derive_chunk_freshness( + *, + entity_search_row_present: bool, + entity_checksum: str | None, + current_file_checksum: str | None, + note_content: NoteContentState | None, + entity_fingerprint_indexed: str | tuple[str, ...] | None, + entity_fingerprint_current: str, + index_behind_rows: bool, + missing_chunk_count: int, +) -> EntityChunkFreshness: + """Derive note-level freshness from file, row, and chunk evidence.""" + # Trigger: the entity has no search row at all (pre-first-index, or cleared for reindex). + # Why: with two empty projections every comparison is vacuously "matching", which would + # report fresh while the exact layer this inspector diagnoses is absent. + # Outcome: name the missing projection instead of claiming freshness. + if not entity_search_row_present: + return ChunkNotIndexed() + + file_evidence = _file_freshness_evidence( + entity_checksum=entity_checksum, + current_file_checksum=current_file_checksum, + note_content=note_content, + ) + + # Trigger: the entity has no final checksum, or current/observed file bytes differ from it. + # Why: the file feeds search rows, which in turn feed the chunk index. + # Outcome: report the upstream rows-behind-file divergence even if the manifest is stale too. + if entity_checksum is None: + return ChunkRowsBehindFile(evidence=file_evidence) + if current_file_checksum is not None: + if current_file_checksum != entity_checksum: + return ChunkRowsBehindFile(evidence=file_evidence) + elif lineage_shows_rows_behind_file( + entity_checksum=entity_checksum, + note_content=note_content, + ): + return ChunkRowsBehindFile(evidence=file_evidence) + else: + return ChunkFreshnessUnknown(evidence=file_evidence) + + # Trigger: stored fingerprints disagree with current rows, or current chunks have no + # manifest row at all (e.g. only the first shard of an over-limit entity was scheduled). + # Why: matching fingerprints cannot prove coverage — a missing key never enters the + # stored-row comparison, so per-chunk checks are blind to it. + # Outcome: report the index as behind the rows, with the uncovered count as evidence. + if index_behind_rows or missing_chunk_count: + return ChunkIndexBehindRows( + entity_fingerprint_indexed=entity_fingerprint_indexed, + entity_fingerprint_current=entity_fingerprint_current, + missing_chunk_count=missing_chunk_count, + ) + return ChunkFresh() + + +async def _load_note_content_state( + repository: SearchRepository, + entity: Entity, +) -> NoteContentState | None: + """Read the project-scoped note-content lineage once for this inspection.""" + note_content_repository = NoteContentRepository(project_id=entity.project_id) + async with db.scoped_session(repository.session_maker) as session: + note_content = await note_content_repository.get_by_entity_id(session, entity.id) + return note_content_state_from_model(note_content) if note_content is not None else None + + +async def _read_current_file_checksum( + file_service: FileService, + entity: Entity, +) -> str | None: + """Read one project-scoped file checksum, returning absence when storage is unavailable.""" + try: + _, checksum = await file_service.read_file(entity.file_path) + except FileOperationError: + return None + return checksum + + +async def inspect_entity_chunks( + repository: SearchRepository, + entity: Entity, + file_service: FileService, +) -> EntityChunkInspection: + """Inspect current search rows and stored vector chunks without running retrieval stages.""" + # SQLite FTS can hold duplicate copies of one logical (type, id) row. + # build_vector_chunk_records() collapses them when chunking, so this view must + # too — otherwise displayed chunks duplicate while readiness counts each once. + search_rows_by_key = { + (row.type, row.id): row for row in await repository.get_entity_search_rows(entity.id) + } + stored_rows = await repository.get_entity_chunk_manifest(entity.id) + physical_chunk_keys = await repository.get_entity_physical_chunk_keys(entity.id) + note_content = await _load_note_content_state(repository, entity) + current_file_checksum = await _read_current_file_checksum(file_service, entity) + + current_records = build_vector_chunk_records(list(search_rows_by_key.values())).records + current_source_hashes = CurrentSourceHashes( + by_chunk_key={record["chunk_key"]: record["source_hash"] for record in current_records}, + entity_fingerprint=build_entity_fingerprint(current_records), + ) + configured_identity = ConfiguredVectorIdentity( + embedding_model=repository.configured_embedding_model, + vector_index=repository.configured_vector_index, + ) + + inspected_chunks: list[InspectedChunk] = [] + chunks_by_search_row: dict[tuple[str, int], list[InspectedChunk]] = {} + for stored_row in stored_rows: + row_key = SearchRepositoryBase._parse_chunk_key(stored_row.chunk_key) + ordinal = int(stored_row.chunk_key.split(":")[2]) + inspected_chunk = InspectedChunk( + stored_row=stored_row, + ordinal=ordinal, + status=classify_chunk_status( + stored_row, + current_source_hashes, + configured_identity, + physical_chunk_keys, + ), + ) + inspected_chunks.append(inspected_chunk) + chunks_by_search_row.setdefault(row_key, []).append(inspected_chunk) + + indexed_fingerprints = tuple(sorted({row.entity_fingerprint for row in stored_rows})) + if not indexed_fingerprints: + indexed_fingerprint: str | tuple[str, ...] | None = None + elif len(indexed_fingerprints) == 1: + indexed_fingerprint = indexed_fingerprints[0] + else: + indexed_fingerprint = indexed_fingerprints + + fingerprint_mismatch = any( + fingerprint != current_source_hashes.entity_fingerprint + for fingerprint in indexed_fingerprints + ) + # A current chunk with no manifest row never enters the stored-row loop above, so + # count the uncovered remainder explicitly (e.g. shards beyond the scheduling limit). + # With semantic indexing off — by config, by the runtime keyword-only fallback, or + # by this note's own embed opt-out — no chunks are expected at all, so an empty + # manifest is not missing coverage. + if entity_embeddings_enabled(entity) and await repository.semantic_effectively_enabled(): + stored_chunk_keys = {stored_row.chunk_key for stored_row in stored_rows} + missing_chunk_count = sum( + 1 + for chunk_key in current_source_hashes.by_chunk_key + if chunk_key not in stored_chunk_keys + ) + else: + missing_chunk_count = 0 + inspected_chunk_tuple = tuple(inspected_chunks) + return EntityChunkInspection( + entity=entity, + configured_identity=configured_identity, + readiness=_summarize_readiness( + inspected_chunk_tuple, + missing_current_chunks=missing_chunk_count, + ), + entity_fingerprint_indexed=indexed_fingerprint, + entity_fingerprint_current=current_source_hashes.entity_fingerprint, + stale=fingerprint_mismatch or missing_chunk_count > 0, + freshness=derive_chunk_freshness( + entity_search_row_present=("entity", entity.id) in search_rows_by_key, + entity_checksum=entity.checksum, + current_file_checksum=current_file_checksum, + note_content=note_content, + entity_fingerprint_indexed=indexed_fingerprint, + entity_fingerprint_current=current_source_hashes.entity_fingerprint, + index_behind_rows=fingerprint_mismatch, + missing_chunk_count=missing_chunk_count, + ), + rows=tuple( + InspectedSearchRow( + search_row=search_row, + chunks=tuple(chunks_by_search_row.get((search_row.type, search_row.id), ())), + ) + for search_row in search_rows_by_key.values() + ), + detached=tuple( + InspectedDetachedSearchRow( + row_type=row_type, + row_id=row_id, + chunks=tuple(chunks), + ) + for (row_type, row_id), chunks in chunks_by_search_row.items() + if (row_type, row_id) not in search_rows_by_key + ), + ) diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 4db7d73fc..077b980b9 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -54,6 +54,34 @@ class _PreparedSearchQuery: min_similarity: float | None +def entity_embeddings_enabled(entity: Entity) -> bool: + """Return whether semantic embeddings should be generated for this entity. + + Shared policy: sync uses it to clear and skip opted-out notes, and the retrieval + inspector uses it so an opt-out is never reported as missing vector coverage. + """ + if not entity.entity_metadata: + return True + + embed_value = entity.entity_metadata.get("embed") + if embed_value is None: + return True + if isinstance(embed_value, bool): + return embed_value + if isinstance(embed_value, str): + normalized = embed_value.strip().lower() + if normalized in {"false", "0", "no", "off"}: + return False + if normalized in {"true", "1", "yes", "on"}: + return True + if isinstance(embed_value, (int, float)): + return bool(embed_value) + + # Default unknown values to enabled so malformed metadata does not silently + # remove notes from semantic search. + return True + + def _strip_nul(value: str) -> str: """Strip NUL bytes that PostgreSQL text columns cannot store. @@ -506,7 +534,7 @@ async def sync_entity_vectors(self, entity_id: int) -> None: await self._clear_entity_vectors(entity_id) return - if not self._entity_embeddings_enabled(entity): + if not entity_embeddings_enabled(entity): await self._clear_entity_vectors(entity_id) return @@ -532,7 +560,7 @@ async def sync_entity_vectors_batch( for entity_id in entity_ids if ( (entity := entities_by_id.get(entity_id)) is not None - and not self._entity_embeddings_enabled(entity) + and not entity_embeddings_enabled(entity) ) ] if opted_out_ids: @@ -579,9 +607,7 @@ async def sync_entity_vectors_batch( ), sample_errors=tuple( dict.fromkeys( - error - for result in repository_results - for error in result.sample_errors + error for result in repository_results for error in result.sample_errors ) )[:VECTOR_SYNC_SAMPLE_ERROR_LIMIT], vector_index=next( @@ -589,11 +615,7 @@ async def sync_entity_vectors_batch( "", ), embedding_model=next( - ( - result.embedding_model - for result in repository_results - if result.embedding_model - ), + (result.embedding_model for result in repository_results if result.embedding_model), "", ), chunks_total=sum(result.chunks_total for result in repository_results), @@ -674,30 +696,6 @@ async def _purge_stale_search_rows(self) -> None: "Purged stale search rows", project_id=self.repository.project_id, purged=purged ) - @staticmethod - def _entity_embeddings_enabled(entity: Entity) -> bool: - """Return whether semantic embeddings should be generated for this entity.""" - if not entity.entity_metadata: - return True - - embed_value = entity.entity_metadata.get("embed") - if embed_value is None: - return True - if isinstance(embed_value, bool): - return embed_value - if isinstance(embed_value, str): - normalized = embed_value.strip().lower() - if normalized in {"false", "0", "no", "off"}: - return False - if normalized in {"true", "1", "yes", "on"}: - return True - if isinstance(embed_value, (int, float)): - return bool(embed_value) - - # Default unknown values to enabled so malformed metadata does not silently - # remove notes from semantic search. - return True - 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/tests/api/v2/test_inspect_router.py b/tests/api/v2/test_inspect_router.py new file mode 100644 index 000000000..a92fe889d --- /dev/null +++ b/tests/api/v2/test_inspect_router.py @@ -0,0 +1,252 @@ +"""API contract tests for note-level retrieval inspection.""" + +from pathlib import Path + +import pytest +from httpx import AsyncClient +from sqlalchemy import text + +from basic_memory import db +from basic_memory.models import Project +from basic_memory.repository.semantic_chunking import ( + build_entity_fingerprint, + build_vector_chunk_records, +) +from basic_memory.schemas.inspect import InspectChunksResponse, InspectRowsBehindFileDetail + + +async def _create_indexed_entity( + *, + test_project: Project, + title: str, + file_name: str, + entity_repository, + search_service, + file_service, +): + content = f"# {title}\n\nRetrieval inspection content for {title}." + file_path = Path(test_project.path) / file_name + checksum = await file_service.write_file(file_path, content) + async with db.scoped_session(search_service.session_maker) as session: + entity = await entity_repository.create( + session, + { + "title": title, + "note_type": "note", + "content_type": "text/markdown", + "file_path": file_name, + "permalink": f"notes/{file_name.removesuffix('.md')}", + "checksum": checksum, + }, + ) + await search_service.index_entity(entity) + return entity + + +async def _seed_current_manifest(search_repository, session_maker, entity_id: int) -> int: + rows = await search_repository.get_entity_search_rows(entity_id) + records = build_vector_chunk_records(rows).records + fingerprint = build_entity_fingerprint(records) + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "INSERT INTO search_vector_chunks (" + "project_id, entity_id, chunk_key, chunk_text, source_hash, " + "entity_fingerprint, embedding_model, vector_index, embedding_status" + ") VALUES (" + ":project_id, :entity_id, :chunk_key, :chunk_text, :source_hash, " + ":entity_fingerprint, :embedding_model, :vector_index, 'ready')" + ), + [ + { + "project_id": search_repository.project_id, + "entity_id": entity_id, + "chunk_key": record["chunk_key"], + "chunk_text": record["chunk_text"], + "source_hash": record["source_hash"], + "entity_fingerprint": fingerprint, + "embedding_model": search_repository.configured_embedding_model, + "vector_index": search_repository.configured_vector_index, + } + for record in records + ], + ) + await session.commit() + return len(records) + + +@pytest.mark.asyncio +async def test_inspect_chunks_returns_valid_schema_for_seeded_corpus( + client: AsyncClient, + v2_project_url: str, + test_project: Project, + entity_repository, + search_repository, + search_service, + file_service, + session_maker, +): + entity = await _create_indexed_entity( + test_project=test_project, + title="API Chunk Inspection", + file_name="api-chunk-inspection.md", + entity_repository=entity_repository, + search_service=search_service, + file_service=file_service, + ) + chunk_count = await _seed_current_manifest(search_repository, session_maker, entity.id) + + response = await client.post( + f"{v2_project_url}/inspect/chunks", + json={"identifier": entity.permalink}, + ) + + assert response.status_code == 200, response.text + inspection = InspectChunksResponse.model_validate(response.json()) + assert inspection.entity_id == entity.id + assert inspection.external_id == entity.external_id + assert inspection.entity_checksum == entity.checksum + assert inspection.readiness.total == chunk_count + assert inspection.readiness.ready == chunk_count + assert inspection.stale is False + assert inspection.freshness == "fresh" + assert inspection.freshness_detail is None + assert [row.type for row in inspection.rows] == ["entity"] + assert sum(len(row.chunks) for row in inspection.rows) == chunk_count + assert inspection.detached == [] + + +@pytest.mark.asyncio +async def test_inspect_chunks_displays_manifest_rows_with_missing_sources_as_detached( + client: AsyncClient, + v2_project_url: str, + test_project: Project, + entity_repository, + search_repository, + search_service, + file_service, + session_maker, +): + entity = await _create_indexed_entity( + test_project=test_project, + title="Detached Chunk Inspection", + file_name="detached-chunk-inspection.md", + entity_repository=entity_repository, + search_service=search_service, + file_service=file_service, + ) + chunk_count = await _seed_current_manifest(search_repository, session_maker, entity.id) + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "DELETE FROM search_index WHERE project_id = :project_id " + "AND type = 'entity' AND id = :entity_id" + ), + {"project_id": test_project.id, "entity_id": entity.id}, + ) + await session.commit() + + response = await client.post( + f"{v2_project_url}/inspect/chunks", + json={"identifier": entity.external_id}, + ) + + assert response.status_code == 200, response.text + inspection = InspectChunksResponse.model_validate(response.json()) + assert inspection.rows == [] + assert [(row.type, row.id, row.source_row_gone) for row in inspection.detached] == [ + ("entity", entity.id, True) + ] + assert sum(len(row.chunks) for row in inspection.detached) == chunk_count + assert inspection.readiness.total == chunk_count + + +@pytest.mark.asyncio +async def test_inspect_chunks_reports_rows_behind_file_with_checksum_detail( + client: AsyncClient, + v2_project_url: str, + test_project: Project, + entity_repository, + search_service, + file_service, +): + entity = await _create_indexed_entity( + test_project=test_project, + title="Rows Behind File", + file_name="rows-behind-file.md", + entity_repository=entity_repository, + search_service=search_service, + file_service=file_service, + ) + file_service.get_entity_path(entity).write_bytes(b"# Edited outside the index\n") + + response = await client.post( + f"{v2_project_url}/inspect/chunks", + json={"identifier": entity.external_id}, + ) + + assert response.status_code == 200, response.text + inspection = InspectChunksResponse.model_validate(response.json()) + assert inspection.freshness == "rows_behind_file" + assert isinstance(inspection.freshness_detail, InspectRowsBehindFileDetail) + assert inspection.freshness_detail.entity_checksum == entity.checksum + assert inspection.freshness_detail.current_file_checksum != entity.checksum + + +@pytest.mark.asyncio +async def test_inspect_chunks_returns_404_for_unresolved_identifier( + client: AsyncClient, + v2_project_url: str, +): + response = await client.post( + f"{v2_project_url}/inspect/chunks", + json={"identifier": "notes/does-not-exist"}, + ) + + assert response.status_code == 404 + assert response.json()["detail"] == "Entity not found: 'notes/does-not-exist'" + + +@pytest.mark.asyncio +async def test_inspect_chunks_semantic_disabled_returns_rows_only( + client: AsyncClient, + v2_project_url: str, + test_project: Project, + app_config, + entity_repository, + search_service, + file_service, +): + assert app_config.semantic_search_enabled is False + entity = await _create_indexed_entity( + test_project=test_project, + title="Rows Only Inspection", + file_name="rows-only-inspection.md", + entity_repository=entity_repository, + search_service=search_service, + file_service=file_service, + ) + async with db.scoped_session(search_service.session_maker) as session: + await session.execute(text("DROP TABLE search_vector_chunks")) + await session.commit() + + response = await client.post( + f"{v2_project_url}/inspect/chunks", + json={"identifier": entity.external_id}, + ) + + assert response.status_code == 200, response.text + inspection = InspectChunksResponse.model_validate(response.json()) + assert inspection.readiness.model_dump() == { + "total": 0, + "ready": 0, + "pending": 0, + "stale": 0, + "orphaned": 0, + "missing": 0, + } + assert inspection.rows + assert all(not row.chunks for row in inspection.rows) + assert inspection.detached == [] + assert inspection.entity_fingerprint_indexed is None + assert inspection.stale is False diff --git a/tests/cli/test_inspect_command.py b/tests/cli/test_inspect_command.py new file mode 100644 index 000000000..d85f1335d --- /dev/null +++ b/tests/cli/test_inspect_command.py @@ -0,0 +1,368 @@ +"""CLI tests for ``bm inspect chunks``.""" + +from contextlib import asynccontextmanager +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastmcp.exceptions import ToolError +from pydantic import ValidationError +from typer.testing import CliRunner + +from basic_memory.cli.main import app as cli_app +import basic_memory.cli.commands.inspect as inspect_command +from basic_memory.schemas.inspect import ( + ChunkStatus, + InspectChunk, + InspectChunkReadiness, + InspectChunksResponse, + InspectDetachedSearchRow, + InspectFreshness, + InspectIndexBehindRowsDetail, + InspectRowsBehindFileDetail, + InspectSearchRow, +) + +runner = CliRunner() + + +def _inspection_response() -> InspectChunksResponse: + updated_at = datetime(2026, 8, 12, 12, 30, tzinfo=timezone.utc) + statuses: tuple[ChunkStatus, ...] = ("ready", "pending", "stale", "orphaned") + chunks = [ + InspectChunk( + chunk_key=f"entity:7:{ordinal}", + ordinal=ordinal, + text=("Chunk text " * (ordinal + 1)).strip(), + source_hash=f"source-{ordinal}", + embedding_model="FastEmbedEmbeddingProvider:test-model:384", + vector_index="sqlite-vec", + status=status, + updated_at=updated_at, + ) + for ordinal, status in enumerate(statuses) + ] + return InspectChunksResponse( + entity_id=7, + external_id="11111111-1111-1111-1111-111111111111", + permalink="notes/retrieval-inspection", + file_path="notes/Retrieval Inspection.md", + title="Retrieval [Inspection]", + entity_checksum="checksum-7", + configured_embedding_model="FastEmbedEmbeddingProvider:test-model:384", + configured_vector_index="sqlite-vec", + readiness=InspectChunkReadiness( + total=4, + ready=1, + pending=1, + stale=1, + orphaned=1, + missing=2, + ), + entity_fingerprint_indexed="old-fingerprint", + entity_fingerprint_current="current-fingerprint", + stale=True, + freshness="index_behind_rows", + freshness_detail=InspectIndexBehindRowsDetail( + entity_fingerprint_indexed="old-fingerprint", + entity_fingerprint_current="current-fingerprint", + missing_chunk_count=2, + ), + rows=[ + InspectSearchRow( + type="entity", + id=7, + title="Retrieval [Inspection]", + category=None, + relation_type=None, + content_preview="Entity content", + chunks=chunks[:-1], + ), + InspectSearchRow( + type="observation", + id=8, + title="Retrieval [Inspection]", + category="fact", + relation_type=None, + content_preview="Observation content", + chunks=[], + ), + InspectSearchRow( + type="relation", + id=9, + title="Retrieval [Inspection]", + category=None, + relation_type="supports", + content_preview="Relation content", + chunks=[], + ), + ], + detached=[ + InspectDetachedSearchRow( + type="relation", + id=10, + chunks=[chunks[-1]], + ) + ], + ) + + +def _rows_only_response() -> InspectChunksResponse: + response = _inspection_response() + return response.model_copy( + update={ + "readiness": InspectChunkReadiness( + total=0, + ready=0, + pending=0, + stale=0, + orphaned=0, + missing=0, + ), + "entity_fingerprint_indexed": None, + "stale": False, + "freshness": "fresh", + "freshness_detail": None, + "rows": [row.model_copy(update={"chunks": []}) for row in response.rows], + "detached": [], + } + ) + + +def _response_with_freshness(freshness: InspectFreshness) -> InspectChunksResponse: + """Build a schema-valid CLI fixture for every closed freshness state.""" + response = _inspection_response() + if freshness == "fresh": + detail = None + elif freshness == "index_behind_rows": + detail = InspectIndexBehindRowsDetail( + entity_fingerprint_indexed="old-fingerprint", + entity_fingerprint_current="current-fingerprint", + missing_chunk_count=2, + ) + else: + detail = InspectRowsBehindFileDetail( + entity_checksum="entity-checksum", + current_file_checksum=( + "current-file-checksum" if freshness == "rows_behind_file" else None + ), + db_checksum="db-checksum", + file_checksum="lineage-file-checksum", + file_write_status="external_change_detected", + ) + return InspectChunksResponse.model_validate( + { + **response.model_dump(), + "freshness": freshness, + "freshness_detail": detail, + } + ) + + +@patch("basic_memory.cli.commands.tool._use_rich", return_value=True) +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_rich_rendering(mock_run, _mock_use_rich): + mock_run.return_value = _inspection_response() + + result = runner.invoke(cli_app, ["inspect", "chunks", "notes/retrieval-inspection"]) + + assert result.exit_code == 0, result.output + assert "Retrieval [Inspection]" in result.output + assert "1 ready, 1 pending, 1 stale, 1 orphaned, 2 missing" in result.output + assert "entity:7" in result.output + assert "category=fact" in result.output + assert "relation=supports" in result.output + assert "relation:10 · source row gone" in result.output + assert all(status in result.output for status in ("ready", "pending", "stale", "orphaned")) + assert "Freshness: index_behind_rows" in result.output + + +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_plain_rendering(mock_run): + mock_run.return_value = _inspection_response() + + result = runner.invoke( + cli_app, + ["inspect", "chunks", "notes/retrieval-inspection", "--plain"], + ) + + assert result.exit_code == 0, result.output + assert "Engine: sqlite-vec / FastEmbedEmbeddingProvider:test-model:384" in result.output + assert "Fingerprint match: no" in result.output + assert "Freshness: index_behind_rows" in result.output + assert "Indexed fingerprint: old-fingerprint" in result.output + assert "0 ready" in result.output + assert "relation:10 · source row gone" in result.output + assert "─" not in result.output + assert "│" not in result.output + + +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_json_is_pydantic_schema_locked(mock_run): + """Machine output is exactly the API response schema serialized by Pydantic.""" + expected = _inspection_response() + mock_run.return_value = expected + + result = runner.invoke( + cli_app, + [ + "inspect", + "chunks", + "notes/retrieval-inspection", + "--json", + "--project", + "research", + "--project-id", + "22222222-2222-2222-2222-222222222222", + ], + ) + + assert result.exit_code == 0, result.output + validated = InspectChunksResponse.model_validate_json(result.output) + assert validated == expected + mock_run.assert_awaited_once_with( + "notes/retrieval-inspection", + project="research", + project_id="22222222-2222-2222-2222-222222222222", + ) + + +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_piped_output_defaults_to_json(mock_run): + expected = _inspection_response() + mock_run.return_value = expected + + result = runner.invoke(cli_app, ["inspect", "chunks", "notes/retrieval-inspection"]) + + assert result.exit_code == 0, result.output + assert InspectChunksResponse.model_validate_json(result.output) == expected + + +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_rows_only_note_does_not_error(mock_run): + mock_run.return_value = _rows_only_response() + + result = runner.invoke( + cli_app, + ["inspect", "chunks", "notes/retrieval-inspection", "--plain"], + ) + + assert result.exit_code == 0, result.output + assert "showing search rows only" in result.output + assert "Semantic search may be disabled" in result.output + assert "(no chunks)" in result.output + assert "Fingerprint match: not indexed" in result.output + assert "Freshness: fresh" in result.output + + +@pytest.mark.parametrize( + ("freshness", "expected_detail"), + [ + ("fresh", None), + ("index_behind_rows", "Indexed fingerprint: old-fingerprint"), + ("rows_behind_file", "Current file checksum: current-file-checksum"), + ("unknown", "Current file checksum: -"), + ], +) +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_plain_renders_every_freshness_value( + mock_run, + freshness: InspectFreshness, + expected_detail: str | None, +): + mock_run.return_value = _response_with_freshness(freshness) + + result = runner.invoke(cli_app, ["inspect", "chunks", "note", "--plain"]) + + assert result.exit_code == 0, result.output + assert f"Freshness: {freshness}" in result.output + if expected_detail is not None: + assert expected_detail in result.output + + +@pytest.mark.parametrize( + ("freshness", "style"), + [ + ("fresh", "green"), + ("index_behind_rows", "yellow"), + ("rows_behind_file", "red"), + ("unknown", "dim"), + ], +) +def test_rich_freshness_uses_diagnostic_colors( + freshness: InspectFreshness, + style: str, +): + assert inspect_command._rich_freshness(freshness).style == style + + +def test_inspect_chunks_schema_rejects_fresh_with_divergence_detail(): + payload = _inspection_response().model_dump() + payload["freshness"] = "fresh" + + with pytest.raises(ValidationError, match="Invalid detail for freshness=fresh"): + InspectChunksResponse.model_validate(payload) + + +def test_inspect_chunks_rejects_mutually_exclusive_output_flags(): + result = runner.invoke( + cli_app, + ["inspect", "chunks", "note", "--json", "--plain"], + ) + + assert result.exit_code == 1 + assert "mutually exclusive" in result.output + + +def test_inspect_chunks_rejects_mutually_exclusive_routing_flags(): + result = runner.invoke( + cli_app, + ["inspect", "chunks", "note", "--local", "--cloud"], + ) + + assert result.exit_code == 1 + assert "Cannot specify both --local and --cloud" in result.output + + +@patch("basic_memory.cli.commands.inspect.run_inspect_chunks", new_callable=AsyncMock) +def test_inspect_chunks_api_error_exits_nonzero(mock_run): + mock_run.side_effect = ToolError("Entity not found") + + result = runner.invoke(cli_app, ["inspect", "chunks", "missing", "--plain"]) + + assert result.exit_code == 1 + assert "Error: Entity not found" in result.output + + +@pytest.mark.asyncio +async def test_run_inspect_chunks_uses_typed_client_and_project_route(monkeypatch): + expected = _inspection_response() + http_client = MagicMock() + active_project = SimpleNamespace(external_id="33333333-3333-3333-3333-333333333333") + + @asynccontextmanager + async def fake_get_project_client(*, project=None, project_id=None): + assert project == "research" + assert project_id == "33333333-3333-3333-3333-333333333333" + yield http_client, active_project + + response = MagicMock() + response.json.return_value = expected.model_dump(mode="json") + call_post = AsyncMock(return_value=response) + monkeypatch.setattr(inspect_command, "get_project_client", fake_get_project_client) + monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", call_post) + + result = await inspect_command.run_inspect_chunks( + "notes/retrieval-inspection", + project="research", + project_id="33333333-3333-3333-3333-333333333333", + ) + + assert result == expected + call_post.assert_awaited_once() + await_args = call_post.await_args + assert await_args is not None + assert await_args.args[1] == ( + "/v2/projects/33333333-3333-3333-3333-333333333333/inspect/chunks" + ) diff --git a/tests/repository/test_chunk_inspection.py b/tests/repository/test_chunk_inspection.py new file mode 100644 index 000000000..2b2177c65 --- /dev/null +++ b/tests/repository/test_chunk_inspection.py @@ -0,0 +1,966 @@ +"""Repository and domain tests for note-level chunk inspection.""" + +from dataclasses import replace +from datetime import datetime, timezone + +import pytest +from sqlalchemy import text + +from basic_memory import db +from basic_memory.indexing.note_content_reconciliation import NoteContentState +from basic_memory.models import NoteContent, Project +from basic_memory.repository.note_content_repository import NoteContentRepository +from basic_memory.repository.project_repository import ProjectRepository +from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.config import DatabaseBackend +from basic_memory.repository.postgres_search_repository import PostgresSearchRepository +from basic_memory.repository.search_repository import create_search_repository +from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository +from basic_memory.repository.search_repository_base import ChunkManifestRow +from basic_memory.repository.semantic_chunking import ( + build_entity_fingerprint, + build_vector_chunk_records, +) +from basic_memory.services.retrieval_inspect import ( + ChunkFresh, + ChunkFreshnessUnknown, + ChunkIndexBehindRows, + ChunkRowsBehindFile, + ConfiguredVectorIdentity, + CurrentSourceHashes, + classify_chunk_status, + inspect_entity_chunks, + lineage_shows_rows_behind_file, +) +from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError +from basic_memory.services.exceptions import FileOperationError + + +def _search_rows(project_id: int, entity_id: int) -> list[SearchIndexRow]: + now = datetime(2026, 8, 12, 12, 0, tzinfo=timezone.utc) + return [ + SearchIndexRow( + project_id=project_id, + id=entity_id, + type="entity", + file_path="notes/inspection.md", + metadata={"note_type": "note"}, + created_at=now, + updated_at=now, + title="Inspection Note", + permalink="notes/inspection", + content_snippet="Entity prose for inspection.", + ), + SearchIndexRow( + project_id=project_id, + id=entity_id, + type="observation", + file_path="notes/inspection.md", + metadata={"note_type": "note"}, + created_at=now, + updated_at=now, + title="Inspection Note", + permalink=f"notes/inspection/observations/{entity_id}", + entity_id=entity_id, + category="fact", + content_snippet="A retrieval fact.", + ), + SearchIndexRow( + project_id=project_id, + id=entity_id, + type="relation", + file_path="notes/inspection.md", + metadata={"note_type": "note"}, + created_at=now, + updated_at=now, + title="Inspection Note", + permalink=f"notes/inspection/supports/{entity_id}", + from_id=entity_id, + entity_id=entity_id, + relation_type="supports", + content_snippet="Inspection Note supports Target Note", + ), + ] + + +async def _insert_manifest( + session_maker, + *, + project_id: int, + entity_id: int, + rows: list[SearchIndexRow], + embedding_model: str, + vector_index: str, + pending_key: str | None = None, + omit_chunk_key: str | None = None, +) -> None: + records = build_vector_chunk_records(rows).records + # Fingerprint always covers the full current record set — omitting a record after + # this point reproduces a scheduling pass that stored only part of the chunks. + entity_fingerprint = build_entity_fingerprint(records) + if omit_chunk_key is not None: + records = [record for record in records if record["chunk_key"] != omit_chunk_key] + updated_at = datetime(2026, 8, 12, 12, 30, tzinfo=timezone.utc) + values = [ + { + "project_id": project_id, + "entity_id": entity_id, + "chunk_key": record["chunk_key"], + "chunk_text": record["chunk_text"], + "source_hash": record["source_hash"], + "entity_fingerprint": entity_fingerprint, + "embedding_model": embedding_model, + "vector_index": vector_index, + "embedding_status": "pending" if record["chunk_key"] == pending_key else "ready", + "updated_at": updated_at, + } + for record in records + ] + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "INSERT INTO search_vector_chunks (" + "project_id, entity_id, chunk_key, chunk_text, source_hash, " + "entity_fingerprint, embedding_model, vector_index, embedding_status, updated_at" + ") VALUES (" + ":project_id, :entity_id, :chunk_key, :chunk_text, :source_hash, " + ":entity_fingerprint, :embedding_model, :vector_index, " + ":embedding_status, :updated_at)" + ), + values, + ) + await session.commit() + + +async def _write_current_entity_file(file_service, entity, content: str = "# Inspection\n") -> str: + """Write the entity file and align its in-memory checksum with those bytes.""" + checksum = await file_service.write_file(entity.file_path, content) + entity.checksum = checksum + return checksum + + +async def _insert_note_content( + session_maker, + entity, + *, + db_checksum: str, + file_checksum: str, + file_write_status: str, +) -> None: + """Persist one lineage state through the project-scoped repository.""" + repository = NoteContentRepository(project_id=entity.project_id) + async with db.scoped_session(session_maker) as session: + await repository.create( + session, + NoteContent( + entity_id=entity.id, + markdown_content="# Accepted\n", + db_version=2, + db_checksum=db_checksum, + file_version=1, + file_checksum=file_checksum, + file_write_status=file_write_status, + ), + ) + + +@pytest.mark.asyncio +async def test_inspection_groups_rows_and_normalizes_manifest_timestamps( + search_repository, + session_maker, + sample_entity, + app_config, + tmp_path, + file_service, +): + """Entity, observation, and relation rows retain distinct chunk ownership.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + pending_key=f"observation:{sample_entity.id}:0", + ) + + # A second project uses the same entity/search ids and chunk key. Project-scoped + # repository reads must not admit any of its rows into the inspection. + async with db.scoped_session(session_maker) as session: + second_project = await ProjectRepository().create( + session, + { + "name": "inspection-other-project", + "path": str(tmp_path / "other"), + "is_active": True, + "is_default": False, + }, + ) + assert isinstance(second_project, Project) + second_repository = create_search_repository( + session_maker, + project_id=second_project.id, + app_config=app_config, + ) + foreign_rows = _search_rows(second_project.id, sample_entity.id) + foreign_rows[0].title = "Foreign inspection row" + await second_repository.bulk_index_items(foreign_rows) + await _insert_manifest( + session_maker, + project_id=second_project.id, + entity_id=sample_entity.id, + rows=foreign_rows, + embedding_model=second_repository.configured_embedding_model, + vector_index=second_repository.configured_vector_index, + ) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert [row.search_row.type for row in inspection.rows] == [ + "entity", + "observation", + "relation", + ] + assert [row.chunks[0].stored_row.chunk_key for row in inspection.rows] == [ + f"entity:{sample_entity.id}:0", + f"observation:{sample_entity.id}:0", + f"relation:{sample_entity.id}:0", + ] + assert inspection.readiness.total == 3 + assert inspection.readiness.ready == 2 + assert inspection.readiness.pending == 1 + assert inspection.readiness.stale == 0 + assert inspection.readiness.orphaned == 0 + assert inspection.stale is False + assert all( + chunk.stored_row.updated_at.tzinfo is not None + for row in inspection.rows + for chunk in row.chunks + ) + assert all(row.search_row.title != "Foreign inspection row" for row in inspection.rows) + + +class _UnusedEmbeddingProvider: + """Satisfies repository construction; inspection never embeds.""" + + model_name = "inspection-embedding" + dimensions = 4 + + async def embed_query(self, text: str) -> list[float]: # pragma: no cover - unused + raise NotImplementedError + + async def embed_documents(self, texts: list[str]) -> list[list[float]]: # pragma: no cover + raise NotImplementedError + + def runtime_log_attrs(self) -> dict[str, object]: # pragma: no cover - unused + return {} + + +def _sqlite_vec_loadable() -> bool: + """The keyword-only runtime fallback exists precisely because this can be False.""" + import sqlite3 + + if not hasattr(sqlite3.Connection, "enable_load_extension"): + return False + try: + import sqlite_vec # noqa: F401 + except ImportError: + return False + return True + + +def _skip_unless_healthy_semantic_runtime(app_config) -> None: + """Healthy-runtime semantic tests need real sqlite-vec on the SQLite backend.""" + if app_config.database_backend != DatabaseBackend.POSTGRES and not _sqlite_vec_loadable(): + pytest.skip("healthy-runtime semantic inspection needs loadable sqlite-vec") + + +def _semantic_repository(session_maker, project_id: int, app_config): + """Build a semantic-enabled repository; the shared fixture disables semantic search.""" + config = app_config.model_copy(update={"semantic_search_enabled": True}) + repository_type = ( + PostgresSearchRepository + if config.database_backend == DatabaseBackend.POSTGRES + else SQLiteSearchRepository + ) + return repository_type( + session_maker, + project_id=project_id, + app_config=config, + embedding_provider=_UnusedEmbeddingProvider(), + ) + + +async def _insert_physical_vectors( + session_maker, + repository, + *, + omit_chunk_key: str | None = None, +) -> None: + """Materialize a physical vector row for every stored manifest chunk. + + Callers must run ``repository._ensure_vector_tables()`` before inserting their + manifest: storage creation resets ready manifest rows to pending by design. + """ + async with db.scoped_session(session_maker) as session: + manifest_result = await session.execute( + text( + "SELECT id, chunk_key, source_hash FROM search_vector_chunks " + "WHERE project_id = :project_id" + ), + {"project_id": repository.project_id}, + ) + manifest_rows = [ + row for row in manifest_result.mappings().all() if row["chunk_key"] != omit_chunk_key + ] + if isinstance(repository, PostgresSearchRepository): + for row in manifest_rows: + await session.execute( + text( + "INSERT INTO search_vector_embeddings (" + "chunk_id, project_id, embedding, embedding_dims, source_hash" + ") VALUES (" + ":chunk_id, :project_id, CAST(:embedding AS vector), 4, :source_hash)" + ), + { + "chunk_id": row["id"], + "project_id": repository.project_id, + "embedding": "[0.1, 0.1, 0.1, 0.1]", + "source_hash": row["source_hash"], + }, + ) + else: + import sqlite_vec + + await repository._ensure_sqlite_vec_loaded(session) + embedding = sqlite_vec.serialize_float32([0.1, 0.1, 0.1, 0.1]) + for row in manifest_rows: + await session.execute( + text( + "INSERT INTO search_vector_embeddings (rowid, embedding, source_hash) " + "VALUES (:rowid, :embedding, :source_hash)" + ), + { + "rowid": row["id"], + "embedding": embedding, + "source_hash": row["source_hash"], + }, + ) + await session.commit() + + +@pytest.mark.asyncio +async def test_current_chunks_missing_from_manifest_mark_index_behind( + session_maker, + sample_entity, + app_config, + file_service, +): + """A manifest covering only part of the current chunks must not report fresh.""" + _skip_unless_healthy_semantic_runtime(app_config) + search_repository = _semantic_repository(session_maker, sample_entity.project_id, app_config) + await search_repository._ensure_vector_tables() + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + # The stored rows carry the full current fingerprint, so per-chunk and fingerprint + # comparisons alone would call this entity fresh while a chunk is unavailable. + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + omit_chunk_key=f"relation:{sample_entity.id}:0", + ) + await _insert_physical_vectors(session_maker, search_repository) + await _write_current_entity_file(file_service, sample_entity) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.readiness.total == 2 + assert inspection.readiness.ready == 2 + assert inspection.readiness.missing == 1 + assert inspection.stale is True + assert isinstance(inspection.freshness, ChunkIndexBehindRows) + assert inspection.freshness.missing_chunk_count == 1 + assert inspection.freshness.entity_fingerprint_indexed == ( + inspection.freshness.entity_fingerprint_current + ) + + +@pytest.mark.asyncio +async def test_embed_opt_out_note_is_not_missing_coverage( + session_maker, + sample_entity, + app_config, + file_service, +): + """A note that opts out of embeddings has no expected chunks to miss.""" + search_repository = _semantic_repository(session_maker, sample_entity.project_id, app_config) + sample_entity.entity_metadata = {"embed": False} + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _write_current_entity_file(file_service, sample_entity) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.readiness.missing == 0 + assert inspection.readiness.total == 0 + assert inspection.stale is False + assert isinstance(inspection.freshness, ChunkFresh) + + +@pytest.mark.asyncio +async def test_ready_chunk_without_physical_vector_reports_orphaned( + session_maker, + sample_entity, + app_config, + file_service, +): + """A manifest-ready chunk whose physical vector row is gone cannot be served.""" + _skip_unless_healthy_semantic_runtime(app_config) + search_repository = _semantic_repository(session_maker, sample_entity.project_id, app_config) + await search_repository._ensure_vector_tables() + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + await _insert_physical_vectors( + session_maker, + search_repository, + omit_chunk_key=f"entity:{sample_entity.id}:0", + ) + await _write_current_entity_file(file_service, sample_entity) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + statuses = { + chunk.stored_row.chunk_key: chunk.status for row in inspection.rows for chunk in row.chunks + } + assert statuses[f"entity:{sample_entity.id}:0"] == "orphaned" + assert statuses[f"observation:{sample_entity.id}:0"] == "ready" + assert statuses[f"relation:{sample_entity.id}:0"] == "ready" + assert inspection.readiness.orphaned == 1 + assert inspection.readiness.ready == 2 + assert inspection.readiness.missing == 0 + + +@pytest.mark.asyncio +async def test_runtime_semantic_fallback_keeps_inspection_manifest_only( + session_maker, + sample_entity, + app_config, + file_service, + monkeypatch, +): + """A vec-less host with enabled config must not invent missing chunks or crash.""" + if app_config.database_backend == DatabaseBackend.POSTGRES: + pytest.skip("the keyword-only runtime fallback is SQLite-specific") + search_repository = _semantic_repository(session_maker, sample_entity.project_id, app_config) + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + # A vec-less host keeps the embeddings table from an earlier working install; a + # plain stand-in reproduces that on-disk shape without loading sqlite-vec, so + # this test runs on the exact keyword-only environment it models. + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "CREATE TABLE IF NOT EXISTS search_vector_embeddings " + "(embedding BLOB, source_hash TEXT)" + ) + ) + await session.commit() + await _write_current_entity_file(file_service, sample_entity) + + async def vec_unavailable(session): + raise SemanticDependenciesMissingError("sqlite-vec unavailable on this host") + + monkeypatch.setattr(search_repository, "_ensure_sqlite_vec_loaded", vec_unavailable) + + assert await search_repository.semantic_effectively_enabled() is False + assert await search_repository.get_entity_physical_chunk_keys(sample_entity.id) is None + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.readiness.missing == 0 + assert inspection.readiness.ready == 3 + assert isinstance(inspection.freshness, ChunkFresh) + + +@pytest.mark.asyncio +async def test_effective_semantic_signal_defaults_to_config_without_runtime_probe( + session_maker, + sample_entity, + app_config, +): + """Backends without a runtime fallback answer from configuration alone.""" + enabled_config = app_config.model_copy(update={"semantic_search_enabled": True}) + enabled = PostgresSearchRepository( + session_maker, + project_id=sample_entity.project_id, + app_config=enabled_config, + embedding_provider=_UnusedEmbeddingProvider(), + ) + assert await enabled.semantic_effectively_enabled() is True + + disabled = PostgresSearchRepository( + session_maker, + project_id=sample_entity.project_id, + app_config=app_config.model_copy(update={"semantic_search_enabled": False}), + ) + assert await disabled.semantic_effectively_enabled() is False + + +@pytest.mark.asyncio +async def test_inspection_marks_manifest_stale_after_search_row_changes( + search_repository, + session_maker, + sample_entity, + file_service, +): + """Changing source search text without re-chunking exposes stale stored chunks.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + await _write_current_entity_file(file_service, sample_entity) + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "UPDATE search_index SET content_snippet = :content " + "WHERE project_id = :project_id AND type = 'observation' " + "AND id = :row_id" + ), + { + "content": "The search projection changed after chunking.", + "project_id": sample_entity.project_id, + "row_id": sample_entity.id, + }, + ) + await session.commit() + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.stale is True + assert inspection.readiness.stale == 3 + assert inspection.readiness.ready == 0 + assert isinstance(inspection.freshness, ChunkIndexBehindRows) + + +@pytest.mark.asyncio +async def test_inspection_uses_all_stored_fingerprints_for_note_staleness( + search_repository, + session_maker, + sample_entity, + file_service, +): + """A mixed shard manifest is stale even when its first row has the current fingerprint.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "UPDATE search_vector_chunks SET entity_fingerprint = :fingerprint " + "WHERE project_id = :project_id AND chunk_key = :chunk_key" + ), + { + "fingerprint": "older-shard-fingerprint", + "project_id": sample_entity.project_id, + "chunk_key": f"relation:{sample_entity.id}:0", + }, + ) + await session.commit() + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.entity_fingerprint_indexed == tuple( + sorted({inspection.entity_fingerprint_current, "older-shard-fingerprint"}) + ) + assert inspection.stale is True + assert inspection.readiness.ready == 2 + assert inspection.readiness.stale == 1 + + +@pytest.mark.asyncio +async def test_inspection_marks_wrong_configured_identity_orphaned( + search_repository, + session_maker, + sample_entity, + file_service, +): + """A manifest row owned by another model is invisible to current retrieval.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "UPDATE search_vector_chunks SET embedding_model = 'LegacyEmbedding:model' " + "WHERE project_id = :project_id AND chunk_key = :chunk_key" + ), + { + "project_id": sample_entity.project_id, + "chunk_key": f"entity:{sample_entity.id}:0", + }, + ) + await session.commit() + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.readiness.orphaned == 1 + assert inspection.readiness.ready == 2 + assert inspection.rows[0].chunks[0].status == "orphaned" + + +@pytest.mark.asyncio +async def test_freshness_marks_missing_entity_checksum_as_rows_behind_file( + search_repository, + sample_entity, + file_service, +): + """A missing entity checksum means the indexing pass has not finalized its rows.""" + await search_repository.bulk_index_items( + _search_rows(sample_entity.project_id, sample_entity.id)[:1] + ) + await file_service.write_file(sample_entity.file_path, "# Unfinished sync\n") + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert isinstance(inspection.freshness, ChunkRowsBehindFile) + assert inspection.freshness.evidence.entity_checksum is None + + +@pytest.mark.asyncio +async def test_freshness_is_fresh_for_untouched_file_rows_and_manifest( + search_repository, + session_maker, + sample_entity, + file_service, +): + """Matching file, entity rows, and manifest produce the positive fresh state.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + await _write_current_entity_file(file_service, sample_entity) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert isinstance(inspection.freshness, ChunkFresh) + + +@pytest.mark.asyncio +async def test_freshness_marks_file_edit_after_index_as_rows_behind_file( + search_repository, + sample_entity, + file_service, +): + """An unsynchronized direct edit makes the file checksum newer than entity rows.""" + await search_repository.bulk_index_items( + _search_rows(sample_entity.project_id, sample_entity.id)[:1] + ) + indexed_checksum = await _write_current_entity_file(file_service, sample_entity) + file_service.get_entity_path(sample_entity).write_bytes(b"# Edited after indexing\n") + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert isinstance(inspection.freshness, ChunkRowsBehindFile) + assert inspection.freshness.evidence.entity_checksum == indexed_checksum + assert inspection.freshness.evidence.current_file_checksum != indexed_checksum + + +@pytest.mark.asyncio +async def test_freshness_uses_conclusive_lineage_when_file_cannot_be_read( + search_repository, + session_maker, + sample_entity, + file_service, + monkeypatch, +): + """A recorded external-file checksum proves the rows trail inaccessible storage.""" + await search_repository.bulk_index_items( + _search_rows(sample_entity.project_id, sample_entity.id)[:1] + ) + sample_entity.checksum = "rows-checksum" + await _insert_note_content( + session_maker, + sample_entity, + db_checksum="accepted-db-checksum", + file_checksum="external-file-checksum", + file_write_status="external_change_detected", + ) + + async def fail_read(_path): + raise FileOperationError("storage unavailable") + + monkeypatch.setattr(file_service, "read_file", fail_read) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert isinstance(inspection.freshness, ChunkRowsBehindFile) + assert inspection.freshness.evidence.current_file_checksum is None + assert inspection.freshness.evidence.db_checksum == "accepted-db-checksum" + assert inspection.freshness.evidence.file_checksum == "external-file-checksum" + assert inspection.freshness.evidence.file_write_status == "external_change_detected" + + +@pytest.mark.asyncio +async def test_freshness_rows_behind_file_takes_precedence_over_stale_manifest( + search_repository, + session_maker, + sample_entity, + file_service, +): + """The upstream file divergence dominates a simultaneous manifest mismatch.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + await _write_current_entity_file(file_service, sample_entity) + file_service.get_entity_path(sample_entity).write_bytes(b"# Newer file\n") + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "UPDATE search_index SET content_snippet = :content " + "WHERE project_id = :project_id AND type = 'entity' AND id = :entity_id" + ), + { + "content": "Newer rows than the stored chunks.", + "project_id": sample_entity.project_id, + "entity_id": sample_entity.id, + }, + ) + await session.commit() + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.stale is True + assert isinstance(inspection.freshness, ChunkRowsBehindFile) + + +@pytest.mark.asyncio +async def test_freshness_is_unknown_when_file_read_fails_with_clean_lineage( + search_repository, + session_maker, + sample_entity, + file_service, + monkeypatch, +): + """Historical agreement cannot prove current inaccessible file bytes are unchanged.""" + await search_repository.bulk_index_items( + _search_rows(sample_entity.project_id, sample_entity.id)[:1] + ) + sample_entity.checksum = "synced-checksum" + await _insert_note_content( + session_maker, + sample_entity, + db_checksum="synced-checksum", + file_checksum="synced-checksum", + file_write_status="synced", + ) + + async def fail_read(_path): + raise FileOperationError("permission denied") + + monkeypatch.setattr(file_service, "read_file", fail_read) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert isinstance(inspection.freshness, ChunkFreshnessUnknown) + + +@pytest.mark.parametrize( + ("state", "expected"), + [ + (NoteContentState(1, "file", 1, "file", "synced"), True), + ( + NoteContentState(2, "accepted", 1, "external", "external_change_detected"), + True, + ), + (NoteContentState(2, "accepted", 1, "old", "pending"), False), + (NoteContentState(2, "accepted", 1, "old", "writing"), False), + (NoteContentState(2, "accepted", 1, "old", "failed"), False), + (NoteContentState(1, "accepted", 1, "different", "synced"), False), + ( + NoteContentState(1, "accepted", 1, "accepted", "external_change_detected"), + False, + ), + ], +) +def test_lineage_only_uses_checksums_observed_by_reconciliation( + state: NoteContentState, + expected: bool, +): + """Only synced or conflict-observed file lineage can prove an upstream mismatch.""" + assert ( + lineage_shows_rows_behind_file( + entity_checksum="rows", + note_content=state, + ) + is expected + ) + + +def test_lineage_without_note_content_is_inconclusive(): + assert not lineage_shows_rows_behind_file( + entity_checksum="rows", + note_content=None, + ) + + +def test_classify_chunk_status_covers_closed_status_space(): + """The pure classifier owns all four mutually exclusive chunk states.""" + current = CurrentSourceHashes( + by_chunk_key={"entity:1:0": "current-source"}, + entity_fingerprint="current-entity", + ) + identity = ConfiguredVectorIdentity( + embedding_model="ConfiguredEmbedding:model", + vector_index="sqlite-vec", + ) + stored = ChunkManifestRow( + entity_id=1, + chunk_key="entity:1:0", + chunk_text="content", + source_hash="current-source", + entity_fingerprint="current-entity", + embedding_model=identity.embedding_model, + vector_index=identity.vector_index, + embedding_status="ready", + updated_at=datetime.now(timezone.utc), + ) + + assert classify_chunk_status(stored, current, identity, {"entity:1:0"}) == "ready" + # None = physical storage not inspectable (semantic disabled or external index): + # status stays manifest-only. + assert classify_chunk_status(stored, current, identity, None) == "ready" + # A ready manifest row whose physical vector row is gone can never be served. + assert classify_chunk_status(stored, current, identity, set()) == "orphaned" + assert classify_chunk_status( + replace(stored, embedding_status="pending"), current, identity, set() + ) == ("pending") + assert ( + classify_chunk_status(replace(stored, source_hash="old"), current, identity, None) + == "stale" + ) + assert ( + classify_chunk_status(replace(stored, vector_index="milvus"), current, identity, None) + == "orphaned" + ) + + +def test_chunk_manifest_row_hydrates_string_timestamp_and_rejects_unknown_status(): + """Portable hydration normalizes SQLite timestamps and validates persisted status.""" + row = { + "entity_id": 1, + "chunk_key": "entity:1:0", + "chunk_text": "content", + "source_hash": "source", + "entity_fingerprint": "entity", + "embedding_model": "model", + "vector_index": "sqlite-vec", + "embedding_status": "ready", + "updated_at": "2026-08-12T12:30:00+00:00", + } + + hydrated = ChunkManifestRow.from_mapping(row) + assert hydrated.updated_at.tzinfo is not None + + with pytest.raises(ValueError, match="Unknown vector chunk embedding status"): + ChunkManifestRow.from_mapping({**row, "embedding_status": "broken"}) + + +@pytest.mark.asyncio +async def test_duplicate_logical_search_rows_collapse_to_one_inspected_row( + search_repository, + session_maker, + sample_entity, + file_service, +): + """SQLite FTS duplicates of one logical row must not double displayed chunks.""" + rows = _search_rows(sample_entity.project_id, sample_entity.id) + await search_repository.bulk_index_items(rows) + # bulk_index_items assumes prior deletion; indexing again fabricates the duplicate + # logical (type, id) copies the finding describes. + await search_repository.bulk_index_items(rows) + await _insert_manifest( + session_maker, + project_id=sample_entity.project_id, + entity_id=sample_entity.id, + rows=rows, + embedding_model=search_repository.configured_embedding_model, + vector_index=search_repository.configured_vector_index, + ) + + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + row_keys = [(row.search_row.type, row.search_row.id) for row in inspection.rows] + assert len(row_keys) == len(set(row_keys)) + displayed_chunks = sum(len(row.chunks) for row in inspection.rows) + sum( + len(detached.chunks) for detached in inspection.detached + ) + assert displayed_chunks == inspection.readiness.total + + +@pytest.mark.asyncio +async def test_absent_search_projection_reports_not_indexed_not_fresh( + search_repository, + sample_entity, + file_service, +): + """Two empty projections are a missing layer, never a vacuous freshness match.""" + inspection = await inspect_entity_chunks(search_repository, sample_entity, file_service) + + assert inspection.freshness.value == "not_indexed" + assert inspection.rows == () + assert inspection.readiness.total == 0 diff --git a/tests/repository/test_hybrid_fusion.py b/tests/repository/test_hybrid_fusion.py index 63806e95f..0b849974b 100644 --- a/tests/repository/test_hybrid_fusion.py +++ b/tests/repository/test_hybrid_fusion.py @@ -60,6 +60,10 @@ def __init__(self): async def init_search_index(self): pass # pragma: no cover + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + return None # physical storage is not inspectable in this double + @override def _prepare_search_term(self, term, is_prefix=True): return term # pragma: no cover diff --git a/tests/repository/test_semantic_search_base.py b/tests/repository/test_semantic_search_base.py index 3e63186af..70d44e425 100644 --- a/tests/repository/test_semantic_search_base.py +++ b/tests/repository/test_semantic_search_base.py @@ -66,6 +66,10 @@ def __init__(self): async def init_search_index(self): pass + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + return None # physical storage is not inspectable in this double + @override def _prepare_search_term(self, term, is_prefix=True): return term diff --git a/tests/repository/test_semantic_vector_sync.py b/tests/repository/test_semantic_vector_sync.py index 4bba2e200..465ded2a6 100644 --- a/tests/repository/test_semantic_vector_sync.py +++ b/tests/repository/test_semantic_vector_sync.py @@ -35,6 +35,10 @@ def __init__(self): async def init_search_index(self): pass + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + return None # physical storage is not inspectable in this double + @override def _prepare_search_term(self, term, is_prefix=True): return term @@ -304,9 +308,7 @@ async def test_vector_sync_handles_final_flush_errors_and_orphan_runtime( ) assert orphan_result.failed_entity_ids == (1,) - assert orphan_result.sample_errors == ( - "Vector sync left unfinished entities after flushes.", - ) + assert orphan_result.sample_errors == ("Vector sync left unfinished entities after flushes.",) def test_vector_shard_planning_and_logging_edges(monkeypatch) -> None: diff --git a/tests/repository/test_vector_pagination.py b/tests/repository/test_vector_pagination.py index dfc8faf3c..0764c404b 100644 --- a/tests/repository/test_vector_pagination.py +++ b/tests/repository/test_vector_pagination.py @@ -45,6 +45,10 @@ def __init__(self): async def init_search_index(self): pass # pragma: no cover + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + return None # physical storage is not inspectable in this double + @override def _prepare_search_term(self, term, is_prefix=True): return term # pragma: no cover diff --git a/tests/repository/test_vector_threshold.py b/tests/repository/test_vector_threshold.py index 35864e521..dcd5db25c 100644 --- a/tests/repository/test_vector_threshold.py +++ b/tests/repository/test_vector_threshold.py @@ -49,6 +49,10 @@ def __init__(self): async def init_search_index(self): pass # pragma: no cover + @override + async def get_entity_physical_chunk_keys(self, entity_id: int) -> set[str] | None: + return None # physical storage is not inspectable in this double + @override def _prepare_search_term(self, term, is_prefix=True): return term # pragma: no cover