admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:basicmachines-co/basic-memory

Browse Source
This commit is contained in:
phernandez
2026-04-10 12:23:19 -05:00
parent cc104f761f a872947e03
commit abd4a5a6da
11 changed files with 1106 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/cli/commands/cloud/project_sync.py
-32
View File
@@ -124,22 +124,6 @@ def sync_project_command(
if success:
console.print(f"[green]{name} synced successfully[/green]")
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client(project_name=name) as client:
return await ProjectClient(client).sync(
project_data.external_id, force_full=False
)
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} sync failed[/red]")
raise typer.Exit(1)
@@ -202,22 +186,6 @@ def bisync_project_command(
sync_entry.last_sync = datetime.now()
sync_entry.bisync_initialized = True
ConfigManager().save_config(config)
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client(project_name=name) as client:
return await ProjectClient(client).sync(
project_data.external_id, force_full=False
)
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} bisync failed[/red]")
raise typer.Exit(1)
src/basic_memory/cli/commands/project.py
+231 -4
View File
@@ -14,6 +14,7 @@ from rich.text import Text
from basic_memory.cli.app import app
from basic_memory.cli.auth import CLIAuth
from basic_memory.cli.commands.cloud.api_client import CloudAPIError, make_api_request
from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
from basic_memory.cli.commands.cloud.project_sync import (
_has_cloud_credentials,
@@ -26,9 +27,13 @@ from basic_memory.cli.commands.cloud.rclone_commands import (
from basic_memory.cli.commands.command_utils import get_project_info, run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.async_client import get_client, resolve_configured_workspace
from basic_memory.mcp.clients import ProjectClient
from basic_memory.schemas.cloud import ProjectVisibility
from basic_memory.schemas.cloud import (
CloudProjectIndexStatus,
CloudTenantIndexStatusResponse,
ProjectVisibility,
)
from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.utils import generate_permalink, normalize_project_path
@@ -58,6 +63,211 @@ def make_bar(value: int, max_value: int, width: int = 40) -> Text:
return bar
def _uses_cloud_project_info_route(project_name: str, *, local: bool, cloud: bool) -> bool:
"""Return whether project info should attempt cloud augmentation."""
if local:
return False
if cloud:
return True
config_manager = ConfigManager()
resolved_name, _ = config_manager.get_project(project_name)
effective_name = resolved_name or project_name
return config_manager.config.get_project_mode(effective_name) == ProjectMode.CLOUD
def _resolve_cloud_status_workspace_id(project_name: str) -> str:
"""Resolve the tenant/workspace for cloud index status lookup."""
config_manager = ConfigManager()
config = config_manager.config
if not _has_cloud_credentials(config):
raise RuntimeError(
"Cloud credentials not found. Run `bm cloud api-key save <key>` or `bm cloud login` first."
)
configured_name, _ = config_manager.get_project(project_name)
effective_name = configured_name or project_name
workspace_id = resolve_configured_workspace(config=config, project_name=effective_name)
if workspace_id is not None:
return workspace_id
workspace_id = _resolve_workspace_id(config, None)
if workspace_id is not None:
return workspace_id
raise RuntimeError(
f"Cloud workspace could not be resolved for project '{effective_name}'. "
"Set a project workspace with `bm project set-cloud --workspace ...` or configure a "
"default workspace with `bm cloud workspace set-default ...`."
)
async def _resolve_cloud_status_workspace_id_async(project_name: str) -> str:
"""Resolve the tenant/workspace for cloud index status lookup in async contexts."""
config_manager = ConfigManager()
config = config_manager.config
if not _has_cloud_credentials(config):
raise RuntimeError(
"Cloud credentials not found. Run `bm cloud api-key save <key>` or `bm cloud login` first."
)
configured_name, _ = config_manager.get_project(project_name)
effective_name = configured_name or project_name
workspace_id = resolve_configured_workspace(config=config, project_name=effective_name)
if workspace_id is not None:
return workspace_id
from basic_memory.mcp.project_context import get_available_workspaces
workspaces = await get_available_workspaces()
if len(workspaces) == 1:
return workspaces[0].tenant_id
raise RuntimeError(
f"Cloud workspace could not be resolved for project '{effective_name}'. "
"Set a project workspace with `bm project set-cloud --workspace ...` or configure a "
"default workspace with `bm cloud workspace set-default ...`."
)
def _match_cloud_index_status_project(
project_name: str, projects: list[CloudProjectIndexStatus]
) -> CloudProjectIndexStatus | None:
"""Match the requested project against the tenant index-status payload."""
exact_match = next(
(project for project in projects if project.project_name == project_name), None
)
if exact_match is not None:
return exact_match
project_permalink = generate_permalink(project_name)
permalink_matches = [
project
for project in projects
if generate_permalink(project.project_name) == project_permalink
]
if len(permalink_matches) == 1:
return permalink_matches[0]
return None
def _format_cloud_index_status_error(error: Exception) -> str:
"""Convert cloud lookup failures into concise user-facing text."""
if isinstance(error, CloudAPIError):
detail_message: str | None = None
detail = error.detail.get("detail")
if isinstance(detail, str):
detail_message = detail
elif isinstance(detail, dict):
if isinstance(detail.get("message"), str):
detail_message = detail["message"]
elif isinstance(detail.get("detail"), str):
detail_message = detail["detail"]
if error.status_code and detail_message:
return f"HTTP {error.status_code}: {detail_message}"
if error.status_code:
return f"HTTP {error.status_code}"
return str(error)
async def _fetch_cloud_project_index_status(project_name: str) -> CloudProjectIndexStatus:
"""Fetch cloud index freshness for one project from the admin tenant endpoint."""
workspace_id = await _resolve_cloud_status_workspace_id_async(project_name)
host_url = ConfigManager().config.cloud_host.rstrip("/")
try:
response = await make_api_request(
method="GET",
url=f"{host_url}/admin/tenants/{workspace_id}/index-status",
)
except typer.Exit as exc:
if exc.exit_code not in (None, 0):
raise RuntimeError(
"Cloud credentials not found. Run `bm cloud api-key save <key>` or "
"`bm cloud login` first."
) from exc
raise
tenant_status = CloudTenantIndexStatusResponse.model_validate(response.json())
if tenant_status.error:
raise RuntimeError(tenant_status.error)
project_status = _match_cloud_index_status_project(project_name, tenant_status.projects)
if project_status is None:
raise RuntimeError(
f"Project '{project_name}' was not found in workspace index status "
f"for tenant '{workspace_id}'."
)
return project_status
def _load_cloud_project_index_status(
project_name: str,
) -> tuple[CloudProjectIndexStatus | None, str | None]:
"""Best-effort wrapper around the cloud index freshness lookup."""
try:
return run_with_cleanup(_fetch_cloud_project_index_status(project_name)), None
except Exception as exc:
return None, _format_cloud_index_status_error(exc)
def _build_cloud_index_status_section(
cloud_index_status: CloudProjectIndexStatus | None,
cloud_index_status_error: str | None,
) -> Table | None:
"""Render the optional Cloud Index Status block for rich project info."""
if cloud_index_status is None and cloud_index_status_error is None:
return None
table = Table.grid(padding=(0, 2))
table.add_column("property", style="cyan")
table.add_column("value", style="green")
table.add_row("[bold]Cloud Index Status[/bold]", "")
if cloud_index_status_error is not None:
table.add_row("[yellow]●[/yellow] Warning", f"[yellow]{cloud_index_status_error}[/yellow]")
return table
if cloud_index_status is None:
return table
table.add_row("Files", str(cloud_index_status.current_file_count))
table.add_row(
"Note content",
f"{cloud_index_status.note_content_synced}/{cloud_index_status.current_file_count}",
)
table.add_row(
"Search",
f"{cloud_index_status.total_indexed_entities}/{cloud_index_status.current_file_count}",
)
table.add_row("Embeddable", str(cloud_index_status.embeddable_indexed_entities))
table.add_row(
"Vectorized",
(
f"{cloud_index_status.total_entities_with_chunks}/"
f"{cloud_index_status.embeddable_indexed_entities}"
),
)
if cloud_index_status.reindex_recommended:
table.add_row("[yellow]●[/yellow] Status", "[yellow]Reindex recommended[/yellow]")
if cloud_index_status.reindex_reason:
table.add_row("Reason", f"[yellow]{cloud_index_status.reindex_reason}[/yellow]")
else:
table.add_row("[green]●[/green] Status", "[green]Up to date[/green]")
return table
def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:
"""Normalize CLI visibility input to the cloud API contract."""
if visibility is None:
@@ -856,9 +1066,20 @@ def display_project_info(
with force_routing(local=local, cloud=cloud):
info = run_with_cleanup(get_project_info(name))
cloud_index_status: CloudProjectIndexStatus | None = None
cloud_index_status_error: str | None = None
if _uses_cloud_project_info_route(info.project_name, local=local, cloud=cloud):
cloud_index_status, cloud_index_status_error = _load_cloud_project_index_status(
info.project_name
)
if json_output:
# Convert to JSON and print
print(json.dumps(info.model_dump(), indent=2, default=str))
output = info.model_dump()
output["cloud_index_status"] = (
cloud_index_status.model_dump() if cloud_index_status is not None else None
)
output["cloud_index_status_error"] = cloud_index_status_error
print(json.dumps(output, indent=2, default=str))
else:
# --- Left column: Knowledge Graph stats ---
left = Table.grid(padding=(0, 2))
@@ -916,6 +1137,10 @@ def display_project_info(
columns = Table.grid(padding=(0, 4), expand=False)
columns.add_row(left, right)
cloud_section = _build_cloud_index_status_section(
cloud_index_status, cloud_index_status_error
)
# --- Note Types bar chart (top 5 by count) ---
bars_section = None
if info.statistics.note_types:
@@ -954,6 +1179,8 @@ def display_project_info(
# --- Assemble dashboard ---
parts: list = [columns, ""]
if cloud_section is not None:
parts.extend([cloud_section, ""])
if bars_section:
parts.extend([bars_section, ""])
parts.append(footer)
src/basic_memory/repository/search_repository.py
+4
View File
@@ -70,6 +70,10 @@ class SearchRepository(Protocol):
"""Sync semantic vector chunks for an entity."""
...
async def delete_entity_vector_rows(self, entity_id: int) -> None:
"""Delete semantic vector chunks and embeddings for one entity."""
...
async def sync_entity_vectors_batch(
self,
entity_ids: list[int],
src/basic_memory/repository/search_repository_base.py
+9
View File
@@ -454,6 +454,15 @@ class SearchRepositoryBase(ABC):
logger.debug(f"Query executed successfully in {elapsed_time:.2f}s.")
return result
async def delete_entity_vector_rows(self, entity_id: int) -> None:
"""Delete one entity's derived vector rows using the backend's cleanup path."""
await self._ensure_vector_tables()
async with db.scoped_session(self.session_maker) as session:
await self._prepare_vector_session(session)
await self._delete_entity_chunks(session, entity_id)
await session.commit()
# ------------------------------------------------------------------
# Shared semantic search: guard, text processing, chunking
# ------------------------------------------------------------------
src/basic_memory/repository/sqlite_search_repository.py
-15
View File
@@ -565,21 +565,6 @@ class SQLiteSearchRepository(SearchRepositoryBase):
stale_params,
)
async def delete_entity_vector_rows(self, entity_id: int) -> None:
"""Delete one entity's vec rows on a sqlite-vec-enabled connection."""
await self._ensure_vector_tables()
async with db.scoped_session(self.session_maker) as session:
await self._ensure_sqlite_vec_loaded(session)
# Constraint: sqlite-vec virtual tables are only visible after vec0 is
# loaded on this exact connection.
# Why: generic repository sessions can reach search_vector_chunks but still
# fail with "no such module: vec0" when touching embeddings.
# Outcome: service-level cleanup routes vec-table deletes through this helper.
await self._delete_entity_chunks(session, entity_id)
await session.commit()
async def delete_project_vector_rows(self) -> None:
"""Delete all vector rows for this project on a sqlite-vec-enabled connection."""
await self._ensure_vector_tables()
src/basic_memory/schemas/cloud.py
+50
View File
@@ -81,3 +81,53 @@ class WorkspaceListResponse(BaseModel):
current_workspace_id: str | None = Field(
default=None, description="Current workspace tenant ID when available"
)
class CloudProjectIndexStatus(BaseModel):
"""Index freshness summary for one cloud project."""
project_name: str = Field(..., description="Project name")
project_id: int = Field(..., description="Project database identifier")
last_scan_timestamp: float | None = Field(
default=None, description="Last scan timestamp from project metadata"
)
last_file_count: int | None = Field(default=None, description="Last observed file count")
current_file_count: int = Field(..., description="Current markdown file count")
total_entities: int = Field(..., description="Current markdown entity count")
total_note_content_rows: int = Field(..., description="Rows present in note_content")
note_content_synced: int = Field(..., description="Files fully materialized into note_content")
note_content_pending: int = Field(..., description="Pending note_content rows")
note_content_failed: int = Field(..., description="Failed note_content rows")
note_content_external_changes: int = Field(
..., description="Rows flagged with external file changes"
)
total_indexed_entities: int = Field(..., description="Files represented in search_index")
embedding_opt_out_entities: int = Field(..., description="Files opted out of vector embeddings")
embeddable_indexed_entities: int = Field(
..., description="Indexed files eligible for vector embeddings"
)
total_entities_with_chunks: int = Field(..., description="Embeddable files with vector chunks")
total_chunks: int = Field(..., description="Vector chunk row count")
total_embeddings: int = Field(..., description="Vector embedding row count")
orphaned_chunks: int = Field(..., description="Chunks missing embeddings")
vector_tables_exist: bool = Field(..., description="Whether vector tables exist")
materialization_current: bool = Field(
..., description="Whether note content matches the current file set"
)
search_current: bool = Field(..., description="Whether search coverage is current")
embeddings_current: bool = Field(..., description="Whether embedding coverage is current")
project_current: bool = Field(..., description="Whether all freshness checks are current")
reindex_recommended: bool = Field(..., description="Whether a reindex is recommended")
reindex_reason: str | None = Field(default=None, description="Reason a reindex is recommended")
class CloudTenantIndexStatusResponse(BaseModel):
"""Index freshness summary for all projects in one cloud tenant."""
tenant_id: str = Field(..., description="Workspace tenant identifier")
fly_app_name: str = Field(..., description="Cloud tenant application identifier")
email: str | None = Field(default=None, description="Owner email when available")
projects: list[CloudProjectIndexStatus] = Field(
default_factory=list, description="Per-project freshness summaries"
)
error: str | None = Field(default=None, description="Tenant-level lookup error")
src/basic_memory/services/search_service.py
+7 -13
View File
@@ -660,7 +660,6 @@ class SearchService:
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
from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
# Trigger: semantic indexing is disabled for this repository instance.
# Why: repositories only create vector tables when semantic search is enabled.
@@ -671,17 +670,7 @@ class SearchService:
):
return
params = {"project_id": self.repository.project_id, "entity_id": entity_id}
if isinstance(self.repository, SQLiteSearchRepository):
await self.repository.delete_entity_vector_rows(entity_id)
else:
await self.repository.execute_query(
text(
"DELETE FROM search_vector_chunks "
"WHERE project_id = :project_id AND entity_id = :entity_id"
),
params,
)
await self.repository.delete_entity_vector_rows(entity_id)
async def index_entity_file(
self,
@@ -889,7 +878,7 @@ class SearchService:
await self.repository.delete_by_entity_id(entity_id)
async def handle_delete(self, entity: Entity):
"""Handle complete entity deletion from search index including observations and relations.
"""Handle complete entity deletion from search and semantic index state.
This replicates the logic from sync_service.handle_delete() to properly clean up
all search index entries for an entity and its related data.
@@ -916,3 +905,8 @@ class SearchService:
await self.delete_by_permalink(permalink)
else:
await self.delete_by_entity_id(entity.id)
# Trigger: entity deletion removes the source rows for this note.
# Why: semantic chunks/embeddings are stored separately from search_index rows.
# Outcome: deleting an entity clears both full-text and vector-derived search state.
await self._clear_entity_vectors(entity.id)
tests/cli/cloud/test_project_sync_command.py
+3 -25
View File
@@ -1,7 +1,6 @@
"""Tests for cloud sync and bisync command behavior."""
import importlib
from contextlib import asynccontextmanager
from types import SimpleNamespace
import pytest
@@ -20,11 +19,10 @@ runner = CliRunner()
["cloud", "bisync", "--name", "research"],
],
)
def test_cloud_sync_commands_use_incremental_db_sync(monkeypatch, argv, config_manager):
"""Cloud sync commands should not force a full database re-index after file sync."""
def test_cloud_sync_commands_skip_explicit_cloud_project_sync(monkeypatch, argv, config_manager):
"""Cloud sync commands should not trigger an extra explicit cloud project sync."""
project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
seen: dict[str, object] = {}
config = config_manager.load_config()
config.set_project_mode("research", ProjectMode.CLOUD)
config_manager.save_config(config)
@@ -50,30 +48,10 @@ def test_cloud_sync_commands_use_incremental_db_sync(monkeypatch, argv, config_m
monkeypatch.setattr(project_sync_command, "project_sync", lambda *args, **kwargs: True)
monkeypatch.setattr(project_sync_command, "project_bisync", lambda *args, **kwargs: True)
@asynccontextmanager
async def fake_get_client(*, project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
yield object()
class FakeProjectClient:
def __init__(self, _client):
pass
async def sync(self, external_id: str, force_full: bool = False):
seen["external_id"] = external_id
seen["force_full"] = force_full
return {"message": "queued"}
monkeypatch.setattr(project_sync_command, "get_client", fake_get_client)
monkeypatch.setattr(project_sync_command, "ProjectClient", FakeProjectClient)
result = runner.invoke(app, argv)
assert result.exit_code == 0, result.output
assert seen["project_name"] == "research"
assert seen["external_id"] == "external-project-id"
assert seen["force_full"] is False
assert "Database sync initiated" not in result.output
def test_cloud_bisync_fails_fast_when_sync_entry_disappears(monkeypatch, config_manager):
tests/cli/test_project_info_cloud_status.py
+652
View File
@@ -0,0 +1,652 @@
"""Tests for cloud index status in `bm project info`."""
import json
from datetime import datetime
from pathlib import Path
import httpx
import pytest
import typer
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.commands.cloud.api_client import CloudAPIError
from basic_memory.schemas.cloud import CloudProjectIndexStatus, WorkspaceInfo
from basic_memory.schemas.project_info import (
ActivityMetrics,
EmbeddingStatus,
ProjectInfoResponse,
ProjectStatistics,
SystemStatus,
)
# Importing registers project subcommands on the shared app instance.
import basic_memory.cli.commands.project as project_cmd # noqa: F401
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
@pytest.fixture
def write_config(tmp_path, monkeypatch):
"""Write config.json under a temporary HOME and return the file path."""
from basic_memory import config as config_module
def _write(config_data: dict) -> Path:
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
config_file = config_dir / "config.json"
config_file.write_text(json.dumps(config_data, indent=2), encoding="utf-8")
monkeypatch.setenv("HOME", str(tmp_path))
return config_file
return _write
def _project_info(project_name: str = "demo") -> ProjectInfoResponse:
return ProjectInfoResponse(
project_name=project_name,
project_path=f"/tmp/{project_name}",
available_projects={
project_name: {
"path": f"/tmp/{project_name}",
"active": True,
"id": 1,
"is_default": True,
"permalink": project_name,
}
},
default_project=project_name,
statistics=ProjectStatistics(
total_entities=10,
total_observations=20,
total_relations=5,
total_unresolved_relations=1,
note_types={"note": 10},
observation_categories={"fact": 20},
relation_types={"relates_to": 5},
most_connected_entities=[],
isolated_entities=2,
),
activity=ActivityMetrics(
recently_created=[],
recently_updated=[],
monthly_growth={},
),
system=SystemStatus(
version="0.0.0-test",
database_path="/tmp/memory.db",
database_size="1.00 MB",
watch_status=None,
timestamp=datetime(2026, 4, 9, 12, 0, 0),
),
embedding_status=EmbeddingStatus(
semantic_search_enabled=True,
embedding_provider="fastembed",
embedding_model="bge-small-en-v1.5",
total_indexed_entities=10,
total_entities_with_chunks=10,
total_chunks=30,
total_embeddings=30,
vector_tables_exist=True,
reindex_recommended=False,
reindex_reason=None,
),
)
def _cloud_index_status(
*,
project_name: str = "demo",
reindex_recommended: bool = False,
reindex_reason: str | None = None,
) -> CloudProjectIndexStatus:
return CloudProjectIndexStatus(
project_name=project_name,
project_id=1,
last_scan_timestamp=1234.5,
last_file_count=12,
current_file_count=12,
total_entities=12,
total_note_content_rows=12,
note_content_synced=11,
note_content_pending=1,
note_content_failed=0,
note_content_external_changes=0,
total_indexed_entities=10,
embedding_opt_out_entities=2,
embeddable_indexed_entities=8,
total_entities_with_chunks=7,
total_chunks=21,
total_embeddings=21,
orphaned_chunks=0,
vector_tables_exist=True,
materialization_current=False,
search_current=False,
embeddings_current=False,
project_current=not reindex_recommended,
reindex_recommended=reindex_recommended,
reindex_reason=reindex_reason,
)
def test_project_info_local_output_is_unchanged(runner: CliRunner, write_config, monkeypatch):
"""Local project info should not attempt cloud augmentation."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "local"}},
"default_project": "demo",
}
)
async def fake_get_project_info(_project_name: str) -> ProjectInfoResponse:
return _project_info()
def fail_if_called(_project_name: str):
raise AssertionError("cloud index status should not be fetched for local projects")
monkeypatch.setattr(project_cmd, "get_project_info", fake_get_project_info)
monkeypatch.setattr(project_cmd, "_load_cloud_project_index_status", fail_if_called)
result = runner.invoke(app, ["project", "info", "demo"], env={"COLUMNS": "240"})
assert result.exit_code == 0
assert "Knowledge Graph" in result.stdout
assert "Cloud Index Status" not in result.stdout
def test_project_info_cloud_output_includes_index_status(
runner: CliRunner, write_config, monkeypatch
):
"""Cloud project info should render the extra Cloud Index Status block."""
write_config(
{
"env": "dev",
"projects": {
"demo": {
"path": "/tmp/demo",
"mode": "cloud",
"workspace_id": "11111111-1111-1111-1111-111111111111",
}
},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
async def fake_get_project_info(_project_name: str) -> ProjectInfoResponse:
return _project_info()
def fake_load_cloud_project_index_status(_project_name: str):
return (
_cloud_index_status(
reindex_recommended=True,
reindex_reason="Search index coverage does not match the current file count",
),
None,
)
monkeypatch.setattr(project_cmd, "get_project_info", fake_get_project_info)
monkeypatch.setattr(
project_cmd, "_load_cloud_project_index_status", fake_load_cloud_project_index_status
)
result = runner.invoke(app, ["project", "info", "demo"], env={"COLUMNS": "240"})
assert result.exit_code == 0
assert "Cloud Index Status" in result.stdout
assert "Files" in result.stdout
assert "12" in result.stdout
assert "Note content" in result.stdout
assert "11/12" in result.stdout
assert "Search" in result.stdout
assert "10/12" in result.stdout
assert "Embeddable" in result.stdout
assert "8" in result.stdout
assert "Vectorized" in result.stdout
assert "7/8" in result.stdout
assert "Reindex recommended" in result.stdout
assert "Search index coverage does not match the current file count" in result.stdout
def test_project_info_cloud_output_warns_when_index_lookup_fails(
runner: CliRunner, write_config, monkeypatch
):
"""Cloud project info should keep rendering when the admin lookup fails."""
write_config(
{
"env": "dev",
"projects": {
"demo": {
"path": "/tmp/demo",
"mode": "cloud",
"workspace_id": "11111111-1111-1111-1111-111111111111",
}
},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
async def fake_get_project_info(_project_name: str) -> ProjectInfoResponse:
return _project_info()
def fake_load_cloud_project_index_status(_project_name: str):
return None, "HTTP 503: index-status endpoint unavailable"
monkeypatch.setattr(project_cmd, "get_project_info", fake_get_project_info)
monkeypatch.setattr(
project_cmd, "_load_cloud_project_index_status", fake_load_cloud_project_index_status
)
result = runner.invoke(app, ["project", "info", "demo"], env={"COLUMNS": "240"})
assert result.exit_code == 0
assert "Knowledge Graph" in result.stdout
assert "Cloud Index Status" in result.stdout
assert "Warning" in result.stdout
assert "HTTP 503: index-status endpoint unavailable" in result.stdout
def test_project_info_json_includes_cloud_index_status(
runner: CliRunner, write_config, monkeypatch
):
"""JSON output should include the matched cloud index status block."""
write_config(
{
"env": "dev",
"projects": {
"demo": {
"path": "/tmp/demo",
"mode": "cloud",
"workspace_id": "11111111-1111-1111-1111-111111111111",
}
},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
async def fake_get_project_info(_project_name: str) -> ProjectInfoResponse:
return _project_info()
def fake_load_cloud_project_index_status(_project_name: str):
return _cloud_index_status(), None
monkeypatch.setattr(project_cmd, "get_project_info", fake_get_project_info)
monkeypatch.setattr(
project_cmd, "_load_cloud_project_index_status", fake_load_cloud_project_index_status
)
result = runner.invoke(app, ["project", "info", "demo", "--json"])
assert result.exit_code == 0
data = json.loads(result.stdout)
assert data["project_name"] == "demo"
assert data["cloud_index_status"]["project_name"] == "demo"
assert data["cloud_index_status"]["current_file_count"] == 12
assert data["cloud_index_status"]["note_content_synced"] == 11
assert data["cloud_index_status"]["embeddable_indexed_entities"] == 8
assert data["cloud_index_status"]["total_entities_with_chunks"] == 7
assert data["cloud_index_status_error"] is None
def test_project_info_json_includes_cloud_error_when_lookup_fails(
runner: CliRunner, write_config, monkeypatch
):
"""JSON output should preserve project info when the cloud status lookup fails."""
write_config(
{
"env": "dev",
"projects": {
"demo": {
"path": "/tmp/demo",
"mode": "cloud",
"workspace_id": "11111111-1111-1111-1111-111111111111",
}
},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
async def fake_get_project_info(_project_name: str) -> ProjectInfoResponse:
return _project_info()
def fake_load_cloud_project_index_status(_project_name: str):
return None, "HTTP 503: index-status endpoint unavailable"
monkeypatch.setattr(project_cmd, "get_project_info", fake_get_project_info)
monkeypatch.setattr(
project_cmd, "_load_cloud_project_index_status", fake_load_cloud_project_index_status
)
result = runner.invoke(app, ["project", "info", "demo", "--json"])
assert result.exit_code == 0
data = json.loads(result.stdout)
assert data["project_name"] == "demo"
assert data["cloud_index_status"] is None
assert data["cloud_index_status_error"] == "HTTP 503: index-status endpoint unavailable"
def test_uses_cloud_project_info_route_respects_flags_and_project_mode(write_config, monkeypatch):
"""Route detection should stay local unless flags or cloud mode require augmentation."""
write_config(
{
"env": "dev",
"projects": {
"local-demo": {"path": "/tmp/local-demo", "mode": "local"},
"cloud-demo": {"path": "/tmp/cloud-demo", "mode": "cloud"},
},
"default_project": "local-demo",
"cloud_api_key": "bmc_test_key_123",
}
)
assert (
project_cmd._uses_cloud_project_info_route("cloud-demo", local=False, cloud=False) is True
)
assert (
project_cmd._uses_cloud_project_info_route("local-demo", local=False, cloud=False) is False
)
assert project_cmd._uses_cloud_project_info_route("local-demo", local=False, cloud=True) is True
assert (
project_cmd._uses_cloud_project_info_route("cloud-demo", local=True, cloud=False) is False
)
def test_resolve_cloud_status_workspace_id_prefers_project_workspace(write_config):
"""Cloud status lookup should use the project workspace before any fallback lookup."""
write_config(
{
"env": "dev",
"projects": {
"demo": {
"path": "/tmp/demo",
"mode": "cloud",
"workspace_id": "project-workspace",
}
},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
"default_workspace": "default-workspace",
}
)
assert project_cmd._resolve_cloud_status_workspace_id("demo") == "project-workspace"
def test_resolve_cloud_status_workspace_id_uses_fallback_resolution(write_config, monkeypatch):
"""Cloud status lookup should fall back to workspace discovery when config has no workspace."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
monkeypatch.setattr(
project_cmd, "_resolve_workspace_id", lambda _config, _workspace: "resolved"
)
assert project_cmd._resolve_cloud_status_workspace_id("demo") == "resolved"
def test_resolve_cloud_status_workspace_id_requires_credentials(write_config):
"""Cloud status lookup should fail fast when no credentials are available."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
}
)
with pytest.raises(RuntimeError, match="Cloud credentials not found"):
project_cmd._resolve_cloud_status_workspace_id("demo")
@pytest.mark.asyncio
async def test_resolve_cloud_status_workspace_id_async_auto_discovers_single_workspace(
write_config, monkeypatch
):
"""Async cloud status lookup should auto-select a single available workspace."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
}
)
async def fake_get_available_workspaces():
return [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
)
]
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fake_get_available_workspaces,
)
workspace_id = await project_cmd._resolve_cloud_status_workspace_id_async("demo")
assert workspace_id == "11111111-1111-1111-1111-111111111111"
def test_match_cloud_index_status_project_prefers_exact_then_permalink():
"""Project matching should use exact names first, then a unique permalink match."""
exact = _cloud_index_status(project_name="Demo Project")
permalink_match = _cloud_index_status(project_name="demo-project")
unrelated = _cloud_index_status(project_name="other")
assert (
project_cmd._match_cloud_index_status_project("Demo Project", [exact, unrelated]) is exact
)
assert (
project_cmd._match_cloud_index_status_project("Demo Project", [permalink_match, unrelated])
is permalink_match
)
assert (
project_cmd._match_cloud_index_status_project(
"Demo Project",
[permalink_match, _cloud_index_status(project_name="Demo Project!!")],
)
is None
)
def test_format_cloud_index_status_error_prefers_cloud_api_detail():
"""Cloud API errors should surface the most useful available detail."""
assert project_cmd._format_cloud_index_status_error(RuntimeError("boom")) == "boom"
assert (
project_cmd._format_cloud_index_status_error(
CloudAPIError("fail", status_code=503, detail={"detail": "down"})
)
== "HTTP 503: down"
)
assert (
project_cmd._format_cloud_index_status_error(
CloudAPIError("fail", status_code=503, detail={"detail": {"message": "nested"}})
)
== "HTTP 503: nested"
)
assert (
project_cmd._format_cloud_index_status_error(
CloudAPIError("fail", status_code=503, detail={"detail": {"detail": "nested-detail"}})
)
== "HTTP 503: nested-detail"
)
assert (
project_cmd._format_cloud_index_status_error(CloudAPIError("fail", status_code=503))
== "HTTP 503"
)
@pytest.mark.asyncio
async def test_fetch_cloud_project_index_status_returns_matching_project(write_config, monkeypatch):
"""Cloud index status fetch should validate the tenant payload and return the matched project."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
"cloud_host": "https://cloud.example.test",
}
)
async def fake_resolve_workspace(_project_name: str) -> str:
return "11111111-1111-1111-1111-111111111111"
monkeypatch.setattr(
project_cmd, "_resolve_cloud_status_workspace_id_async", fake_resolve_workspace
)
async def fake_make_api_request(**kwargs):
assert kwargs["method"] == "GET"
assert (
kwargs["url"]
== "https://cloud.example.test/admin/tenants/11111111-1111-1111-1111-111111111111/index-status"
)
return httpx.Response(
200,
json={
"tenant_id": "11111111-1111-1111-1111-111111111111",
"fly_app_name": "demo-app",
"email": "demo@example.com",
"projects": [_cloud_index_status().model_dump()],
"error": None,
},
)
monkeypatch.setattr(project_cmd, "make_api_request", fake_make_api_request)
status = await project_cmd._fetch_cloud_project_index_status("demo")
assert status.project_name == "demo"
assert status.current_file_count == 12
@pytest.mark.asyncio
async def test_fetch_cloud_project_index_status_handles_exit_and_missing_project(
write_config, monkeypatch
):
"""Cloud fetch should convert auth exits and fail when the project is missing."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
"cloud_host": "https://cloud.example.test",
}
)
async def fake_resolve_workspace(_project_name: str) -> str:
return "tenant-1"
monkeypatch.setattr(
project_cmd, "_resolve_cloud_status_workspace_id_async", fake_resolve_workspace
)
async def fake_make_api_request_exit(**_kwargs):
raise typer.Exit(1)
monkeypatch.setattr(project_cmd, "make_api_request", fake_make_api_request_exit)
with pytest.raises(RuntimeError, match="Cloud credentials not found"):
await project_cmd._fetch_cloud_project_index_status("demo")
async def fake_make_api_request_missing(**_kwargs):
return httpx.Response(
200,
json={
"tenant_id": "tenant-1",
"fly_app_name": "demo-app",
"email": "demo@example.com",
"projects": [_cloud_index_status(project_name="other").model_dump()],
"error": None,
},
)
monkeypatch.setattr(project_cmd, "make_api_request", fake_make_api_request_missing)
with pytest.raises(RuntimeError, match="was not found in workspace index status"):
await project_cmd._fetch_cloud_project_index_status("demo")
@pytest.mark.asyncio
async def test_fetch_cloud_project_index_status_preserves_successful_exit_and_tenant_error(
write_config, monkeypatch
):
"""Only non-zero typer exits should be converted; tenant-level errors should bubble clearly."""
write_config(
{
"env": "dev",
"projects": {"demo": {"path": "/tmp/demo", "mode": "cloud"}},
"default_project": "demo",
"cloud_api_key": "bmc_test_key_123",
"cloud_host": "https://cloud.example.test",
}
)
async def fake_resolve_workspace(_project_name: str) -> str:
return "tenant-1"
monkeypatch.setattr(
project_cmd, "_resolve_cloud_status_workspace_id_async", fake_resolve_workspace
)
async def fake_make_api_request_success_exit(**_kwargs):
raise typer.Exit(0)
monkeypatch.setattr(project_cmd, "make_api_request", fake_make_api_request_success_exit)
with pytest.raises(typer.Exit) as exc_info:
await project_cmd._fetch_cloud_project_index_status("demo")
assert exc_info.value.exit_code == 0
async def fake_make_api_request_tenant_error(**_kwargs):
return httpx.Response(
200,
json={
"tenant_id": "tenant-1",
"fly_app_name": "demo-app",
"email": "demo@example.com",
"projects": [],
"error": "tenant is unavailable",
},
)
monkeypatch.setattr(project_cmd, "make_api_request", fake_make_api_request_tenant_error)
with pytest.raises(RuntimeError, match="tenant is unavailable"):
await project_cmd._fetch_cloud_project_index_status("demo")
def test_build_cloud_index_status_section_handles_missing_status():
"""The renderer should return a safe header-only table if invariants are broken."""
table = project_cmd._build_cloud_index_status_section(None, None)
assert table is None
warning_table = project_cmd._build_cloud_index_status_section(
None, "HTTP 503: index-status endpoint unavailable"
)
assert warning_table is not None
tests/services/test_entity_service.py
+150 -1
View File
@@ -6,8 +6,10 @@ from textwrap import dedent
import pytest
import yaml
from sqlalchemy import text
from basic_memory.config import ProjectConfig, BasicMemoryConfig
from basic_memory import db
from basic_memory.config import ProjectConfig, BasicMemoryConfig, DatabaseBackend
from basic_memory.markdown import EntityParser
from basic_memory.models import Entity as EntityModel
from basic_memory.repository import EntityRepository
@@ -19,6 +21,98 @@ from basic_memory.services.search_service import SearchService
from basic_memory.utils import generate_permalink
class _DeleteTestEmbeddingProvider:
"""Deterministic embedding provider for entity delete cleanup tests."""
model_name = "delete-test"
dimensions = 4
async def embed_query(self, text: str) -> list[float]:
return self._vectorize(text)
async def embed_documents(self, texts: list[str]) -> list[list[float]]:
return [self._vectorize(text) for text in texts]
@staticmethod
def _vectorize(text: str) -> list[float]:
normalized = text.lower()
if "semantic" in normalized:
return [1.0, 0.0, 0.0, 0.0]
if "cleanup" in normalized:
return [0.0, 1.0, 0.0, 0.0]
return [0.0, 0.0, 1.0, 0.0]
async def _count_entity_search_state(
session_maker,
app_config: BasicMemoryConfig,
project_id: int,
entity_id: int,
) -> tuple[int, int, int]:
"""Return counts for all derived search rows tied to one entity."""
embedding_join = (
"e.chunk_id = c.id"
if app_config.database_backend == DatabaseBackend.POSTGRES
else "e.rowid = c.id"
)
params = {"project_id": project_id, "entity_id": entity_id}
async with db.scoped_session(session_maker) as session:
search_index_rows = await session.execute(
text(
"SELECT COUNT(*) FROM search_index "
"WHERE project_id = :project_id AND entity_id = :entity_id"
),
params,
)
vector_chunk_rows = await session.execute(
text(
"SELECT COUNT(*) FROM search_vector_chunks "
"WHERE project_id = :project_id AND entity_id = :entity_id"
),
params,
)
vector_embedding_rows = await session.execute(
text(
"SELECT COUNT(*) FROM search_vector_embeddings e "
"JOIN search_vector_chunks c ON "
f"{embedding_join} "
"WHERE c.project_id = :project_id AND c.entity_id = :entity_id"
),
params,
)
return (
int(search_index_rows.scalar_one()),
int(vector_chunk_rows.scalar_one()),
int(vector_embedding_rows.scalar_one()),
)
@pytest.fixture
def entity_service_with_search(
entity_repository: EntityRepository,
observation_repository,
relation_repository,
entity_parser: EntityParser,
file_service: FileService,
link_resolver,
search_service: SearchService,
app_config: BasicMemoryConfig,
) -> EntityService:
"""Create EntityService with a real attached search service."""
return EntityService(
entity_parser=entity_parser,
entity_repository=entity_repository,
observation_repository=observation_repository,
relation_repository=relation_repository,
file_service=file_service,
link_resolver=link_resolver,
search_service=search_service,
app_config=app_config,
)
@pytest.mark.asyncio
async def test_create_entity(
entity_service: EntityService, file_service: FileService, project_config: ProjectConfig
@@ -227,6 +321,61 @@ async def test_delete_entity_by_id(entity_service: EntityService):
await entity_service.get_by_permalink(entity_data.permalink)
@pytest.mark.asyncio
async def test_delete_entity_removes_search_and_vector_state(
entity_service_with_search: EntityService,
search_service: SearchService,
session_maker,
app_config: BasicMemoryConfig,
):
"""Deleting an entity should clear all of its full-text and semantic search state."""
if app_config.database_backend == DatabaseBackend.SQLITE:
pytest.importorskip("sqlite_vec")
repository = search_service.repository
repository._semantic_enabled = True
repository._embedding_provider = _DeleteTestEmbeddingProvider()
repository._vector_dimensions = repository._embedding_provider.dimensions
repository._vector_tables_initialized = False
await search_service.init_search_index()
entity = await entity_service_with_search.create_entity(
EntitySchema(
title="Semantic Delete Target",
directory="test",
note_type="note",
content=dedent("""
# Semantic Delete Target
- [note] Semantic cleanup should remove every derived row
- references [[Cleanup Target]]
""").strip(),
)
)
await search_service.index_entity(entity)
await search_service.sync_entity_vectors(entity.id)
search_rows, chunk_rows, embedding_rows = await _count_entity_search_state(
session_maker,
app_config,
search_service.repository.project_id,
entity.id,
)
assert search_rows >= 3
assert chunk_rows > 0
assert embedding_rows > 0
assert await entity_service_with_search.delete_entity(entity.id) is True
assert await _count_entity_search_state(
session_maker,
app_config,
search_service.repository.project_id,
entity.id,
) == (0, 0, 0)
@pytest.mark.asyncio
async def test_get_entity_by_permalink_not_found(entity_service: EntityService):
"""Test handling of non-existent entity retrieval."""
tests/services/test_initialization.py
-1
View File
@@ -196,4 +196,3 @@ async def test_initialize_app_no_precedence_warning_when_not_conflicting(
"ensure_frontmatter_on_sync=True overrides disable_permalinks=True" in message
for message in warnings
)