perf(core): streamline entity write upserts

Signed-off-by: phernandez <paul@basicmachines.co>

.github/workflows/test.yml

@@ -6,7 +6,6 @@ concurrency:
 
 on:
   push:
-    branches: [ "main" ]
   pull_request:
     branches: [ "main" ]
 
@@ -52,7 +51,6 @@ jobs:
   test-sqlite-unit:
     name: Test SQLite Unit (${{ matrix.os }}, Python ${{ matrix.python-version }})
     timeout-minutes: 30
-    needs: [static-checks]
     strategy:
       fail-fast: false
       matrix:
@@ -99,7 +97,6 @@ jobs:
   test-sqlite-integration:
     name: Test SQLite Integration (${{ matrix.os }}, Python ${{ matrix.python-version }})
     timeout-minutes: 45
-    needs: [static-checks]
     strategy:
       fail-fast: false
       matrix:
@@ -146,7 +143,6 @@ jobs:
   test-postgres-unit:
     name: Test Postgres Unit (Python ${{ matrix.python-version }})
     timeout-minutes: 30
-    needs: [static-checks]
     strategy:
       fail-fast: false
       matrix:
@@ -155,8 +151,22 @@ jobs:
           - python-version: "3.13"
           - python-version: "3.14"
     runs-on: ubuntu-latest
-
-    # Note: No services section needed - testcontainers handles Postgres in Docker
+    services:
+      postgres:
+        image: pgvector/pgvector:pg16
+        env:
+          POSTGRES_USER: basic_memory_user
+          POSTGRES_PASSWORD: dev_password
+          POSTGRES_DB: basic_memory_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U basic_memory_user -d basic_memory_test"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      BASIC_MEMORY_TEST_POSTGRES_URL: postgresql://basic_memory_user:dev_password@127.0.0.1:5432/basic_memory_test
 
     steps:
       - uses: actions/checkout@v4
@@ -190,7 +200,6 @@ jobs:
   test-postgres-integration:
     name: Test Postgres Integration (Python ${{ matrix.python-version }})
     timeout-minutes: 45
-    needs: [static-checks]
     strategy:
       fail-fast: false
       matrix:
@@ -199,8 +208,22 @@ jobs:
           - python-version: "3.13"
           - python-version: "3.14"
     runs-on: ubuntu-latest
-
-    # Note: No services section needed - testcontainers handles Postgres in Docker
+    services:
+      postgres:
+        image: pgvector/pgvector:pg16
+        env:
+          POSTGRES_USER: basic_memory_user
+          POSTGRES_PASSWORD: dev_password
+          POSTGRES_DB: basic_memory_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U basic_memory_user -d basic_memory_test"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      BASIC_MEMORY_TEST_POSTGRES_URL: postgresql://basic_memory_user:dev_password@127.0.0.1:5432/basic_memory_test
 
     steps:
       - uses: actions/checkout@v4
@@ -234,7 +257,6 @@ jobs:
   test-semantic:
     name: Test Semantic (Python 3.12)
     timeout-minutes: 45
-    needs: [static-checks]
     runs-on: ubuntu-latest
 
     steps:

AGENTS.md

@@ -442,5 +442,9 @@ With GitHub integration, the development workflow includes:
 3. **Branch management** - Claude can create feature branches for implementations
 4. **Documentation maintenance** - Claude can keep documentation updated as the code evolves
 5. **Code Commits**: ALWAYS sign off commits with `git commit -s`
+6. **Pull Request Titles**: PR titles must follow the semantic format enforced by `.github/workflows/pr-title.yml`: `type(scope): summary`
+   - Allowed types: `feat`, `fix`, `chore`, `docs`, `style`, `refactor`, `perf`, `test`, `build`, `ci`
+   - Allowed scopes: `core`, `cli`, `api`, `mcp`, `sync`, `ui`, `deps`, `installer`
+   - Example: `fix(cli): propagate cloud workspace routing`
 
 This level of integration represents a new paradigm in AI-human collaboration, where the AI assistant becomes a full-fledged team member rather than just a tool for generating code snippets.

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

@@ -306,8 +306,13 @@ async def create_entity(
         ):
             if fast:
                 entity = await entity_service.fast_write_entity(data)
+                written_content = None
+                search_content = None
             else:
