feat: min_similarity override, cloud promo improvements (#570)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

src/basic_memory/api/v2/routers/schema_router.py

@@ -8,12 +8,11 @@ observations and relations.
 Flow: Entity loaded with eager observations/relations -> convert to tuples -> core functions.
 """
 
+from pathlib import Path as FilePath
+
 from fastapi import APIRouter, Path, Query
 
-from basic_memory.deps import (
-    SearchServiceV2ExternalDep,
-    EntityRepositoryV2ExternalDep,
-)
+from basic_memory.deps import EntityRepositoryV2ExternalDep
 from basic_memory.models.knowledge import Entity
 from basic_memory.schemas.schema import (
     ValidationReport,
@@ -24,11 +23,11 @@ from basic_memory.schemas.schema import (
     FieldFrequencyResponse,
     DriftFieldResponse,
 )
-from basic_memory.schemas.search import SearchQuery
 from basic_memory.schema.resolver import resolve_schema
 from basic_memory.schema.validator import validate_note
 from basic_memory.schema.inference import infer_schema, NoteData, ObservationData, RelationData
 from basic_memory.schema.diff import diff_schema
+from basic_memory.utils import generate_permalink
 
 # Note: No prefix here -- it's added during registration as /v2/{project_id}/schema
 router = APIRouter(tags=["schema"])
@@ -81,7 +80,6 @@ def _entity_frontmatter(entity: Entity) -> dict:
 @router.post("/schema/validate", response_model=ValidationReport)
 async def validate_schema(
     entity_repository: EntityRepositoryV2ExternalDep,
-    search_service: SearchServiceV2ExternalDep,
     project_id: str = Path(..., description="Project external UUID"),
     entity_type: str | None = Query(None, description="Entity type to validate"),
     identifier: str | None = Query(None, description="Specific note identifier"),
@@ -93,26 +91,24 @@ async def validate_schema(
     """
     results: list[NoteValidationResponse] = []
 
-    async def search_fn(query: str) -> list:
-        # Search for schema notes, then load full entity_metadata from the entity table.
-        # The search index only stores minimal metadata (e.g., {"entity_type": "schema"}),
-        # but parse_schema_note needs the full frontmatter with entity/schema/version keys.
-        results = await search_service.search(SearchQuery(text=query, types=["schema"]), limit=5)
-        frontmatters = []
-        for row in results:
-            if row.permalink:
-                entity = await entity_repository.get_by_permalink(row.permalink)
-                if entity:
-                    frontmatters.append(_entity_frontmatter(entity))
-        return frontmatters
-
     # --- Single note validation ---
     if identifier:
         entity = await entity_repository.get_by_permalink(identifier)
         if not entity:
             return ValidationReport(entity_type=entity_type, total_notes=0, results=[])
 
-        schema_def = await resolve_schema(_entity_frontmatter(entity), search_fn)
+        frontmatter = _entity_frontmatter(entity)
+        schema_ref = frontmatter.get("schema")
+
+        async def search_fn(query: str) -> list[dict]:
+            entities = await _find_schema_entities(
+                entity_repository,
+                query,
+                allow_reference_match=isinstance(schema_ref, str) and query == schema_ref,
+            )
+            return [_entity_frontmatter(e) for e in entities]
+
+        schema_def = await resolve_schema(frontmatter, search_fn)
         if schema_def:
             result = validate_note(
                 entity.permalink or identifier,
@@ -135,7 +131,18 @@ async def validate_schema(
     entities = await _find_by_entity_type(entity_repository, entity_type) if entity_type else []
 
     for entity in entities:
-        schema_def = await resolve_schema(_entity_frontmatter(entity), search_fn)
+        frontmatter = _entity_frontmatter(entity)
+        schema_ref = frontmatter.get("schema")
+
+        async def search_fn(query: str) -> list[dict]:
+            entities = await _find_schema_entities(
+                entity_repository,
+                query,
+                allow_reference_match=isinstance(schema_ref, str) and query == schema_ref,
+            )
+            return [_entity_frontmatter(e) for e in entities]
+
+        schema_def = await resolve_schema(frontmatter, search_fn)
         if schema_def:
             result = validate_note(
                 entity.permalink or entity.file_path,
@@ -149,6 +156,7 @@ async def validate_schema(
     return ValidationReport(
         entity_type=entity_type,
         total_notes=len(results),
+        total_entities=len(entities),
         valid_count=valid,
         warning_count=sum(len(r.warnings) for r in results),
         error_count=sum(len(r.errors) for r in results),
@@ -205,7 +213,6 @@ async def infer_schema_endpoint(
 @router.get("/schema/diff/{entity_type}", response_model=DriftReport)
 async def diff_schema_endpoint(
     entity_repository: EntityRepositoryV2ExternalDep,
-    search_service: SearchServiceV2ExternalDep,
     entity_type: str = Path(..., description="Entity type to check for drift"),
     project_id: str = Path(..., description="Project external UUID"),
 ):
@@ -216,25 +223,16 @@ async def diff_schema_endpoint(
     fields, and cardinality changes.
     """
 
-    async def search_fn(query: str) -> list:
-        # Search for schema notes, then load full entity_metadata from the entity table.
-        # The search index only stores minimal metadata (e.g., {"entity_type": "schema"}),
-        # but parse_schema_note needs the full frontmatter with entity/schema/version keys.
-        results = await search_service.search(SearchQuery(text=query, types=["schema"]), limit=5)
-        frontmatters = []
-        for row in results:
-            if row.permalink:
-                entity = await entity_repository.get_by_permalink(row.permalink)
-                if entity:
-                    frontmatters.append(_entity_frontmatter(entity))
-        return frontmatters
+    async def search_fn(query: str) -> list[dict]:
+        entities = await _find_schema_entities(entity_repository, query)
+        return [_entity_frontmatter(e) for e in entities]
 
     # Resolve schema by entity type
     schema_frontmatter = {"type": entity_type}
     schema_def = await resolve_schema(schema_frontmatter, search_fn)
 
     if not schema_def:
-        return DriftReport(entity_type=entity_type)
+        return DriftReport(entity_type=entity_type, schema_found=False)
 
     # Collect all notes of this type
     entities = await _find_by_entity_type(entity_repository, entity_type)
@@ -281,6 +279,54 @@ async def _find_by_entity_type(
     return list(result.scalars().all())
 
 
+async def _find_schema_entities(
+    entity_repository: EntityRepositoryV2ExternalDep,
+    target_entity_type: str,
+    *,
+    allow_reference_match: bool = False,
+) -> list[Entity]:
+    """Find schema entities for resolver lookups.
+
+    Resolution strategy:
+    1) Always try exact entity_metadata['entity'] match (for implicit type lookup
+       and explicit references that use entity names)
+    2) Only when allow_reference_match=True and no entity match was found, try
+       exact reference matching by title/permalink (explicit schema references)
+    """
+    query = entity_repository.select().where(Entity.entity_type == "schema")
+    result = await entity_repository.execute_query(query)
+    entities = list(result.scalars().all())
+
+    normalized_target = generate_permalink(target_entity_type)
+
+    entity_matches = [
+        e
+        for e in entities
+        if e.entity_metadata
+        and isinstance(e.entity_metadata.get("entity"), str)
+        and generate_permalink(e.entity_metadata["entity"]) == normalized_target
+    ]
+    if entity_matches:
+        return entity_matches
+
+    if not allow_reference_match:
+        return []
+
+    reference_matches: list[Entity] = []
+    for entity in entities:
+        candidate_refs: list[str] = []
+        if entity.title:
+            candidate_refs.append(entity.title)
+        if entity.permalink:
+            candidate_refs.append(entity.permalink)
+            candidate_refs.append(FilePath(entity.permalink).name)
+
+        if any(generate_permalink(ref) == normalized_target for ref in candidate_refs):
+            reference_matches.append(entity)
+
+    return reference_matches
+
+
 def _to_note_validation_response(result) -> NoteValidationResponse:
     """Convert a core ValidationResult to a Pydantic response model."""
     return NoteValidationResponse(

src/basic_memory/cli/app.py

@@ -9,7 +9,7 @@ from typing import Optional  # noqa: E402
 import typer  # noqa: E402
 
 from basic_memory.cli.container import CliContainer, set_container  # noqa: E402
-from basic_memory.cli.promo import maybe_show_cloud_promo  # noqa: E402
+from basic_memory.cli.promo import maybe_show_cloud_promo, maybe_show_init_line  # noqa: E402
 from basic_memory.config import init_cli_logging  # noqa: E402
 
 
@@ -47,7 +47,15 @@ def app_callback(
     container = CliContainer.create()
     set_container(container)
 
-    maybe_show_cloud_promo(ctx.invoked_subcommand)
+    # Trigger: first-run init confirmation before command output.
+    # Why: informational "initialized" message belongs above command results, not in the upsell panel.
+    # Outcome: one-time plain line printed before the subcommand runs.
+    maybe_show_init_line(ctx.invoked_subcommand)
+
+    # Trigger: register promo as a post-command callback.
+    # Why: promo output should appear after the command's own output, not before.
+    # Outcome: promo panel renders below the command results (status tree, table, etc.).
+    ctx.call_on_close(lambda: maybe_show_cloud_promo(ctx.invoked_subcommand))
 
     # Run initialization for commands that don't use the API
     # Skip for 'mcp' command - it has its own lifespan that handles initialization

src/basic_memory/cli/commands/cloud/core_commands.py

@@ -59,8 +59,7 @@ def login():
             console.print("\n[red]Subscription Required[/red]\n")
             console.print(f"[yellow]{e.args[0]}[/yellow]\n")
             console.print(
-                f"OSS discount code: [bold]{OSS_DISCOUNT_CODE}[/bold] "
-                "(20% off for 3 months)\n"
+                f"OSS discount code: [bold]{OSS_DISCOUNT_CODE}[/bold] (20% off for 3 months)\n"
             )
             console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
             console.print(

src/basic_memory/cli/commands/command_utils.py

@@ -9,8 +9,8 @@ import typer
 from rich.console import Console
 
 from basic_memory import db
+from basic_memory.config import ConfigManager
 from basic_memory.mcp.async_client import get_client
-
 from basic_memory.mcp.tools.utils import call_post, call_get
 from basic_memory.mcp.project_context import get_active_project
 from basic_memory.schemas import ProjectInfoResponse
@@ -55,8 +55,11 @@ async def run_sync(
         run_in_background: If True, return immediately; if False, wait for completion
     """
 
+    # Resolve default project so get_client() can route per-project
+    project = project or ConfigManager().default_project
+
     try:
-        async with get_client() as client:
+        async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
             url = f"/v2/projects/{project_item.external_id}/sync"
             params = []
@@ -88,9 +91,8 @@ async def run_sync(
 
 async def get_project_info(project: str):
     """Get project information via API endpoint."""
-
     try:
-        async with get_client() as client:
+        async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
             response = await call_get(client, f"/v2/projects/{project_item.external_id}/info")
             return ProjectInfoResponse.model_validate(response.json())

src/basic_memory/cli/commands/project.py

@@ -13,7 +13,7 @@ from rich.table import Table
 from basic_memory.cli.app import app
 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, ProjectMode
+from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
 from basic_memory.mcp.async_client import get_client
 from basic_memory.mcp.tools.utils import call_delete, call_get, call_patch, call_post, call_put
 from basic_memory.schemas.project_info import ProjectList, ProjectStatusResponse
@@ -79,34 +79,38 @@ def list_projects(
         table.add_column("Path", style="green")
         table.add_column("Mode", style="blue")
 
-        # Add Local Path column if in cloud mode and not forcing local
+        # Add cloud-specific columns when in cloud mode
         if config.cloud_mode_enabled and not local:
             table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
+            table.add_column("Sync", style="green")
 
-        # Show Default column in local mode or if default_project_mode is enabled in cloud mode
-        show_default_column = local or not config.cloud_mode_enabled or config.default_project_mode
-        if show_default_column:
-            table.add_column("Default", style="magenta")
+        table.add_column("Default", style="magenta")
 
         for project in result.projects:
             is_default = "[X]" if project.is_default else ""
             normalized_path = normalize_project_path(project.path)
-            project_mode = config.get_project_mode(project.name).value
+            # Trigger: cloud mode and project not in local config
+            # Why: cloud-discovered projects default to LOCAL in get_project_mode
+            # Outcome: show "cloud" for projects only known to the cloud API
+            entry = config.projects.get(project.name)
+            if config.cloud_mode_enabled and not local and entry is None:
+                project_mode = ProjectMode.CLOUD.value
+            else:
+                project_mode = config.get_project_mode(project.name).value
 
             # Build row based on mode
             row = [project.name, format_path(normalized_path), project_mode]
 
-            # Add local path if in cloud mode and not forcing local
+            # Add cloud-specific columns
             if config.cloud_mode_enabled and not local:
                 local_path = ""
-                if project.name in config.cloud_projects:
-                    local_path = config.cloud_projects[project.name].local_path or ""
-                    local_path = format_path(local_path)
+                if entry:
+                    local_path = format_path(entry.cloud_sync_path or entry.path)
                 row.append(local_path)
+                has_sync = "[X]" if entry and entry.cloud_sync_path else ""
+                row.append(has_sync)
 
-            # Add default indicator if showing default column
-            if show_default_column:
-                row.append(is_default)
+            row.append(is_default)
 
             table.add_row(*row)
 
@@ -194,18 +198,20 @@ def add_project(
 
         # Save local sync path to config if in cloud mode
         if effective_cloud_mode and local_sync_path:
-            from basic_memory.config import CloudProjectConfig
-
             # Create local directory if it doesn't exist
             local_dir = Path(local_sync_path)
             local_dir.mkdir(parents=True, exist_ok=True)
 
-            # Update config with sync path
-            config.cloud_projects[name] = CloudProjectConfig(
-                local_path=local_sync_path,
-                last_sync=None,
-                bisync_initialized=False,
-            )
+            # Update project entry with sync path
+            entry = config.projects.get(name)
+            if entry:
+                entry.cloud_sync_path = local_sync_path
+            else:
+                # Project may not be in local config yet (cloud-only add)
+                config.projects[name] = ProjectEntry(
+                    path=local_sync_path,
+                    cloud_sync_path=local_sync_path,
+                )
             ConfigManager().save_config(config)
 
             console.print(f"\n[green]Local sync path configured: {local_sync_path}[/green]")
@@ -252,14 +258,17 @@ def setup_project_sync(
         resolved_path = Path(os.path.abspath(os.path.expanduser(local_path)))
         resolved_path.mkdir(parents=True, exist_ok=True)
 
-        # Update local config with sync path
-        from basic_memory.config import CloudProjectConfig
-
-        config.cloud_projects[name] = CloudProjectConfig(
-            local_path=resolved_path.as_posix(),
-            last_sync=None,
-            bisync_initialized=False,
-        )
+        # Update project entry with sync path
+        entry = config.projects.get(name)
+        if entry:
+            entry.cloud_sync_path = resolved_path.as_posix()
+            entry.bisync_initialized = False
+            entry.last_sync = None
+        else:
+            config.projects[name] = ProjectEntry(
+                path=resolved_path.as_posix(),
+                cloud_sync_path=resolved_path.as_posix(),
+            )
         config_manager.save_config(config)
 
         console.print(f"[green]Sync configured for project '{name}'[/green]")
@@ -316,8 +325,9 @@ def remove_project(
         local_path_config = None
         has_bisync_state = False
 
-        if config.cloud_mode_enabled and not local and name in config.cloud_projects:
-            local_path_config = config.cloud_projects[name].local_path
+        entry = config.projects.get(name)
+        if config.cloud_mode_enabled and not local and entry and entry.cloud_sync_path:
+            local_path_config = entry.cloud_sync_path
 
             # Check for bisync state
             from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
@@ -349,9 +359,11 @@ def remove_project(
                 shutil.rmtree(bisync_state_path)
                 console.print("[green]Removed bisync state[/green]")
 
-        # Clean up cloud_projects config entry
-        if config.cloud_mode_enabled and not local and name in config.cloud_projects:
-            del config.cloud_projects[name]
+        # Clean up cloud sync fields on the project entry
+        if config.cloud_mode_enabled and not local and entry and entry.cloud_sync_path:
+            entry.cloud_sync_path = None
+            entry.bisync_initialized = False
+            entry.last_sync = None
             ConfigManager().save_config(config)
 
         # Show informative message if files were not deleted
@@ -371,7 +383,7 @@ def set_default_project(
         False, "--local", help="Force local API routing (required in cloud mode)"
     ),
 ) -> None:
-    """Set the default project when 'config.default_project_mode' is set.
+    """Set the default project used as fallback when no project is specified.
 
     In cloud mode, use --local to modify the local configuration.
     """
@@ -618,10 +630,9 @@ def sync_project_command(
             console.print(f"[red]Error: Project '{name}' not found[/red]")
             raise typer.Exit(1)
 
-        # Get local_sync_path from cloud_projects config
-        local_sync_path = None
-        if name in config.cloud_projects:
-            local_sync_path = config.cloud_projects[name].local_path
+        # Get local_sync_path from project entry
+        sync_entry = config.projects.get(name)
+        local_sync_path = sync_entry.cloud_sync_path if sync_entry else None
 
         if not local_sync_path:
             console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
@@ -710,10 +721,9 @@ def bisync_project_command(
             console.print(f"[red]Error: Project '{name}' not found[/red]")
             raise typer.Exit(1)
 
-        # Get local_sync_path from cloud_projects config
-        local_sync_path = None
-        if name in config.cloud_projects:
-            local_sync_path = config.cloud_projects[name].local_path
+        # Get local_sync_path from project entry
+        sync_entry = config.projects.get(name)
+        local_sync_path = sync_entry.cloud_sync_path if sync_entry else None
 
         if not local_sync_path:
             console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
@@ -736,9 +746,11 @@ def bisync_project_command(
         if success:
             console.print(f"[green]{name} bisync completed successfully[/green]")
 
-            # Update config
-            config.cloud_projects[name].last_sync = datetime.now()
-            config.cloud_projects[name].bisync_initialized = True
+            # Update config — sync_entry is guaranteed non-None because
+            # we checked local_sync_path above (which comes from sync_entry)
+            assert sync_entry is not None
+            sync_entry.last_sync = datetime.now()
+            sync_entry.bisync_initialized = True
             ConfigManager().save_config(config)
 
             # Trigger database sync if not a dry run
@@ -805,10 +817,9 @@ def check_project_command(
             console.print(f"[red]Error: Project '{name}' not found[/red]")
             raise typer.Exit(1)
 
-        # Get local_sync_path from cloud_projects config
-        local_sync_path = None
-        if name in config.cloud_projects:
-            local_sync_path = config.cloud_projects[name].local_path
+        # Get local_sync_path from project entry
+        check_entry = config.projects.get(name)
+        local_sync_path = check_entry.cloud_sync_path if check_entry else None
 
         if not local_sync_path:
             console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")

src/basic_memory/cli/commands/routing.py

@@ -9,6 +9,8 @@ behavior (determined by cloud_mode_enabled in config). This allows users to:
 
 The routing is controlled via environment variables:
 - BASIC_MEMORY_FORCE_LOCAL: When "true", forces local ASGI transport
+- BASIC_MEMORY_EXPLICIT_ROUTING: When "true", signals that --local/--cloud
+  was explicitly passed, overriding per-project routing in get_client()
 - These are checked in basic_memory.mcp.async_client.get_client()
 """
 
@@ -24,6 +26,11 @@ def force_routing(local: bool = False, cloud: bool = False) -> Generator[None, N
     Sets environment variables that are checked by get_client() to determine
     whether to use local ASGI transport or cloud proxy transport.
 
+    When either flag is set, BASIC_MEMORY_EXPLICIT_ROUTING is also set so
+    that get_client() skips per-project routing and honors the flag directly.
+    This only affects CLI commands — the MCP server sets FORCE_LOCAL directly
+    (without EXPLICIT_ROUTING), so per-project routing still works for MCP tools.
+
     Args:
         local: If True, force local ASGI transport (ignores cloud_mode_enabled)
         cloud: If True, clear force_local to allow cloud routing
@@ -41,23 +48,30 @@ def force_routing(local: bool = False, cloud: bool = False) -> Generator[None, N
 
     # Save original values
     original_force_local = os.environ.get("BASIC_MEMORY_FORCE_LOCAL")
+    original_explicit = os.environ.get("BASIC_MEMORY_EXPLICIT_ROUTING")
 
     try:
         if local:
-            # Force local routing by setting the env var
             os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true"
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = "true"
         elif cloud:
             # Ensure force_local is NOT set, let cloud_mode_enabled take effect
             os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None)
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = "true"
         # If neither is set, don't change anything (use default behavior)
         yield
     finally:
-        # Restore original value
+        # Restore original values
         if original_force_local is None:
             os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None)
         else:
             os.environ["BASIC_MEMORY_FORCE_LOCAL"] = original_force_local
 
+        if original_explicit is None:
+            os.environ.pop("BASIC_MEMORY_EXPLICIT_ROUTING", None)
+        else:
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = original_explicit
+
 
 def validate_routing_flags(local: bool, cloud: bool) -> None:
     """Validate that --local and --cloud flags are not both specified.

src/basic_memory/cli/commands/schema.py

@@ -10,7 +10,6 @@ from typing import Annotated, Optional
 import typer
 from loguru import logger
 from rich.console import Console
-from rich.panel import Panel
 from rich.table import Table
 
 from basic_memory.cli.app import app
@@ -49,11 +48,11 @@ async def _run_validate(
     """Run schema validation via the API."""
     from basic_memory.mcp.clients.schema import SchemaClient
 
-    async with get_client() as client:
+    async with get_client(project_name=project) as client:
         active_project = await get_active_project(client, project, None)
         schema_client = SchemaClient(client, active_project.external_id)
 
-        # Determine if target is a note identifier or entity type
+        # Determine if target is a note identifier or note type
         # Heuristic: if target contains / or ., treat as identifier
         entity_type = None
         identifier = None
@@ -70,7 +69,13 @@ async def _run_validate(
 
         # --- Display results ---
         if report.total_notes == 0:
-            console.print("[yellow]No notes matched for validation.[/yellow]")
+            if report.total_entities == 0:
+                console.print(f"[yellow]No notes of type '{entity_type}' found.[/yellow]")
+            else:
+                console.print(
+                    f"[yellow]Found {report.total_entities} notes but no schema "
+                    f"defined for '{entity_type}'.[/yellow]"
+                )
             return
 
         table = Table(title=f"Schema Validation: {entity_type or identifier or 'all'}")
@@ -109,7 +114,7 @@ async def _run_validate(
 def validate(
     target: Annotated[
         Optional[str],
-        typer.Argument(help="Note path or entity type to validate"),
+        typer.Argument(help="Note path or note type to validate"),
     ] = None,
     project: Annotated[
         Optional[str],
@@ -123,8 +128,8 @@ def validate(
 ):
     """Validate notes against their schemas.
 
-    TARGET can be a note path (e.g., people/ada-lovelace.md) or an entity type
-    (e.g., Person). If omitted, validates all notes that have schemas.
+    TARGET can be a note path (e.g., people/ada-lovelace.md) or a note type
+    (e.g., person). If omitted, validates all notes that have schemas.
 
     Use --strict to exit with error code 1 if any validation errors are found.
     Use --local to force local routing when cloud mode is enabled.
@@ -158,7 +163,7 @@ async def _run_infer(
     """Run schema inference via the API."""
     from basic_memory.mcp.clients.schema import SchemaClient
 
-    async with get_client() as client:
+    async with get_client(project_name=project) as client:
         active_project = await get_active_project(client, project, None)
         schema_client = SchemaClient(client, active_project.external_id)
 
@@ -168,6 +173,27 @@ async def _run_infer(
             console.print(f"[yellow]No notes found with type: {entity_type}[/yellow]")
             return
 
+        # --- Empty schema guard ---
+        # Trigger: notes were analyzed but no fields met the threshold
+        # Why: dumping hundreds of excluded fields is not useful output
+        # Outcome: show count and suggest a more specific type
+        if not report.suggested_schema:
+            console.print(
+                f"\n[yellow]Analyzed {report.notes_analyzed} notes of type '{entity_type}', "
+                f"but no fields met the {threshold:.0%} threshold.[/yellow]\n"
+            )
+            console.print(
+                f"This usually means '{entity_type}' is too broad — "
+                f"the notes don't share a consistent structure.\n"
+            )
+            console.print("[bold]Suggestions:[/bold]")
+            console.print("  1. Use a more specific type")
+            console.print(
+                f"  2. Lower the threshold: bm schema infer {entity_type} --threshold 0.1"
+            )
+            console.print("  3. Create typed notes with write_note using a specific note_type")
+            return
+
         # --- Display frequency analysis ---
         console.print(
             f"\n[bold]Analyzing {report.notes_analyzed} notes with type: {entity_type}...[/bold]\n"
@@ -201,7 +227,7 @@ async def _run_infer(
 
         # --- Display suggested schema ---
         console.print("\n[bold]Suggested schema:[/bold]")
-        console.print(Panel(json.dumps(report.suggested_schema, indent=2), title="Picoschema"))
+        console.print(json.dumps(report.suggested_schema, indent=2))
 
         if save:
             console.print(
@@ -214,7 +240,7 @@ async def _run_infer(
 def infer(
     entity_type: Annotated[
         str,
-        typer.Argument(help="Entity type to analyze (e.g., Person, meeting)"),
+        typer.Argument(help="Note type to analyze (e.g., person, meeting)"),
     ],
     project: Annotated[
         Optional[str],
@@ -231,7 +257,7 @@ def infer(
 ):
     """Infer schema from existing notes of a type.
 
-    Analyzes all notes with the given entity type and suggests a Picoschema
+    Analyzes all notes with the given type and suggests a Picoschema
     definition based on observation and relation frequency.
 
     Fields present in 95%+ of notes become required. Fields above the
@@ -266,7 +292,7 @@ async def _run_diff(
     """Run schema drift detection via the API."""
     from basic_memory.mcp.clients.schema import SchemaClient
 
-    async with get_client() as client:
+    async with get_client(project_name=project) as client:
         active_project = await get_active_project(client, project, None)
         schema_client = SchemaClient(client, active_project.external_id)
 
@@ -300,7 +326,7 @@ async def _run_diff(
 def diff(
     entity_type: Annotated[
         str,
-        typer.Argument(help="Entity type to check for drift"),
+        typer.Argument(help="Note type to check for drift"),
     ],
     project: Annotated[
         Optional[str],
@@ -313,8 +339,8 @@ def diff(
 ):
     """Show drift between schema and actual usage.
 
-    Compares the existing schema definition for an entity type against
-    how notes of that type are actually structured. Identifies new fields,
+    Compares the existing schema definition against how notes of that type
+    are actually structured. Identifies new fields,
     dropped fields, and cardinality changes.
 
     Use --local to force local routing when cloud mode is enabled.

src/basic_memory/cli/commands/status.py

@@ -12,6 +12,7 @@ from rich.tree import Tree
 
 from basic_memory.cli.app import app
 from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
+from basic_memory.config import ConfigManager
 from basic_memory.mcp.async_client import get_client
 from basic_memory.mcp.tools.utils import call_post
 from basic_memory.schemas import SyncReportResponse
@@ -142,9 +143,11 @@ def display_changes(
 
 async def run_status(project: Optional[str] = None, verbose: bool = False):  # pragma: no cover
     """Check sync status of files vs database."""
+    # Resolve default project so get_client() can route per-project
+    project = project or ConfigManager().default_project
 
     try:
-        async with get_client() as client:
+        async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
             response = await call_post(client, f"/v2/projects/{project_item.external_id}/status")
             sync_report = SyncReportResponse.model_validate(response.json())

src/basic_memory/cli/commands/tool.py

@@ -50,7 +50,7 @@ async def _write_note_json(
     await mcp_write_note.fn(title, content, folder, project_name, tags)
 
     # Resolve the entity to get metadata back
-    async with get_client() as client:
+    async with get_client(project_name=project_name) as client:
         active_project = await get_active_project(client, project_name)
         knowledge_client = KnowledgeClient(client, active_project.external_id)
 
@@ -72,7 +72,7 @@ async def _read_note_json(
     identifier: str, project_name: Optional[str], page: int, page_size: int
 ) -> dict:
     """Read a note and return structured JSON with content and metadata."""
-    async with get_client() as client:
+    async with get_client(project_name=project_name) as client:
         active_project = await get_active_project(client, project_name)
         knowledge_client = KnowledgeClient(client, active_project.external_id)
         resource_client = ResourceClient(client, active_project.external_id)
@@ -120,7 +120,7 @@ async def _recent_activity_json(
     page_size: int = 50,
 ) -> list:
     """Get recent activity and return structured JSON list."""
-    async with get_client() as client:
+    async with get_client(project_name=project_name) as client:
         # Build query params matching the MCP tool's logic
         params: dict = {"page": page, "page_size": page_size, "max_related": 10}
         if depth:
@@ -364,7 +364,7 @@ def build_context(
         project_name = project_name or config_manager.default_project
 
         with force_routing(local=local, cloud=cloud):
-            context = run_with_cleanup(
+            result = run_with_cleanup(
                 mcp_build_context.fn(
                     project=project_name,
                     url=url,
@@ -375,8 +375,8 @@ def build_context(
                     max_related=max_related,
                 )
             )
-        context_dict = context.model_dump(exclude_none=True)
-        print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str))
+        # build_context now returns a slimmed dict (already serializable)
+        print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)

src/basic_memory/cli/main.py

@@ -5,6 +5,7 @@ import warnings
 
 from basic_memory.cli.app import app  # pragma: no cover
 
+
 def _version_only_invocation(argv: list[str]) -> bool:
     # Trigger: invocation is exactly `bm --version` or `bm -v`
     # Why: avoid importing command modules on the hot version path

src/basic_memory/cli/promo.py

@@ -2,14 +2,15 @@
 
 import os
 import sys
-from collections.abc import Callable
 
-import typer
+from rich.console import Console
+from rich.panel import Panel
 
+import basic_memory
 from basic_memory.config import ConfigManager
 
-CLOUD_PROMO_VERSION = "2026-02-06"
-OSS_DISCOUNT_CODE = "{{OSS_DISCOUNT_CODE}}"
+OSS_DISCOUNT_CODE = "BMFOSS"
+CLOUD_LEARN_MORE_URL = "https://basicmemory.com"
 
 
 def _promos_disabled_by_env() -> bool:
@@ -23,24 +24,44 @@ def _is_interactive_session() -> bool:
     return sys.stdin.isatty() and sys.stdout.isatty()
 
 
-def _build_first_run_message() -> str:
-    """Build first-run cloud promo copy."""
+def _build_cloud_promo_message() -> str:
+    """Build benefit-led cloud upsell copy with Rich markup."""
     return (
-        "Basic Memory initialized (local mode).\n"
-        "Cloud is optional and keeps your workflow local-first.\n"
-        "Cloud adds cross-device sync + mobile/web access.\n"
-        f"OSS discount: {OSS_DISCOUNT_CODE} (20% off for 3 months).\n"
-        "Run `bm cloud login` to enable."
+        "☁️ [bold]Your knowledge, everywhere.[/bold] ✨\n"
+        "Stop losing context when you switch machines.\n"
+        "Basic Memory Cloud syncs your memory across every device, including mobile and web.\n"
+        "Try it free for 7 days.\n"
+        f"Use [bold cyan]{OSS_DISCOUNT_CODE}[/bold cyan] for 20% off when you subscribe.\n"
+        "[bold green]→ bm cloud login[/bold green]"
     )
 
 
-def _build_version_message() -> str:
-    """Build cloud promo copy shown after promo-version bumps."""
-    return (
-        "New in Basic Memory Cloud: cross-device sync + mobile/web access.\n"
-        f"OSS discount: {OSS_DISCOUNT_CODE} (20% off for 3 months).\n"
-        "Run `bm cloud login` to enable."
-    )
+def maybe_show_init_line(
+    invoked_subcommand: str | None,
+    *,
+    config_manager: ConfigManager | None = None,
+    is_interactive: bool | None = None,
+    console: Console | None = None,
+) -> None:
+    """Show a one-time init confirmation line before command output."""
+    manager = config_manager or ConfigManager()
+    config = manager.load_config()
+
+    interactive = _is_interactive_session() if is_interactive is None else is_interactive
+
+    # Same gates as the cloud promo — suppress in non-interactive, env kill-switch,
+    # mcp/root-help contexts, or when already shown.
+    if _promos_disabled_by_env() or not interactive:
+        return
+
+    if invoked_subcommand in {None, "mcp"}:
+        return
+
+    if config.cloud_promo_first_run_shown:
+        return
+
+    out = console or Console()
+    out.print("Basic Memory initialized ✓")
 
 
 def maybe_show_cloud_promo(
@@ -48,7 +69,7 @@ def maybe_show_cloud_promo(
     *,
     config_manager: ConfigManager | None = None,
     is_interactive: bool | None = None,
-    echo: Callable[[str], None] = typer.echo,
+    console: Console | None = None,
 ) -> None:
     """Show cloud promo copy when discovery gates are satisfied."""
     manager = config_manager or ConfigManager()
@@ -72,13 +93,22 @@ def maybe_show_cloud_promo(
         return
 
     show_first_run = not config.cloud_promo_first_run_shown
-    show_version_notice = config.cloud_promo_last_version_shown != CLOUD_PROMO_VERSION
+    show_version_notice = config.cloud_promo_last_version_shown != basic_memory.__version__
     if not show_first_run and not show_version_notice:
         return
 
-    message = _build_first_run_message() if show_first_run else _build_version_message()
-    echo(message)
+    out = console or Console()
+    out.print(
+        Panel(
+            _build_cloud_promo_message(),
+            title="Basic Memory Cloud",
+            border_style="cyan",
+            expand=False,
+        )
+    )
+    out.print(f"Learn more at [link={CLOUD_LEARN_MORE_URL}]{CLOUD_LEARN_MORE_URL}[/link]")
+    out.print("[dim]Disable with: bm cloud promo --off[/dim]")
 
     config.cloud_promo_first_run_shown = True
-    config.cloud_promo_last_version_shown = CLOUD_PROMO_VERSION
+    config.cloud_promo_last_version_shown = basic_memory.__version__
     manager.save_config(config)

src/basic_memory/config.py

@@ -60,6 +60,9 @@ class CloudProjectConfig(BaseModel):
 
     This tracks the local working directory and sync state for a project
     that is synced with Basic Memory Cloud.
+
+    DEPRECATED: Kept for backward-compatible migration only. New code should
+    use ProjectEntry fields (cloud_sync_path, bisync_initialized, last_sync).
     """
 
     local_path: str = Field(description="Local working directory path for this cloud project")
@@ -71,26 +74,52 @@ class CloudProjectConfig(BaseModel):
     )
 
 
+class ProjectEntry(BaseModel):
+    """Unified project configuration entry.
+
+    Replaces the old triple of projects (Dict[str, str]), project_modes
+    (Dict[str, ProjectMode]), and cloud_projects (Dict[str, CloudProjectConfig])
+    with a single structure per project.
+    """
+
+    path: str = Field(description="Local filesystem path for the project")
+    mode: ProjectMode = Field(
+        default=ProjectMode.LOCAL,
+        description="Routing mode: local (in-process ASGI) or cloud (remote API)",
+    )
+    # Cloud sync state (replaces CloudProjectConfig)
+    cloud_sync_path: Optional[str] = Field(
+        default=None,
+        description="Local working directory for bisync (formerly CloudProjectConfig.local_path)",
+    )
+    bisync_initialized: bool = Field(
+        default=False,
+        description="Whether rclone bisync baseline has been established",
+    )
+    last_sync: Optional[datetime] = Field(
+        default=None,
+        description="Timestamp of last successful sync operation",
+    )
+
+
 class BasicMemoryConfig(BaseSettings):
     """Pydantic model for Basic Memory global configuration."""
 
     env: Environment = Field(default="dev", description="Environment name")
 
-    projects: Dict[str, str] = Field(
+    projects: Dict[str, ProjectEntry] = Field(
         default_factory=lambda: {
-            "main": str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
+            "main": ProjectEntry(
+                path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
+            )
         }
         if os.getenv("BASIC_MEMORY_HOME")
         else {},
-        description="Mapping of project names to their filesystem paths",
+        description="Mapping of project names to their ProjectEntry configuration",
     )
-    default_project: str = Field(
+    default_project: Optional[str] = Field(
         default="main",
-        description="Name of the default project to use",
-    )
-    default_project_mode: bool = Field(
-        default=True,
-        description="When True, MCP tools automatically use default_project when no project parameter is specified. Enables simplified UX for single-project workflows.",
+        description="Name of the default project to use. When set, acts as fallback when no project parameter is specified. Set to null to disable automatic project resolution.",
     )
 
     # overridden by ~/.basic-memory/config.json
@@ -134,6 +163,12 @@ class BasicMemoryConfig(BaseSettings):
         description="Vector candidate count for vector and hybrid retrieval.",
         gt=0,
     )
+    semantic_min_similarity: float = Field(
+        default=0.55,
+        description="Minimum similarity score for vector search results. Results below this threshold are filtered out. 0.0 disables filtering.",
+        ge=0.0,
+        le=1.0,
+    )
 
     # Database connection pool configuration (Postgres only)
     db_pool_size: int = Field(
@@ -257,11 +292,6 @@ class BasicMemoryConfig(BaseSettings):
         description="Enable cloud mode - all requests go to cloud instead of local (config file value)",
     )
 
-    cloud_projects: Dict[str, CloudProjectConfig] = Field(
-        default_factory=dict,
-        description="Cloud project sync configuration mapping project names to their local paths and sync state",
-    )
-
     cloud_promo_opt_out: bool = Field(
         default=False,
         description="Disable CLI cloud promo messages when true.",
@@ -282,10 +312,77 @@ class BasicMemoryConfig(BaseSettings):
         description="API key for cloud access (bmc_ prefixed). Account-level, not per-project.",
     )
 
-    project_modes: Dict[str, ProjectMode] = Field(
-        default_factory=dict,
-        description="Per-project routing mode. Projects not listed default to LOCAL.",
-    )
+    @model_validator(mode="before")
+    @classmethod
+    def migrate_legacy_projects(cls, data: Any) -> Any:
+        """Migrate old-format config (Dict[str, str]) to new ProjectEntry format.
+
+        Old format stored projects as three separate dicts:
+          projects:      {"name": "/path"}
+          project_modes: {"name": "cloud"}
+          cloud_projects: {"name": {"local_path": "...", ...}}
+
+        New format unifies them into:
+          projects: {"name": {"path": "/path", "mode": "cloud", ...}}
+
+        Also removes stale keys (default_project_mode, permalinks_include_project)
+        that are no longer part of the config model.
+        """
+        if not isinstance(data, dict):
+            return data
+
+        # --- Remove stale keys from old config versions ---
+        data.pop("default_project_mode", None)
+
+        projects = data.get("projects", {})
+        if not projects:
+            return data
+
+        # Check if already in new format — peek at first value
+        first_value = next(iter(projects.values()), None)
+        if isinstance(first_value, str):
+            # Old format: {"name": "/path"} → convert
+            project_modes = data.pop("project_modes", {})
+            cloud_projects = data.pop("cloud_projects", {})
+            new_projects: Dict[str, Any] = {}
+            for name, path in projects.items():
+                entry: Dict[str, Any] = {"path": path}
+                if name in project_modes:
+                    entry["mode"] = project_modes[name]
+                if name in cloud_projects:
+                    cp = cloud_projects[name]
+                    if isinstance(cp, dict):
+                        entry["cloud_sync_path"] = cp.get("local_path")
+                        entry["bisync_initialized"] = cp.get("bisync_initialized", False)
+                        entry["last_sync"] = cp.get("last_sync")
+                    else:
+                        # Already a CloudProjectConfig-like object
+                        entry["cloud_sync_path"] = getattr(cp, "local_path", None)
+                        entry["bisync_initialized"] = getattr(cp, "bisync_initialized", False)
+                        entry["last_sync"] = getattr(cp, "last_sync", None)
+                new_projects[name] = entry
+
+            # Pick up cloud_projects entries not already in projects
+            # These are cloud-only projects — path is the cloud permalink,
+            # local_path goes into cloud_sync_path for bisync
+            for name, cp in cloud_projects.items():
+                if name not in new_projects:
+                    if isinstance(cp, dict):
+                        new_projects[name] = {
+                            "path": generate_permalink(name),
+                            "mode": project_modes.get(name, "cloud"),
+                            "cloud_sync_path": cp.get("local_path"),
+                            "bisync_initialized": cp.get("bisync_initialized", False),
+                            "last_sync": cp.get("last_sync"),
+                        }
+
+            data["projects"] = new_projects
+        else:
+            # New format or dict-based — just clean up stale keys
+            data.pop("project_modes", None)
+            data.pop("cloud_projects", None)
+
+        return data
 
     @property
     def is_test_env(self) -> bool:
@@ -325,21 +422,26 @@ class BasicMemoryConfig(BaseSettings):
 
         Returns the per-project mode if set, otherwise LOCAL.
         """
-        return self.project_modes.get(project_name, ProjectMode.LOCAL)
+        entry = self.projects.get(project_name)
+        return entry.mode if entry else ProjectMode.LOCAL
 
     def set_project_mode(self, project_name: str, mode: ProjectMode) -> None:
-        """Set the routing mode for a project."""
-        if mode == ProjectMode.LOCAL:
-            # Remove from dict to keep config clean — LOCAL is the default
-            self.project_modes.pop(project_name, None)
+        """Set the routing mode for a project.
+
+        Creates a minimal ProjectEntry if the project doesn't already exist,
+        preserving backward compatibility with code that sets mode before
+        adding a full project entry.
+        """
+        if project_name in self.projects:
+            self.projects[project_name].mode = mode
         else:
-            self.project_modes[project_name] = mode
+            self.projects[project_name] = ProjectEntry(path="", mode=mode)
 
     @classmethod
     def for_cloud_tenant(
         cls,
         database_url: str,
-        projects: Optional[Dict[str, str]] = None,
+        projects: Optional[Dict[str, "ProjectEntry"]] = None,
     ) -> "BasicMemoryConfig":
         """Create config for cloud tenant - no config.json, database is source of truth.
 
@@ -377,7 +479,7 @@ class BasicMemoryConfig(BaseSettings):
         if name not in self.projects:
             raise ValueError(f"Project '{name}' not found in configuration")
 
-        return Path(self.projects[name])
+        return Path(self.projects[name].path)
 
     def model_post_init(self, __context: Any) -> None:
         """Ensure configuration is valid after initialization."""
@@ -387,12 +489,15 @@ class BasicMemoryConfig(BaseSettings):
 
         # Ensure at least one project exists; if none exist then create main
         if not self.projects:  # pragma: no cover
-            self.projects["main"] = str(
-                Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory"))
+            self.projects["main"] = ProjectEntry(
+                path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
             )
 
         # Ensure default project is valid (i.e. points to an existing project)
-        if self.default_project not in self.projects:  # pragma: no cover
+        # None means "no default" — intentionally left unset
+        if (
+            self.default_project is not None and self.default_project not in self.projects
+        ):  # pragma: no cover
             # Set default to first available project
             self.default_project = next(iter(self.projects.keys()))
 
@@ -429,8 +534,8 @@ class BasicMemoryConfig(BaseSettings):
     def project_list(self) -> List[ProjectConfig]:  # pragma: no cover
         """Get all configured projects as ProjectConfig objects."""
         return [
-            ProjectConfig(name=name, home=Path(path), mode=self.get_project_mode(name))
-            for name, path in self.projects.items()
+            ProjectConfig(name=name, home=Path(entry.path), mode=entry.mode)
+            for name, entry in self.projects.items()
         ]
 
     @model_validator(mode="after")
@@ -444,8 +549,8 @@ class BasicMemoryConfig(BaseSettings):
         if self.database_backend == DatabaseBackend.POSTGRES:
             return self
 
-        for name, path_value in self.projects.items():
-            path = Path(path_value)
+        for name, entry in self.projects.items():
+            path = Path(entry.path)
             if not path.exists():
                 try:
                     path.mkdir(parents=True)
@@ -511,6 +616,17 @@ class ConfigManager:
             try:
                 file_data = json.loads(self.config_file.read_text(encoding="utf-8"))
 
+                # Detect legacy format before model validators strip stale keys
+                _STALE_KEYS = {"default_project_mode", "project_modes", "cloud_projects"}
+                needs_resave = bool(_STALE_KEYS & file_data.keys())
+
+                # Check if projects dict uses old string-value format
+                projects_raw = file_data.get("projects", {})
+                if projects_raw:
+                    first_val = next(iter(projects_raw.values()), None)
+                    if isinstance(first_val, str):
+                        needs_resave = True
+
                 # First, create config from environment variables (Pydantic will read them)
                 # Then overlay with file data for fields that aren't set via env vars
                 # This ensures env vars take precedence
@@ -532,6 +648,12 @@ class ConfigManager:
                         merged_data[field_name] = env_dict[field_name]
 
                 _CONFIG_CACHE = BasicMemoryConfig(**merged_data)
+
+                # Re-save to normalize legacy config into current format
+                if needs_resave:
+                    logger.info("Migrating config to current format")
+                    save_basic_memory_config(self.config_file, _CONFIG_CACHE)
+
                 return _CONFIG_CACHE
             except Exception as e:  # pragma: no cover
                 logger.exception(f"Failed to load config: {e}")
@@ -550,11 +672,15 @@ class ConfigManager:
 
     @property
     def projects(self) -> Dict[str, str]:
-        """Get all configured projects."""
-        return self.config.projects.copy()
+        """Get all configured projects as name -> path mapping.
+
+        Returns the legacy Dict[str, str] format for backward compatibility
+        with code that expects project name -> filesystem path.
+        """
+        return {name: entry.path for name, entry in self.config.projects.items()}
 
     @property
-    def default_project(self) -> str:
+    def default_project(self) -> Optional[str]:
         """Get the default project name."""
         return self.config.default_project
 
@@ -570,7 +696,7 @@ class ConfigManager:
 
         # Load config, modify it, and save it
         config = self.load_config()
-        config.projects[name] = str(project_path)
+        config.projects[name] = ProjectEntry(path=str(project_path))
         self.save_config(config)
         return ProjectConfig(name=name, home=project_path)
 
@@ -602,12 +728,15 @@ class ConfigManager:
         self.save_config(config)
 
     def get_project(self, name: str) -> Tuple[str, str] | Tuple[None, None]:
-        """Look up a project from the configuration by name or permalink"""
+        """Look up a project from the configuration by name or permalink.
+
+        Returns (project_name, path_string) for backward compatibility.
+        """
         project_permalink = generate_permalink(name)
         app_config = self.config
-        for project_name, path in app_config.projects.items():
+        for project_name, entry in app_config.projects.items():
             if project_permalink == generate_permalink(project_name):
-                return project_name, path
+                return project_name, entry.path
         return None, None
 
 
@@ -642,9 +771,9 @@ def get_project_config(project_name: Optional[str] = None) -> ProjectConfig:
 
     project_permalink = generate_permalink(actual_project_name)
 
-    for name, path in app_config.projects.items():
+    for name, entry in app_config.projects.items():
         if project_permalink == generate_permalink(name):
-            return ProjectConfig(name=name, home=Path(path))
+            return ProjectConfig(name=name, home=Path(entry.path))
 
     # otherwise raise error
     raise ValueError(f"Project '{actual_project_name}' not found")  # pragma: no cover

src/basic_memory/db.py

@@ -344,24 +344,33 @@ async def engine_session_factory(
 
     global _engine, _session_maker
 
-    # Use the same helper function as production code
-    _engine, _session_maker = _create_engine_and_session(db_path, db_type, config)
+    # Use the same helper function as production code.
+    #
+    # Keep local references so teardown can deterministically dispose the
+    # specific engine created by this context manager, even if other code calls
+    # shutdown_db() and mutates module-level globals mid-test.
+    created_engine, created_session_maker = _create_engine_and_session(db_path, db_type, config)
+    _engine, _session_maker = created_engine, created_session_maker
 
     try:
         # Verify that engine and session maker are initialized
-        if _engine is None:  # pragma: no cover
+        if created_engine is None:  # pragma: no cover
             logger.error("Database engine is None in engine_session_factory")
             raise RuntimeError("Database engine initialization failed")
 
-        if _session_maker is None:  # pragma: no cover
+        if created_session_maker is None:  # pragma: no cover
             logger.error("Session maker is None in engine_session_factory")
             raise RuntimeError("Session maker initialization failed")
 
-        yield _engine, _session_maker
+        yield created_engine, created_session_maker
     finally:
-        if _engine:
-            await _engine.dispose()
+        await created_engine.dispose()
+
+        # Only clear module-level globals if they still point to this context's
+        # engine/session. This avoids clobbering newer globals from other callers.
+        if _engine is created_engine:
             _engine = None
+        if _session_maker is created_session_maker:
             _session_maker = None
 
 

src/basic_memory/deps/services.py

@@ -8,6 +8,7 @@ This module provides service-layer dependencies:
 """
 
 import asyncio
+import os
 from typing import Annotated, Any, Callable, Coroutine, Mapping, Protocol
 
 from fastapi import Depends
@@ -446,13 +447,22 @@ def _log_task_failure(completed: asyncio.Task) -> None:
 
 
 class LocalTaskScheduler:
-    """Default scheduler that runs tasks in-process via asyncio.create_task."""
+    """Default scheduler that runs tasks in-process via asyncio.create_task.
+
+    In test mode (BASIC_MEMORY_ENV=test), tasks run as no-ops to avoid
+    background asyncio tasks racing against test teardown and causing
+    SQLite 'cannot commit transaction' errors.
+    """
 
     def __init__(
         self,
         handlers: Mapping[str, Callable[..., Coroutine[Any, Any, None]]],
+        test_mode: bool | None = None,
     ) -> None:
         self._handlers = handlers
+        self._test_mode = (
+            test_mode if test_mode is not None else os.environ.get("BASIC_MEMORY_ENV") == "test"
+        )
 
     def schedule(self, task_name: str, **payload: Any) -> None:
         handler = self._handlers.get(task_name)
@@ -461,6 +471,15 @@ class LocalTaskScheduler:
         # Outcome: fail fast to surface misconfiguration
         if not handler:
             raise ValueError(f"Unknown task name: {task_name}")
+
+        # Trigger: running inside pytest (BASIC_MEMORY_ENV=test)
+        # Why: background create_task() outlives test fixtures and races
+        #      against engine disposal, causing flaky SQLite errors
+        # Outcome: skip background scheduling; tests exercise the sync
+        #          codepaths directly when they need to
+        if self._test_mode:
+            return
+
         task = asyncio.create_task(handler(**payload))
         task.add_done_callback(_log_task_failure)
 
@@ -516,7 +535,8 @@ async def get_task_scheduler(
             "sync_entity_vectors": _sync_entity_vectors,
             "sync_project": _sync_project,
             "reindex_project": _reindex_project,
-        }
+        },
+        test_mode=app_config.is_test_env,
     )
     return scheduler
 

src/basic_memory/importers/claude_conversations_importer.py

@@ -54,9 +54,9 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
             for chat in conversations:
                 # Get name, providing default for unnamed conversations
                 chat_name = chat.get("name") or f"Conversation {chat.get('uuid', 'untitled')}"
-                date_prefix = datetime.fromisoformat(chat["created_at"].replace("Z", "+00:00")).strftime(
-                    "%Y%m%d"
-                )
+                date_prefix = datetime.fromisoformat(
+                    chat["created_at"].replace("Z", "+00:00")
+                ).strftime("%Y%m%d")
                 clean_title = clean_filename(chat_name)
                 relative_path = (
                     f"{destination_folder}/{date_prefix}-{clean_title}"

src/basic_memory/mcp/async_client.py

@@ -22,6 +22,19 @@ def _force_local_mode() -> bool:
     return os.environ.get("BASIC_MEMORY_FORCE_LOCAL", "").lower() in ("true", "1", "yes")
 
 
+def _explicit_routing() -> bool:
+    """Check if CLI --local/--cloud flag was explicitly passed.
+
+    Set by force_routing() in CLI commands. When active, --local/--cloud
+    flags override per-project routing. The MCP server sets FORCE_LOCAL
+    directly (without this flag), so per-project routing still works there.
+
+    Returns:
+        True if BASIC_MEMORY_EXPLICIT_ROUTING is set to a truthy value
+    """
+    return os.environ.get("BASIC_MEMORY_EXPLICIT_ROUTING", "").lower() in ("true", "1", "yes")
+
+
 # Optional factory override for dependency injection
 _client_factory: Optional[Callable[[], AbstractAsyncContextManager[AsyncClient]]] = None
 
@@ -56,23 +69,29 @@ async def get_client(
     1. **Factory injection** (cloud app, tests):
        If a custom factory is set via set_client_factory(), use that.
 
-    2. **Per-project cloud mode** (project_name provided):
-       If the project's mode is CLOUD, routes to cloud using API key or
-       OAuth token. Honored even when FORCE_LOCAL is set, because the user
-       explicitly declared this project as cloud.
+    2. **CLI explicit override** (BASIC_MEMORY_EXPLICIT_ROUTING env var):
+       When --local or --cloud is explicitly passed via CLI, skip per-project
+       routing and fall through to force-local / global cloud mode handling.
+       This allows users to override per-project mode for commands like
+       `bm status --project specs --local` (check local copy of a cloud project).
 
-    3. **Per-project local mode** (project_name provided):
+    3. **Per-project cloud mode** (project_name provided, no explicit override):
+       If the project's mode is CLOUD, routes to cloud using API key or
+       OAuth token. Honored even when FORCE_LOCAL is set (e.g. MCP server),
+       because the user explicitly declared this project as cloud.
+
+    4. **Per-project local mode** (project_name provided, no explicit override):
        If the project's mode is LOCAL (or unspecified, default LOCAL), route
        to local ASGI transport. This allows mixed local/cloud routing even when
        global cloud mode is enabled.
 
-    4. **Force-local** (BASIC_MEMORY_FORCE_LOCAL env var):
+    5. **Force-local** (BASIC_MEMORY_FORCE_LOCAL env var):
        Routes to local ASGI transport, ignoring global cloud settings.
 
-    5. **Global cloud mode** (deprecated fallback):
-       When cloud_mode_enabled is True, uses OAuth JWT token.
+    6. **Global cloud mode**:
+       When cloud_mode_enabled is True, uses OAuth JWT token or API key.
 
-    6. **Local mode** (default):
+    7. **Local mode** (default):
        Use ASGI transport for in-process requests to local FastAPI app.
 
     Args:
@@ -108,53 +127,65 @@ async def get_client(
             pool=30.0,  # 30 seconds for connection pool
         )
 
-        # Trigger: project has per-project cloud mode set
-        # Why: per-project CLOUD is an explicit user declaration that should be
+        # --- Per-project routing (when project_name given and no CLI override) ---
+        # Trigger: CLI --local/--cloud flag was NOT explicitly passed
+        # Why: per-project routing is an explicit user declaration that should be
         #      honored even from the MCP server (which sets FORCE_LOCAL)
-        # Outcome: HTTP client with API key or OAuth auth to cloud proxy
-        if project_name and config.get_project_mode(project_name) == ProjectMode.CLOUD:
-            # Try API key first (explicit, no network)
-            token = config.cloud_api_key
-            if not token:
-                # Fall back to OAuth session (may refresh token)
-                from basic_memory.cli.auth import CLIAuth
+        # Outcome: route based on project's configured mode (CLOUD or LOCAL)
+        if project_name is not None and not _explicit_routing():
+            project_mode = config.get_project_mode(project_name)
 
-                auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
-                token = await auth.get_valid_token()
+            if project_mode == ProjectMode.CLOUD:
+                # Try API key first (explicit, no network)
+                token = config.cloud_api_key
+                if not token:
+                    # Fall back to OAuth session (may refresh token)
+                    from basic_memory.cli.auth import CLIAuth
 
-            if not token:
-                raise RuntimeError(
-                    f"Project '{project_name}' is set to cloud mode but no credentials found. "
-                    "Run 'bm cloud set-key <key>' or 'bm cloud login' first."
+                    auth = CLIAuth(
+                        client_id=config.cloud_client_id, authkit_domain=config.cloud_domain
+                    )
+                    token = await auth.get_valid_token()
+
+                if not token:
+                    raise RuntimeError(
+                        f"Project '{project_name}' is set to cloud mode but no credentials "
+                        "found. Run 'bm cloud set-key <key>' or 'bm cloud login' first."
+                    )
+
+                proxy_base_url = f"{config.cloud_host}/proxy"
+                logger.info(
+                    f"Creating HTTP client for cloud project '{project_name}' at: {proxy_base_url}"
                 )
+                async with AsyncClient(
+                    base_url=proxy_base_url,
+                    headers={"Authorization": f"Bearer {token}"},
+                    timeout=timeout,
+                ) as client:
+                    yield client
+                return
 
-            proxy_base_url = f"{config.cloud_host}/proxy"
-            logger.info(
-                f"Creating HTTP client for cloud project '{project_name}' at: {proxy_base_url}"
-            )
-            async with AsyncClient(
-                base_url=proxy_base_url,
-                headers={"Authorization": f"Bearer {token}"},
-                timeout=timeout,
-            ) as client:
-                yield client
+            # Trigger: project is LOCAL (the default, no CLI override)
+            # Why: project-scoped routing should honor local mode even when global
+            #      cloud mode is enabled for backward compatibility
+            # Outcome: uses ASGI transport for in-process local API calls
+            else:
+                logger.info(f"Project '{project_name}' is set to local mode - using ASGI transport")
+                async with AsyncClient(
+                    transport=ASGITransport(app=fastapi_app),
+                    base_url="http://test",
+                    timeout=timeout,
+                ) as client:
+                    yield client
+                return
 
-        # Trigger: project is not explicitly cloud (LOCAL is the default)
-        # Why: project-scoped routing should honor local mode even when global
-        #      cloud mode is enabled for backward compatibility
-        # Outcome: uses ASGI transport for in-process local API calls
-        elif project_name and config.get_project_mode(project_name) == ProjectMode.LOCAL:
-            logger.info(f"Project '{project_name}' is set to local mode - using ASGI transport")
-            async with AsyncClient(
-                transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout
-            ) as client:
-                yield client
+        # --- Fallback routing (no per-project routing applies) ---
 
         # Trigger: BASIC_MEMORY_FORCE_LOCAL env var is set
         # Why: allows local MCP server and CLI commands to route locally
         #      even when cloud_mode_enabled is True
         # Outcome: uses ASGI transport for in-process local API calls
-        elif _force_local_mode():
+        if _force_local_mode():
             logger.info("Force local mode enabled - using ASGI client for local Basic Memory API")
             async with AsyncClient(
                 transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout

src/basic_memory/mcp/formatting.py

@@ -84,7 +84,9 @@ def format_search_results_ascii(
     if query:
         lines.append(f"Query: {query}")
 
-    summary = f"Results: {len(results)} | Page: {result.current_page} | Page size: {result.page_size}"
+    summary = (
+        f"Results: {len(results)} | Page: {result.current_page} | Page size: {result.page_size}"
+    )
     lines.append(_apply_style(summary, ANSI_DIM, color))
 
     if not results:

src/basic_memory/mcp/project_context.py

@@ -31,7 +31,6 @@ async def resolve_project_parameter(
     project: Optional[str] = None,
     allow_discovery: bool = False,
     cloud_mode: Optional[bool] = None,
-    default_project_mode: Optional[bool] = None,
     default_project: Optional[str] = None,
 ) -> Optional[str]:
     """Resolve project parameter using unified linear priority chain.
@@ -43,7 +42,7 @@ async def resolve_project_parameter(
     Resolution order (same for local and cloud modes):
     1. ENV_CONSTRAINT: BASIC_MEMORY_MCP_PROJECT env var (highest priority)
     2. EXPLICIT: project parameter passed directly
-    3. DEFAULT: default project when default_project_mode=true
+    3. DEFAULT: default_project from config (if set)
     4. Fallback: cloud → CLOUD_DISCOVERY or ValueError; local → NONE
 
     Args:
@@ -51,26 +50,22 @@ async def resolve_project_parameter(
         allow_discovery: If True, allows returning None in cloud mode for discovery mode
             (used by tools like recent_activity that can operate across all projects)
         cloud_mode: Optional explicit cloud mode. If not provided, reads from ConfigManager.
-        default_project_mode: Optional explicit default project mode. If not provided, reads from ConfigManager.
         default_project: Optional explicit default project. If not provided, reads from ConfigManager.
 
     Returns:
         Resolved project name or None if no resolution possible
     """
     # Load config for any values not explicitly provided
-    if cloud_mode is None or default_project_mode is None or default_project is None:
+    if cloud_mode is None or default_project is None:
         config = ConfigManager().config
         if cloud_mode is None:
             cloud_mode = config.cloud_mode
-        if default_project_mode is None:
-            default_project_mode = config.default_project_mode
         if default_project is None:
             default_project = config.default_project
 
     # Create resolver with configuration and resolve
     resolver = ProjectResolver.from_env(
         cloud_mode=cloud_mode,
-        default_project_mode=default_project_mode,
         default_project=default_project,
     )
     result = resolver.resolve(project=project, allow_discovery=allow_discovery)
@@ -114,7 +109,7 @@ async def get_active_project(
         project_names = await get_project_names(client, headers)
         raise ValueError(
             "No project specified. "
-            "Either set 'default_project_mode=true' in config, or use 'project' argument.\n"
+            "Either set 'default_project' in config, or use 'project' argument.\n"
             f"Available projects: {project_names}"
         )
 
@@ -225,9 +220,7 @@ async def resolve_project_and_path(
             if context:
                 context.set_state("active_project", active_project)
 
-            resolved_path = (
-                f"{resolved.permalink}/{remainder}" if include_project else remainder
-            )
+            resolved_path = f"{resolved.permalink}/{remainder}" if include_project else remainder
             return active_project, resolved_path, True
 
     # Trigger: no resolvable project prefix in the memory URL
@@ -295,7 +288,7 @@ async def get_project_client(
             project_names = await get_project_names(client)
             raise ValueError(
                 "No project specified. "
-                "Either set 'default_project_mode=true' in config, or use 'project' argument.\n"
+                "Either set 'default_project' in config, or use 'project' argument.\n"
                 f"Available projects: {project_names}"
             )
 

src/basic_memory/mcp/prompts/ai_assistant_guide.py

@@ -14,8 +14,8 @@ def ai_assistant_guide() -> str:
     """Return a concise guide on Basic Memory tools and how to use them.
 
     Dynamically adapts instructions based on configuration:
-    - Default project mode: Simplified instructions with automatic project
-    - Regular mode: Project discovery and selection guidance
+    - Default project set: Simplified instructions with automatic project fallback
+    - No default project: Project discovery and selection guidance
     - CLI constraint mode: Single project constraint information
 
     Returns:
@@ -30,34 +30,32 @@ def ai_assistant_guide() -> str:
     # Check configuration for mode-specific instructions
     config = ConfigManager().config
 
-    # Add mode-specific header
-    mode_info = ""
-    if config.default_project_mode:
+    # Add mode-specific header based on whether a default project is configured
+    if config.default_project:
         mode_info = f"""
-# 🎯 Default Project Mode Active
+# Default Project Active
 
-**Current Configuration**: All operations automatically use project '{config.default_project}'
+**Current Configuration**: Operations automatically fall back to project '{config.default_project}'
 
 **Simplified Usage**: You don't need to specify the project parameter in tool calls.
-- `write_note(title="Note", content="...", folder="docs")` ✅
-- Project parameter is optional and will default to '{config.default_project}'
+- `write_note(title="Note", content="...", folder="docs")` - uses '{config.default_project}'
 - To use a different project, explicitly specify: `project="other-project"`
 
-────────────────────────────────────────
+---
 
 """
     else:  # pragma: no cover
         mode_info = """
-# 🔧 Multi-Project Mode Active
+# Multi-Project Mode
 
-**Current Configuration**: Project parameter required for all operations
+**Current Configuration**: No default project set — project parameter required for all operations
 
 **Project Discovery Required**: Use these tools to select a project:
 - `list_memory_projects()` - See all available projects
 - `recent_activity()` - Get project activity and recommendations
 - Remember the user's project choice throughout the conversation
 
-────────────────────────────────────────
+---
 
 """
 
@@ -65,6 +63,7 @@ def ai_assistant_guide() -> str:
     enhanced_content = mode_info + content
 
     logger.info(
-        f"Loaded AI assistant guide ({len(enhanced_content)} chars) with mode: {'default_project' if config.default_project_mode else 'multi_project'}"
+        f"Loaded AI assistant guide ({len(enhanced_content)} chars) "
+        f"with default_project: {config.default_project or 'none'}"
     )
     return enhanced_content

src/basic_memory/mcp/resources/__init__.py

@@ -1,25 +1,27 @@
 """MCP resources for Basic Memory."""
 
 from basic_memory.mcp.resources.project_info import project_info
-from basic_memory.mcp.resources.ui import (
-    note_preview_ui,
-    note_preview_ui_mcp_ui,
-    note_preview_ui_tool_ui,
-    note_preview_ui_vanilla,
-    search_results_ui,
-    search_results_ui_mcp_ui,
-    search_results_ui_tool_ui,
-    search_results_ui_vanilla,
-)
+
+# TODO: re-enable once MCP client rendering is working
+# from basic_memory.mcp.resources.ui import (
+#     note_preview_ui,
+#     note_preview_ui_mcp_ui,
+#     note_preview_ui_tool_ui,
+#     note_preview_ui_vanilla,
+#     search_results_ui,
+#     search_results_ui_mcp_ui,
+#     search_results_ui_tool_ui,
+#     search_results_ui_vanilla,
+# )
 
 __all__ = [
     "project_info",
-    "note_preview_ui",
-    "note_preview_ui_mcp_ui",
-    "note_preview_ui_tool_ui",
-    "note_preview_ui_vanilla",
-    "search_results_ui",
-    "search_results_ui_mcp_ui",
-    "search_results_ui_tool_ui",
-    "search_results_ui_vanilla",
+    # "note_preview_ui",
+    # "note_preview_ui_mcp_ui",
+    # "note_preview_ui_tool_ui",
+    # "note_preview_ui_vanilla",
+    # "search_results_ui",
+    # "search_results_ui_mcp_ui",
+    # "search_results_ui_tool_ui",
+    # "search_results_ui_vanilla",
 ]

src/basic_memory/mcp/resources/ai_assistant_guide.md

@@ -14,36 +14,30 @@ Basic Memory creates a semantic knowledge graph from markdown files. Focus on bu
 
 **Your role**: You're helping humans build enduring knowledge they'll own forever. The semantic graph (observations, relations, context) helps you provide better assistance by understanding connections and maintaining continuity. Think: lasting insights worth keeping, not disposable chat logs.
 
-## Project Management 
+## Project Management
 
-All tools require explicit project specification.
-
-**Three-tier resolution:**
-1. CLI constraint: `--project name` (highest priority)
+**Resolution priority:**
+1. CLI constraint: `BASIC_MEMORY_MCP_PROJECT` env var (highest priority)
 2. Explicit parameter: `project="name"` in tool calls
-3. Default mode: `default_project_mode=true` in config (fallback)
+3. Default project: `default_project` in config (fallback)
 
 ### Quick Setup Check
 
 ```python
 # Discover projects
 projects = await list_memory_projects()
-
-# Check if default_project_mode enabled
-# If yes: project parameter optional
-# If no: project parameter required
 ```
 
-### Default Project Mode
+### Default Project
 
-When `default_project_mode=true`:
+When `default_project` is set in config:
 ```python
 # These are equivalent:
 await write_note("Note", "Content", "folder")
 await write_note("Note", "Content", "folder", project="main")
 ```
 
-When `default_project_mode=false`:
+When no `default_project` is configured:
 ```python
 # Project required:
 await write_note("Note", "Content", "folder", project="main")  # ✓
@@ -59,7 +53,7 @@ await write_note(
     title="Topic",
     content="# Topic\n## Observations\n- [category] fact\n## Relations\n- relates_to [[Other]]",
     folder="notes",
-    project="main"  # Required unless default_project_mode=true
+    project="main"  # Optional if default_project is set in config
 )
 ```
 
@@ -143,12 +137,11 @@ await write_note(
 ### 1. Project Management
 
 **Single-project users:**
-- Enable `default_project_mode=true`
-- Simpler tool calls
+- Set `default_project` in config (e.g., `"main"`)
+- Simpler tool calls — project parameter is optional
 
 **Multi-project users:**
-- Keep `default_project_mode=false`
-- Always specify project explicitly
+- Always specify project explicitly in tool calls
 
 **Discovery:**
 ```python
@@ -200,7 +193,7 @@ Background information
 **Missing project:**
 ```python
 try:
-    await search_notes(query="test")  # Missing project parameter - will error
+    await search_notes(query="test")  # Fails if no default_project configured
 except:
     # Show available projects
     projects = await list_memory_projects()

src/basic_memory/mcp/resources/project_info.py

@@ -38,8 +38,8 @@ async def project_info(
 
     Args:
         project: Optional project name. If not provided, uses default_project
-                (if default_project_mode=true) or CLI constraint. If unknown,
-                use list_memory_projects() to discover available projects.
+                from config or CLI constraint. If unknown, use
+                list_memory_projects() to discover available projects.
         context: Optional FastMCP context for performance caching.
 
     Returns:

src/basic_memory/mcp/server.py

@@ -2,12 +2,14 @@
 Basic Memory FastMCP server.
 """
 
+import time
 from contextlib import asynccontextmanager
 
 from fastmcp import FastMCP
 from loguru import logger
 
 from basic_memory import db
+from basic_memory.cli.auth import CLIAuth
 from basic_memory.mcp.container import McpContainer, set_container
 from basic_memory.services.initialization import initialize_app
 
@@ -26,7 +28,43 @@ async def lifespan(app: FastMCP):
     container = McpContainer.create()
     set_container(container)
 
-    logger.debug(f"Starting Basic Memory MCP server (mode={container.mode.name})")
+    config = container.config
+    logger.info(f"Starting Basic Memory MCP server (mode={container.mode.name})")
+    logger.info(
+        f"Config: database_backend={config.database_backend.value}, "
+        f"semantic_search_enabled={config.semantic_search_enabled}, "
+        f"default_project={config.default_project}"
+    )
+    if config.semantic_search_enabled:
+        logger.info(
+            f"Semantic search: provider={config.semantic_embedding_provider}, "
+            f"model={config.semantic_embedding_model}, "
+            f"dimensions={config.semantic_embedding_dimensions or 'auto'}, "
+            f"batch_size={config.semantic_embedding_batch_size}"
+        )
+
+    # Log configured projects with their routing mode
+    for name, entry in config.projects.items():
+        default = " (default)" if name == config.default_project else ""
+        logger.info(f"Project: {name} -> {entry.path} [mode={entry.mode.value}]{default}")
+
+    # Check cloud login status (local file check, no network call)
+    if config.cloud_mode:
+        auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
+        tokens = auth.load_tokens()
+        if tokens is None:
+            logger.warning("Cloud mode enabled but not authenticated - run 'bm cloud login'")
+        elif not auth.is_token_valid(tokens):
+            expires_at = tokens.get("expires_at", 0)
+            expired_ago = int(time.time() - expires_at)
+            logger.warning(f"Cloud token expired {expired_ago}s ago - may need 'bm cloud login'")
+        else:
+            logger.info("Cloud: authenticated (token valid)")
+
+        if config.cloud_api_key:
+            logger.info("Cloud: API key configured (preferred for per-project routing)")
+        else:
+            logger.info("Cloud: no API key set (will use OAuth token for cloud projects)")
 
     # Track if we created the engine (vs test fixtures providing it)
     # This prevents disposing an engine provided by test fixtures when

src/basic_memory/mcp/tools/__init__.py

@@ -11,7 +11,9 @@ from basic_memory.mcp.tools.read_content import read_content
 from basic_memory.mcp.tools.build_context import build_context
 from basic_memory.mcp.tools.recent_activity import recent_activity
 from basic_memory.mcp.tools.read_note import read_note
-from basic_memory.mcp.tools.ui_sdk import read_note_ui, search_notes_ui
+
+# TODO: re-enable once MCP client rendering is working
+# from basic_memory.mcp.tools.ui_sdk import read_note_ui, search_notes_ui
 from basic_memory.mcp.tools.view_note import view_note
 from basic_memory.mcp.tools.write_note import write_note
 from basic_memory.mcp.tools.cloud_info import cloud_info
@@ -48,7 +50,7 @@ __all__ = [
     "read_content",
     "read_note",
     "release_notes",
-    "read_note_ui",
+    # "read_note_ui",
     "recent_activity",
     "schema_diff",
     "schema_infer",
@@ -56,7 +58,7 @@ __all__ = [
     "search",
     "search_by_metadata",
     "search_notes",
-    "search_notes_ui",
+    # "search_notes_ui",
     "view_note",
     "write_note",
 ]

src/basic_memory/mcp/tools/build_context.py

@@ -8,7 +8,168 @@ from fastmcp import Context
 from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
 from basic_memory.mcp.server import mcp
 from basic_memory.schemas.base import TimeFrame
-from basic_memory.schemas.memory import GraphContext, MemoryUrl
+from basic_memory.schemas.memory import (
+    ContextResult,
+    EntitySummary,
+    GraphContext,
+    MemoryUrl,
+    ObservationSummary,
+    RelationSummary,
+)
+
+# --- Fields to strip from each model (redundant with parent entity) ---
+
+_OBSERVATION_STRIP = {
+    "observation_id",
+    "entity_id",
+    "entity_external_id",
+    "title",
+    "file_path",
+    "created_at",
+}
+_RELATION_STRIP = {
+    "relation_id",
+    "entity_id",
+    "from_entity_id",
+    "from_entity_external_id",
+    "to_entity_id",
+    "to_entity_external_id",
+    "title",
+    "file_path",
+    "created_at",
+}
+_ENTITY_STRIP = {"entity_id", "created_at"}
+_METADATA_STRIP = {"total_results", "generated_at"}
+
+
+def _slim_summary(summary: EntitySummary | RelationSummary | ObservationSummary) -> dict:
+    """Strip redundant fields from a summary model based on its type."""
+    if isinstance(summary, ObservationSummary):
+        strip = _OBSERVATION_STRIP
+    elif isinstance(summary, RelationSummary):
+        strip = _RELATION_STRIP
+    else:
+        strip = _ENTITY_STRIP
+
+    data = summary.model_dump()
+    for key in strip:
+        data.pop(key, None)
+    return data
+
+
+def _slim_context(graph: GraphContext) -> dict:
+    """Transform GraphContext into a slimmed dict, stripping redundant fields.
+
+    Reduces payload size ~40% by removing fields on nested objects that
+    duplicate information already present on the parent entity (IDs,
+    timestamps, file paths).
+    """
+    slimmed_results = []
+    for result in graph.results:
+        slimmed_results.append(
+            {
+                "primary_result": _slim_summary(result.primary_result),
+                "observations": [_slim_summary(obs) for obs in result.observations],
+                "related_results": [_slim_summary(rel) for rel in result.related_results],
+            }
+        )
+
+    metadata = graph.metadata.model_dump()
+    for key in _METADATA_STRIP:
+        metadata.pop(key, None)
+
+    return {
+        "results": slimmed_results,
+        "metadata": metadata,
+        "page": graph.page,
+        "page_size": graph.page_size,
+    }
+
+
+def _format_entity_block(result: ContextResult) -> str:
+    """Format a single context result as a markdown block."""
+    primary = result.primary_result
+    lines = []
+
+    # --- Header ---
+    lines.append(f"## {primary.title}")
+    if primary.permalink:
+        lines.append(f"permalink: {primary.permalink}")
+    # RelationSummary has no content field; Entity/Observation do
+    if not isinstance(primary, RelationSummary) and primary.content:
+        lines.append("")
+        lines.append(primary.content)
+
+    # --- Observations ---
+    if result.observations:
+        lines.append("")
+        lines.append("### Observations")
+        for obs in result.observations:
+            lines.append(f"- [{obs.category}] {obs.content}")
+
+    # --- Relations (from primary's related_results that are RelationSummary) ---
+    relation_items: list[RelationSummary] = [
+        r for r in result.related_results if isinstance(r, RelationSummary)
+    ]
+    if relation_items:
+        lines.append("")
+        lines.append("### Relations")
+        for rel in relation_items:
+            lines.append(f"- {rel.relation_type} [[{rel.to_entity}]]")
+
+    # --- Related entities (non-relation related results) ---
+    related_entities: list[EntitySummary | ObservationSummary] = [
+        r for r in result.related_results if not isinstance(r, RelationSummary)
+    ]
+    if related_entities:
+        lines.append("")
+        lines.append("### Related")
+        for item in related_entities:
+            permalink = item.permalink if item.permalink else ""
+            lines.append(f"- [[{item.title}]] ({permalink})")
+
+    return "\n".join(lines)
+
+
+def _format_context_markdown(graph: GraphContext, project: str) -> str:
+    """Format GraphContext as compact markdown text.
+
+    Produces a human-readable markdown representation that is much smaller
+    than the equivalent JSON, suitable for LLM consumption when structured
+    data isn't needed.
+    """
+    if not graph.results:
+        uri = graph.metadata.uri or ""
+        return f"No results found for '{uri}' in project '{project}'."
+
+    parts = []
+
+    # --- Title from first primary result ---
+    first_title = graph.results[0].primary_result.title
+    if len(graph.results) == 1:
+        parts.append(f"# Context: {first_title}")
+    else:
+        uri = graph.metadata.uri or ""
+        parts.append(f"# Context: {uri}")
+
+    parts.append("")
+
+    # --- Entity blocks separated by --- ---
+    entity_blocks = [_format_entity_block(result) for result in graph.results]
+    parts.append("\n\n---\n\n".join(entity_blocks))
+
+    # --- Footer ---
+    meta = graph.metadata
+    primary_count = meta.primary_count or 0
+    related_count = meta.related_count or 0
+    parts.append("")
+    parts.append("---")
+    parts.append(
+        f"*{primary_count} primary, {related_count} related"
+        f" | depth={meta.depth} | project: {project}*"
+    )
+
+    return "\n".join(parts)
 
 
 @mcp.tool(
@@ -26,6 +187,10 @@ from basic_memory.schemas.memory import GraphContext, MemoryUrl
     Timeframes support natural language like:
     - "2 days ago", "last week", "today", "3 months ago"
     - Or standard formats like "7d", "24h"
+
+    Format options:
+    - "json" (default): Slimmed JSON with redundant fields removed
+    - "markdown": Compact markdown text for LLM consumption
     """,
 )
 async def build_context(
@@ -36,8 +201,9 @@ async def build_context(
     page: int = 1,
     page_size: int = 10,
     max_related: int = 10,
+    format: str = "json",
     context: Context | None = None,
-) -> GraphContext:
+) -> dict | str:
     """Get context needed to continue a discussion within a specific project.
 
     This tool enables natural continuation of discussions by loading relevant context
@@ -58,13 +224,12 @@ async def build_context(
         page: Page number of results to return (default: 1)
         page_size: Number of results to return per page (default: 10)
         max_related: Maximum number of related results to return (default: 10)
+        format: Response format - "json" for slimmed JSON dict, "markdown" for compact text
         context: Optional FastMCP context for performance caching.
 
     Returns:
-        GraphContext containing:
-            - primary_results: Content matching the memory:// URI
-            - related_results: Connected content via relations
-            - metadata: Context building details
+        dict (format="json"): Slimmed JSON with redundant fields removed
+        str (format="markdown"): Compact markdown representation
 
     Examples:
         # Continue a specific discussion
@@ -73,11 +238,8 @@ async def build_context(
         # Get deeper context about a component
         build_context("work-docs", "memory://components/memory-service", depth=2)
 
-        # Look at recent changes to a specification
-        build_context("research", "memory://specs/document-format", timeframe="today")
-
-        # Research the history of a feature
-        build_context("dev-notes", "memory://features/knowledge-graph", timeframe="3 months ago")
+        # Get markdown output for compact context
+        build_context("research", "memory://specs/search", format="markdown")
 
     Raises:
         ToolError: If project doesn't exist or depth parameter is invalid
@@ -97,16 +259,14 @@ async def build_context(
 
     async with get_project_client(project, context) as (client, active_project):
         # Resolve memory:// identifier with project-prefix awareness
-        _, resolved_path, _ = await resolve_project_and_path(
-            client, url, project, context
-        )
+        _, resolved_path, _ = await resolve_project_and_path(client, url, project, context)
 
         # Import here to avoid circular import
         from basic_memory.mcp.clients import MemoryClient
 
         # Use typed MemoryClient for API calls
         memory_client = MemoryClient(client, active_project.external_id)
-        return await memory_client.build_context(
+        graph = await memory_client.build_context(
             resolved_path,
             depth=depth or 1,
             timeframe=timeframe,
@@ -114,3 +274,8 @@ async def build_context(
             page_size=page_size,
             max_related=max_related,
         )
+
+        if format == "markdown":
+            return _format_context_markdown(graph, active_project.name)
+
+        return _slim_context(graph)

src/basic_memory/mcp/tools/read_note.py

@@ -16,7 +16,8 @@ from basic_memory.utils import validate_project_path
 
 @mcp.tool(
     description="Read a markdown note by title or permalink.",
-    meta={"ui/resourceUri": "ui://basic-memory/note-preview"},
+    # TODO: re-enable once MCP client rendering is working
+    # meta={"ui/resourceUri": "ui://basic-memory/note-preview"},
 )
 async def read_note(
     identifier: str,
@@ -83,9 +84,7 @@ async def read_note(
     """
     async with get_project_client(project, context) as (client, active_project):
         # Resolve identifier with project-prefix awareness for memory:// URLs
-        _, entity_path, _ = await resolve_project_and_path(
-            client, identifier, project, context
-        )
+        _, entity_path, _ = await resolve_project_and_path(client, identifier, project, context)
 
         # Validate identifier to prevent path traversal attacks
         # For memory:// URLs, validate the extracted path (not the raw URL which

src/basic_memory/mcp/tools/schema.py

@@ -15,11 +15,66 @@ from basic_memory.mcp.server import mcp
 from basic_memory.schemas.schema import ValidationReport, InferenceReport, DriftReport
 
 
+def _no_notes_guidance(note_type: str, tool_name: str) -> str:
+    """Build guidance string when no notes of a given type exist.
+
+    Used by schema_validate when the project has zero notes of the
+    requested type — a different situation from "notes exist but no schema".
+    """
+    return (
+        f"# No Notes Found of Type '{note_type}'\n\n"
+        f"`{tool_name}` found no notes with type '{note_type}' in the project.\n\n"
+        f"## Next Steps\n\n"
+        f"1. **Create notes of this type** — use `write_note` with "
+        f'`note_type="{note_type}"` to create notes\n'
+        f"2. **Check existing types** — use `search_notes` with `entity_types` "
+        f"filter to see what types exist\n"
+        f"3. **Browse content** — use `list_directory` or `recent_activity` to "
+        f"see what's in the project\n"
+    )
+
+
+def _no_schema_guidance(note_type: str, tool_name: str) -> str:
+    """Build guidance string when no schema exists for a note type.
+
+    Used by schema_validate and schema_diff to explain what happened
+    and how to create a schema.
+    """
+    return (
+        f"# No Schema Found for '{note_type}'\n\n"
+        f"`{tool_name}` requires a schema note to exist for type '{note_type}'.\n\n"
+        f"## How to Create a Schema\n\n"
+        f'1. **Infer from existing notes** — run `schema_infer("{note_type}")` to '
+        f"analyze your notes and get a suggested schema\n"
+        f"2. **Create a schema note** — write a markdown file with this frontmatter:\n\n"
+        f"```yaml\n"
+        f"---\n"
+        f"title: {note_type.title()}\n"
+        f"type: schema\n"
+        f"entity: {note_type}\n"
+        f"version: 1\n"
+        f"schema:\n"
+        f"  name: string, full name\n"
+        f"  role?: string, job title\n"
+        f"settings:\n"
+        f"  validation: warn\n"
+        f"---\n"
+        f"```\n\n"
+        f"Schema fields use Picoschema notation:\n"
+        f"- `field_name: type, description` — required field\n"
+        f"- `field_name?: type, description` — optional field\n"
+        f"- Supported types: `string`, `number`, `boolean`, `string[]`\n\n"
+        f"3. **Sync** — run `basic-memory sync` or wait for auto-sync to pick up "
+        f"the new schema note\n"
+        f'4. **Re-run** — call `{tool_name}("{note_type}")` again\n'
+    )
+
+
 @mcp.tool(
     description="Validate notes against their Picoschema definitions.",
 )
 async def schema_validate(
-    entity_type: Optional[str] = None,
+    note_type: Optional[str] = None,
     identifier: Optional[str] = None,
     project: Optional[str] = None,
     context: Context | None = None,
@@ -32,7 +87,7 @@ async def schema_validate(
     Schemas are resolved in priority order:
     1. Inline schema (dict in frontmatter)
     2. Explicit reference (string in frontmatter)
-    3. Implicit by type (type field matches schema note entity field)
+    3. Implicit by type (type field matches schema note's entity field)
     4. No schema (no validation)
 
     Project Resolution:
@@ -40,7 +95,7 @@ async def schema_validate(
     If project unknown, use list_memory_projects() first.
 
     Args:
-        entity_type: Entity type to batch-validate (e.g., "Person").
+        note_type: Note type to batch-validate (e.g., "person", "meeting").
             If provided, validates all notes of this type.
         identifier: Specific note to validate (permalink, title, or path).
             If provided, validates only this note.
@@ -51,20 +106,20 @@ async def schema_validate(
         ValidationReport with per-note results, or error guidance string
 
     Examples:
-        # Validate all Person notes
-        schema_validate(entity_type="Person")
+        # Validate all person notes
+        schema_validate(note_type="person")
 
         # Validate a specific note
         schema_validate(identifier="people/paul-graham")
 
         # Validate in a specific project
-        schema_validate(entity_type="Person", project="my-research")
+        schema_validate(note_type="person", project="my-research")
     """
     async with get_client() as client:
         active_project = await get_active_project(client, project, context)
         logger.info(
             f"MCP tool call tool=schema_validate project={active_project.name} "
-            f"entity_type={entity_type} identifier={identifier}"
+            f"note_type={note_type} identifier={identifier}"
         )
 
         try:
@@ -72,7 +127,7 @@ async def schema_validate(
 
             schema_client = SchemaClient(client, active_project.external_id)
             result = await schema_client.validate(
-                entity_type=entity_type,
+                entity_type=note_type,
                 identifier=identifier,
             )
 
@@ -81,6 +136,21 @@ async def schema_validate(
                 f"total={result.total_notes} valid={result.valid_count} "
                 f"warnings={result.warning_count} errors={result.error_count}"
             )
+
+            # --- No notes guard ---
+            # Trigger: no entities of this type exist in the project
+            # Why: can't validate notes that don't exist yet
+            # Outcome: return guidance on creating notes of this type
+            if note_type and result.total_entities == 0:
+                return _no_notes_guidance(note_type, "schema_validate")
+
+            # --- No schema guard ---
+            # Trigger: entities exist but none were validated (no schema found)
+            # Why: notes of this type exist but no schema was found, so none were validated
+            # Outcome: return guidance on how to create a schema
+            if note_type and result.total_notes == 0:
+                return _no_schema_guidance(note_type, "schema_validate")
+
             return result
 
         except Exception as e:
@@ -89,7 +159,7 @@ async def schema_validate(
                 f"# Schema Validation Failed\n\n"
                 f"Error validating schemas: {e}\n\n"
                 f"## Troubleshooting\n"
-                f"1. Ensure schema notes exist (type: schema) for the target entity type\n"
+                f"1. Ensure schema notes exist (type: schema) for the target note type\n"
                 f"2. Check that notes have the correct type in frontmatter\n"
                 f"3. Verify the project has been synced: `basic-memory status`\n"
             )
@@ -99,7 +169,7 @@ async def schema_validate(
     description="Analyze existing notes and suggest a Picoschema definition.",
 )
 async def schema_infer(
-    entity_type: str,
+    note_type: str,
     threshold: float = 0.25,
     project: Optional[str] = None,
     context: Context | None = None,
@@ -120,7 +190,7 @@ async def schema_infer(
     If project unknown, use list_memory_projects() first.
 
     Args:
-        entity_type: The entity type to analyze (e.g., "Person", "meeting").
+        note_type: The note type to analyze (e.g., "person", "meeting").
         threshold: Minimum frequency (0-1) for a field to be suggested as optional.
             Default 0.25 (25%). Fields above 95% become required.
         project: Project name. Optional -- server will resolve.
@@ -130,44 +200,68 @@ async def schema_infer(
         InferenceReport with frequency data and suggested schema, or error string
 
     Examples:
-        # Infer schema for Person notes
-        schema_infer("Person")
+        # Infer schema for person notes
+        schema_infer("person")
 
         # Use a higher threshold (50% minimum)
         schema_infer("meeting", threshold=0.5)
 
         # Infer in a specific project
-        schema_infer("Person", project="my-research")
+        schema_infer("person", project="my-research")
     """
     async with get_client() as client:
         active_project = await get_active_project(client, project, context)
         logger.info(
             f"MCP tool call tool=schema_infer project={active_project.name} "
-            f"entity_type={entity_type} threshold={threshold}"
+            f"note_type={note_type} threshold={threshold}"
         )
 
         try:
             from basic_memory.mcp.clients.schema import SchemaClient
 
             schema_client = SchemaClient(client, active_project.external_id)
-            result = await schema_client.infer(entity_type, threshold=threshold)
+            result = await schema_client.infer(note_type, threshold=threshold)
 
             logger.info(
                 f"MCP tool response: tool=schema_infer project={active_project.name} "
-                f"entity_type={entity_type} notes_analyzed={result.notes_analyzed} "
+                f"note_type={note_type} notes_analyzed={result.notes_analyzed} "
                 f"required={len(result.suggested_required)} "
                 f"optional={len(result.suggested_optional)}"
             )
+
+            # --- Empty schema guard ---
+            # Trigger: notes were analyzed but no fields met the threshold
+            # Why: returning hundreds of excluded fields overwhelms the LLM context
+            # Outcome: return actionable guidance instead of a massive empty result
+            if result.notes_analyzed > 0 and not result.suggested_schema:
+                return (
+                    f"# No Schema Pattern Found\n\n"
+                    f"Analyzed {result.notes_analyzed} notes of type '{note_type}', "
+                    f"but no observation or relation appeared in enough notes to suggest "
+                    f"a schema (threshold: {threshold:.0%}).\n\n"
+                    f"This usually means '{note_type}' is too broad — the notes don't "
+                    f"share a consistent structure.\n\n"
+                    f"## Suggestions\n"
+                    f"1. **Use a more specific type** — try `search_notes` with "
+                    f"`entity_types` filter to see what types exist\n"
+                    f"2. **Lower the threshold** — "
+                    f'`schema_infer("{note_type}", threshold=0.1)` to include '
+                    f"rarer fields\n"
+                    f"3. **Create typed notes** — use `write_note` with a specific "
+                    f'`note_type` (e.g., "person", "meeting") to build consistent '
+                    f"structure\n"
+                )
+
             return result
 
         except Exception as e:
             logger.error(f"Schema inference failed: {e}, project: {active_project.name}")
             return (
                 f"# Schema Inference Failed\n\n"
-                f"Error inferring schema for '{entity_type}': {e}\n\n"
+                f"Error inferring schema for type '{note_type}': {e}\n\n"
                 f"## Troubleshooting\n"
-                f"1. Ensure notes of type '{entity_type}' exist in the project\n"
-                f'2. Try searching: `search_notes("{entity_type}", types=["{entity_type}"])`\n'
+                f"1. Ensure notes of type '{note_type}' exist in the project\n"
+                f'2. Try searching: `search_notes("{note_type}", types=["{note_type}"])`\n'
                 f"3. Verify the project has been synced: `basic-memory status`\n"
             )
 
@@ -176,13 +270,13 @@ async def schema_infer(
     description="Detect drift between a schema definition and actual note usage.",
 )
 async def schema_diff(
-    entity_type: str,
+    note_type: str,
     project: Optional[str] = None,
     context: Context | None = None,
 ) -> DriftReport | str:
     """Detect drift between a schema definition and actual note usage.
 
-    Compares the existing schema for an entity type against how notes of
+    Compares the existing schema for a note type against how notes of
     that type are actually structured. Identifies new fields that have
     appeared, declared fields that are rarely used, and cardinality changes
     (single-value vs array).
@@ -195,7 +289,7 @@ async def schema_diff(
     If project unknown, use list_memory_projects() first.
 
     Args:
-        entity_type: The entity type to check for drift (e.g., "Person").
+        note_type: The note type to check for drift (e.g., "person").
         project: Project name. Optional -- server will resolve.
         context: Optional FastMCP context for performance caching.
 
@@ -204,41 +298,48 @@ async def schema_diff(
         or error guidance string
 
     Examples:
-        # Check drift for Person schema
-        schema_diff("Person")
+        # Check drift for person schema
+        schema_diff("person")
 
         # Check drift in a specific project
-        schema_diff("Person", project="my-research")
+        schema_diff("person", project="my-research")
     """
     async with get_client() as client:
         active_project = await get_active_project(client, project, context)
         logger.info(
-            f"MCP tool call tool=schema_diff project={active_project.name} "
-            f"entity_type={entity_type}"
+            f"MCP tool call tool=schema_diff project={active_project.name} note_type={note_type}"
         )
 
         try:
             from basic_memory.mcp.clients.schema import SchemaClient
 
             schema_client = SchemaClient(client, active_project.external_id)
-            result = await schema_client.diff(entity_type)
+            result = await schema_client.diff(note_type)
 
             logger.info(
                 f"MCP tool response: tool=schema_diff project={active_project.name} "
-                f"entity_type={entity_type} "
+                f"note_type={note_type} schema_found={result.schema_found} "
                 f"new_fields={len(result.new_fields)} "
                 f"dropped_fields={len(result.dropped_fields)} "
                 f"cardinality_changes={len(result.cardinality_changes)}"
             )
+
+            # --- No schema guard ---
+            # Trigger: API reports no schema was found for this type
+            # Why: diff requires a schema to compare against
+            # Outcome: return guidance on how to create a schema
+            if not result.schema_found:
+                return _no_schema_guidance(note_type, "schema_diff")
+
             return result
 
         except Exception as e:
             logger.error(f"Schema diff failed: {e}, project: {active_project.name}")
             return (
                 f"# Schema Diff Failed\n\n"
-                f"Error detecting drift for '{entity_type}': {e}\n\n"
+                f"Error detecting drift for type '{note_type}': {e}\n\n"
                 f"## Troubleshooting\n"
-                f"1. Ensure a schema note exists for entity type '{entity_type}'\n"
-                f"2. Ensure notes of type '{entity_type}' exist in the project\n"
+                f"1. Ensure a schema note exists for type '{note_type}'\n"
+                f"2. Ensure notes of type '{note_type}' exist in the project\n"
                 f"3. Verify the project has been synced: `basic-memory status`\n"
             )

src/basic_memory/mcp/tools/search.py

@@ -6,6 +6,7 @@ from typing import List, Optional, Dict, Any, Literal
 from loguru import logger
 from fastmcp import Context
 
+from basic_memory.mcp.container import get_container
 from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
 from basic_memory.mcp.formatting import format_search_results_ascii
 from basic_memory.mcp.server import mcp
@@ -231,7 +232,8 @@ Error searching for '{query}': {error_message}
 
 @mcp.tool(
     description="Search across all content in the knowledge base with advanced syntax support.",
-    meta={"ui/resourceUri": "ui://basic-memory/search-results"},
+    # TODO: re-enable once MCP client rendering is working
+    # meta={"ui/resourceUri": "ui://basic-memory/search-results"},
 )
 async def search_notes(
     query: str,
@@ -246,6 +248,7 @@ async def search_notes(
     metadata_filters: Optional[Dict[str, Any]] = None,
     tags: Optional[List[str]] = None,
     status: Optional[str] = None,
+    min_similarity: Optional[float] = None,
     context: Context | None = None,
 ) -> SearchResponse | str:
     """Search across all content in the knowledge base with comprehensive syntax support.
@@ -322,7 +325,7 @@ async def search_notes(
         page: The page number of results to return (default 1)
         page_size: The number of results to return per page (default 10)
         search_type: Type of search to perform, one of:
-                    "text", "title", "permalink", "vector", "hybrid" (default: "text")
+                    "text", "title", "permalink", "vector", "semantic", "hybrid" (default: "text")
         output_format: "default" returns structured data, "ascii" returns a plain text table,
             "ansi" returns a colorized table for TUI clients.
         types: Optional list of note types to search (e.g., ["note", "person"])
@@ -331,6 +334,9 @@ async def search_notes(
         metadata_filters: Optional structured frontmatter filters (e.g., {"status": "in-progress"})
         tags: Optional tag filter (frontmatter tags); shorthand for metadata_filters["tags"]
         status: Optional status filter (frontmatter status); shorthand for metadata_filters["status"]
+        min_similarity: Optional float to override the global semantic_min_similarity threshold
+                       for this query. E.g., 0.0 to see all vector results, or 0.8 for high precision.
+                       Only applies to vector and hybrid search types.
         context: Optional FastMCP context for performance caching.
 
     Returns:
@@ -408,44 +414,57 @@ async def search_notes(
             query = resolved_query
             search_type = "permalink"
 
-        # Create a SearchQuery object based on the parameters
-        search_query = SearchQuery()
-
-        # Set the appropriate search field based on search_type
-        if search_type == "text":
-            search_query.text = query
-        elif search_type == "vector":
-            search_query.text = query
-            search_query.retrieval_mode = SearchRetrievalMode.VECTOR
-        elif search_type == "hybrid":
-            search_query.text = query
-            search_query.retrieval_mode = SearchRetrievalMode.HYBRID
-        elif search_type == "title":
-            search_query.title = query
-        elif search_type == "permalink" and "*" in query:
-            search_query.permalink_match = query
-        elif search_type == "permalink":
-            search_query.permalink = query
-        else:  # pragma: no cover
-            search_query.text = query  # Default to text search
-
-        # Add optional filters if provided (empty lists are treated as no filter)
-        if entity_types:
-            search_query.entity_types = [SearchItemType(t) for t in entity_types]
-        if types:
-            search_query.types = types
-        if after_date:
-            search_query.after_date = after_date
-        if metadata_filters:
-            search_query.metadata_filters = metadata_filters
-        if tags:
-            search_query.tags = tags
-        if status:
-            search_query.status = status
-
-        logger.info(f"Searching for {search_query} in project {active_project.name}")
-
         try:
+            # Create a SearchQuery object based on the parameters
+            search_query = SearchQuery()
+
+            # Map search_type to the appropriate query field and retrieval mode
+            valid_search_types = {"text", "title", "permalink", "vector", "semantic", "hybrid"}
+            if search_type == "text":
+                search_query.text = query
+                # Upgrade to hybrid when semantic search is available —
+                # combines FTS keyword matching with vector similarity for better results
+                try:
+                    container = get_container()
+                    if container.config.semantic_search_enabled:
+                        search_query.retrieval_mode = SearchRetrievalMode.HYBRID
+                except RuntimeError:
+                    pass  # Container not initialized (e.g., CLI context) — stay with FTS
+            elif search_type in ("vector", "semantic"):
+                search_query.text = query
+                search_query.retrieval_mode = SearchRetrievalMode.VECTOR
+            elif search_type == "hybrid":
+                search_query.text = query
+                search_query.retrieval_mode = SearchRetrievalMode.HYBRID
+            elif search_type == "title":
+                search_query.title = query
+            elif search_type == "permalink" and "*" in query:
+                search_query.permalink_match = query
+            elif search_type == "permalink":
+                search_query.permalink = query
+            else:
+                raise ValueError(
+                    f"Invalid search_type '{search_type}'. "
+                    f"Valid options: {', '.join(sorted(valid_search_types))}"
+                )
+
+            # Add optional filters if provided (empty lists are treated as no filter)
+            if entity_types:
+                search_query.entity_types = [SearchItemType(t) for t in entity_types]
+            if types:
+                search_query.types = types
+            if after_date:
+                search_query.after_date = after_date
+            if metadata_filters:
+                search_query.metadata_filters = metadata_filters
+            if tags:
+                search_query.tags = tags
+            if status:
+                search_query.status = status
+            if min_similarity is not None:
+                search_query.min_similarity = min_similarity
+
+            logger.info(f"Searching for {search_query} in project {active_project.name}")
             # Import here to avoid circular import (tools → clients → utils → tools)
             from basic_memory.mcp.clients import SearchClient
 

src/basic_memory/mcp/tools/write_note.py

@@ -24,6 +24,7 @@ async def write_note(
     project: Optional[str] = None,
     tags: list[str] | str | None = None,
     note_type: str = "note",
+    metadata: dict | None = None,
     context: Context | None = None,
 ) -> str:
     """Write a markdown note to the knowledge base.
@@ -67,6 +68,9 @@ async def write_note(
               Note: If passing from external MCP clients, use a string format (e.g. "tag1,tag2,tag3")
         note_type: Type of note to create (stored in frontmatter). Defaults to "note".
                    Can be "guide", "report", "config", "person", etc.
+        metadata: Optional dict of extra frontmatter fields merged into entity_metadata.
+                  Useful for schema notes or any note that needs custom YAML frontmatter
+                  beyond title/type/tags. Nested dicts are supported.
         context: Optional FastMCP context for performance caching.
 
     Returns:
@@ -105,6 +109,20 @@ async def write_note(
             content="# Weekly Standup\\n\\n- [decision] Use PostgreSQL instead #tech"
         )
 
+        # Create a schema note with custom frontmatter via metadata
+        write_note(
+            title="Person",
+            directory="schemas",
+            note_type="schema",
+            content="# Person\\n\\nSchema for person entities.",
+            metadata={
+                "entity": "person",
+                "version": 1,
+                "schema": {"name": "string", "role?": "string"},
+                "settings": {"validation": "warn"},
+            },
+        )
+
     Raises:
         HTTPError: If project doesn't exist or is inaccessible
         SecurityError: If directory path attempts path traversal
@@ -130,15 +148,22 @@ async def write_note(
 
         # Process tags using the helper function
         tag_list = parse_tags(tags)
-        # Create the entity request
-        metadata = {"tags": tag_list} if tag_list else None
+
+        # Build entity_metadata from optional metadata, then explicit tags on top
+        # Order matters: explicit tags parameter takes precedence over metadata["tags"]
+        entity_metadata = {}
+        if metadata:
+            entity_metadata.update(metadata)
+        if tag_list:
+            entity_metadata["tags"] = tag_list
+
         entity = Entity(
             title=title,
             directory=directory,
             entity_type=note_type,
             content_type="text/markdown",
             content=content,
-            entity_metadata=metadata,
+            entity_metadata=entity_metadata or None,
         )
 
         # Import here to avoid circular import

src/basic_memory/project_resolver.py

@@ -8,7 +8,7 @@ identically in both local and cloud modes:
 
 1. ENV_CONSTRAINT: BASIC_MEMORY_MCP_PROJECT env var (highest priority)
 2. EXPLICIT: Project passed directly to operation
-3. DEFAULT: Default project when default_project_mode=true
+3. DEFAULT: default_project from config (if set)
 4. Fallback: cloud → CLOUD_DISCOVERY or ValueError; local → NONE
 """
 
@@ -27,7 +27,7 @@ class ResolutionMode(Enum):
     CLOUD_DISCOVERY = auto()  # Discovery mode allowed in cloud (no project)
     ENV_CONSTRAINT = auto()  # BASIC_MEMORY_MCP_PROJECT env var
     EXPLICIT = auto()  # Explicit project parameter
-    DEFAULT = auto()  # default_project with default_project_mode=true
+    DEFAULT = auto()  # default_project from config
     NONE = auto()  # No resolution possible
 
 
@@ -70,14 +70,12 @@ class ProjectResolver:
 
     Args:
         cloud_mode: Whether running in cloud mode
-        default_project_mode: Whether to use default project when not specified
-        default_project: The default project name
+        default_project: The default project name (used as fallback when set)
         constrained_project: Optional env-constrained project override
             (typically from BASIC_MEMORY_MCP_PROJECT)
     """
 
     cloud_mode: bool = False
-    default_project_mode: bool = False
     default_project: Optional[str] = None
     constrained_project: Optional[str] = None
 
@@ -85,14 +83,12 @@ class ProjectResolver:
     def from_env(
         cls,
         cloud_mode: bool = False,
-        default_project_mode: bool = False,
         default_project: Optional[str] = None,
     ) -> "ProjectResolver":
         """Create resolver with constrained_project from environment.
 
         Args:
             cloud_mode: Whether running in cloud mode
-            default_project_mode: Whether to use default project when not specified
             default_project: The default project name
 
         Returns:
@@ -101,7 +97,6 @@ class ProjectResolver:
         constrained = os.environ.get("BASIC_MEMORY_MCP_PROJECT")
         return cls(
             cloud_mode=cloud_mode,
-            default_project_mode=default_project_mode,
             default_project=default_project,
             constrained_project=constrained,
         )
@@ -116,7 +111,7 @@ class ProjectResolver:
         The same resolution order applies in both local and cloud modes:
         1. ENV_CONSTRAINT — BASIC_MEMORY_MCP_PROJECT env var (highest priority)
         2. EXPLICIT — project parameter passed directly
-        3. DEFAULT — default project when default_project_mode=true
+        3. DEFAULT — default_project from config (if set)
         4. Fallback — cloud: CLOUD_DISCOVERY or ValueError; local: NONE
 
         Args:
@@ -150,13 +145,13 @@ class ProjectResolver:
                 reason=f"Explicit parameter: {project}",
             )
 
-        # --- Priority 3: Default project mode ---
-        if self.default_project_mode and self.default_project:
+        # --- Priority 3: Default project from config ---
+        if self.default_project:
             logger.debug(f"Using default project from config: {self.default_project}")
             return ResolvedProject(
                 project=self.default_project,
                 mode=ResolutionMode.DEFAULT,
-                reason=f"Default project mode: {self.default_project}",
+                reason=f"Default project: {self.default_project}",
             )
 
         # --- Fallback: mode-dependent behavior ---
@@ -168,7 +163,7 @@ class ProjectResolver:
                     mode=ResolutionMode.CLOUD_DISCOVERY,
                     reason="Discovery mode enabled in cloud",
                 )
-            raise ValueError("No project specified. Project is required for cloud mode.")
+            raise ValueError("No project specified. Project is required.")
 
         # Local mode: no resolution possible
         logger.debug("No project resolution possible")
@@ -200,7 +195,7 @@ class ProjectResolver:
         result = self.resolve(project, allow_discovery=False)
         if not result.is_resolved:
             msg = error_message or (
-                "No project specified. Either set 'default_project_mode=true' in config, "
+                "No project specified. Either set 'default_project' in config, "
                 "or provide a 'project' argument."
             )
             raise ValueError(msg)

src/basic_memory/repository/postgres_search_repository.py

@@ -51,6 +51,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
         self._app_config = app_config or ConfigManager().config
         self._semantic_enabled = self._app_config.semantic_search_enabled
         self._semantic_vector_k = self._app_config.semantic_vector_k
+        self._semantic_min_similarity = self._app_config.semantic_min_similarity
         self._embedding_provider = embedding_provider
         self._vector_dimensions = 384
         self._vector_tables_initialized = False
@@ -64,17 +65,16 @@ class PostgresSearchRepository(SearchRepositoryBase):
     async def init_search_index(self):
         """Create Postgres table with tsvector column and GIN indexes.
 
-        Note: This is handled by Alembic migrations. This method is a no-op
-        for Postgres as the schema is created via migrations.
+        Note: FTS schema is handled by Alembic migrations. Vector tables are
+        created here at startup so missing pgvector or provider errors surface
+        immediately.
         """
         logger.info("PostgreSQL search index initialization handled by migrations")
-        # Table creation is done via Alembic migrations
-        # This includes:
-        # - CREATE TABLE search_index (...)
-        # - ADD COLUMN textsearchable_index_col tsvector GENERATED ALWAYS AS (...)
-        # - CREATE INDEX USING GIN on textsearchable_index_col
-        # - CREATE INDEX USING GIN on metadata jsonb_path_ops
-        pass
+
+        # Fail fast: create vector tables at startup so missing pgvector
+        # or embedding provider errors surface immediately
+        if self._semantic_enabled:
+            await self._ensure_vector_tables()
 
     async def index_item(self, search_index_row: SearchIndexRow) -> None:
         """Index or update a single item using UPSERT.
@@ -260,6 +260,8 @@ class PostgresSearchRepository(SearchRepositoryBase):
         if self._vector_tables_initialized:
             return
 
+        logger.info("Ensuring Postgres vector tables exist for semantic search")
+
         async with self._vector_tables_lock:
             if self._vector_tables_initialized:
                 return
@@ -349,6 +351,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
                 )
                 await session.commit()
 
+            logger.info(f"Postgres vector tables ready (dimensions={self._vector_dimensions})")
             self._vector_tables_initialized = True
 
     async def _get_existing_embedding_dims(self, session: AsyncSession) -> int | None:
@@ -587,6 +590,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
         search_item_types: Optional[List[SearchItemType]] = None,
         metadata_filters: Optional[dict] = None,
         retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS,
+        min_similarity: Optional[float] = None,
         limit: int = 10,
         offset: int = 0,
     ) -> List[SearchIndexRow]:
@@ -602,6 +606,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
             search_item_types=search_item_types,
             metadata_filters=metadata_filters,
             retrieval_mode=retrieval_mode,
+            min_similarity=min_similarity,
             limit=limit,
             offset=offset,
         )

src/basic_memory/repository/search_repository.py

@@ -42,6 +42,7 @@ class SearchRepository(Protocol):
         search_item_types: Optional[List[SearchItemType]] = None,
         metadata_filters: Optional[dict] = None,
         retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS,
+        min_similarity: Optional[float] = None,
         limit: int = 10,
         offset: int = 0,
     ) -> List[SearchIndexRow]:

src/basic_memory/repository/search_repository_base.py

@@ -49,6 +49,7 @@ class SearchRepositoryBase(ABC):
     # --- Subclass-populated attributes ---
     _semantic_enabled: bool
     _semantic_vector_k: int
+    _semantic_min_similarity: float
     _embedding_provider: Optional[EmbeddingProvider]
     _vector_dimensions: int
     _vector_tables_initialized: bool
@@ -112,6 +113,7 @@ class SearchRepositoryBase(ABC):
         search_item_types: Optional[List[SearchItemType]] = None,
         metadata_filters: Optional[Dict[str, Any]] = None,
         retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS,
+        min_similarity: Optional[float] = None,
         limit: int = 10,
         offset: int = 0,
     ) -> List[SearchIndexRow]:
@@ -753,6 +755,7 @@ class SearchRepositoryBase(ABC):
         search_item_types: Optional[List[SearchItemType]],
         metadata_filters: Optional[dict],
         retrieval_mode: SearchRetrievalMode,
+        min_similarity: Optional[float] = None,
         limit: int,
         offset: int,
     ) -> Optional[List[SearchIndexRow]]:
@@ -784,6 +787,7 @@ class SearchRepositoryBase(ABC):
                 after_date=after_date,
                 search_item_types=search_item_types,
                 metadata_filters=metadata_filters,
+                min_similarity=min_similarity,
                 limit=limit,
                 offset=offset,
             )
@@ -802,6 +806,7 @@ class SearchRepositoryBase(ABC):
                 after_date=after_date,
                 search_item_types=search_item_types,
                 metadata_filters=metadata_filters,
+                min_similarity=min_similarity,
                 limit=limit,
                 offset=offset,
             )
@@ -830,6 +835,7 @@ class SearchRepositoryBase(ABC):
         after_date: Optional[datetime],
         search_item_types: Optional[List[SearchItemType]],
         metadata_filters: Optional[dict],
+        min_similarity: Optional[float] = None,
         limit: int,
         offset: int,
     ) -> List[SearchIndexRow]:
@@ -843,7 +849,7 @@ class SearchRepositoryBase(ABC):
         await self._ensure_vector_tables()
         assert self._embedding_provider is not None
         query_embedding = await self._embedding_provider.embed_query(search_text.strip())
-        candidate_limit = max(self._semantic_vector_k, (limit + offset) * 5)
+        candidate_limit = max(self._semantic_vector_k, (limit + offset) * 10)
 
         async with db.scoped_session(self.session_maker) as session:
             await self._prepare_vector_session(session)
@@ -872,6 +878,18 @@ class SearchRepositoryBase(ABC):
         if not similarity_by_si_id:
             return []
 
+        # Filter out results below the minimum similarity threshold.
+        # Per-query min_similarity overrides the instance-level default.
+        effective_min_similarity = (
+            min_similarity if min_similarity is not None else self._semantic_min_similarity
+        )
+        if effective_min_similarity > 0.0:
+            similarity_by_si_id = {
+                k: v for k, v in similarity_by_si_id.items() if v >= effective_min_similarity
+            }
+            if not similarity_by_si_id:
+                return []
+
         # Fetch the actual search_index rows
         si_ids = list(similarity_by_si_id.keys())
         search_index_rows = await self._fetch_search_index_rows_by_ids(si_ids)
@@ -1029,6 +1047,7 @@ class SearchRepositoryBase(ABC):
         after_date: Optional[datetime],
         search_item_types: Optional[List[SearchItemType]],
         metadata_filters: Optional[dict],
+        min_similarity: Optional[float] = None,
         limit: int,
         offset: int,
     ) -> List[SearchIndexRow]:
@@ -1061,26 +1080,39 @@ class SearchRepositoryBase(ABC):
             after_date=after_date,
             search_item_types=search_item_types,
             metadata_filters=metadata_filters,
+            min_similarity=min_similarity,
             limit=candidate_limit,
             offset=0,
         )
 
-        # RRF fusion keyed on search_index row id for granular results.
-        # This allows observations and relations to surface as individual results,
-        # not collapsed into their parent entity.
+        # Score-weighted RRF fusion keyed on search_index row id.
+        # Multiplies the standard 1/(k+rank) score by the normalized original score
+        # so that high-confidence matches contribute more than weak ones at the same rank.
         fused_scores: dict[int, float] = {}
         rows_by_id: dict[int, SearchIndexRow] = {}
 
+        # Normalize FTS scores to [0, 1] — handles both SQLite (negative bm25)
+        # and Postgres (positive ts_rank) by using absolute values
+        fts_abs = [abs(row.score or 0.0) for row in fts_results]
+        fts_max = max(fts_abs) if fts_abs else 1.0
+
         for rank, row in enumerate(fts_results, start=1):
             if row.id is None:
                 continue
-            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (1.0 / (RRF_K + rank))
+            norm = abs(row.score or 0.0) / fts_max if fts_max > 0 else 0.0
+            weight = max(norm, 0.1)  # floor preserves RRF stability
+            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + weight * (1.0 / (RRF_K + rank))
             rows_by_id[row.id] = row
 
+        # Vector scores already in [0, 1] from the similarity formula
+        vec_max = max((row.score or 0.0) for row in vector_results) if vector_results else 1.0
+
         for rank, row in enumerate(vector_results, start=1):
             if row.id is None:
                 continue
-            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (1.0 / (RRF_K + rank))
+            norm = (row.score or 0.0) / vec_max if vec_max > 0 else 0.0
+            weight = max(norm, 0.1)  # floor preserves RRF stability
+            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + weight * (1.0 / (RRF_K + rank))
             rows_by_id[row.id] = row
 
         ranked = sorted(fused_scores.items(), key=lambda item: item[1], reverse=True)

src/basic_memory/repository/sqlite_search_repository.py

@@ -51,6 +51,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
         self._app_config = app_config or ConfigManager().config
         self._semantic_enabled = self._app_config.semantic_search_enabled
         self._semantic_vector_k = self._app_config.semantic_vector_k
+        self._semantic_min_similarity = self._app_config.semantic_min_similarity
         self._embedding_provider = embedding_provider
         self._sqlite_vec_lock = asyncio.Lock()
         self._vector_tables_initialized = False
@@ -72,7 +73,8 @@ class SQLiteSearchRepository(SearchRepositoryBase):
         """Create FTS5 virtual table for search if it doesn't exist.
 
         Uses CREATE VIRTUAL TABLE IF NOT EXISTS to preserve existing indexed data
-        across server restarts.
+        across server restarts. Also creates vector tables when semantic search
+        is enabled so missing dependencies are caught at startup, not first query.
         """
         logger.info("Initializing SQLite FTS5 search index")
         try:
@@ -84,6 +86,11 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             logger.error(f"Error initializing search index: {e}")
             raise e
 
+        # Fail fast: create vector tables at startup so missing sqlite-vec
+        # or embedding provider errors surface immediately
+        if self._semantic_enabled:
+            await self._ensure_vector_tables()
+
     # ------------------------------------------------------------------
     # FTS5 query preparation (backend-specific)
     # ------------------------------------------------------------------
@@ -367,6 +374,8 @@ class SQLiteSearchRepository(SearchRepositoryBase):
         if self._vector_tables_initialized:
             return
 
+        logger.info("Ensuring SQLite vector tables exist for semantic search")
+
         async with db.scoped_session(self.session_maker) as session:
             await self._ensure_sqlite_vec_loaded(session)
 
@@ -386,6 +395,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             }
             schema_mismatch = bool(chunks_columns) and set(chunks_columns) != expected_columns
             if schema_mismatch:
+                logger.warning("search_vector_chunks schema mismatch, recreating vector tables")
                 await session.execute(text("DROP TABLE IF EXISTS search_vector_embeddings"))
                 await session.execute(text("DROP TABLE IF EXISTS search_vector_chunks"))
 
@@ -408,11 +418,16 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             expected_dimension_sql = f"float[{self._vector_dimensions}]"
 
             if vector_sql and expected_dimension_sql not in vector_sql:
+                logger.warning(
+                    f"Embedding dimension mismatch (expected {self._vector_dimensions}), "
+                    "recreating search_vector_embeddings"
+                )
                 await session.execute(text("DROP TABLE IF EXISTS search_vector_embeddings"))
 
             await session.execute(create_sqlite_search_vector_embeddings(self._vector_dimensions))
             await session.commit()
 
+        logger.info(f"SQLite vector tables ready (dimensions={self._vector_dimensions})")
         self._vector_tables_initialized = True
 
     async def _prepare_vector_session(self, session: AsyncSession) -> None:
@@ -566,6 +581,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
         search_item_types: Optional[List[SearchItemType]] = None,
         metadata_filters: Optional[dict] = None,
         retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS,
+        min_similarity: Optional[float] = None,
         limit: int = 10,
         offset: int = 0,
     ) -> List[SearchIndexRow]:
@@ -581,6 +597,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             search_item_types=search_item_types,
             metadata_filters=metadata_filters,
             retrieval_mode=retrieval_mode,
+            min_similarity=min_similarity,
             limit=limit,
             offset=offset,
         )

src/basic_memory/schemas/project_info.py

@@ -88,7 +88,7 @@ class ProjectInfoResponse(BaseModel):
     available_projects: Dict[str, Dict[str, Any]] = Field(
         description="Map of configured project names to detailed project information"
     )
-    default_project: str = Field(description="Name of the default project")
+    default_project: Optional[str] = Field(description="Name of the default project")
 
     # Statistics
     statistics: ProjectStatistics = Field(description="Statistics about the knowledge base")
@@ -196,7 +196,7 @@ class ProjectList(BaseModel):
     """Response model for listing projects."""
 
     projects: List[ProjectItem]
-    default_project: str
+    default_project: Optional[str]
 
 
 class ProjectStatusResponse(BaseModel):

src/basic_memory/schemas/schema.py

@@ -47,6 +47,7 @@ class ValidationReport(BaseModel):
 
     entity_type: str | None = None
     total_notes: int = 0
+    total_entities: int = 0
     valid_count: int = 0
     warning_count: int = 0
     error_count: int = 0
@@ -110,6 +111,10 @@ class DriftReport(BaseModel):
     """Schema drift analysis comparing schema definition to actual usage."""
 
     entity_type: str
+    schema_found: bool = Field(
+        default=True,
+        description="Whether a schema was found for this type",
+    )
     new_fields: list[DriftFieldResponse] = Field(
         default_factory=list,
         description="Fields common in notes but not in schema",

src/basic_memory/schemas/search.py

@@ -68,6 +68,7 @@ class SearchQuery(BaseModel):
     tags: Optional[List[str]] = None  # Convenience tag filter
     status: Optional[str] = None  # Convenience status filter
     retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS
+    min_similarity: Optional[float] = None  # Per-query override for semantic_min_similarity
 
     @field_validator("after_date")
     @classmethod

src/basic_memory/services/initialization.py

@@ -13,7 +13,7 @@ from pathlib import Path
 from loguru import logger
 
 from basic_memory import db
-from basic_memory.config import BasicMemoryConfig, ProjectMode
+from basic_memory.config import BasicMemoryConfig, DatabaseBackend, ProjectMode
 from basic_memory.models import Project
 from basic_memory.repository import (
     ProjectRepository,
@@ -174,9 +174,13 @@ async def initialize_app(
     Args:
         app_config: The Basic Memory project configuration
     """
-    # Skip initialization in cloud mode - cloud manages its own projects
-    if app_config.cloud_mode_enabled:
-        logger.debug("Skipping initialization in cloud mode - projects managed by cloud")
+    # Trigger: database backend is Postgres (cloud deployment)
+    # Why: cloud deployments manage their own projects and migrations via the cloud platform.
+    # The local MCP server always uses SQLite and needs initialization even when
+    # cloud_mode is enabled (for per-project cloud routing).
+    # Outcome: skip initialization only for actual cloud Postgres deployments.
+    if app_config.database_backend == DatabaseBackend.POSTGRES:
+        logger.info("Skipping local initialization - Postgres backend manages its own schema")
         return
 
     logger.info("Initializing app...")
@@ -186,7 +190,7 @@ async def initialize_app(
     # Reconcile projects from config.json with projects table
     await reconcile_projects_with_config(app_config)
 
-    logger.info("App initialization completed (migration running in background if needed)")
+    logger.info("App initialization completed")
 
 
 def ensure_initialization(app_config: BasicMemoryConfig) -> None:
@@ -195,14 +199,13 @@ def ensure_initialization(app_config: BasicMemoryConfig) -> None:
     This is a wrapper for the async initialize_app function that can be
     called from synchronous code like CLI entry points.
 
-    No-op if app_config.cloud_mode == True. Cloud basic memory manages it's own projects
+    No-op if database backend is Postgres (cloud deployment manages its own schema).
 
     Args:
         app_config: The Basic Memory project configuration
     """
-    # Skip initialization in cloud mode - cloud manages its own projects
-    if app_config.cloud_mode_enabled:
-        logger.debug("Skipping initialization in cloud mode - projects managed by cloud")
+    if app_config.database_backend == DatabaseBackend.POSTGRES:
+        logger.info("Skipping local initialization - Postgres backend manages its own schema")
         return
 
     async def _init_and_cleanup():

src/basic_memory/services/link_resolver.py

@@ -340,7 +340,9 @@ class LinkResolver:
         if not project:
             project = await self._project_repository.get_by_name_case_insensitive(identifier)
         if not project:
-            project = await self._project_repository.get_by_permalink(generate_permalink(identifier))
+            project = await self._project_repository.get_by_permalink(
+                generate_permalink(identifier)
+            )
 
         if project:
             self._project_cache_by_identifier[cache_key] = project

src/basic_memory/services/project_service.py

@@ -20,7 +20,13 @@ from basic_memory.schemas import (
     ProjectStatistics,
     SystemStatus,
 )
-from basic_memory.config import WATCH_STATUS_JSON, ConfigManager, get_project_config, ProjectConfig
+from basic_memory.config import (
+    WATCH_STATUS_JSON,
+    ConfigManager,
+    ProjectEntry,
+    get_project_config,
+    ProjectConfig,
+)
 from basic_memory.utils import generate_permalink
 
 
@@ -62,20 +68,20 @@ class ProjectService:
         return self.config_manager.projects
 
     @property
-    def default_project(self) -> str:
+    def default_project(self) -> Optional[str]:
         """Get the name of the default project.
 
         Returns:
-            The name of the default project
+            The name of the default project, or None if not set
         """
         return self.config_manager.default_project
 
     @property
-    def current_project(self) -> str:
+    def current_project(self) -> Optional[str]:
         """Get the name of the currently active project.
 
         Returns:
-            The name of the current project
+            The name of the current project, or None if not set
         """
         return os.environ.get("BASIC_MEMORY_PROJECT", self.config_manager.default_project)
 
@@ -340,7 +346,9 @@ class ProjectService:
             # No default project - set the config default as default
             # This is defensive code for edge cases where no default exists
             config_default = self.config_manager.default_project  # pragma: no cover
-            config_project = await self.repository.get_by_name(config_default)  # pragma: no cover
+            config_project = (
+                await self.repository.get_by_name(config_default) if config_default else None
+            )  # pragma: no cover
             if config_project:  # pragma: no cover
                 await self.repository.set_as_default(config_project.id)  # pragma: no cover
                 logger.info(
@@ -364,11 +372,12 @@ class ProjectService:
         db_projects_by_permalink = {p.permalink: p for p in db_projects}
 
         # Get all projects from configuration and normalize names if needed
-        config_projects = self.config_manager.projects.copy()
-        updated_config = {}
+        # Use .config property (not load_config()) so tests can patch ConfigManager.config
+        config = self.config_manager.config
+        updated_config: Dict[str, ProjectEntry] = {}
         config_updated = False
 
-        for name, path in config_projects.items():
+        for name, entry in config.projects.items():
             # Generate normalized name (what the database expects)
             normalized_name = generate_permalink(name)
 
@@ -376,25 +385,24 @@ class ProjectService:
                 logger.info(f"Normalizing project name in config: '{name}' -> '{normalized_name}'")
                 config_updated = True
 
-            updated_config[normalized_name] = path
+            updated_config[normalized_name] = entry
 
         # Update the configuration if any changes were made
         if config_updated:
-            config = self.config_manager.load_config()
             config.projects = updated_config
             self.config_manager.save_config(config)
             logger.info("Config updated with normalized project names")
 
-        # Use the normalized config for further processing
-        config_projects = updated_config
+        # Use the normalized config for further processing — keys are now project names
+        config_project_names = updated_config
 
         # Add projects that exist in config but not in DB
-        for name, path in config_projects.items():
+        for name, entry in config_project_names.items():
             if name not in db_projects_by_permalink:
                 logger.info(f"Adding project '{name}' to database")
                 project_data = {
                     "name": name,
-                    "path": path,
+                    "path": entry.path,
                     "permalink": generate_permalink(name),
                     "is_active": True,
                     # Don't set is_default here - let the enforcement logic handle it
@@ -405,7 +413,7 @@ class ProjectService:
         # Config is the source of truth - if a project was deleted from config,
         # it should be deleted from DB too (fixes issue #193)
         for name, project in db_projects_by_permalink.items():
-            if name not in config_projects:
+            if name not in config_project_names:
                 logger.info(
                     f"Removing project '{name}' from database (deleted from config, source of truth)"
                 )
@@ -456,8 +464,8 @@ class ProjectService:
 
         # Update in configuration
         config = self.config_manager.load_config()
-        old_path = config.projects[name]
-        config.projects[name] = resolved_path
+        old_path = config.projects[name].path
+        config.projects[name].path = resolved_path
         self.config_manager.save_config(config)
 
         # Update in database using robust lookup
@@ -468,7 +476,7 @@ class ProjectService:
         else:
             logger.error(f"Project '{name}' exists in config but not in database")
             # Restore the old path in config since DB update failed
-            config.projects[name] = old_path
+            config.projects[name].path = old_path
             self.config_manager.save_config(config)
             raise ValueError(f"Project '{name}' not found in database")
 
@@ -504,7 +512,7 @@ class ProjectService:
 
             # Update in config
             config = self.config_manager.load_config()
-            config.projects[name] = resolved_path
+            config.projects[name].path = resolved_path
             self.config_manager.save_config(config)
 
             # Update in database

src/basic_memory/services/search_service.py

@@ -135,6 +135,7 @@ class SearchService:
             after_date=after_date,
             metadata_filters=metadata_filters,
             retrieval_mode=query.retrieval_mode or SearchRetrievalMode.FTS,
+            min_similarity=query.min_similarity,
             limit=limit,
             offset=offset,
         )