-                entity = await entity_service.create_entity(data)
+                write_result = await entity_service.create_entity_with_content(data)
+                entity = write_result.entity
+                written_content = write_result.content
+                search_content = write_result.search_content
 
         if fast:
             with telemetry.scope(
@@ -329,7 +334,7 @@ async def create_entity(
                 action="create_entity",
                 phase="search_index",
             ):
-                await search_service.index_entity(entity)
+                await search_service.index_entity(entity, content=search_content)
             with telemetry.scope(
                 "api.knowledge.create_entity.vector_sync",
                 domain="knowledge",
@@ -352,8 +357,15 @@ async def create_entity(
             domain="knowledge",
             action="create_entity",
             phase="read_content",
+            source="file" if fast else "memory",
         ):
-            content = await file_service.read_file_content(entity.file_path)
+            if fast:
+                content = await file_service.read_file_content(entity.file_path)
+            else:
+                # Non-fast writes already captured the markdown in memory. Reuse it here
+                # instead of re-reading the file; format_on_save is the one config that can
+                # still make the persisted file diverge because write_file only returns a checksum.
+                content = written_content
         result = result.model_copy(update={"content": content})
 
         logger.info(
@@ -421,18 +433,28 @@ async def update_entity_by_id(
         ):
             if fast:
                 entity = await entity_service.fast_write_entity(data, external_id=entity_id)
+                written_content = None
+                search_content = None
                 response.status_code = 200 if existing else 201
             else:
                 if existing:
-                    entity = await entity_service.update_entity(existing, data)
+                    write_result = await entity_service.update_entity_with_content(existing, data)
+                    entity = write_result.entity
+                    written_content = write_result.content
+                    search_content = write_result.search_content
                     response.status_code = 200
                 else:
-                    entity = await entity_service.create_entity(data)
+                    write_result = await entity_service.create_entity_with_content(data)
+                    entity = write_result.entity
+                    written_content = write_result.content
+                    search_content = write_result.search_content
                     if entity.external_id != entity_id:
                         entity = await entity_repository.update(
                             entity.id,
                             {"external_id": entity_id},
                         )
+                        # external_id fixup only changes the DB row. The file content is unchanged,
+                        # so the markdown captured during the write remains valid downstream.
                         if not entity:
                             raise HTTPException(
                                 status_code=404,
@@ -461,7 +483,7 @@ async def update_entity_by_id(
                 action="update_entity",
                 phase="search_index",
             ):
-                await search_service.index_entity(entity)
+                await search_service.index_entity(entity, content=search_content)
             with telemetry.scope(
                 "api.knowledge.update_entity.vector_sync",
                 domain="knowledge",
@@ -484,8 +506,15 @@ async def update_entity_by_id(
             domain="knowledge",
             action="update_entity",
             phase="read_content",
+            source="file" if fast else "memory",
         ):
-            content = await file_service.read_file_content(entity.file_path)
+            if fast:
+                content = await file_service.read_file_content(entity.file_path)
+            else:
+                # Non-fast writes already captured the markdown in memory. Reuse it here
+                # instead of re-reading the file; format_on_save is the one config that can
+                # still make the persisted file diverge because write_file only returns a checksum.
+                content = written_content
         result = result.model_copy(update={"content": content})
 
         logger.info(
@@ -563,9 +592,11 @@ async def edit_entity_by_id(
                         find_text=data.find_text,
                         expected_replacements=data.expected_replacements,
                     )
+                    written_content = None
+                    search_content = None
                 else:
                     identifier = entity.permalink or entity.file_path
-                    updated_entity = await entity_service.edit_entity(
+                    write_result = await entity_service.edit_entity_with_content(
                         identifier=identifier,
                         operation=data.operation,
                         content=data.content,
@@ -573,6 +604,9 @@ async def edit_entity_by_id(
                         find_text=data.find_text,
                         expected_replacements=data.expected_replacements,
                     )
+                    updated_entity = write_result.entity
+                    written_content = write_result.content
+                    search_content = write_result.search_content
 
             if fast:
                 with telemetry.scope(
@@ -594,7 +628,7 @@ async def edit_entity_by_id(
                     action="edit_entity",
                     phase="search_index",
                 ):
-                    await search_service.index_entity(updated_entity)
+                    await search_service.index_entity(updated_entity, content=search_content)
                 with telemetry.scope(
                     "api.knowledge.edit_entity.vector_sync",
                     domain="knowledge",
@@ -617,8 +651,15 @@ async def edit_entity_by_id(
                 domain="knowledge",
                 action="edit_entity",
                 phase="read_content",
+                source="file" if fast else "memory",
             ):
-                content = await file_service.read_file_content(updated_entity.file_path)
+                if fast:
+                    content = await file_service.read_file_content(updated_entity.file_path)
+                else:
+                    # Non-fast writes already captured the markdown in memory. Reuse it here
+                    # instead of re-reading the file; format_on_save is the one config that can
+                    # still make the persisted file diverge because write_file only returns a checksum.
+                    content = written_content
             result = result.model_copy(update={"content": content})
 
             logger.info(

src/basic_memory/api/v2/utils.py

@@ -1,6 +1,7 @@
 from typing import Optional, List
 
 from basic_memory import telemetry
+from basic_memory.models import Entity as EntityModel
 from basic_memory.repository import EntityRepository
 from basic_memory.repository.search_repository import SearchIndexRow
 from basic_memory.schemas.memory import (
@@ -177,30 +178,26 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
         phase="hydrate_results",
         result_count=len(results),
     ):
-        # First pass: collect all unique entity IDs needed across all results.
-        # This covers entity_id (owning entity), from_id and to_id (relation endpoints).
-        entity_ids_needed: set[int] = set()
+        # Collect all unique entity IDs across all results in a single pass
+        # This avoids N+1 queries — one batch fetch instead of one per result
+        all_entity_ids: set[int] = set()
         for result in results:
-            if result.entity_id is not None:
-                entity_ids_needed.add(result.entity_id)
-            if result.from_id is not None:
-                entity_ids_needed.add(result.from_id)
-            if result.to_id is not None:
-                entity_ids_needed.add(result.to_id)
+            for eid in (result.entity_id, result.from_id, result.to_id):
+                if eid is not None:
+                    all_entity_ids.add(eid)
 
-        # Batch fetch all entities in a single query instead of one per result.
-        entity_permalink_lookup: dict[int, str] = {}
-        if entity_ids_needed:
-            with telemetry.scope(
-                "search.hydrate_results.fetch_entities",
-                domain="search",
-                action="search",
-                phase="fetch_entities",
-                result_count=len(entity_ids_needed),
-            ):
-                entities = await entity_service.get_entities_by_id(list(entity_ids_needed))
-            for e in entities:
-                entity_permalink_lookup[e.id] = e.permalink
+        # Single batch fetch for all entities
+        entities_by_id: dict[int, EntityModel] = {}
+        with telemetry.scope(
+            "search.hydrate_results.fetch_entities",
+            domain="search",
+            action="search",
+            phase="fetch_entities",
+            result_count=len(all_entity_ids),
+        ):
+            if all_entity_ids:
+                entities = await entity_service.get_entities_by_id(list(all_entity_ids))
+                entities_by_id = {e.id: e for e in entities}
 
         search_results = []
         with telemetry.scope(
@@ -224,21 +221,10 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
                     relation_id = result.id
                     entity_id = result.entity_id
 
-                entity_permalink = (
-                    entity_permalink_lookup.get(result.entity_id)  # pyright: ignore
-                    if result.entity_id is not None
-                    else None
-                )
-                from_permalink = (
-                    entity_permalink_lookup.get(result.from_id)  # pyright: ignore
-                    if result.from_id is not None
-                    else None
-                )
-                to_permalink = (
-                    entity_permalink_lookup.get(result.to_id)
-                    if result.to_id is not None
-                    else None
-                )
+                # Look up entities by their specific IDs
+                parent_entity = entities_by_id.get(result.entity_id) if result.entity_id else None  # pyright: ignore
+                from_entity = entities_by_id.get(result.from_id) if result.from_id else None  # pyright: ignore
+                to_entity = entities_by_id.get(result.to_id) if result.to_id else None
 
                 search_results.append(
                     SearchResult(
@@ -246,7 +232,7 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
                         type=result.type,  # pyright: ignore
                         permalink=result.permalink,
                         score=result.score,  # pyright: ignore
-                        entity=entity_permalink,
+                        entity=parent_entity.permalink if parent_entity else None,
                         content=result.content,
                         matched_chunk=result.matched_chunk_text,
                         file_path=result.file_path,
@@ -255,8 +241,8 @@ async def to_search_results(entity_service: EntityService, results: List[SearchI
                         observation_id=observation_id,
                         relation_id=relation_id,
                         category=result.category,
-                        from_entity=from_permalink,
-                        to_entity=to_permalink,
+                        from_entity=from_entity.permalink if from_entity else None,
+                        to_entity=to_entity.permalink if to_entity else None,
                         relation_type=result.relation_type,
                     )
                 )

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

@@ -2,10 +2,12 @@
 
 from basic_memory.cli.commands.cloud.api_client import make_api_request
 from basic_memory.config import ConfigManager
+from basic_memory.mcp.async_client import resolve_configured_workspace
 from basic_memory.schemas.cloud import (
     CloudProjectList,
     CloudProjectCreateRequest,
     CloudProjectCreateResponse,
+    ProjectVisibility,
 )
 from basic_memory.utils import generate_permalink
 
@@ -16,8 +18,25 @@ class CloudUtilsError(Exception):
     pass
 
 
+def _workspace_headers(
+    *,
+    project_name: str | None = None,
+    workspace: str | None = None,
+) -> dict[str, str]:
+    """Build optional workspace headers using the CLI config resolution chain."""
+    resolved_workspace = resolve_configured_workspace(
+        project_name=project_name,
+        workspace=workspace,
+    )
+    if resolved_workspace is None:
+        return {}
+    return {"X-Workspace-ID": resolved_workspace}
+
+
 async def fetch_cloud_projects(
     *,
+    project_name: str | None = None,
+    workspace: str | None = None,
     api_request=make_api_request,
 ) -> CloudProjectList:
     """Fetch list of projects from cloud API.
@@ -30,7 +49,11 @@ async def fetch_cloud_projects(
         config = config_manager.config
         host_url = config.cloud_host.rstrip("/")
 
-        response = await api_request(method="GET", url=f"{host_url}/proxy/v2/projects/")
+        response = await api_request(
+            method="GET",
+            url=f"{host_url}/proxy/v2/projects/",
+            headers=_workspace_headers(project_name=project_name, workspace=workspace),
+        )
 
         return CloudProjectList.model_validate(response.json())
     except Exception as e:
@@ -40,12 +63,16 @@ async def fetch_cloud_projects(
 async def create_cloud_project(
     project_name: str,
     *,
+    workspace: str | None = None,
+    visibility: ProjectVisibility = "workspace",
     api_request=make_api_request,
 ) -> CloudProjectCreateResponse:
     """Create a new project on cloud.
 
     Args:
         project_name: Name of project to create
+        workspace: Optional workspace override for tenant-scoped project creation
+        visibility: Visibility for the created cloud project
 
     Returns:
         CloudProjectCreateResponse with project details from API
@@ -62,12 +89,16 @@ async def create_cloud_project(
             name=project_name,
             path=project_path,
             set_default=False,
+            visibility=visibility,
         )
 
         response = await api_request(
             method="POST",
             url=f"{host_url}/proxy/v2/projects/",
-            headers={"Content-Type": "application/json"},
+            headers={
+                "Content-Type": "application/json",
+                **_workspace_headers(project_name=project_name, workspace=workspace),
+            },
             json_data=project_data.model_dump(),
         )
 
@@ -91,18 +122,28 @@ async def sync_project(project_name: str, force_full: bool = False) -> None:
         raise CloudUtilsError(f"Failed to sync project '{project_name}': {e}") from e
 
 
-async def project_exists(project_name: str, *, api_request=make_api_request) -> bool:
+async def project_exists(
+    project_name: str,
+    *,
+    workspace: str | None = None,
+    api_request=make_api_request,
+) -> bool:
     """Check if a project exists on cloud.
 
     Args:
         project_name: Name of project to check
+        workspace: Optional workspace override for tenant-scoped project lookup
 
     Returns:
         True if project exists, False otherwise
+
+    Raises:
+        CloudUtilsError: If the project list cannot be fetched from cloud
     """
-    try:
-        projects = await fetch_cloud_projects(api_request=api_request)
-        project_names = {p.name for p in projects.projects}
-        return project_name in project_names
-    except Exception:
-        return False
+    projects = await fetch_cloud_projects(
+        project_name=project_name,
+        workspace=workspace,
+        api_request=api_request,
+    )
+    project_names = {p.name for p in projects.projects}
+    return project_name in project_names

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

@@ -54,7 +54,7 @@ def _require_cloud_credentials(config) -> None:
 
 async def _get_cloud_project(name: str) -> ProjectItem | None:
     """Fetch a project by name from the cloud API."""
-    async with get_client() as client:
+    async with get_client(project_name=name) as client:
         projects_list = await ProjectClient(client).list_projects()
         for proj in projects_list.projects:
             if generate_permalink(proj.name) == generate_permalink(name):
@@ -129,9 +129,9 @@ def sync_project_command(
             if not dry_run:
 
                 async def _trigger_db_sync():
-                    async with get_client() as client:
+                    async with get_client(project_name=name) as client:
                         return await ProjectClient(client).sync(
-                            project_data.external_id, force_full=True
+                            project_data.external_id, force_full=False
                         )
 
                 try:
@@ -195,7 +195,10 @@ def bisync_project_command(
             # Update config — sync_entry is guaranteed non-None because
             # _get_sync_project validated local_sync_path (which comes from sync_entry)
             sync_entry = config.projects.get(name)
-            assert sync_entry is not None
+            if sync_entry is None:
+                raise RuntimeError(
+                    f"Sync entry for project '{name}' unexpectedly missing after validation"
+                )
             sync_entry.last_sync = datetime.now()
             sync_entry.bisync_initialized = True
             ConfigManager().save_config(config)
@@ -204,9 +207,9 @@ def bisync_project_command(
             if not dry_run:
 
                 async def _trigger_db_sync():
-                    async with get_client() as client:
+                    async with get_client(project_name=name) as client:
                         return await ProjectClient(client).sync(
-                            project_data.external_id, force_full=True
+                            project_data.external_id, force_full=False
                         )
 
                 try:
@@ -320,7 +323,7 @@ def setup_project_sync(
 
     async def _verify_project_exists():
         """Verify the project exists on cloud by listing all projects."""
-        async with get_client() as client:
+        async with get_client(project_name=name) as client:
             projects_list = await ProjectClient(client).list_projects()
             project_names = [p.name for p in projects_list.projects]
             if name not in project_names:

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

@@ -1,5 +1,6 @@
 """Upload CLI commands for basic-memory projects."""
 
+from functools import partial
 from pathlib import Path
 
 import typer
@@ -8,12 +9,16 @@ from rich.console import Console
 from basic_memory.cli.app import cloud_app
 from basic_memory.cli.commands.command_utils import run_with_cleanup
 from basic_memory.cli.commands.cloud.cloud_utils import (
+    CloudUtilsError,
     create_cloud_project,
     project_exists,
     sync_project,
 )
 from basic_memory.cli.commands.cloud.upload import upload_path
-from basic_memory.mcp.async_client import get_cloud_control_plane_client
+from basic_memory.mcp.async_client import (
+    get_cloud_control_plane_client,
+    resolve_configured_workspace,
+)
 
 console = Console()
 
@@ -73,12 +78,20 @@ def upload(
     """
 
     async def _upload():
+        resolved_workspace = resolve_configured_workspace(project_name=project)
+
+        try:
+            project_already_exists = await project_exists(project, workspace=resolved_workspace)
+        except CloudUtilsError as e:
+            console.print(f"[red]Failed to check cloud project '{project}': {e}[/red]")
+            raise typer.Exit(1)
+
         # Check if project exists
-        if not await project_exists(project):
+        if not project_already_exists:
             if create_project:
                 console.print(f"[blue]Creating cloud project '{project}'...[/blue]")
                 try:
-                    await create_cloud_project(project)
+                    await create_cloud_project(project, workspace=resolved_workspace)
                     console.print(f"[green]Created project '{project}'[/green]")
                 except Exception as e:
                     console.print(f"[red]Failed to create project: {e}[/red]")
@@ -106,7 +119,10 @@ def upload(
             verbose=verbose,
             use_gitignore=not no_gitignore,
             dry_run=dry_run,
-            client_cm_factory=get_cloud_control_plane_client,
+            client_cm_factory=partial(
+                get_cloud_control_plane_client,
+                workspace=resolved_workspace,
+            ),
         )
         if not success:
             console.print("[red]Upload failed[/red]")
@@ -117,8 +133,10 @@ def upload(
         else:
             console.print(f"[green]Successfully uploaded to '{project}'[/green]")
 
-        # Sync project if requested (skip on dry run)
-        # Force full scan after bisync to ensure database is up-to-date with synced files
+        # Sync project if requested (skip on dry run).
+        # Trigger: upload adds new files the watcher has not observed locally.
+        # Why: force_full ensures those freshly uploaded files are indexed immediately.
+        # Outcome: upload keeps its eager reindex while sync/bisync stay incremental.
         if sync and not dry_run:
             console.print(f"[blue]Syncing project '{project}'...[/blue]")
             try:

src/basic_memory/cli/commands/project.py

@@ -4,6 +4,7 @@ import json
 import os
 from datetime import datetime
 from pathlib import Path
+from typing import cast
 
 import typer
 from rich.console import Console, Group
@@ -27,6 +28,7 @@ from basic_memory.cli.commands.routing import force_routing, validate_routing_fl
 from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
 from basic_memory.mcp.async_client import get_client
 from basic_memory.mcp.clients import ProjectClient
+from basic_memory.schemas.cloud import ProjectVisibility
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.utils import generate_permalink, normalize_project_path
 
@@ -56,6 +58,57 @@ def make_bar(value: int, max_value: int, width: int = 40) -> Text:
     return bar
 
 
+def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:
+    """Normalize CLI visibility input to the cloud API contract."""
+    if visibility is None:
+        return "workspace"
+
+    normalized = visibility.strip().lower()
+    if normalized in {"workspace", "shared", "private"}:
+        return cast(ProjectVisibility, normalized)
+
+    raise ValueError("Invalid visibility. Expected one of: workspace, shared, private.")
+
+
+def _resolve_workspace_id(config, workspace: str | None) -> str | None:
+    """Resolve a workspace name or tenant_id to a tenant_id."""
+    from basic_memory.mcp.project_context import (
+        _workspace_choices,
+        _workspace_matches_identifier,
+        get_available_workspaces,
+    )
+
+    if workspace is not None:
+        workspaces = run_with_cleanup(get_available_workspaces())
+        matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
+        if not matches:
+            console.print(f"[red]Error: Workspace '{workspace}' not found[/red]")
+            if workspaces:
+                console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+            raise typer.Exit(1)
+        if len(matches) > 1:
+            console.print(
+                f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. "
+                f"Use tenant_id instead.[/red]"
+            )
+            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+            raise typer.Exit(1)
+        return matches[0].tenant_id
+
+    if config.default_workspace:
+        return config.default_workspace
+
+    try:
+        workspaces = run_with_cleanup(get_available_workspaces())
+        if len(workspaces) == 1:
+            return workspaces[0].tenant_id
+    except Exception:
+        # Workspace resolution is optional until a command needs a specific tenant.
+        pass
+
+    return None
+
+
 @project_app.command("list")
 def list_projects(
     local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
@@ -257,6 +310,16 @@ def add_project(
     local_path: str = typer.Option(
         None, "--local-path", help="Local sync path for cloud mode (optional)"
     ),
+    workspace: str = typer.Option(
+        None,
+        "--workspace",
+        help="Cloud workspace name or tenant_id (cloud mode only)",
+    ),
+    visibility: str = typer.Option(
+        None,
+        "--visibility",
+        help="Cloud project visibility: workspace, shared, or private",
+    ),
     set_default: bool = typer.Option(False, "--default", help="Set as default project"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -271,6 +334,8 @@ def add_project(
     Cloud mode examples:\n
         bm project add research                           # No local sync\n
         bm project add research --local-path ~/docs       # With local sync\n
+        bm project add research --cloud --visibility shared\n
+        bm project add research --cloud --workspace Personal --visibility shared\n
 
     Local mode example:\n
         bm project add research ~/Documents/research
@@ -285,6 +350,7 @@ def add_project(
 
     # Determine effective mode: default local, cloud only when explicitly requested.
     effective_cloud_mode = cloud and not local
+    resolved_workspace_id: str | None = None
 
     # Resolve local sync path early (needed for both cloud and local mode)
     local_sync_path: str | None = None
@@ -293,18 +359,31 @@ def add_project(
 
     if effective_cloud_mode:
         _require_cloud_credentials(config)
+        try:
+            resolved_visibility = _normalize_project_visibility(visibility)
+        except ValueError as e:
+            console.print(f"[red]Error: {e}[/red]")
+            raise typer.Exit(1)
+        resolved_workspace_id = _resolve_workspace_id(config, workspace)
         # Cloud mode: path auto-generated from name, local sync is optional
 
         async def _add_project():
-            async with get_client() as client:
+            async with get_client(workspace=resolved_workspace_id) as client:
                 data = {
                     "name": name,
                     "path": generate_permalink(name),
                     "local_sync_path": local_sync_path,
                     "set_default": set_default,
+                    "visibility": resolved_visibility,
                 }
                 return await ProjectClient(client).create_project(data)
     else:
+        if workspace is not None:
+            console.print("[red]Error: --workspace is only supported in cloud mode[/red]")
+            raise typer.Exit(1)
+        if visibility is not None:
+            console.print("[red]Error: --visibility is only supported in cloud mode[/red]")
+            raise typer.Exit(1)
         # Local mode: path is required
         if path is None:
             console.print("[red]Error: path argument is required in local mode[/red]")
@@ -323,25 +402,34 @@ def add_project(
             result = run_with_cleanup(_add_project())
         console.print(f"[green]{result.message}[/green]")
 
+        # Trigger: local config needs enough metadata to route future commands back to cloud.
+        # Why: explicit workspace selection and local sync state should persist across CLI sessions.
+        # Outcome: cloud-backed projects keep cloud mode, workspace_id, and optional local sync path.
+        if effective_cloud_mode and (local_sync_path or resolved_workspace_id):
+            entry = config.projects.get(name)
+            if entry:
+                entry.mode = ProjectMode.CLOUD
+                if local_sync_path:
+                    entry.path = local_sync_path
+                    entry.local_sync_path = local_sync_path
+                if resolved_workspace_id:
+                    entry.workspace_id = resolved_workspace_id
+            else:
+                # Project may not be in local config yet (cloud-only add)
+                config.projects[name] = ProjectEntry(
+                    path=local_sync_path or "",
+                    mode=ProjectMode.CLOUD,
+                    local_sync_path=local_sync_path,
+                    workspace_id=resolved_workspace_id,
+                )
+            ConfigManager().save_config(config)
+
         # Save local sync path to config if in cloud mode
         if effective_cloud_mode and local_sync_path:
             # Create local directory if it doesn't exist
             local_dir = Path(local_sync_path)
             local_dir.mkdir(parents=True, exist_ok=True)
 
-            # Update project entry — path is always the local directory
-            entry = config.projects.get(name)
-            if entry:
-                entry.path = local_sync_path
-                entry.local_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,
-                    local_sync_path=local_sync_path,
-                )
-            ConfigManager().save_config(config)
-
             console.print(f"\n[green]Local sync path configured: {local_sync_path}[/green]")
             console.print("\nNext steps:")
             console.print(f"  1. Preview: bm cloud bisync --name {name} --resync --dry-run")
@@ -575,45 +663,7 @@ def set_cloud(
         console.print("[dim]Run 'bm cloud api-key save <key>' or 'bm cloud login' first[/dim]")
         raise typer.Exit(1)
 
-    # --- Resolve workspace to tenant_id ---
-    resolved_workspace_id: str | None = None
-
-    if workspace is not None:
-        # Explicit --workspace: resolve to tenant_id via cloud lookup
-        from basic_memory.mcp.project_context import (
-            get_available_workspaces,
-            _workspace_matches_identifier,
-            _workspace_choices,
-        )
-
-        workspaces = run_with_cleanup(get_available_workspaces())
-        matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
-        if not matches:
-            console.print(f"[red]Error: Workspace '{workspace}' not found[/red]")
-            if workspaces:
-                console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
-            raise typer.Exit(1)
-        if len(matches) > 1:
-            console.print(
-                f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. "
-                f"Use tenant_id instead.[/red]"
-            )
-            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
-            raise typer.Exit(1)
-        resolved_workspace_id = matches[0].tenant_id
-    elif config.default_workspace:
-        # Fall back to global default
-        resolved_workspace_id = config.default_workspace
-    else:
-        # Try auto-select if single workspace
-        try:
-            from basic_memory.mcp.project_context import get_available_workspaces
-
-            workspaces = run_with_cleanup(get_available_workspaces())
-            if len(workspaces) == 1:
-                resolved_workspace_id = workspaces[0].tenant_id
-        except Exception:
-            pass  # Workspace resolution is optional at set-cloud time
+    resolved_workspace_id = _resolve_workspace_id(config, workspace)
 
     config.set_project_mode(name, ProjectMode.CLOUD)
     if resolved_workspace_id:

src/basic_memory/mcp/async_client.py

@@ -66,6 +66,27 @@ async def _resolve_cloud_token(config) -> str:
         )
 
 
+def resolve_configured_workspace(
+    *,
+    config=None,
+    project_name: Optional[str] = None,
+    workspace: Optional[str] = None,
+) -> Optional[str]:
+    """Resolve workspace from explicit input, per-project config, then global default."""
+    if workspace is not None:
+        return workspace
+
+    if config is None:
+        config = ConfigManager().config
+
+    if project_name is not None:
+        project_entry = config.projects.get(project_name)
+        if project_entry and project_entry.workspace_id:
+            return project_entry.workspace_id
+
+    return config.default_workspace
+
+
 @asynccontextmanager
 async def _cloud_client(
     config,
@@ -88,15 +109,20 @@ async def _cloud_client(
 
 
 @asynccontextmanager
-async def get_cloud_control_plane_client() -> AsyncIterator[AsyncClient]:
+async def get_cloud_control_plane_client(
+    workspace: Optional[str] = None,
+) -> AsyncIterator[AsyncClient]:
     """Create a control-plane cloud client for endpoints outside /proxy."""
     config = ConfigManager().config
     timeout = _build_timeout()
     token = await _resolve_cloud_token(config)
+    headers = {"Authorization": f"Bearer {token}"}
+    if workspace:
+        headers["X-Workspace-ID"] = workspace
     logger.info(f"Creating HTTP client for cloud control plane at: {config.cloud_host}")
     async with AsyncClient(
         base_url=config.cloud_host,
-        headers={"Authorization": f"Bearer {token}"},
+        headers=headers,
         timeout=timeout,
     ) as client:
         yield client
@@ -167,7 +193,12 @@ async def get_client(
 
         if _force_cloud_mode():
             logger.debug("Explicit cloud routing enabled - using cloud proxy client")
-            async with _cloud_client(config, timeout, workspace=workspace) as client:
+            effective_workspace = resolve_configured_workspace(
+                config=config,
+                project_name=project_name,
+                workspace=workspace,
+            )
+            async with _cloud_client(config, timeout, workspace=effective_workspace) as client:
                 yield client
             return
 
@@ -179,8 +210,13 @@ async def get_client(
         project_mode = config.get_project_mode(project_name)
         if project_mode == ProjectMode.CLOUD:
             logger.debug(f"Project '{project_name}' is cloud mode - using cloud proxy client")
+            effective_workspace = resolve_configured_workspace(
+                config=config,
+                project_name=project_name,
+                workspace=workspace,
+            )
             try:
-                async with _cloud_client(config, timeout, workspace=workspace) as client:
+                async with _cloud_client(config, timeout, workspace=effective_workspace) as client:
                     yield client
             except RuntimeError as exc:
                 raise RuntimeError(

src/basic_memory/repository/entity_repository.py

@@ -59,16 +59,27 @@ class EntityRepository(Repository[Entity]):
         )
         return await self.find_one(query)
 
-    async def get_by_permalink(self, permalink: str) -> Optional[Entity]:
+    async def _find_one_by_query(self, query, *, load_relations: bool) -> Optional[Entity]:
+        """Return one entity row with optional eager loading."""
+        if load_relations:
+            query = query.options(*self.get_load_options())
+            return await self.find_one(query)
+
+        result = await self.execute_query(query, use_query_options=False)
+        return result.scalars().one_or_none()
+
+    async def get_by_permalink(
+        self, permalink: str, *, load_relations: bool = True
+    ) -> Optional[Entity]:
         """Get entity by permalink.
 
         Args:
             permalink: Unique identifier for the entity
         """
-        query = self.select().where(Entity.permalink == permalink).options(*self.get_load_options())
-        return await self.find_one(query)
+        query = self.select().where(Entity.permalink == permalink)
+        return await self._find_one_by_query(query, load_relations=load_relations)
 
-    async def get_by_title(self, title: str) -> Sequence[Entity]:
+    async def get_by_title(self, title: str, *, load_relations: bool = True) -> Sequence[Entity]:
         """Get entities by title, ordered by shortest path first.
 
         When multiple entities share the same title (in different folders),
@@ -82,23 +93,20 @@ class EntityRepository(Repository[Entity]):
             self.select()
             .where(Entity.title == title)
             .order_by(func.length(Entity.file_path), Entity.file_path)
-            .options(*self.get_load_options())
         )
-        result = await self.execute_query(query)
+        result = await self.execute_query(query, use_query_options=load_relations)
         return list(result.scalars().all())
 
-    async def get_by_file_path(self, file_path: Union[Path, str]) -> Optional[Entity]:
+    async def get_by_file_path(
+        self, file_path: Union[Path, str], *, load_relations: bool = True
+    ) -> Optional[Entity]:
         """Get entity by file_path.
 
         Args:
             file_path: Path to the entity file (will be converted to string internally)
         """
-        query = (
-            self.select()
-            .where(Entity.file_path == Path(file_path).as_posix())
-            .options(*self.get_load_options())
-        )
-        return await self.find_one(query)
+        query = self.select().where(Entity.file_path == Path(file_path).as_posix())
+        return await self._find_one_by_query(query, load_relations=load_relations)
 
     # -------------------------------------------------------------------------
     # Lightweight methods for permalink resolution (no eager loading)
@@ -306,7 +314,7 @@ class EntityRepository(Repository[Entity]):
         result = await self.execute_query(query)
         return list(result.scalars().all())
 
-    async def upsert_entity(self, entity: Entity) -> Entity:
+    async def upsert_entity(self, entity: Entity, *, reload: bool = True) -> Entity:
         """Insert or update entity using simple try/catch with database-level conflict resolution.
 
         Handles file_path race conditions by checking for existing entity on IntegrityError.
@@ -327,6 +335,9 @@ class EntityRepository(Repository[Entity]):
                 session.add(entity)
                 await session.flush()
 
+                if not reload:
+                    return entity
+
                 # Return with relationships loaded
                 query = (
                     self.select()
@@ -363,13 +374,12 @@ class EntityRepository(Repository[Entity]):
                 await session.rollback()
 
                 # Re-query after rollback to get a fresh, attached entity
-                existing_result = await session.execute(
-                    select(Entity)
-                    .where(
-                        Entity.file_path == entity.file_path, Entity.project_id == entity.project_id
-                    )
-                    .options(*self.get_load_options())
+                existing_query = select(Entity).where(
+                    Entity.file_path == entity.file_path, Entity.project_id == entity.project_id
                 )
+                if reload:
+                    existing_query = existing_query.options(*self.get_load_options())
+                existing_result = await session.execute(existing_query)
                 existing_entity = existing_result.scalar_one_or_none()
 
                 if existing_entity:
@@ -393,6 +403,9 @@ class EntityRepository(Repository[Entity]):
 
                     await session.commit()
 
+                    if not reload:
+                        return merged_entity
+
                     # Re-query to get proper relationships loaded
                     final_result = await session.execute(
                         select(Entity)

src/basic_memory/repository/repository.py

@@ -268,8 +268,21 @@ class Repository[T: Base]:
 
             return await self.select_by_ids(session, [model.id for model in model_list])  # pyright: ignore [reportAttributeAccessIssue]
 
-    async def update(self, entity_id: int, entity_data: dict | T) -> Optional[T]:
-        """Update an entity with the given data."""
+    async def update(
+        self,
+        entity_id: int,
+        entity_data: dict | T,
+        *,
+        reload: bool = True,
+    ) -> Optional[T]:
+        """Update an entity with the given data.
+
+        Args:
+            entity_id: Primary key to update
+            entity_data: Column values or a model instance to copy from
+            reload: When True, re-select the entity with repository load options.
+                When False, return the attached row after flush/refresh.
+        """
         logger.debug(f"Updating {self.Model.__name__} {entity_id} with data: {entity_data}")
         async with db.scoped_session(self.session_maker) as session:
             try:
@@ -291,6 +304,8 @@ class Repository[T: Base]:
                 await session.refresh(entity)  # Refresh
 
                 logger.debug(f"Updated {self.Model.__name__}: {entity_id}")
+                if not reload:
+                    return entity
                 return await self.select_by_id(session, entity.id)  # pyright: ignore [reportAttributeAccessIssue]
 
             except NoResultFound:

src/basic_memory/schemas/cloud.py

@@ -1,7 +1,11 @@
 """Schemas for cloud-related API responses."""
 
+from typing import Literal
+
 from pydantic import BaseModel, Field
 
+type ProjectVisibility = Literal["workspace", "shared", "private"]
+
 
 class TenantMountInfo(BaseModel):
     """Response from /tenant/mount/info endpoint."""
@@ -36,6 +40,10 @@ class CloudProjectCreateRequest(BaseModel):
     name: str = Field(..., description="Project name")
     path: str = Field(..., description="Project path (permalink)")
     set_default: bool = Field(default=False, description="Set as default project")
+    visibility: ProjectVisibility = Field(
+        default="workspace",
+        description="Project visibility for team workspaces",
+    )
 
 
 class CloudProjectCreateResponse(BaseModel):

src/basic_memory/services/entity_service.py

@@ -1,6 +1,7 @@
 """Service for managing entities in the database."""
 
 from collections.abc import Callable
+from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
 from typing import List, Optional, Sequence, Tuple, Union
@@ -50,6 +51,15 @@ from basic_memory.services.search_service import SearchService
 from basic_memory.utils import build_canonical_permalink
 
 
+@dataclass(frozen=True)
+class EntityWriteResult:
+    """Persisted entity plus the response/search content produced during this call."""
+
+    entity: EntityModel
+    content: str
+    search_content: str
+
+
 class EntityService(BaseService[EntityModel]):
     """Service for managing entities in the database."""
 
@@ -79,7 +89,7 @@ class EntityService(BaseService[EntityModel]):
 
     async def detect_file_path_conflicts(
         self, file_path: str, skip_check: bool = False
-    ) -> List[Entity]:
+    ) -> List[str]:
         """Detect potential file path conflicts for a given file path.
 
         This checks for entities with similar file paths that might cause conflicts:
@@ -93,28 +103,19 @@ class EntityService(BaseService[EntityModel]):
             skip_check: If True, skip the check and return empty list (optimization for bulk operations)
 
         Returns:
-            List of entities that might conflict with the given file path
+            List of file paths that might conflict with the given file path
         """
         if skip_check:
             return []
 
         from basic_memory.utils import detect_potential_file_conflicts
 
-        conflicts = []
-
-        # Get all existing file paths
-        all_entities = await self.repository.find_all()
-        existing_paths = [entity.file_path for entity in all_entities]
+        # Load only file paths. Conflict detection is on the hot write path and
+        # does not need observations or relations.
+        existing_paths = await self.repository.get_all_file_paths()
 
         # Use the enhanced conflict detection utility
-        conflicting_paths = detect_potential_file_conflicts(file_path, existing_paths)
-
-        # Find the entities corresponding to conflicting paths
-        for entity in all_entities:
-            if entity.file_path in conflicting_paths:
-                conflicts.append(entity)
-
-        return conflicts
+        return detect_potential_file_conflicts(file_path, existing_paths)
 
     async def resolve_permalink(
         self,
@@ -143,8 +144,7 @@ class EntityService(BaseService[EntityModel]):
         )
         if conflicts:
             logger.warning(
-                f"Detected potential file path conflicts for '{file_path_str}': "
-                f"{[entity.file_path for entity in conflicts]}"
+                f"Detected potential file path conflicts for '{file_path_str}': {conflicts}"
             )
 
         # If markdown has explicit permalink, try to validate it
@@ -242,9 +242,17 @@ class EntityService(BaseService[EntityModel]):
 
         # Try to find existing entity using strict resolution (no fuzzy search)
         # This prevents incorrectly matching similar file paths like "Node A.md" and "Node C.md"
-        existing = await self.link_resolver.resolve_link(schema.file_path, strict=True)
+        existing = await self.link_resolver.resolve_link(
+            schema.file_path,
+            strict=True,
+            load_relations=False,
+        )
         if not existing and schema.permalink:
-            existing = await self.link_resolver.resolve_link(schema.permalink, strict=True)
+            existing = await self.link_resolver.resolve_link(
+                schema.permalink,
+                strict=True,
+                load_relations=False,
+            )
 
         if existing:
             logger.debug(f"Found existing entity: {existing.file_path}")
@@ -255,6 +263,10 @@ class EntityService(BaseService[EntityModel]):
 
     async def create_entity(self, schema: EntitySchema) -> EntityModel:
         """Create a new entity and write to filesystem."""
+        return (await self.create_entity_with_content(schema)).entity
+
+    async def create_entity_with_content(self, schema: EntitySchema) -> EntityWriteResult:
+        """Create a new entity and return both the entity row and written markdown."""
         logger.debug(f"Creating entity: {schema.title}")
 
         # Get file path and ensure it's a Path object
@@ -320,18 +332,28 @@ class EntityService(BaseService[EntityModel]):
             action="create",
             phase="upsert_entity",
         ):
-            entity = await self.upsert_entity_from_markdown(file_path, entity_markdown, is_new=True)
-
-        with telemetry.scope(
-            "entity_service.create.update_checksum",
-            domain="entity_service",
-            action="create",
-            phase="update_checksum",
-        ):
-            return await self.repository.update(entity.id, {"checksum": checksum})
+            updated = await self.upsert_entity_from_markdown(
+                file_path,
+                entity_markdown,
+                is_new=True,
+                checksum=checksum,
+            )
+        if not updated:  # pragma: no cover
+            raise ValueError(f"Failed to persist entity after create: {file_path}")
+        return EntityWriteResult(
+            entity=updated,
+            content=final_content,
+            search_content=remove_frontmatter(final_content),
+        )
 
     async def update_entity(self, entity: EntityModel, schema: EntitySchema) -> EntityModel:
         """Update an entity's content and metadata."""
+        return (await self.update_entity_with_content(entity, schema)).entity
+
+    async def update_entity_with_content(
+        self, entity: EntityModel, schema: EntitySchema
+    ) -> EntityWriteResult:
+        """Update an entity and return both the entity row and written markdown."""
         logger.debug(
             f"Updating entity with permalink: {entity.permalink} content-type: {schema.content_type}"
         )
@@ -434,18 +456,19 @@ class EntityService(BaseService[EntityModel]):
             phase="upsert_entity",
         ):
             entity = await self.upsert_entity_from_markdown(
-                file_path, entity_markdown, is_new=False
+                file_path,
+                entity_markdown,
+                is_new=False,
+                checksum=checksum,
             )
+        if not entity:  # pragma: no cover
+            raise ValueError(f"Failed to persist entity after update: {file_path}")
 
-        with telemetry.scope(
-            "entity_service.update.update_checksum",
-            domain="entity_service",
-            action="update",
-            phase="update_checksum",
-        ):
-            entity = await self.repository.update(entity.id, {"checksum": checksum})
-
-        return entity
+        return EntityWriteResult(
+            entity=entity,
+            content=final_content,
+            search_content=remove_frontmatter(final_content),
+        )
 
     async def fast_write_entity(
         self,
@@ -825,7 +848,7 @@ class EntityService(BaseService[EntityModel]):
 
         # Use UPSERT to handle conflicts cleanly
         try:
-            return await self.repository.upsert_entity(model)
+            return await self.repository.upsert_entity(model, reload=False)
         except Exception as e:
             logger.error(f"Failed to upsert entity for {file_path}: {e}")
             raise EntityCreationError(f"Failed to create entity: {str(e)}") from e
@@ -840,7 +863,12 @@ class EntityService(BaseService[EntityModel]):
         """
         logger.debug(f"Updating entity and observations: {file_path}")
 
-        db_entity = await self.repository.get_by_file_path(file_path.as_posix())
+        db_entity = await self.repository.get_by_file_path(
+            file_path.as_posix(),
+            load_relations=False,
+        )
+        if not db_entity:  # pragma: no cover
+            raise EntityNotFoundError(f"Entity not found: {file_path}")
 
         # Clear observations for entity
         await self.observation_repository.delete_by_fields(entity_id=db_entity.id)
@@ -857,23 +885,37 @@ class EntityService(BaseService[EntityModel]):
             )
             for obs in markdown.observations
         ]
-        await self.observation_repository.add_all(observations)
+        if observations:
+            await self.observation_repository.add_all(observations)
 
-        # update values from markdown
-        db_entity = entity_model_from_markdown(file_path, markdown, db_entity)
+        # Trigger: the lightweight lookup above returns a detached row without loaded collections
+        # Why: assigning a new observation list onto that detached ORM object would trigger lazy loads
+        # Outcome: rebuild a fresh model from markdown, then copy over stable identity fields
+        db_entity_data = entity_model_from_markdown(
+            file_path,
+            markdown,
+            project_id=self.repository.project_id,
+        )
+        db_entity_data.id = db_entity.id
+        db_entity_data.project_id = db_entity.project_id
+        db_entity_data.external_id = db_entity.external_id
+        db_entity_data.created_by = db_entity.created_by
 
         # checksum value is None == not finished with sync
-        db_entity.checksum = None
+        db_entity_data.checksum = None
 
         # Set last_updated_by for cloud usage (preserve existing created_by)
         user_id = self.get_user_id()
         if user_id is not None:
-            db_entity.last_updated_by = user_id
+            db_entity_data.last_updated_by = user_id
+        else:
+            db_entity_data.last_updated_by = db_entity.last_updated_by
 
         # update entity
         return await self.repository.update(
             db_entity.id,
-            db_entity,
+            db_entity_data,
+            reload=False,
         )
 
     async def upsert_entity_from_markdown(
@@ -882,26 +924,76 @@ class EntityService(BaseService[EntityModel]):
         markdown: EntityMarkdown,
         *,
         is_new: bool,
+        checksum: Optional[str] = None,
     ) -> EntityModel:
         """Create/update entity and relations from parsed markdown."""
-        if is_new:
-            created = await self.create_entity_from_markdown(file_path, markdown)
-        else:
-            created = await self.update_entity_and_observations(file_path, markdown)
-        return await self.update_entity_relations(created.file_path, markdown)
+        # --- Base Entity Row ---
+        # Trigger: writes rebuild the entity row before touching relation edges
+        # Why: relations need a stable source entity ID, but not a fully hydrated graph
+        # Outcome: create/update the row with a lightweight return value
+        with telemetry.scope(
+            "entity_service.upsert.base_entity",
+            domain="entity_service",
+            action="upsert",
+            phase="base_entity",
+        ):
+            if is_new:
+                created = await self.create_entity_from_markdown(file_path, markdown)
+            else:
+                created = await self.update_entity_and_observations(file_path, markdown)
+
+        # --- Relation Edges ---
+        with telemetry.scope(
+            "entity_service.upsert.relations",
+            domain="entity_service",
+            action="upsert",
+            phase="relations",
+        ):
+            await self.update_entity_relations(created, markdown)
+
+        # --- Final Entity State ---
+        # Trigger: create/update/edit already computed the final file checksum
+        # Why: fold the checksum write into the upsert flow so callers do one hydrated read
+        # Outcome: the write path returns the final entity state without an extra checksum step
+        if checksum is not None:
+            with telemetry.scope(
+                "entity_service.upsert.persist_checksum",
+                domain="entity_service",
+                action="upsert",
+                phase="persist_checksum",
+            ):
+                updated = await self.repository.update(created.id, {"checksum": checksum})
+            if not updated:  # pragma: no cover
+                raise ValueError(f"Failed to update entity checksum after upsert: {file_path}")
+            return updated
+
+        with telemetry.scope(
+            "entity_service.upsert.hydrate_entity",
+            domain="entity_service",
+            action="upsert",
+            phase="hydrate_entity",
+        ):
+            hydrated = await self.repository.get_by_file_path(created.file_path)
+        if not hydrated:  # pragma: no cover
+            raise EntityNotFoundError(f"Entity not found after upsert: {created.file_path}")
+        return hydrated
 
     async def update_entity_relations(
         self,
-        path: str,
+        db_entity: EntityModel,
         markdown: EntityMarkdown,
-    ) -> EntityModel:
+    ) -> None:
         """Update relations for entity"""
-        logger.debug(f"Updating relations for entity: {path}")
-
-        db_entity = await self.repository.get_by_file_path(path)
+        logger.debug(f"Updating relations for entity: {db_entity.file_path}")
 
         # Clear existing relations first
-        await self.relation_repository.delete_outgoing_relations_from_entity(db_entity.id)
+        with telemetry.scope(
+            "entity_service.upsert.delete_relations",
+            domain="entity_service",
+            action="upsert",
+            phase="delete_relations",
+        ):
+            await self.relation_repository.delete_outgoing_relations_from_entity(db_entity.id)
 
         # Batch resolve all relation targets in parallel
         if markdown.relations:
@@ -910,13 +1002,23 @@ class EntityService(BaseService[EntityModel]):
             # Create tasks for all relation lookups
             # Use strict=True to disable fuzzy search - only exact matches should create resolved relations
             # This ensures forward references (links to non-existent entities) remain unresolved (to_id=NULL)
-            lookup_tasks = [
-                self.link_resolver.resolve_link(rel.target, strict=True)
-                for rel in markdown.relations
-            ]
+            with telemetry.scope(
+                "entity_service.upsert.resolve_relation_targets",
+                domain="entity_service",
+                action="upsert",
+                phase="resolve_relation_targets",
+            ):
+                lookup_tasks = [
+                    self.link_resolver.resolve_link(
+                        rel.target,
+                        strict=True,
+                        load_relations=False,
+                    )
+                    for rel in markdown.relations
+                ]
 
-            # Execute all lookups in parallel
-            resolved_entities = await asyncio.gather(*lookup_tasks, return_exceptions=True)
+                # Execute all lookups in parallel
+                resolved_entities = await asyncio.gather(*lookup_tasks, return_exceptions=True)
 
             # Process results and create relation records
             relations_to_add = []
@@ -945,22 +1047,26 @@ class EntityService(BaseService[EntityModel]):
 
             # Batch insert all relations
             if relations_to_add:
-                try:
-                    await self.relation_repository.add_all(relations_to_add)
-                except IntegrityError:
-                    # Some relations might be duplicates - fall back to individual inserts
-                    logger.debug("Batch relation insert failed, trying individual inserts")
-                    for relation in relations_to_add:
-                        try:
-                            await self.relation_repository.add(relation)
-                        except IntegrityError:
-                            # Unique constraint violation - relation already exists
-                            logger.debug(
-                                f"Skipping duplicate relation {relation.relation_type} from {db_entity.permalink}"
-                            )
-                            continue
-
-        return await self.repository.get_by_file_path(path)
+                with telemetry.scope(
+                    "entity_service.upsert.insert_relations",
+                    domain="entity_service",
+                    action="upsert",
+                    phase="insert_relations",
+                ):
+                    try:
+                        await self.relation_repository.add_all(relations_to_add)
+                    except IntegrityError:
+                        # Some relations might be duplicates - fall back to individual inserts
+                        logger.debug("Batch relation insert failed, trying individual inserts")
+                        for relation in relations_to_add:
+                            try:
+                                await self.relation_repository.add(relation)
+                            except IntegrityError:
+                                # Unique constraint violation - relation already exists
+                                logger.debug(
+                                    f"Skipping duplicate relation {relation.relation_type} from {db_entity.permalink}"
+                                )
+                                continue
 
     async def edit_entity(
         self,
@@ -988,6 +1094,27 @@ class EntityService(BaseService[EntityModel]):
             EntityNotFoundError: If the entity cannot be found
             ValueError: If required parameters are missing for the operation or replacement count doesn't match expected
         """
+        return (
+            await self.edit_entity_with_content(
+                identifier=identifier,
+                operation=operation,
+                content=content,
+                section=section,
+                find_text=find_text,
+                expected_replacements=expected_replacements,
+            )
+        ).entity
+
+    async def edit_entity_with_content(
+        self,
+        identifier: str,
+        operation: str,
+        content: str,
+        section: Optional[str] = None,
+        find_text: Optional[str] = None,
+        expected_replacements: int = 1,
+    ) -> EntityWriteResult:
+        """Edit an entity and return both the entity row and written markdown."""
         logger.debug(f"Editing entity: {identifier}, operation: {operation}")
 
         with telemetry.scope(
@@ -996,7 +1123,11 @@ class EntityService(BaseService[EntityModel]):
             action="edit",
             phase="resolve_entity",
         ):
-            entity = await self.link_resolver.resolve_link(identifier, strict=True)
+            entity = await self.link_resolver.resolve_link(
+                identifier,
+                strict=True,
+                load_relations=False,
+            )
         if not entity:
             raise EntityNotFoundError(f"Entity not found: {identifier}")
 
@@ -1045,18 +1176,19 @@ class EntityService(BaseService[EntityModel]):
             phase="upsert_entity",
         ):
             entity = await self.upsert_entity_from_markdown(
-                file_path, entity_markdown, is_new=False
+                file_path,
+                entity_markdown,
+                is_new=False,
+                checksum=checksum,
             )
+        if not entity:  # pragma: no cover
+            raise ValueError(f"Failed to persist entity after edit: {file_path}")
 
-        with telemetry.scope(
-            "entity_service.edit.update_checksum",
-            domain="entity_service",
-            action="edit",
-            phase="update_checksum",
-        ):
-            entity = await self.repository.update(entity.id, {"checksum": checksum})
-
-        return entity
+        return EntityWriteResult(
+            entity=entity,
+            content=new_content,
+            search_content=remove_frontmatter(new_content),
+        )
 
     def apply_edit_operation(
         self,

src/basic_memory/services/link_resolver.py

@@ -47,6 +47,7 @@ class LinkResolver:
         use_search: bool = True,
         strict: bool = False,
         source_path: Optional[str] = None,
+        load_relations: bool = True,
     ) -> Optional[Entity]:
         """Resolve a markdown link to a permalink.
 
@@ -56,6 +57,7 @@ class LinkResolver:
             strict: If True, only exact matches are allowed (no fuzzy search fallback)
             source_path: Optional path of the source file containing the link.
                         Used to prefer notes closer to the source (context-aware resolution).
+            load_relations: When False, skip eager loading and return a lightweight entity row.
         """
         logger.trace(f"Resolving link: {link_text} (source: {source_path})")
 
@@ -98,6 +100,7 @@ class LinkResolver:
                 strict=strict,
                 source_path=None,
                 project_permalink=project.permalink,
+                load_relations=load_relations,
             )
 
         current_project_permalink = await self._get_current_project_permalink()
@@ -109,6 +112,7 @@ class LinkResolver:
             strict=strict,
             source_path=source_path,
             project_permalink=current_project_permalink,
+            load_relations=load_relations,
         )
         if resolved:
             return resolved
@@ -136,6 +140,7 @@ class LinkResolver:
             strict=strict,
             source_path=None,
             project_permalink=project.permalink,
+            load_relations=load_relations,
         )
 
     def _normalize_link_text(self, link_text: str) -> Tuple[str, Optional[str]]:
@@ -176,6 +181,7 @@ class LinkResolver:
         strict: bool,
         source_path: Optional[str],
         project_permalink: Optional[str],
+        load_relations: bool,
     ) -> Optional[Entity]:
         """Resolve a link within a specific project scope."""
         clean_text = link_text
@@ -223,12 +229,18 @@ class LinkResolver:
                     # Try with .md extension
                     if not relative_path.endswith(".md"):
                         relative_path_md = f"{relative_path}.md"
-                        entity = await entity_repository.get_by_file_path(relative_path_md)
+                        entity = await entity_repository.get_by_file_path(
+                            relative_path_md,
+                            load_relations=load_relations,
+                        )
                         if entity:
                             return entity
 
                     # Try as-is (already has extension or is a permalink)
-                    entity = await entity_repository.get_by_file_path(relative_path)
+                    entity = await entity_repository.get_by_file_path(
+                        relative_path,
+                        load_relations=load_relations,
+                    )
                     if entity:
                         return entity
 
@@ -242,12 +254,18 @@ class LinkResolver:
 
             # Check permalink match
             for candidate_permalink in permalink_candidates:
-                permalink_entity = await entity_repository.get_by_permalink(candidate_permalink)
+                permalink_entity = await entity_repository.get_by_permalink(
+                    candidate_permalink,
+                    load_relations=load_relations,
+                )
                 if permalink_entity and permalink_entity.id not in [c.id for c in candidates]:
                     candidates.append(permalink_entity)
 
             # Check title matches
-            title_entities = await entity_repository.get_by_title(clean_text)
+            title_entities = await entity_repository.get_by_title(
+                clean_text,
+                load_relations=load_relations,
+            )
             for entity in title_entities:
                 # Avoid duplicates (permalink match might also be in title matches)
                 if entity.id not in [c.id for c in candidates]:
@@ -263,13 +281,19 @@ class LinkResolver:
         # Standard resolution (no source context): permalink first, then title
         # 1. Try exact permalink match first (most efficient)
         for candidate_permalink in permalink_candidates:
-            entity = await entity_repository.get_by_permalink(candidate_permalink)
+            entity = await entity_repository.get_by_permalink(
+                candidate_permalink,
+                load_relations=load_relations,
+            )
             if entity:
                 logger.debug(f"Found exact permalink match: {entity.permalink}")
                 return entity
 
         # 2. Try exact title match
-        found = await entity_repository.get_by_title(clean_text)
+        found = await entity_repository.get_by_title(
+            clean_text,
+            load_relations=load_relations,
+        )
         if found:
             # Return first match (shortest path) if no source context
             entity = found[0]
@@ -277,7 +301,10 @@ class LinkResolver:
             return entity
 
         # 3. Try file path
-        found_path = await entity_repository.get_by_file_path(clean_text)
+        found_path = await entity_repository.get_by_file_path(
+            clean_text,
+            load_relations=load_relations,
+        )
         if found_path:
             logger.debug(f"Found entity with path: {found_path.file_path}")
             return found_path
@@ -285,7 +312,10 @@ class LinkResolver:
         # 4. Try file path with .md extension if not already present
         if not clean_text.endswith(".md") and "/" in clean_text:
             file_path_with_md = f"{clean_text}.md"
-            found_path_md = await entity_repository.get_by_file_path(file_path_with_md)
+            found_path_md = await entity_repository.get_by_file_path(
+                file_path_with_md,
+                load_relations=load_relations,
+            )
             if found_path_md:
                 logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
                 return found_path_md
@@ -309,7 +339,10 @@ class LinkResolver:
                     f"Selected best match from {len(results)} results: {best_match.permalink}"
                 )
                 if best_match.permalink:
-                    return await entity_repository.get_by_permalink(best_match.permalink)
+                    return await entity_repository.get_by_permalink(
+                        best_match.permalink,
+                        load_relations=load_relations,
+                    )
 
         # if we couldn't find anything then return None
         return None

test-int/conftest.py

@@ -51,7 +51,7 @@ The `app` fixture ensures FastAPI dependency overrides are active, and
 """
 
 import os
-from typing import AsyncGenerator, Literal
+from typing import AsyncGenerator, Generator, Literal
 
 import pytest
 import pytest_asyncio
@@ -63,7 +63,13 @@ from testcontainers.postgres import PostgresContainer
 
 from httpx import AsyncClient, ASGITransport
 
-from basic_memory.config import BasicMemoryConfig, ProjectConfig, ConfigManager, DatabaseBackend
+from basic_memory.config import (
+    BasicMemoryConfig,
+    ProjectConfig,
+    ProjectEntry,
+    ConfigManager,
+    DatabaseBackend,
+)
 from basic_memory.db import engine_session_factory, DatabaseType
 from basic_memory.models import Project
 from basic_memory.models.base import Base
@@ -103,7 +109,7 @@ def postgres_container(db_backend):
     Uses testcontainers to spin up a real Postgres instance.
     Only starts if db_backend is "postgres".
     """
-    if db_backend != "postgres":
+    if db_backend != "postgres" or _configured_postgres_sync_url():
         yield None
         return
 
@@ -112,6 +118,70 @@ def postgres_container(db_backend):
         yield postgres
 
 
+POSTGRES_EPHEMERAL_TABLES = [
+    "search_vector_embeddings",
+    "search_vector_chunks",
+    "search_vector_index",
+]
+
+
+def _configured_postgres_sync_url() -> str | None:
+    """Prefer an externally managed Postgres server when CI provides one."""
+    configured_url = os.environ.get("BASIC_MEMORY_TEST_POSTGRES_URL") or os.environ.get(
+        "POSTGRES_TEST_URL"
+    )
+    if not configured_url:
+        return None
+
+    return (
+        configured_url.replace("postgresql+asyncpg://", "postgresql+psycopg2://", 1)
+        .replace("postgresql://", "postgresql+psycopg2://", 1)
+        .replace("postgres://", "postgresql+psycopg2://", 1)
+    )
+
+
+def _postgres_reset_tables() -> list[str]:
+    """Resolve the current ORM table set at reset time."""
+    return [table.name for table in Base.metadata.sorted_tables] + ["search_index"]
+
+
+def _resolve_postgres_sync_url(postgres_container) -> str:
+    """Use CI's shared service when configured, otherwise fall back to testcontainers."""
+    configured_url = _configured_postgres_sync_url()
+    if configured_url:
+        return configured_url
+    assert postgres_container is not None
+    return postgres_container.get_connection_url()
+
+
+async def _reset_postgres_integration_schema(engine) -> None:
+    """Restore the shared Postgres integration schema to a clean baseline."""
+    from basic_memory.models.search import (
+        CREATE_POSTGRES_SEARCH_INDEX_FTS,
+        CREATE_POSTGRES_SEARCH_INDEX_METADATA,
+        CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
+        CREATE_POSTGRES_SEARCH_INDEX_TABLE,
+    )
+
+    async with engine.begin() as conn:
+        # Trigger: integration tests may leave behind temporary search/vector tables while
+        # exercising full-stack recovery paths.
+        # Why: recreating only the missing schema is much cheaper than dropping every table.
+        # Outcome: each integration test gets the same baseline without paying repeated full DDL cost.
+        await conn.run_sync(Base.metadata.create_all)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)
+
+        for table_name in POSTGRES_EPHEMERAL_TABLES:
+            await conn.execute(text(f"DROP TABLE IF EXISTS {table_name} CASCADE"))
+
+        await conn.execute(
+            text(f"TRUNCATE TABLE {', '.join(_postgres_reset_tables())} RESTART IDENTITY CASCADE")
+        )
+
+
 @pytest_asyncio.fixture
 async def engine_factory(
     app_config,
@@ -121,18 +191,12 @@ async def engine_factory(
     tmp_path,
 ) -> AsyncGenerator[tuple, None]:
     """Create engine and session factory for the configured database backend."""
-    from basic_memory.models.search import (
-        CREATE_SEARCH_INDEX,
-        CREATE_POSTGRES_SEARCH_INDEX_TABLE,
-        CREATE_POSTGRES_SEARCH_INDEX_FTS,
-        CREATE_POSTGRES_SEARCH_INDEX_METADATA,
-        CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
-    )
+    from basic_memory.models.search import CREATE_SEARCH_INDEX
     from basic_memory import db
 
     if db_backend == "postgres":
         # Postgres mode using testcontainers
-        sync_url = postgres_container.get_connection_url()
+        sync_url = _resolve_postgres_sync_url(postgres_container)
         async_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
 
         engine = create_async_engine(
@@ -153,16 +217,7 @@ async def engine_factory(
         db._engine = engine
         db._session_maker = session_maker
 
-        # Drop and recreate all tables for test isolation
-        async with engine.begin() as conn:
-            await conn.execute(text("DROP TABLE IF EXISTS search_index CASCADE"))
-            await conn.run_sync(Base.metadata.drop_all)
-            await conn.run_sync(Base.metadata.create_all)
-            # asyncpg requires separate execute calls for each statement
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)
+        await _reset_postgres_integration_schema(engine)
 
         yield engine, session_maker
 
@@ -228,13 +283,15 @@ def app_config(
     monkeypatch.setenv("BASIC_MEMORY_CLOUD_MODE", "false")
 
     # Create a basic config with test-project like unit tests do
-    projects = {"test-project": str(config_home)}
+    projects = {"test-project": ProjectEntry(path=str(config_home))}
 
     # Configure database backend based on env var
     if db_backend == "postgres":
         database_backend = DatabaseBackend.POSTGRES
-        # Get URL from testcontainer and convert to asyncpg driver
-        sync_url = postgres_container.get_connection_url()
+        # Trigger: CI jobs can provide a shared Postgres service instead of per-session containers.
+        # Why: reusing one pgvector-enabled server avoids Docker startup churn on every job.
+        # Outcome: local runs keep using testcontainers, while CI injects a stable service URL.
+        sync_url = _resolve_postgres_sync_url(postgres_container)
         database_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
     else:
         database_backend = DatabaseBackend.SQLITE
@@ -285,7 +342,9 @@ def project_config(test_project):
 
 
 @pytest.fixture
-def app(app_config, project_config, engine_factory, test_project, config_manager) -> FastAPI:
+def app(
+    app_config, project_config, engine_factory, test_project, config_manager
+) -> Generator[FastAPI, None, None]:
     """Create test FastAPI application with single project."""
 
     # Import the FastAPI app AFTER the config_manager has written the test config to disk

tests/api/v2/test_knowledge_router.py

@@ -355,6 +355,7 @@ async def test_update_entity_by_id(
     response = await client.put(
         f"{v2_project_url}/knowledge/entities/{original_external_id}",
         json=update_data,
+        params={"fast": False},
     )
 
     assert response.status_code == 200
@@ -363,6 +364,8 @@ async def test_update_entity_by_id(
     # V2 update must return external_id field
     assert updated_entity.external_id is not None
     assert updated_entity.api_version == "v2"
+    assert updated_entity.content is not None
+    assert "Updated content via V2" in updated_entity.content
 
     # Verify file was updated
     file_path = file_service.get_entity_path(updated_entity)
@@ -532,6 +535,7 @@ async def test_edit_entity_by_id_append(
     response = await client.patch(
         f"{v2_project_url}/knowledge/entities/{original_external_id}",
         json=edit_data,
+        params={"fast": False},
     )
 
     assert response.status_code == 200
@@ -540,6 +544,8 @@ async def test_edit_entity_by_id_append(
     # V2 patch must return external_id field
     assert edited_entity.external_id is not None
     assert edited_entity.api_version == "v2"
+    assert edited_entity.content is not None
+    assert "Appended content" in edited_entity.content
 
     # Verify file has both original and appended content
     file_path = file_service.get_entity_path(edited_entity)

tests/api/v2/test_knowledge_router_telemetry.py

@@ -59,13 +59,21 @@ async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
     monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
 
     entity = _fake_entity()
+    response_content = (
+        "---\ntitle: Telemetry Entity\ntype: note\npermalink: notes/test\n---\n\ntelemetry content"
+    )
 
     class FakeEntityService:
-        async def create_entity(self, data):
-            return entity
+        async def create_entity_with_content(self, data):
+            return SimpleNamespace(
+                entity=entity,
+                content=response_content,
+                search_content="telemetry content",
+            )
 
     class FakeSearchService:
-        async def index_entity(self, entity):
+        async def index_entity(self, entity, content=None):
+            assert content == "telemetry content"
             return None
 
     class FakeTaskScheduler:
@@ -74,7 +82,7 @@ async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
 
     class FakeFileService:
         async def read_file_content(self, path):
-            return "telemetry content"
+            raise AssertionError("non-fast create should not re-read file content")
 
     result = await knowledge_router_module.create_entity(
         project_id="project-123",
@@ -94,7 +102,7 @@ async def test_create_entity_emits_root_and_nested_spans(monkeypatch) -> None:
         fast=False,
     )
 
-    assert result.content == "telemetry content"
+    assert result.content == response_content
     _assert_names_in_order(
         [name for name, _ in spans],
         [
@@ -113,13 +121,19 @@ async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
     monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
 
     entity = _fake_entity()
+    response_content = "---\ntitle: Telemetry Entity\ntype: note\npermalink: notes/test\n---\n\nupdated telemetry content"
 
     class FakeEntityService:
-        async def update_entity(self, existing, data):
-            return entity
+        async def update_entity_with_content(self, existing, data):
+            return SimpleNamespace(
+                entity=entity,
+                content=response_content,
+                search_content="updated telemetry content",
+            )
 
     class FakeSearchService:
-        async def index_entity(self, entity):
+        async def index_entity(self, entity, content=None):
+            assert content == "updated telemetry content"
             return None
 
     class FakeEntityRepository:
@@ -132,7 +146,7 @@ async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
 
     class FakeFileService:
         async def read_file_content(self, path):
-            return "updated telemetry content"
+            raise AssertionError("non-fast update should not re-read file content")
 
     response = Response()
     result = await knowledge_router_module.update_entity_by_id(
@@ -156,7 +170,7 @@ async def test_update_entity_emits_root_and_nested_spans(monkeypatch) -> None:
         fast=False,
     )
 
-    assert result.content == "updated telemetry content"
+    assert result.content == response_content
     _assert_names_in_order(
         [name for name, _ in spans],
         [
@@ -176,13 +190,19 @@ async def test_edit_entity_emits_root_and_nested_spans(monkeypatch) -> None:
     monkeypatch.setattr(knowledge_router_module.telemetry, "span", fake_span)
 
     entity = _fake_entity()
+    response_content = "---\ntitle: Telemetry Entity\ntype: note\npermalink: notes/test\n---\n\nedited telemetry content"
 
     class FakeEntityService:
-        async def edit_entity(self, **kwargs):
-            return entity
+        async def edit_entity_with_content(self, **kwargs):
+            return SimpleNamespace(
+                entity=entity,
+                content=response_content,
+                search_content="edited telemetry content",
+            )
 
     class FakeSearchService:
-        async def index_entity(self, entity):
+        async def index_entity(self, entity, content=None):
+            assert content == "edited telemetry content"
             return None
 
     class FakeEntityRepository:
@@ -195,7 +215,7 @@ async def test_edit_entity_emits_root_and_nested_spans(monkeypatch) -> None:
 
     class FakeFileService:
         async def read_file_content(self, path):
-            return "edited telemetry content"
+            raise AssertionError("non-fast edit should not re-read file content")
 
     result = await knowledge_router_module.edit_entity_by_id(
         data=EditEntityRequest(operation="append", content="edited telemetry content"),
@@ -211,7 +231,7 @@ async def test_edit_entity_emits_root_and_nested_spans(monkeypatch) -> None:
         fast=False,
     )
 
-    assert result.content == "edited telemetry content"
+    assert result.content == response_content
     _assert_names_in_order(
         [name for name, _ in spans],
         [

tests/api/v2/test_memory_hydration.py

@@ -0,0 +1,187 @@
+"""Tests for graph context hydration in to_graph_context().
+
+Proves that recent-activity/build-context hydration batches entity lookups
+for entities, observations, and relations in a single repository call.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from basic_memory.api.v2.utils import to_graph_context
+from basic_memory.schemas.search import SearchItemType
+from basic_memory.services.context_service import (
+    ContextMetadata,
+    ContextResult as ServiceContextResult,
+    ContextResultItem,
+    ContextResultRow,
+)
+
+
+# --- Helpers ---
+
+
+def _make_entity(id: int, title: str, external_id: str) -> SimpleNamespace:
+    return SimpleNamespace(id=id, title=title, external_id=external_id)
+
+
+def _make_row(*, type: str, id: int, root_id: int, **kwargs) -> ContextResultRow:
+    now = kwargs.pop("created_at", datetime.now(timezone.utc))
+    defaults = dict(
+        title=f"Item {id}",
+        permalink=f"notes/{id}",
+        file_path=f"notes/{id}.md",
+        depth=0,
+        root_id=root_id,
+        created_at=now,
+    )
+    defaults.update(kwargs)
+    return ContextResultRow(type=type, id=id, **defaults)
+
+
+class SpyEntityRepository:
+    """Tracks batched ID lookups and returns entities from a preset map."""
+
+    def __init__(self, entities_by_id: dict[int, SimpleNamespace]):
+        self.entities_by_id = entities_by_id
+        self.calls: list[list[int]] = []
+
+    async def find_by_ids(self, ids: list[int]):
+        self.calls.append(ids)
+        return [self.entities_by_id[i] for i in ids if i in self.entities_by_id]
+
+
+# --- Single batch fetch (N+1 elimination) ---
+
+
+@pytest.mark.asyncio
+async def test_to_graph_context_batches_entity_hydration_for_recent_activity():
+    """Mixed entity, observation, and relation items must hydrate in one lookup."""
+    repo = SpyEntityRepository(
+        {
+            1: _make_entity(1, "Root", "ext-root"),
+            2: _make_entity(2, "Child", "ext-child"),
+            3: _make_entity(3, "Peer", "ext-peer"),
+        }
+    )
+    now = datetime.now(timezone.utc)
+
+    root_entity = _make_row(
+        type="entity",
+        id=1,
+        root_id=1,
+        title="Root",
+        permalink="notes/root",
+        file_path="notes/root.md",
+        created_at=now,
+    )
+    root_observation = _make_row(
+        type="observation",
+        id=10,
+        root_id=1,
+        title="fact: observed",
+        permalink="notes/root/observations/fact/observed",
+        file_path="notes/root.md",
+        category="fact",
+        content="observed",
+        entity_id=1,
+        created_at=now,
+    )
+    root_relation = _make_row(
+        type="relation",
+        id=20,
+        root_id=1,
+        title="links_to: Child",
+        permalink="notes/root",
+        file_path="notes/root.md",
+        relation_type="links_to",
+        from_id=1,
+        to_id=2,
+        depth=1,
+        created_at=now,
+    )
+    child_observation = _make_row(
+        type="observation",
+        id=11,
+        root_id=11,
+        title="note: child update",
+        permalink="notes/child/observations/note/update",
+        file_path="notes/child.md",
+        category="note",
+        content="child update",
+        entity_id=2,
+        created_at=now,
+    )
+    peer_entity = _make_row(
+        type="entity",
+        id=3,
+        root_id=11,
+        title="Peer",
+        permalink="notes/peer",
+        file_path="notes/peer.md",
+        depth=1,
+        created_at=now,
+    )
+
+    context = ServiceContextResult(
+        results=[
+            ContextResultItem(
+                primary_result=root_entity,
+                observations=[root_observation],
+                related_results=[root_relation],
+            ),
+            ContextResultItem(
+                primary_result=child_observation,
+                observations=[],
+                related_results=[peer_entity],
+            ),
+        ],
+        metadata=ContextMetadata(
+            types=[
+                SearchItemType.ENTITY,
+                SearchItemType.OBSERVATION,
+                SearchItemType.RELATION,
+            ],
+            depth=1,
+            primary_count=2,
+            related_count=2,
+            total_relations=1,
+            total_observations=1,
+        ),
+    )
+
+    graph = await to_graph_context(context, entity_repository=repo, page=1, page_size=10)
+
+    assert len(repo.calls) == 1, f"Expected 1 entity lookup, got {len(repo.calls)}"
+    assert set(repo.calls[0]) == {1, 2, 3}
+
+    first_result = graph.results[0]
+    assert first_result.primary_result.external_id == "ext-root"
+    assert first_result.observations[0].entity_external_id == "ext-root"
+    assert first_result.observations[0].title == "Root"
+
+    relation = first_result.related_results[0]
+    assert relation.from_entity == "Root"
+    assert relation.from_entity_external_id == "ext-root"
+    assert relation.to_entity == "Child"
+    assert relation.to_entity_external_id == "ext-child"
+
+    second_result = graph.results[1]
+    assert second_result.primary_result.entity_external_id == "ext-child"
+    assert second_result.primary_result.title == "Child"
+    assert second_result.related_results[0].external_id == "ext-peer"
+
+
+@pytest.mark.asyncio
+async def test_to_graph_context_empty_results_skip_entity_lookup():
+    """An empty context result should not perform any entity hydration lookup."""
+    repo = SpyEntityRepository({})
+    context = ServiceContextResult(results=[], metadata=ContextMetadata(depth=1))
+
+    graph = await to_graph_context(context, entity_repository=repo)
+
+    assert repo.calls == []
+    assert list(graph.results) == []

tests/api/v2/test_search_hydration.py

@@ -0,0 +1,305 @@
+"""Tests for search result hydration in to_search_results().
+
+Proves that the batch fetch eliminates N+1 queries and that
+entity ID lookups are correct across all result types.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from basic_memory.api.v2.utils import to_search_results
+from basic_memory.repository.search_index_row import SearchIndexRow
+
+
+# --- Helpers ---
+
+
+def _make_entity(id: int, permalink: str) -> SimpleNamespace:
+    return SimpleNamespace(id=id, permalink=permalink)
+
+
+def _make_row(*, type: str, id: int, **kwargs) -> SearchIndexRow:
+    now = datetime.now(timezone.utc)
+    defaults = dict(
+        project_id=1,
+        file_path=f"notes/{id}.md",
+        created_at=now,
+        updated_at=now,
+        score=1.0,
+        title=f"Item {id}",
+        permalink=f"notes/{id}",
+    )
+    defaults.update(kwargs)
+    return SearchIndexRow(type=type, id=id, **defaults)
+
+
+class SpyEntityService:
+    """Tracks calls to get_entities_by_id and returns from a preset lookup."""
+
+    def __init__(self, entities_by_id: dict[int, SimpleNamespace]):
+        self.entities_by_id = entities_by_id
+        self.calls: list[list[int]] = []
+
+    async def get_entities_by_id(self, ids: list[int]):
+        self.calls.append(ids)
+        return [self.entities_by_id[i] for i in ids if i in self.entities_by_id]
+
+
+# --- Single batch fetch (N+1 elimination) ---
+
+
+@pytest.mark.asyncio
+async def test_single_db_call_for_multiple_results():
+    """Multiple search results must trigger exactly one get_entities_by_id call."""
+    service = SpyEntityService(
+        {
+            1: _make_entity(1, "notes/a"),
+            2: _make_entity(2, "notes/b"),
+            3: _make_entity(3, "notes/c"),
+        }
+    )
+    results = [
+        _make_row(type="entity", id=1, entity_id=1),
+        _make_row(type="entity", id=2, entity_id=2),
+        _make_row(type="entity", id=3, entity_id=3),
+    ]
+
+    await to_search_results(service, results)
+
+    assert len(service.calls) == 1, f"Expected 1 DB call, got {len(service.calls)}"
+
+
+@pytest.mark.asyncio
+async def test_no_db_call_for_empty_results():
+    """Empty result list should not make any DB call."""
+    service = SpyEntityService({})
+
+    search_results = await to_search_results(service, [])
+
+    assert len(service.calls) == 0
+    assert search_results == []
+
+
+# --- ID deduplication ---
+
+
+@pytest.mark.asyncio
+async def test_deduplicates_entity_ids():
+    """Shared entity IDs across results should be fetched once, not per-result."""
+    # entity_id=1 appears in all three results, from_id=1 overlaps with entity_id
+    service = SpyEntityService(
+        {
+            1: _make_entity(1, "notes/shared"),
+            2: _make_entity(2, "notes/target-a"),
+            3: _make_entity(3, "notes/target-b"),
+        }
+    )
+    results = [
+        _make_row(type="relation", id=10, entity_id=1, from_id=1, to_id=2, relation_type="links"),
+        _make_row(type="relation", id=11, entity_id=1, from_id=1, to_id=3, relation_type="links"),
+    ]
+
+    await to_search_results(service, results)
+
+    # Single call with deduplicated IDs: {1, 2, 3}
+    assert len(service.calls) == 1
+    fetched_ids = set(service.calls[0])
+    assert fetched_ids == {1, 2, 3}
+
+
+# --- Correct entity-to-field mapping ---
+
+
+@pytest.mark.asyncio
+async def test_entity_result_maps_permalink():
+    """Entity results should populate the 'entity' field with the entity's permalink."""
+    service = SpyEntityService({5: _make_entity(5, "notes/my-entity")})
+    results = [_make_row(type="entity", id=5, entity_id=5)]
+
+    search_results = await to_search_results(service, results)
+
+    assert len(search_results) == 1
+    r = search_results[0]
+    assert r.entity == "notes/my-entity"
+    assert r.entity_id == 5
+    assert r.from_entity is None
+    assert r.to_entity is None
+
+
+@pytest.mark.asyncio
+async def test_observation_result_maps_parent_entity():
+    """Observation results should populate 'entity' with the parent entity's permalink."""
+    service = SpyEntityService({10: _make_entity(10, "notes/parent")})
+    results = [_make_row(type="observation", id=20, entity_id=10)]
+
+    search_results = await to_search_results(service, results)
+
+    r = search_results[0]
+    assert r.entity == "notes/parent"
+    assert r.entity_id == 10
+    assert r.observation_id == 20
+    assert r.from_entity is None
+    assert r.to_entity is None
+
+
+@pytest.mark.asyncio
+async def test_relation_result_maps_from_and_to():
+    """Relation results should populate entity, from_entity, and to_entity correctly."""
+    service = SpyEntityService(
+        {
+            1: _make_entity(1, "notes/parent"),
+            2: _make_entity(2, "notes/source"),
+            3: _make_entity(3, "notes/target"),
+        }
+    )
+    results = [
+        _make_row(
+            type="relation",
+            id=99,
+            entity_id=1,
+            from_id=2,
+            to_id=3,
+            relation_type="references",
+        )
+    ]
+
+    search_results = await to_search_results(service, results)
+
+    r = search_results[0]
+    assert r.entity == "notes/parent"
+    assert r.from_entity == "notes/source"
+    assert r.to_entity == "notes/target"
+    assert r.relation_id == 99
+    assert r.relation_type == "references"
+
+
+@pytest.mark.asyncio
+async def test_relation_with_distinct_entity_and_from_ids():
+    """When entity_id != from_id, from_entity must use from_id's permalink, not entity_id's.
+
+    This was a bug in the old positional-index code: entities[0] was used for both
+    'entity' and 'from_entity', which was wrong when entity_id != from_id.
+    """
+    service = SpyEntityService(
+        {
+            10: _make_entity(10, "notes/parent-entity"),
+            20: _make_entity(20, "notes/actual-source"),
+            30: _make_entity(30, "notes/target"),
+        }
+    )
+    results = [
+        _make_row(
+            type="relation",
+            id=50,
+            entity_id=10,
+            from_id=20,
+            to_id=30,
+            relation_type="derived_from",
+        )
+    ]
+
+    search_results = await to_search_results(service, results)
+
+    r = search_results[0]
+    # entity should be the parent entity (entity_id=10)
+    assert r.entity == "notes/parent-entity"
+    # from_entity must be from_id=20, NOT entity_id=10
+    assert r.from_entity == "notes/actual-source"
+    assert r.to_entity == "notes/target"
+
+
+# --- Mixed result types ---
+
+
+@pytest.mark.asyncio
+async def test_mixed_result_types_single_fetch():
+    """A mix of entity, observation, and relation results should all hydrate in one fetch."""
+    service = SpyEntityService(
+        {
+            1: _make_entity(1, "notes/entity-one"),
+            2: _make_entity(2, "notes/entity-two"),
+            3: _make_entity(3, "notes/entity-three"),
+        }
+    )
+    results = [
+        _make_row(type="entity", id=1, entity_id=1),
+        _make_row(type="observation", id=10, entity_id=2, category="fact"),
+        _make_row(type="relation", id=20, entity_id=1, from_id=1, to_id=3, relation_type="links"),
+    ]
+
+    search_results = await to_search_results(service, results)
+
+    # Single DB call
+    assert len(service.calls) == 1
+
+    # Entity result
+    assert search_results[0].entity == "notes/entity-one"
+    assert search_results[0].entity_id == 1
+
+    # Observation result
+    assert search_results[1].entity == "notes/entity-two"
+    assert search_results[1].observation_id == 10
+
+    # Relation result
+    assert search_results[2].from_entity == "notes/entity-one"
+    assert search_results[2].to_entity == "notes/entity-three"
+
+
+# --- Graceful handling of missing entities ---
+
+
+@pytest.mark.asyncio
+async def test_missing_entity_returns_none_permalink():
+    """If an entity ID isn't found in the DB, permalink fields should be None."""
+    # Only entity 1 exists; entity 99 (to_id) is missing
+    service = SpyEntityService({1: _make_entity(1, "notes/source")})
+    results = [
+        _make_row(type="relation", id=5, entity_id=1, from_id=1, to_id=99, relation_type="links")
+    ]
+
+    search_results = await to_search_results(service, results)
+
+    r = search_results[0]
+    assert r.entity == "notes/source"
+    assert r.from_entity == "notes/source"
+    assert r.to_entity is None  # entity 99 not found
+
+
+@pytest.mark.asyncio
+async def test_null_ids_handled_gracefully():
+    """Results with None entity_id/from_id/to_id should not cause errors."""
+    service = SpyEntityService({})
+    # Entity result: entity_id is the row id itself, from_id/to_id are None
+    results = [_make_row(type="entity", id=1)]
+
+    search_results = await to_search_results(service, results)
+
+    # No entity_id on the row means no fetch needed, all fields None
+    r = search_results[0]
+    assert r.entity is None
+    assert r.from_entity is None
+    assert r.to_entity is None
+
+
+# --- Scaling: prove O(1) DB calls ---
+
+
+@pytest.mark.asyncio
+async def test_single_db_call_scales_to_many_results():
+    """Even with many results, only one DB call should be made."""
+    n = 50
+    entities = {i: _make_entity(i, f"notes/e-{i}") for i in range(1, n + 1)}
+    service = SpyEntityService(entities)
+    results = [_make_row(type="entity", id=i, entity_id=i) for i in range(1, n + 1)]
+
+    search_results = await to_search_results(service, results)
+
+    assert len(service.calls) == 1, f"Expected 1 DB call for {n} results, got {len(service.calls)}"
+    assert len(search_results) == n
+    # Every result got its permalink
+    for i, r in enumerate(search_results, start=1):
+        assert r.entity == f"notes/e-{i}"

tests/api/v2/test_utils_telemetry.py

@@ -32,7 +32,6 @@ async def test_to_search_results_emits_hydration_spans(monkeypatch) -> None:
 
     class FakeEntityService:
         async def get_entities_by_id(self, ids):
-            # Return entities with .id so the batch lookup dict can be built
             return [
                 SimpleNamespace(id=1, permalink="notes/root"),
                 SimpleNamespace(id=2, permalink="notes/child"),
@@ -60,9 +59,6 @@ async def test_to_search_results_emits_hydration_spans(monkeypatch) -> None:
     search_results = await utils_module.to_search_results(FakeEntityService(), results)
 
     assert search_results[0].relation_type == "relates_to"
-    # from_entity and to_entity are now correctly resolved via lookup dict
-    assert search_results[0].from_entity == "notes/root"
-    assert search_results[0].to_entity == "notes/child"
     assert [name for name, _ in spans] == [
         "search.hydrate_results",
         "search.hydrate_results.fetch_entities",

tests/cli/cloud/test_cloud_api_client_and_utils.py

@@ -10,10 +10,12 @@ from basic_memory.cli.commands.cloud.api_client import (
     make_api_request,
 )
 from basic_memory.cli.commands.cloud.cloud_utils import (
+    CloudUtilsError,
     create_cloud_project,
     fetch_cloud_projects,
     project_exists,
 )
+from basic_memory.config import ProjectMode
 
 
 @pytest.mark.asyncio
@@ -163,6 +165,109 @@ async def test_cloud_utils_fetch_and_exists_and_create_project(
     assert created.new_project["name"] == "My Project"
     # Path should be permalink-like (kebab)
     assert seen["create_payload"]["path"] == "my-project"
+    assert seen["create_payload"]["visibility"] == "workspace"
+
+
+@pytest.mark.asyncio
+async def test_create_cloud_project_accepts_visibility_override(config_home, config_manager):
+    """Shared cloud helper should pass explicit visibility through to the API payload."""
+    config = config_manager.load_config()
+    config.cloud_host = "https://cloud.example.test"
+    config_manager.save_config(config)
+
+    seen_payload: dict | None = None
+
+    async def api_request(**kwargs):
+        nonlocal seen_payload
+        seen_payload = kwargs["json_data"]
+        return httpx.Response(
+            200,
+            json={
+                "message": "created",
+                "status": "success",
+                "default": False,
+                "old_project": None,
+                "new_project": {"name": "shared-project", "path": "shared-project"},
+            },
+        )
+
+    created = await create_cloud_project(
+        "Shared Project",
+        visibility="shared",
+        api_request=api_request,
+    )
+
+    assert created.new_project is not None
+    assert seen_payload == {
+        "name": "Shared Project",
+        "path": "shared-project",
+        "set_default": False,
+        "visibility": "shared",
+    }
+
+
+@pytest.mark.asyncio
+async def test_cloud_utils_use_configured_workspace_headers(config_home, config_manager):
+    """Workspace-aware cloud helpers should prefer project workspace over global default."""
+    config = config_manager.load_config()
+    config.cloud_host = "https://cloud.example.test"
+    config.default_workspace = "default-workspace"
+    config.set_project_mode("alpha", ProjectMode.CLOUD)
+    config.projects["alpha"].workspace_id = "project-workspace"
+    config_manager.save_config(config)
+
+    seen: list[tuple[str, str | None]] = []
+
+    async def api_request(**kwargs):
+        seen.append(
+            (
+                kwargs["method"],
+                (kwargs.get("headers") or {}).get("X-Workspace-ID"),
+            )
+        )
+
+        if kwargs["method"] == "GET":
+            return httpx.Response(
+                200,
+                json={
+                    "projects": [{"id": 1, "name": "alpha", "path": "alpha", "is_default": True}]
+                },
+            )
+
+        return httpx.Response(
+            200,
+            json={
+                "message": "created",
+                "status": "success",
+                "default": False,
+                "old_project": None,
+                "new_project": {"name": "alpha", "path": "alpha"},
+            },
+        )
+
+    assert await project_exists("alpha", api_request=api_request) is True
+    await create_cloud_project("alpha", api_request=api_request)
+    await fetch_cloud_projects(project_name="missing", api_request=api_request)
+
+    assert seen == [
+        ("GET", "project-workspace"),
+        ("POST", "project-workspace"),
+        ("GET", "default-workspace"),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_project_exists_surfaces_cloud_lookup_failures(config_home, config_manager):
+    """project_exists should surface lookup failures instead of pretending the project is missing."""
+    config = config_manager.load_config()
+    config.cloud_host = "https://cloud.example.test"
+    config_manager.save_config(config)
+
+    async def api_request(**_kwargs):
+        raise httpx.ConnectError("boom")
+
+    with pytest.raises(CloudUtilsError, match="Failed to fetch cloud projects"):
+        await project_exists("alpha", api_request=api_request)
 
 
 @pytest.mark.asyncio

tests/cli/cloud/test_project_sync_command.py

@@ -0,0 +1,114 @@
+"""Tests for cloud sync and bisync command behavior."""
+
+import importlib
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+
+import pytest
+from typer.testing import CliRunner
+
+from basic_memory.cli.app import app
+from basic_memory.config import ProjectMode
+
+runner = CliRunner()
+
+
+@pytest.mark.parametrize(
+    "argv",
+    [
+        ["cloud", "sync", "--name", "research"],
+        ["cloud", "bisync", "--name", "research"],
+    ],
+)
+def test_cloud_sync_commands_use_incremental_db_sync(monkeypatch, argv, config_manager):
+    """Cloud sync commands should not force a full database re-index after file sync."""
+    project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
+
+    seen: dict[str, object] = {}
+    config = config_manager.load_config()
+    config.set_project_mode("research", ProjectMode.CLOUD)
+    config_manager.save_config(config)
+
+    monkeypatch.setattr(project_sync_command, "_require_cloud_credentials", lambda _config: None)
+    monkeypatch.setattr(
+        project_sync_command,
+        "get_mount_info",
+        lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
+    )
+    monkeypatch.setattr(
+        project_sync_command,
+        "_get_cloud_project",
+        lambda _name: _async_value(
+            SimpleNamespace(name="research", external_id="external-project-id", path="research")
+        ),
+    )
+    monkeypatch.setattr(
+        project_sync_command,
+        "_get_sync_project",
+        lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
+    )
+    monkeypatch.setattr(project_sync_command, "project_sync", lambda *args, **kwargs: True)
+    monkeypatch.setattr(project_sync_command, "project_bisync", lambda *args, **kwargs: True)
+
+    @asynccontextmanager
+    async def fake_get_client(*, project_name=None, workspace=None):
+        seen["project_name"] = project_name
+        seen["workspace"] = workspace
+        yield object()
+
+    class FakeProjectClient:
+        def __init__(self, _client):
+            pass
+
+        async def sync(self, external_id: str, force_full: bool = False):
+            seen["external_id"] = external_id
+            seen["force_full"] = force_full
+            return {"message": "queued"}
+
+    monkeypatch.setattr(project_sync_command, "get_client", fake_get_client)
+    monkeypatch.setattr(project_sync_command, "ProjectClient", FakeProjectClient)
+
+    result = runner.invoke(app, argv)
+
+    assert result.exit_code == 0, result.output
+    assert seen["project_name"] == "research"
+    assert seen["external_id"] == "external-project-id"
+    assert seen["force_full"] is False
+
+
+def test_cloud_bisync_fails_fast_when_sync_entry_disappears(monkeypatch, config_manager):
+    """Bisync should raise a runtime error when validated sync config vanishes before persistence."""
+    project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
+
+    config = config_manager.load_config()
+    config.projects.pop("research", None)
+    config_manager.save_config(config)
+
+    monkeypatch.setattr(project_sync_command, "_require_cloud_credentials", lambda _config: None)
+    monkeypatch.setattr(
+        project_sync_command,
+        "get_mount_info",
+        lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
+    )
+    monkeypatch.setattr(
+        project_sync_command,
+        "_get_cloud_project",
+        lambda _name: _async_value(
+            SimpleNamespace(name="research", external_id="external-project-id", path="research")
+        ),
+    )
+    monkeypatch.setattr(
+        project_sync_command,
+        "_get_sync_project",
+        lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
+    )
+    monkeypatch.setattr(project_sync_command, "project_bisync", lambda *args, **kwargs: True)
+
+    result = runner.invoke(app, ["cloud", "bisync", "--name", "research"])
+
+    assert result.exit_code == 1, result.output
+    assert "unexpectedly missing after validation" in result.output
+
+
+async def _async_value(value):
+    return value

tests/cli/cloud/test_upload_command_routing.py

@@ -6,6 +6,8 @@ import httpx
 from typer.testing import CliRunner
 
 from basic_memory.cli.app import app
+from basic_memory.cli.commands.cloud.cloud_utils import CloudUtilsError
+from basic_memory.config import ProjectMode
 
 runner = CliRunner()
 
@@ -20,11 +22,11 @@ def test_cloud_upload_uses_control_plane_client(monkeypatch, tmp_path):
 
     seen: dict[str, str] = {}
 
-    async def fake_project_exists(_project_name: str) -> bool:
+    async def fake_project_exists(_project_name: str, workspace: str | None = None) -> bool:
         return True
 
     @asynccontextmanager
-    async def fake_get_client():
+    async def fake_get_client(workspace: str | None = None):
         async with httpx.AsyncClient(base_url="https://cloud.example.test") as client:
             yield client
 
@@ -53,3 +55,89 @@ def test_cloud_upload_uses_control_plane_client(monkeypatch, tmp_path):
 
     assert result.exit_code == 0, result.output
     assert seen["base_url"] == "https://cloud.example.test"
+
+
+def test_cloud_upload_uses_project_workspace_for_api_and_webdav(
+    monkeypatch, tmp_path, config_manager
+):
+    """Upload command should reuse the configured workspace across API and WebDAV calls."""
+    import basic_memory.cli.commands.cloud.upload_command as upload_command
+
+    config = config_manager.load_config()
+    config.default_workspace = "default-workspace"
+    config.set_project_mode("routing-test", ProjectMode.CLOUD)
+    config.projects["routing-test"].workspace_id = "project-workspace"
+    config_manager.save_config(config)
+
+    upload_dir = tmp_path / "upload"
+    upload_dir.mkdir()
+    (upload_dir / "note.md").write_text("hello", encoding="utf-8")
+
+    seen: dict[str, str | None] = {}
+
+    async def fake_project_exists(_project_name: str, workspace: str | None = None) -> bool:
+        seen["project_exists_workspace"] = workspace
+        return True
+
+    @asynccontextmanager
+    async def fake_get_client(workspace: str | None = None):
+        seen["control_plane_workspace"] = workspace
+        async with httpx.AsyncClient(base_url="https://cloud.example.test") as client:
+            yield client
+
+    async def fake_upload_path(*args, **kwargs):
+        client_cm_factory = kwargs.get("client_cm_factory")
+        assert client_cm_factory is not None
+        async with client_cm_factory() as client:
+            seen["base_url"] = str(client.base_url).rstrip("/")
+        return True
+
+    monkeypatch.setattr(upload_command, "project_exists", fake_project_exists)
+    monkeypatch.setattr(upload_command, "get_cloud_control_plane_client", fake_get_client)
+    monkeypatch.setattr(upload_command, "upload_path", fake_upload_path)
+
+    result = runner.invoke(
+        app,
+        [
+            "cloud",
+            "upload",
+            str(upload_dir),
+            "--project",
+            "routing-test",
+            "--no-sync",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert seen["project_exists_workspace"] == "project-workspace"
+    assert seen["control_plane_workspace"] == "project-workspace"
+    assert seen["base_url"] == "https://cloud.example.test"
+
+
+def test_cloud_upload_exits_when_project_lookup_fails(monkeypatch, tmp_path):
+    """Upload command should fail fast when cloud project lookup cannot reach the API."""
+    import basic_memory.cli.commands.cloud.upload_command as upload_command
+
+    upload_dir = tmp_path / "upload"
+    upload_dir.mkdir()
+    (upload_dir / "note.md").write_text("hello", encoding="utf-8")
+
+    async def fake_project_exists(_project_name: str, workspace: str | None = None) -> bool:
+        raise CloudUtilsError("lookup failed")
+
+    monkeypatch.setattr(upload_command, "project_exists", fake_project_exists)
+
+    result = runner.invoke(
+        app,
+        [
+            "cloud",
+            "upload",
+            str(upload_dir),
+            "--project",
+            "routing-test",
+            "--no-sync",
+        ],
+    )
+
+    assert result.exit_code == 1, result.output
+    assert "Failed to check cloud project 'routing-test'" in result.output

tests/cli/test_project_add_with_local_path.py

@@ -52,9 +52,11 @@ def mock_config(tmp_path, monkeypatch):
 @pytest.fixture
 def mock_api_client(monkeypatch):
     """Stub the API client for project add without stdlib mocks."""
+    seen_workspaces: list[str | None] = []
 
     @asynccontextmanager
-    async def fake_get_client():
+    async def fake_get_client(*, workspace=None):
+        seen_workspaces.append(workspace)
         yield object()
 
     _response_data = {
@@ -80,7 +82,7 @@ def mock_api_client(monkeypatch):
     monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
     monkeypatch.setattr(ProjectClient, "create_project", fake_create_project)
 
-    return calls
+    return {"calls": calls, "workspaces": seen_workspaces}
 
 
 def test_project_add_with_local_path_saves_to_config(
@@ -113,6 +115,7 @@ def test_project_add_with_local_path_saves_to_config(
     assert "test-project" in config_data["projects"]
     entry = config_data["projects"]["test-project"]
     # Use as_posix() for cross-platform compatibility (Windows uses backslashes)
+    assert entry["mode"] == "cloud"
     assert entry["local_sync_path"] == local_sync_dir.as_posix()
     assert entry.get("last_sync") is None
     assert entry.get("bisync_initialized", False) is False
@@ -173,3 +176,162 @@ def test_project_add_local_path_creates_nested_directories(
     assert result.exit_code == 0
     assert nested_path.exists()
     assert nested_path.is_dir()
+
+
+def test_project_add_cloud_visibility_passes_payload(runner, mock_config, mock_api_client):
+    """Cloud project creation should forward visibility to the API payload."""
+    result = runner.invoke(
+        app,
+        ["project", "add", "test-project", "--cloud", "--visibility", "shared"],
+    )
+
+    assert result.exit_code == 0
+    assert mock_api_client["workspaces"] == [None]
+    assert mock_api_client["calls"] == [
+        {
+            "name": "test-project",
+            "path": "test-project",
+            "local_sync_path": None,
+            "set_default": False,
+            "visibility": "shared",
+        }
+    ]
+
+
+def test_project_add_cloud_workspace_resolves_and_persists(
+    runner, mock_config, mock_api_client, monkeypatch, tmp_path
+):
+    """Cloud project add should resolve workspace names to tenant IDs."""
+    from basic_memory.schemas.cloud import WorkspaceInfo
+
+    local_sync_dir = tmp_path / "sync" / "team-notes"
+
+    async def fake_get_available_workspaces():
+        return [
+            WorkspaceInfo(
+                tenant_id="11111111-1111-1111-1111-111111111111",
+                workspace_type="organization",
+                name="Basic Memory",
+                role="owner",
+            ),
+        ]
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "project",
+            "add",
+            "team-notes",
+            "--cloud",
+            "--workspace",
+            "Basic Memory",
+            "--local-path",
+            str(local_sync_dir),
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert mock_api_client["workspaces"] == ["11111111-1111-1111-1111-111111111111"]
+    assert mock_api_client["calls"] == [
+        {
+            "name": "team-notes",
+            "path": "team-notes",
+            "local_sync_path": local_sync_dir.as_posix(),
+            "set_default": False,
+            "visibility": "workspace",
+        }
+    ]
+
+    config_data = json.loads(mock_config.read_text())
+    entry = config_data["projects"]["team-notes"]
+    assert entry["mode"] == "cloud"
+    assert entry["workspace_id"] == "11111111-1111-1111-1111-111111111111"
+
+
+def test_project_add_cloud_workspace_persists_without_local_path(
+    runner, mock_config, mock_api_client, monkeypatch
+):
+    """Cloud project add should persist workspace routing even without local sync."""
+    from basic_memory.schemas.cloud import WorkspaceInfo
+
+    async def fake_get_available_workspaces():
+        return [
+            WorkspaceInfo(
+                tenant_id="11111111-1111-1111-1111-111111111111",
+                workspace_type="organization",
+                name="Basic Memory",
+                role="owner",
+            ),
+        ]
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "project",
+            "add",
+            "team-notes",
+            "--cloud",
+            "--workspace",
+            "Basic Memory",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert mock_api_client["workspaces"] == ["11111111-1111-1111-1111-111111111111"]
+    assert mock_api_client["calls"] == [
+        {
+            "name": "team-notes",
+            "path": "team-notes",
+            "local_sync_path": None,
+            "set_default": False,
+            "visibility": "workspace",
+        }
+    ]
+
+    config_data = json.loads(mock_config.read_text())
+    entry = config_data["projects"]["team-notes"]
+    assert entry["path"] == ""
+    assert entry["mode"] == "cloud"
+    assert entry["workspace_id"] == "11111111-1111-1111-1111-111111111111"
+    assert entry["local_sync_path"] is None
+
+
+def test_project_add_visibility_requires_cloud_mode(runner, mock_config, tmp_path):
+    """Visibility is a cloud-only option."""
+    project_path = tmp_path / "local-project"
+
+    result = runner.invoke(
+        app,
+        [
+            "project",
+            "add",
+            "local-project",
+            str(project_path),
+            "--visibility",
+            "shared",
+        ],
+    )
+
+    assert result.exit_code == 1
+    assert "--visibility is only supported in cloud mode" in result.stdout
+
+
+def test_project_add_rejects_invalid_visibility(runner, mock_config):
+    """Invalid visibility values should fail fast before the API call."""
+    result = runner.invoke(
+        app,
+        ["project", "add", "test-project", "--cloud", "--visibility", "team-only"],
+    )
+
+    assert result.exit_code == 1
+    assert "Invalid visibility" in result.stdout

tests/conftest.py

@@ -22,7 +22,13 @@ from sqlalchemy.pool import NullPool
 from testcontainers.postgres import PostgresContainer
 
 from basic_memory import db
-from basic_memory.config import ProjectConfig, BasicMemoryConfig, ConfigManager, DatabaseBackend
+from basic_memory.config import (
+    ProjectConfig,
+    ProjectEntry,
+    BasicMemoryConfig,
+    ConfigManager,
+    DatabaseBackend,
+)
 from basic_memory.db import DatabaseType
 from basic_memory.markdown import EntityParser
 from basic_memory.markdown.markdown_processor import MarkdownProcessor
@@ -74,7 +80,7 @@ def postgres_container(db_backend):
     The container is started once per test session and shared across all tests.
     Only starts if db_backend is "postgres".
     """
-    if db_backend != "postgres":
+    if db_backend != "postgres" or _configured_postgres_sync_url():
         yield None
         return
 
@@ -83,6 +89,100 @@ def postgres_container(db_backend):
         yield postgres
 
 
+POSTGRES_EPHEMERAL_TABLES = [
+    "search_vector_embeddings",
+    "search_vector_index",
+]
+
+
+def _configured_postgres_sync_url() -> str | None:
+    """Prefer an externally managed Postgres server when CI provides one."""
+    configured_url = os.environ.get("BASIC_MEMORY_TEST_POSTGRES_URL") or os.environ.get(
+        "POSTGRES_TEST_URL"
+    )
+    if not configured_url:
+        return None
+
+    return (
+        configured_url.replace("postgresql+asyncpg://", "postgresql+psycopg2://", 1)
+        .replace("postgresql://", "postgresql+psycopg2://", 1)
+        .replace("postgres://", "postgresql+psycopg2://", 1)
+    )
+
+
+def _postgres_alembic_config(async_url: str) -> Config:
+    """Build Alembic config for stamping the shared Postgres test schema."""
+    alembic_dir = Path(db.__file__).parent / "alembic"
+    cfg = Config()
+    cfg.set_main_option("script_location", str(alembic_dir))
+    cfg.set_main_option(
+        "file_template",
+        "%%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s",
+    )
+    cfg.set_main_option("timezone", "UTC")
+    cfg.set_main_option("revision_environment", "false")
+    cfg.set_main_option("sqlalchemy.url", async_url)
+    return cfg
+
+
+def _postgres_reset_tables() -> list[str]:
+    """Resolve the current ORM table set at reset time.
+
+    Some tests declare models after conftest import, so the list must stay dynamic.
+    """
+    return [table.name for table in Base.metadata.sorted_tables] + [
+        "search_index",
+        "search_vector_chunks",
+    ]
+
+
+def _resolve_postgres_sync_url(postgres_container) -> str:
+    """Use CI's shared service when configured, otherwise fall back to testcontainers."""
+    configured_url = _configured_postgres_sync_url()
+    if configured_url:
+        return configured_url
+    assert postgres_container is not None
+    return postgres_container.get_connection_url()
+
+
+async def _reset_postgres_test_schema(engine: AsyncEngine, async_url: str) -> None:
+    """Restore the shared Postgres schema to a clean baseline before each test."""
+    from basic_memory.models.search import (
+        CREATE_POSTGRES_SEARCH_INDEX_FTS,
+        CREATE_POSTGRES_SEARCH_INDEX_METADATA,
+        CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
+        CREATE_POSTGRES_SEARCH_INDEX_TABLE,
+        CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_INDEX,
+        CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_TABLE,
+    )
+
+    async with engine.begin() as conn:
+        # Trigger: several tests intentionally drop or stub search tables to exercise recovery code.
+        # Why: TRUNCATE is much cheaper than drop_all/create_all, but it only works when the schema exists.
+        # Outcome: we recreate any missing core tables once, then clear rows for deterministic test setup.
+        await conn.run_sync(Base.metadata.create_all)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
+        await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)
+        await conn.execute(CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_TABLE)
+        await conn.execute(CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_INDEX)
+
+        for table_name in POSTGRES_EPHEMERAL_TABLES:
+            await conn.execute(text(f"DROP TABLE IF EXISTS {table_name} CASCADE"))
+
+        await conn.execute(
+            text(f"TRUNCATE TABLE {', '.join(_postgres_reset_tables())} RESTART IDENTITY CASCADE")
+        )
+
+        alembic_version_exists = (
+            await conn.execute(text("SELECT to_regclass('public.alembic_version')"))
+        ).scalar() is not None
+
+    if not alembic_version_exists:
+        command.stamp(_postgres_alembic_config(async_url), "head")
+
+
 @pytest.fixture
 def anyio_backend():
     return "asyncio"
@@ -114,13 +214,15 @@ def config_home(tmp_path, monkeypatch) -> Path:
 @pytest.fixture(scope="function")
 def app_config(config_home, db_backend, postgres_container, monkeypatch) -> BasicMemoryConfig:
     """Create test app configuration for the appropriate backend."""
-    projects = {"test-project": str(config_home)}
+    projects = {"test-project": ProjectEntry(path=str(config_home))}
 
     # Set backend based on parameterized db_backend fixture
     if db_backend == "postgres":
         backend = DatabaseBackend.POSTGRES
-        # Get URL from testcontainer and convert to asyncpg driver
-        sync_url = postgres_container.get_connection_url()
+        # Trigger: CI jobs can provide a shared Postgres service instead of per-session containers.
+        # Why: reusing one pgvector-enabled server avoids Docker startup churn on every job.
+        # Outcome: local runs keep using testcontainers, while CI injects a stable service URL.
+        sync_url = _resolve_postgres_sync_url(postgres_container)
         database_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
     else:
         backend = DatabaseBackend.SQLITE
@@ -206,7 +308,7 @@ async def engine_factory(
     if db_backend == "postgres":
         # Postgres mode using testcontainers
         # Get async connection URL (asyncpg driver - same as production)
-        sync_url = postgres_container.get_connection_url()
+        sync_url = _resolve_postgres_sync_url(postgres_container)
         async_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
 
         engine = create_async_engine(
@@ -229,46 +331,7 @@ async def engine_factory(
         db._engine = engine
         db._session_maker = session_maker
 
-        from basic_memory.models.search import (
-            CREATE_POSTGRES_SEARCH_INDEX_TABLE,
-            CREATE_POSTGRES_SEARCH_INDEX_FTS,
-            CREATE_POSTGRES_SEARCH_INDEX_METADATA,
-            CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
-            CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_TABLE,
-            CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_INDEX,
-        )
-
-        # Drop and recreate all tables for test isolation
-        async with engine.begin() as conn:
-            # Must drop search_index first (has FK to project, blocks drop_all)
-            await conn.execute(text("DROP TABLE IF EXISTS search_index CASCADE"))
-            await conn.run_sync(Base.metadata.drop_all)
-            await conn.run_sync(Base.metadata.create_all)
-            # Create search_index via DDL (not ORM - uses composite PK + tsvector)
-            # asyncpg requires separate execute calls for each statement
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
-            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)
-            await conn.execute(CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_TABLE)
-            await conn.execute(CREATE_POSTGRES_SEARCH_VECTOR_CHUNKS_INDEX)
-
-            # Mark migrations as already applied for this test-created schema.
-            #
-            # Some codepaths (e.g. ensure_initialization()) invoke Alembic migrations.
-            # If we create tables via ORM directly, alembic_version is missing and migrations
-            # will try to create tables again, causing DuplicateTableError.
-            alembic_dir = Path(db.__file__).parent / "alembic"
-            cfg = Config()
-            cfg.set_main_option("script_location", str(alembic_dir))
-            cfg.set_main_option(
-                "file_template",
-                "%%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s",
-            )
-            cfg.set_main_option("timezone", "UTC")
-            cfg.set_main_option("revision_environment", "false")
-            cfg.set_main_option("sqlalchemy.url", async_url)
-            command.stamp(cfg, "head")
+        await _reset_postgres_test_schema(engine, async_url)
 
         yield engine, session_maker
 

tests/mcp/test_async_client_modes.py

@@ -79,6 +79,35 @@ async def test_get_client_cloud_adds_workspace_header(config_manager):
         assert client.headers.get("X-Workspace-ID") == "tenant-123"
 
 
+@pytest.mark.asyncio
+async def test_get_client_cloud_uses_project_workspace_when_not_explicit(config_manager):
+    cfg = config_manager.load_config()
+    cfg.cloud_host = "https://cloud.example.test"
+    cfg.cloud_api_key = "bmc_test_key_123"
+    cfg.default_workspace = "default-tenant"
+    cfg.set_project_mode("research", ProjectMode.CLOUD)
+    cfg.projects["research"].workspace_id = "project-tenant"
+    config_manager.save_config(cfg)
+
+    async with get_client(project_name="research") as client:
+        assert str(client.base_url).rstrip("/") == "https://cloud.example.test/proxy"
+        assert client.headers.get("X-Workspace-ID") == "project-tenant"
+
+
+@pytest.mark.asyncio
+async def test_get_client_cloud_uses_default_workspace_when_project_has_none(config_manager):
+    cfg = config_manager.load_config()
+    cfg.cloud_host = "https://cloud.example.test"
+    cfg.cloud_api_key = "bmc_test_key_123"
+    cfg.default_workspace = "default-tenant"
+    cfg.set_project_mode("research", ProjectMode.CLOUD)
+    config_manager.save_config(cfg)
+
+    async with get_client(project_name="research") as client:
+        assert str(client.base_url).rstrip("/") == "https://cloud.example.test/proxy"
+        assert client.headers.get("X-Workspace-ID") == "default-tenant"
+
+
 @pytest.mark.asyncio
 async def test_get_client_explicit_cloud_raises_without_credentials(config_manager, monkeypatch):
     cfg = config_manager.load_config()
@@ -253,6 +282,19 @@ async def test_get_cloud_control_plane_client_uses_api_key_when_available(config
         assert client.headers.get("Authorization") == "Bearer bmc_test_key_123"
 
 
+@pytest.mark.asyncio
+async def test_get_cloud_control_plane_client_adds_workspace_header(config_manager):
+    cfg = config_manager.load_config()
+    cfg.cloud_host = "https://cloud.example.test"
+    cfg.cloud_api_key = "bmc_test_key_123"
+    config_manager.save_config(cfg)
+
+    async with get_cloud_control_plane_client(workspace="tenant-123") as client:
+        assert str(client.base_url).rstrip("/") == "https://cloud.example.test"
+        assert client.headers.get("Authorization") == "Bearer bmc_test_key_123"
+        assert client.headers.get("X-Workspace-ID") == "tenant-123"
+
+
 @pytest.mark.asyncio
 async def test_get_cloud_control_plane_client_uses_oauth_token(config_manager):
     cfg = config_manager.load_config()

tests/services/test_entity_service_telemetry.py

@@ -54,7 +54,9 @@ async def test_create_entity_emits_expected_phase_spans(entity_service, monkeypa
             "file_service.write",
             "entity_service.create.parse_markdown",
             "entity_service.create.upsert_entity",
-            "entity_service.create.update_checksum",
+            "entity_service.upsert.base_entity",
+            "entity_service.upsert.relations",
+            "entity_service.upsert.persist_checksum",
         ],
     )
 
@@ -93,7 +95,9 @@ async def test_edit_entity_emits_expected_phase_spans(entity_service, monkeypatc
             "file_service.write",
             "entity_service.edit.parse_markdown",
             "entity_service.edit.upsert_entity",
-            "entity_service.edit.update_checksum",
+            "entity_service.upsert.base_entity",
+            "entity_service.upsert.relations",
+            "entity_service.upsert.persist_checksum",
         ],
     )
 
@@ -124,6 +128,9 @@ async def test_reindex_entity_emits_expected_phase_spans(entity_service, monkeyp
             "file_service.read_content",
             "entity_service.reindex.parse_markdown",
             "entity_service.reindex.upsert_entity",
+            "entity_service.upsert.base_entity",
+            "entity_service.upsert.relations",
+            "entity_service.upsert.hydrate_entity",
             "entity_service.reindex.update_checksum",
         ],
     )

tests/services/test_entity_service_write_result.py

@@ -0,0 +1,86 @@
+"""Tests for EntityWriteResult content variants."""
+
+import pytest
+
+from basic_memory.file_utils import remove_frontmatter
+from basic_memory.schemas import Entity as EntitySchema
+
+
+@pytest.mark.asyncio
+async def test_create_entity_with_content_returns_full_and_search_content(
+    entity_service, file_service
+) -> None:
+    result = await entity_service.create_entity_with_content(
+        EntitySchema(
+            title="Create Write Result",
+            directory="notes",
+            note_type="note",
+            content="Create body content",
+        )
+    )
+
+    file_path = file_service.get_entity_path(result.entity)
+    file_content, _ = await file_service.read_file(file_path)
+
+    assert result.content == file_content
+    assert result.search_content == remove_frontmatter(file_content)
+    assert result.search_content == "Create body content"
+
+
+@pytest.mark.asyncio
+async def test_update_entity_with_content_returns_full_and_search_content(
+    entity_service, file_service
+) -> None:
+    created = await entity_service.create_entity(
+        EntitySchema(
+            title="Update Write Result",
+            directory="notes",
+            note_type="note",
+            content="Original body content",
+        )
+    )
+
+    result = await entity_service.update_entity_with_content(
+        created,
+        EntitySchema(
+            title="Update Write Result",
+            directory="notes",
+            note_type="note",
+            content="Updated body content",
+        ),
+    )
+
+    file_path = file_service.get_entity_path(result.entity)
+    file_content, _ = await file_service.read_file(file_path)
+
+    assert result.content == file_content
+    assert result.search_content == remove_frontmatter(file_content)
+    assert result.search_content == "Updated body content"
+
+
+@pytest.mark.asyncio
+async def test_edit_entity_with_content_returns_full_and_search_content(
+    entity_service, file_service
+) -> None:
+    created = await entity_service.create_entity(
+        EntitySchema(
+            title="Edit Write Result",
+            directory="notes",
+            note_type="note",
+            content="Original body content",
+        )
+    )
+
+    result = await entity_service.edit_entity_with_content(
+        identifier=created.permalink,
+        operation="find_replace",
+        content="Edited body content",
+        find_text="Original body content",
+    )
+
+    file_path = file_service.get_entity_path(result.entity)
+    file_content, _ = await file_service.read_file(file_path)
+
+    assert result.content == file_content
+    assert result.search_content == remove_frontmatter(file_content)
+    assert result.search_content == "Edited body content"

ui/tool-ui-react/package-lock.json

@@ -1769,6 +1769,60 @@
         "node": ">=14.0.0"
       }
     },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": {
+      "version": "1.7.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "@emnapi/wasi-threads": "1.1.0",
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": {
+      "version": "1.7.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": {
+      "version": "1.1.0",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": {
+      "version": "1.1.0",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "@emnapi/core": "^1.7.1",
+        "@emnapi/runtime": "^1.7.1",
+        "@tybys/wasm-util": "^0.10.1"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": {
+      "version": "0.10.1",
+      "inBundle": true,
+      "license": "MIT",
+      "optional": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
+    "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": {
+      "version": "2.8.1",
+      "inBundle": true,
+      "license": "0BSD",
+      "optional": true
+    },
     "node_modules/@tailwindcss/oxide-win32-arm64-msvc": {
       "version": "4.1.18",
       "resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.1.18.tgz",
@@ -2237,9 +2291,9 @@
       "license": "ISC"
     },
     "node_modules/picomatch": {
-      "version": "4.0.3",
-      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
-      "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
+      "version": "4.0.4",
+      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.4.tgz",
+      "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==",
       "license": "MIT",
       "engines": {
         "node": ">=12"

uv.lock

@@ -142,14 +142,14 @@ wheels = [
 
 [[package]]
 name = "authlib"
-version = "1.6.7"
+version = "1.6.9"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "cryptography" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/49/dc/ed1681bf1339dd6ea1ce56136bad4baabc6f7ad466e375810702b0237047/authlib-1.6.7.tar.gz", hash = "sha256:dbf10100011d1e1b34048c9d120e83f13b35d69a826ae762b93d2fb5aafc337b", size = 164950, upload-time = "2026-02-06T14:04:14.171Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/af/98/00d3dd826d46959ad8e32af2dbb2398868fd9fd0683c26e56d0789bd0e68/authlib-1.6.9.tar.gz", hash = "sha256:d8f2421e7e5980cc1ddb4e32d3f5fa659cfaf60d8eaf3281ebed192e4ab74f04", size = 165134, upload-time = "2026-03-02T07:44:01.998Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/f8/00/3ed12264094ec91f534fae429945efbaa9f8c666f3aa7061cc3b2a26a0cd/authlib-1.6.7-py2.py3-none-any.whl", hash = "sha256:c637340d9a02789d2efa1d003a7437d10d3e565237bcb5fcbc6c134c7b95bab0", size = 244115, upload-time = "2026-02-06T14:04:12.141Z" },
+    { url = "https://files.pythonhosted.org/packages/53/23/b65f568ed0c22f1efacb744d2db1a33c8068f384b8c9b482b52ebdbc3ef6/authlib-1.6.9-py2.py3-none-any.whl", hash = "sha256:f08b4c14e08f0861dc18a32357b33fbcfd2ea86cfe3fe149484b4d764c4a0ac3", size = 244197, upload-time = "2026-03-02T07:44:00.307Z" },
 ]
 
 [[package]]
@@ -2744,7 +2744,7 @@ wheels = [
 
 [[package]]
 name = "requests"
-version = "2.32.5"
+version = "2.33.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "certifi" },
@@ -2752,9 +2752,9 @@ dependencies = [
     { name = "idna" },
     { name = "urllib3" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/c9/74/b3ff8e6c8446842c3f5c837e9c3dfcfe2018ea6ecef224c710c85ef728f4/requests-2.32.5.tar.gz", hash = "sha256:dbba0bac56e100853db0ea71b82b4dfd5fe2bf6d3754a8893c3af500cec7d7cf", size = 134517, upload-time = "2025-08-18T20:46:02.573Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/34/64/8860370b167a9721e8956ae116825caff829224fbca0ca6e7bf8ddef8430/requests-2.33.0.tar.gz", hash = "sha256:c7ebc5e8b0f21837386ad0e1c8fe8b829fa5f544d8df3b2253bff14ef29d7652", size = 134232, upload-time = "2026-03-25T15:10:41.586Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1e/db/4254e3eabe8020b458f1a747140d32277ec7a271daf1d235b70dc0b4e6e3/requests-2.32.5-py3-none-any.whl", hash = "sha256:2462f94637a34fd532264295e186976db0f5d453d1cdd31473c85a6a161affb6", size = 64738, upload-time = "2025-08-18T20:46:00.542Z" },
+    { url = "https://files.pythonhosted.org/packages/56/5d/c814546c2333ceea4ba42262d8c4d55763003e767fa169adc693bd524478/requests-2.33.0-py3-none-any.whl", hash = "sha256:3324635456fa185245e24865e810cecec7b4caf933d7eb133dcde67d48cee69b", size = 65017, upload-time = "2026-03-25T15:10:40.382Z" },
 ]
 
 [[package]]