chore(sync): remove duplicate resolver comment

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

AGENTS.md

@@ -56,6 +56,15 @@ Run `just test-smoke` when you specifically need the MCP smoke flow.
 
 If testmon is “cold,” the first run may be long. Subsequent runs get much faster.
 
+### PR CI Gate
+
+Before opening or updating a PR, run the checks that mirror the common required CI failures:
+
+- Run `just typecheck` in addition to targeted `ruff` and `pytest` commands when tests were added or changed.
+- Sign commits with `git commit -s` so DCO passes. If a PR branch already has unsigned commits, rewrite the branch with signed-off commits before asking for review.
+- Use a semantic PR title accepted by `.github/workflows/pr-title.yml`: `type(scope): summary`.
+- Use one of the allowed scopes: `core`, `cli`, `api`, `mcp`, `sync`, `ui`, `deps`, `installer`.
+
 ### Test Structure
 
 - `tests/` - Unit tests for individual components (mocked, fast)
@@ -244,6 +253,31 @@ async_client.set_client_factory(your_custom_factory)
 
 See SPEC-16 for full context manager refactor details.
 
+### Release Process
+
+Releases are driven by `just release` / `just beta` — never by a bare `git tag`. The recipes bump version metadata, run pre-flight checks, commit, tag, and push. GitHub Actions then publishes to PyPI and updates the Homebrew formula.
+
+**Stable release:**
+
+```
+just release v0.21.3
+```
+
+The recipe runs `just lint` + `just typecheck`, then updates `__version__` in `src/basic_memory/__init__.py` and `"version"` in `server.json` (MCP registry metadata), commits as `chore: update version to X.Y.Z for vX.Y.Z release`, creates the `vX.Y.Z` tag, and pushes both the commit and the tag to `origin/main`. After the tag lands, the `Release` workflow builds the package, publishes to PyPI, creates the GitHub release with auto-generated notes, and updates the Homebrew formula. The recipe finishes by printing the post-release tasks the workflow doesn't cover.
+
+**Beta release:** `just beta v0.21.3b1` — same flow with a beta-suffixed tag. PyPI consumers install with `pip install basic-memory --pre`.
+
+**Development builds:** every commit to `main` publishes a `0.21.3.dev26+468a22f`-style version to PyPI automatically via `.github/workflows/dev-release.yml`. No human action.
+
+**Do not tag releases by hand.** A bare `git tag vX.Y.Z` skips the in-code version bump. Package metadata is still correct (uv-dynamic-versioning derives it from the git tag) but `basic-memory --version` reports the previous release, which is what happened with v0.21.2 → v0.21.3.
+
+**Post-release tasks** the recipe surfaces but doesn't run:
+- `docs.basicmemory.com` — add notes to `src/pages/latest-releases.mdx`
+- `basicmachines.co` — bump version in `src/components/sections/hero.tsx`
+- MCP Registry — `mcp-publisher publish` from the repo root
+
+See `.claude/commands/release/release.md` (and `beta.md`, `release-check.md`, `changelog.md` alongside it) for the full release + post-release runbook, including the slash commands.
+
 ## BASIC MEMORY PRODUCT USAGE
 
 ### Knowledge Structure

CHANGELOG.md

@@ -1,5 +1,21 @@
 # CHANGELOG
 
+## v0.21.5 (2026-05-26)
+
+Workspace/project routing fixes for MCP, plus a SQLite vector reindex stability fix.
+
+### Bug Fixes
+
+- **#854**: MCP project listing now keeps duplicate cloud project rows distinct by
+  workspace, so only the selected workspace row inherits local project state.
+- **#853**: `write_note` returns workspace-qualified permalinks for cloud
+  workspace writes, allowing follow-up `memory://` reads to route back to the
+  correct workspace/project.
+- **#852**: Full SQLite vector reindex now loads `sqlite-vec` before dropping
+  `vec0` virtual tables, preventing reindex crashes.
+- **#838**: Local ASGI database initialization is preloaded so MCP routing can
+  safely enter local project contexts.
+
 ## v0.21.1 (2026-05-16)
 
 CI-only release. No user-facing changes.

CONTRIBUTING.md

@@ -224,40 +224,6 @@ See `test-int/BENCHMARKS.md` for detailed benchmark documentation.
 - **Fixtures**: Use async pytest fixtures for setup and teardown
 - **Markers**: Use `@pytest.mark.benchmark` for benchmarks, `@pytest.mark.slow` for slow tests
 
-## Release Process
-
-Basic Memory uses automatic versioning based on git tags with `uv-dynamic-versioning`. Here's how releases work:
-
-### Version Management
-- **Development versions**: Automatically generated from git commits (e.g., `0.12.4.dev26+468a22f`)
-- **Beta releases**: Created by tagging with beta suffixes (e.g., `git tag v0.13.0b1`)
-- **Stable releases**: Created by tagging with version numbers (e.g., `git tag v0.13.0`)
-
-### Release Workflows
-
-#### Development Builds
-- Automatically published to PyPI on every commit to `main`
-- Version format: `0.12.4.dev26+468a22f` (base version + dev + commit count + hash)
-- Users install with: `pip install basic-memory --pre --force-reinstall`
-
-#### Beta Releases
-1. Create and push a beta tag: `git tag v0.13.0b1 && git push origin v0.13.0b1`
-2. GitHub Actions automatically builds and publishes to PyPI
-3. Users install with: `pip install basic-memory --pre`
-
-#### Stable Releases
-1. Create and push a version tag: `git tag v0.13.0 && git push origin v0.13.0`
-2. GitHub Actions automatically:
-   - Builds the package with version `0.13.0`
-   - Creates GitHub release with auto-generated notes
-   - Publishes to PyPI
-3. Users install with: `pip install basic-memory`
-
-### For Contributors
-- No manual version bumping required
-- Versions are automatically derived from git tags
-- Focus on code changes, not version management
-
 ## Creating Issues
 
 If you're planning to work on something, please create an issue first to discuss the approach. Include:

README.md

@@ -123,8 +123,7 @@ phone.
   time — same files, same format, same wikilinks. Cancel anytime, your data
   stays yours.
 
-Built on WorkOS AuthKit, Neon Postgres, and Tigris S3. Platform source at
-[basic-memory-cloud](https://github.com/basicmachines-co/basic-memory-cloud).
+Built on WorkOS AuthKit, Neon Postgres, and Tigris S3.
 
 ### Pricing
 

docker-compose-postgres.yml

@@ -1,5 +1,8 @@
 # Docker Compose configuration for Basic Memory with PostgreSQL
-# Use this for local development and testing with Postgres backend
+# Use this for local development and testing with Postgres backend.
+#
+# The Postgres backend requires the pgvector extension (semantic search).
+# This image bundles pgvector; plain postgres:17 will not work for vector search.
 #
 # Usage:
 #   docker-compose -f docker-compose-postgres.yml up -d
@@ -7,7 +10,7 @@
 
 services:
   postgres:
-    image: postgres:17
+    image: pgvector/pgvector:pg17
     container_name: basic-memory-postgres
     environment:
       # Local development/test credentials - NOT for production

docs/semantic-search.md

@@ -263,6 +263,7 @@ The sqlite-vec extension is loaded per-connection. Vector tables are created laz
 ### Postgres (cloud)
 
 - **Vector storage**: [pgvector](https://github.com/pgvector/pgvector) with HNSW indexing
+- **Local Docker**: use `docker-compose-postgres.yml` (`pgvector/pgvector:pg17`). Plain `postgres:17` lacks the extension; run `CREATE EXTENSION IF NOT EXISTS vector;` on any external instance before first migration.
 - **Chunk metadata table**: Created via Alembic migration (`search_vector_chunks` with `BIGSERIAL` primary key)
 - **Embedding table**: `search_vector_embeddings` created at runtime (dimension-dependent, same pattern as SQLite)
 - **Index**: HNSW index on the embedding column for fast approximate nearest-neighbour queries

server.json

@@ -6,12 +6,12 @@
     "url": "https://github.com/basicmachines-co/basic-memory.git",
     "source": "github"
   },
-  "version": "0.21.1",
+  "version": "0.21.5",
   "packages": [
     {
       "registryType": "pypi",
       "identifier": "basic-memory",
-      "version": "0.21.1",
+      "version": "0.21.5",
       "runtimeHint": "uvx",
       "runtimeArguments": [
         {"type": "positional", "value": "basic-memory"},

src/basic_memory/__init__.py

@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
 # Package version - updated by release automation
-__version__ = "0.21.1"
+__version__ = "0.21.5"
 
 # API version for FastAPI - independent of package version
 __api_version__ = "v0"

src/basic_memory/config.py

@@ -21,7 +21,7 @@ from basic_memory.utils import setup_logging, generate_permalink
 
 DATABASE_NAME = "memory.db"
 APP_DATABASE_NAME = "memory.db"  # Using the same name but in the app directory
-DATA_DIR_NAME = ".basic-memory"
+DATA_DIR_NAME = "basic-memory"
 CONFIG_FILE_NAME = "config.json"
 WATCH_STATUS_JSON = "watch-status.json"
 CONFIG_DIR_MODE = 0o700
@@ -69,15 +69,18 @@ def resolve_data_dir() -> Path:
 
     Single source of truth for the per-user state directory. Honors
     ``BASIC_MEMORY_CONFIG_DIR`` so each process/worktree can isolate config
-    and database state; otherwise falls back to ``<user home>/.basic-memory``.
+    and database state; otherwise falls back to ``<user home>/.basic-memory``,
+    and then to ``XDG_CONFIG_HOME``.
 
     Cross-platform: ``Path.home()`` reads ``$HOME`` on POSIX and
     ``%USERPROFILE%`` on Windows, so there's no need to check ``$HOME``
     explicitly here.
     """
-    if config_dir := os.getenv("BASIC_MEMORY_CONFIG_DIR"):
-        return Path(config_dir)
-    return Path.home() / DATA_DIR_NAME
+    if basic_memory_dir := os.getenv("BASIC_MEMORY_CONFIG_DIR"):
+        return Path(basic_memory_dir)
+    if xdg_config := os.getenv("XDG_CONFIG_HOME"):
+        return Path(xdg_config) / DATA_DIR_NAME
+    return Path.home() / ("." + DATA_DIR_NAME)
 
 
 def default_fastembed_cache_dir() -> str:

src/basic_memory/deps/services.py

@@ -447,8 +447,12 @@ class TaskScheduler(Protocol):
 
 
 def _log_task_failure(completed: asyncio.Task) -> None:
+    if completed.cancelled():
+        return
     try:
         completed.result()
+    except asyncio.CancelledError:
+        return
     except Exception as exc:  # pragma: no cover
         logger.exception("Background task failed", error=str(exc))
 

src/basic_memory/indexing/batch_indexer.py

@@ -493,8 +493,14 @@ class BatchIndexer:
         async def resolve_relation(relation: Relation) -> int:
             async with semaphore:
                 try:
+                    # strict=True for deferred resolution: only fill in to_id on an
+                    # exact permalink/title/file_path match. Fuzzy fallback would silently
+                    # resolve ambiguous links to whichever entity shares tokens with the
+                    # link text, mismatching this with the sync_service forward-reference
+                    # path and producing confidently-wrong graph edges. See
+                    # sync_service.resolve_forward_references for the same change.
                     resolved_entity = await self.entity_service.link_resolver.resolve_link(
-                        relation.to_name
+                        relation.to_name, strict=True
                     )
                     if resolved_entity is None or resolved_entity.id == relation.from_id:
                         return 0

src/basic_memory/mcp/async_client.py

@@ -1,13 +1,36 @@
 import os
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import AsyncIterator, Callable, Optional
+from asyncio import Lock
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from dataclasses import dataclass
+from threading import RLock
+from typing import TYPE_CHECKING, Annotated, Any, AsyncIterator, Callable, Optional
 
+from fastapi import Depends, FastAPI, Request
 from httpx import ASGITransport, AsyncClient, Timeout
 from loguru import logger
 
 import logfire
 from basic_memory.config import ConfigManager, ProjectMode
 
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+LocalDatabaseState = tuple["AsyncEngine", "async_sessionmaker[AsyncSession]"]
+_MISSING_STATE_VALUE = object()
+
+
+@dataclass
+class _PreparedLocalAsgiDatabase:
+    active_count: int
+    previous_engine: object
+    previous_session_maker: object
+    dependency_context: AbstractAsyncContextManager[LocalDatabaseState]
+
+
+_prepared_local_asgi_database_lock = RLock()
+_prepared_local_asgi_database_prepare_locks: dict[FastAPI, Lock] = {}
+_prepared_local_asgi_databases: dict[FastAPI, _PreparedLocalAsgiDatabase] = {}
+
 
 def _force_local_mode() -> bool:
     """Check if local mode is forced via environment variable."""
@@ -34,15 +57,12 @@ def _build_timeout() -> Timeout:
     )
 
 
-def _asgi_client(timeout: Timeout) -> AsyncClient:
-    """Create a local ASGI client."""
-    # Import on first local-client use so CLI help/version paths can import
-    # routing helpers without constructing the full FastAPI router graph.
-    from basic_memory.api.app import app as fastapi_app
+def _build_asgi_client(app: FastAPI, timeout: Timeout) -> AsyncClient:
+    """Create a local ASGI client for an already-prepared FastAPI app."""
     from basic_memory.workspace_context import workspace_permalink_headers
 
     return AsyncClient(
-        transport=ASGITransport(app=fastapi_app),
+        transport=ASGITransport(app=app),
         base_url="http://test",
         timeout=timeout,
         # Local ASGI calls still cross the HTTP boundary, so request handlers need
@@ -51,6 +71,169 @@ def _asgi_client(timeout: Timeout) -> AsyncClient:
     )
 
 
+def _get_prepared_local_asgi_database_prepare_lock(app: FastAPI) -> Lock:
+    """Get the async lock that serializes first-time DB preparation for an app."""
+    with _prepared_local_asgi_database_lock:
+        prepare_lock = _prepared_local_asgi_database_prepare_locks.get(app)
+        if prepare_lock is None:
+            prepare_lock = Lock()
+            _prepared_local_asgi_database_prepare_locks[app] = prepare_lock
+        return prepare_lock
+
+
+@asynccontextmanager
+async def _resolve_local_asgi_database(app: FastAPI) -> AsyncIterator[LocalDatabaseState]:
+    """Resolve database state for a local ASGI request."""
+    from fastapi.dependencies.utils import get_dependant, solve_dependencies
+
+    from basic_memory.deps import get_engine_factory
+
+    async def resolve_database_state(
+        database_state: Annotated[LocalDatabaseState, Depends(get_engine_factory)],
+    ) -> LocalDatabaseState:
+        return database_state
+
+    scope: dict[str, Any] = {
+        "type": "http",
+        "asgi": {"version": "3.0"},
+        "method": "GET",
+        "scheme": "http",
+        "path": "/",
+        "raw_path": b"/",
+        "root_path": "",
+        "query_string": b"",
+        "headers": [],
+        "client": ("testclient", 50000),
+        "server": ("testserver", 80),
+        "app": app,
+        "path_params": {},
+    }
+
+    async with AsyncExitStack() as request_stack, AsyncExitStack() as function_stack:
+        scope["fastapi_inner_astack"] = request_stack
+        scope["fastapi_function_astack"] = function_stack
+        request = Request(scope)
+        dependant = get_dependant(path="/", call=resolve_database_state)
+        solved = await solve_dependencies(
+            request=request,
+            dependant=dependant,
+            dependency_overrides_provider=app,
+            async_exit_stack=request_stack,
+            embed_body_fields=False,
+        )
+        if solved.errors:
+            raise RuntimeError(f"Failed to resolve local ASGI database dependency: {solved.errors}")
+
+        yield await resolve_database_state(**solved.values)
+
+
+def _retain_prepared_local_asgi_database(app: FastAPI) -> bool:
+    """Retain an active local ASGI database preparation if one exists."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is None:
+            return False
+
+        active.active_count += 1
+        return True
+
+
+def _install_prepared_local_asgi_database(
+    app: FastAPI,
+    database_state: LocalDatabaseState,
+    dependency_context: AbstractAsyncContextManager[LocalDatabaseState],
+) -> None:
+    """Install local ASGI database state after dependency resolution."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is not None:
+            raise RuntimeError("Local ASGI database state installed while another state is active")
+
+        previous_engine = getattr(app.state, "engine", _MISSING_STATE_VALUE)
+        previous_session_maker = getattr(app.state, "session_maker", _MISSING_STATE_VALUE)
+        engine, session_maker = database_state
+
+        app.state.engine = engine
+        app.state.session_maker = session_maker
+        _prepared_local_asgi_databases[app] = _PreparedLocalAsgiDatabase(
+            active_count=1,
+            previous_engine=previous_engine,
+            previous_session_maker=previous_session_maker,
+            dependency_context=dependency_context,
+        )
+
+
+def _restore_local_asgi_state_attribute(app: FastAPI, name: str, previous_value: object) -> None:
+    """Restore a FastAPI app.state attribute captured before local ASGI preparation."""
+    if previous_value is _MISSING_STATE_VALUE:
+        if hasattr(app.state, name):
+            delattr(app.state, name)
+    else:
+        setattr(app.state, name, previous_value)
+
+
+def _release_prepared_local_asgi_database(
+    app: FastAPI,
+) -> AbstractAsyncContextManager[LocalDatabaseState] | None:
+    """Release local ASGI database state after a client context exits."""
+    with _prepared_local_asgi_database_lock:
+        active = _prepared_local_asgi_databases.get(app)
+        if active is None:
+            raise RuntimeError("Local ASGI database state released without a matching retain")
+
+        active.active_count -= 1
+        if active.active_count > 0:
+            return None
+
+        del _prepared_local_asgi_databases[app]
+        _restore_local_asgi_state_attribute(app, "engine", active.previous_engine)
+        _restore_local_asgi_state_attribute(
+            app,
+            "session_maker",
+            active.previous_session_maker,
+        )
+        return active.dependency_context
+
+
+@asynccontextmanager
+async def _prepared_local_asgi_database(app: FastAPI) -> AsyncIterator[None]:
+    """Initialize local ASGI database state before the first request."""
+    prepare_lock = _get_prepared_local_asgi_database_prepare_lock(app)
+    async with prepare_lock:
+        if not _retain_prepared_local_asgi_database(app):
+            database_context = _resolve_local_asgi_database(app)
+            database_state = await database_context.__aenter__()
+            try:
+                _install_prepared_local_asgi_database(app, database_state, database_context)
+            except Exception:
+                await database_context.__aexit__(None, None, None)
+                raise
+
+    try:
+        yield
+    finally:
+        database_context = _release_prepared_local_asgi_database(app)
+        if database_context is not None:
+            await database_context.__aexit__(None, None, None)
+
+
+@asynccontextmanager
+async def _asgi_client(timeout: Timeout) -> AsyncIterator[AsyncClient]:
+    """Create a local ASGI client."""
+    # Import on first local-client use so CLI help/version paths can import
+    # routing helpers without constructing the full FastAPI router graph.
+    from basic_memory.api.app import app as fastapi_app
+
+    # Trigger: local ASGITransport does not execute FastAPI lifespan startup.
+    # Why: letting request dependencies initialize Postgres can run asyncpg DDL
+    # under Starlette's request loop and trigger CPython's empty-ready-queue race.
+    # Outcome: request handling sees the same app.state database objects as API
+    # lifespan startup would have provided.
+    async with _prepared_local_asgi_database(fastapi_app):
+        async with _build_asgi_client(fastapi_app, timeout) as client:
+            yield client
+
+
 async def _resolve_cloud_token(config) -> str:
     """Resolve cloud token with API key preferred, OAuth fallback."""
     with logfire.span(
@@ -260,7 +443,12 @@ def create_client() -> AsyncClient:
 
     if _force_local_mode() or not _force_cloud_mode():
         logger.info("Creating ASGI client for local Basic Memory API")
-        return _asgi_client(timeout)
+        # Deprecated sync path: create_client() cannot await the local ASGI
+        # pre-initialization used by get_client(), so callers that need proper
+        # resource setup should use the async context manager instead.
+        from basic_memory.api.app import app as fastapi_app
+
+        return _build_asgi_client(fastapi_app, timeout)
 
     logger.info("Creating HTTP client for cloud proxy (legacy create_client path)")
     config = ConfigManager().config

src/basic_memory/mcp/project_context.py

@@ -142,6 +142,15 @@ async def _set_cached_active_project(
         await context.set_state("default_project_name", active_project.name)
 
 
+async def _clear_cached_active_project(context: Optional[Context]) -> None:
+    """Clear cached project metadata that may no longer match the active route."""
+    if not context:
+        return
+
+    await context.set_state("active_project", None)
+    await context.set_state("default_project_name", None)
+
+
 async def _get_cached_active_workspace(context: Optional[Context]) -> Optional[WorkspaceInfo]:
     """Return the cached active workspace from context when available."""
     if not context:
@@ -167,8 +176,7 @@ async def _set_cached_active_workspace(
         # Why: project names are only unique inside one workspace, so a cached
         #   ProjectItem from the previous tenant can point at the wrong project
         # Outcome: force the next validation call to resolve within the new tenant
-        await context.set_state("active_project", None)
-        await context.set_state("default_project_name", None)
+        await _clear_cached_active_project(context)
 
     await context.set_state("active_workspace", active_workspace.model_dump())
 
@@ -525,6 +533,28 @@ def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool:
     )
 
 
+def _workspace_identifier_discovery_available(
+    identifier: str,
+    config: BasicMemoryConfig,
+) -> bool:
+    """Return True when an identifier is allowed to consult workspace discovery."""
+    if _cloud_workspace_discovery_available(config):
+        return True
+
+    from basic_memory.mcp.async_client import (
+        _explicit_routing,
+        _force_local_mode,
+    )
+
+    if _explicit_routing() and _force_local_mode():
+        return False
+
+    return (
+        has_cloud_credentials(config)
+        and _split_workspace_identifier_segments(identifier) is not None
+    )
+
+
 async def resolve_workspace_qualified_memory_url(
     identifier: str,
     context: Optional[Context] = None,
@@ -1326,11 +1356,23 @@ async def detect_project_from_identifier_prefix(
         # Outcome: keep unqualified search/title input on the active/default project route.
         return None
 
-    if _cloud_workspace_discovery_available(config):
-        workspace_resolution = await resolve_workspace_qualified_identifier(
-            identifier,
-            context=context,
+    if _workspace_identifier_discovery_available(identifier, config):
+        workspace_discovery_fallback_errors = (
+            "not found",
+            "no accessible workspaces",
+            "unable to discover",
         )
+        try:
+            workspace_resolution = await resolve_workspace_qualified_identifier(
+                identifier,
+                context=context,
+            )
+        except ValueError as exc:
+            message = str(exc).lower()
+            if any(error in message for error in workspace_discovery_fallback_errors):
+                return None
+            raise
+
         if workspace_resolution is not None:
             return workspace_resolution.project_identifier
 
@@ -1345,7 +1387,7 @@ async def detect_project_from_identifier_prefix(
             )
         except ValueError as exc:
             message = str(exc).lower()
-            if "not found" in message or "no accessible workspaces" in message:
+            if any(error in message for error in workspace_discovery_fallback_errors):
                 return None
             raise
 
@@ -1462,9 +1504,54 @@ async def get_project_client(
     if project_id and not cloud_available:
         project_mode = ProjectMode.LOCAL
 
+    # Trigger: project_id is a local external_id in a mixed local+cloud setup.
+    # Why: UUIDs are not local config keys, so get_project_mode() treats them as
+    #   cloud projects. A local-first probe avoids making local UUIDs depend on
+    #   healthy cloud workspace discovery.
+    # Outcome: resolve the effective UUID against local ASGI first; if it is not
+    #   local, preserve the existing cloud workspace lookup path.
+    if (
+        project_id
+        and config.projects
+        and not factory_mode
+        and not explicit_cloud_routing
+        and project_mode == ProjectMode.CLOUD
+    ):
+        try:
+            canonical_project_id = str(UUID(resolved_project))
+        except ValueError:
+            pass
+        else:
+            with logfire.span(
+                "routing.local_project_id_probe",
+                project_id=canonical_project_id,
+            ):
+                async with get_client() as client:
+                    try:
+                        active_project = await get_active_project(
+                            client,
+                            canonical_project_id,
+                            context,
+                        )
+                    except ToolError as exc:
+                        if "not found" not in str(exc).lower():
+                            raise
+                    else:
+                        route_mode = "local_asgi"
+                        await _clear_cached_active_workspace_for_local_route(context)
+                        with logfire.span(
+                            "routing.client_session",
+                            project_name=active_project.name,
+                            route_mode=route_mode,
+                        ):
+                            logger.debug("Using local ASGI routing for project_id")
+                            yield client, active_project
+                        return
+
     if factory_mode or project_mode == ProjectMode.CLOUD or explicit_cloud_routing:
         route_mode = "factory" if factory_mode else "cloud_proxy"
         active_ws: WorkspaceInfo | None = None
+        resolved_entry: WorkspaceProjectEntry | None = None
         workspace_id: str
         project_for_api = _unqualified_project_identifier(resolved_project)
 
@@ -1487,6 +1574,13 @@ async def get_project_client(
 
         if active_ws is not None:
             await _set_cached_active_workspace(context, active_ws)
+        if resolved_entry is not None:
+            cached_project = await _get_cached_active_project(context)
+            if (
+                cached_project is not None
+                and cached_project.external_id != resolved_entry.project.external_id
+            ):
+                await _clear_cached_active_project(context)
         with logfire.span(
             "routing.client_session",
             project_name=project_for_api,

src/basic_memory/mcp/tools/edit_note.py

@@ -9,7 +9,7 @@ from pydantic import AliasChoices, Field
 
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import (
-    _cloud_workspace_discovery_available,
+    _workspace_identifier_discovery_available,
     detect_project_from_memory_url_prefix,
     get_project_client,
     add_project_metadata,
@@ -368,8 +368,9 @@ async def edit_note(
                 config,
                 context=context,
             )
-        elif _cloud_workspace_discovery_available(
-            config
+        elif _workspace_identifier_discovery_available(
+            identifier,
+            config,
         ) and is_workspace_qualified_plain_identifier(identifier):
             detected = await detect_project_from_workspace_identifier_prefix(
                 identifier,

src/basic_memory/mcp/tools/project_management.py

@@ -10,7 +10,12 @@ from typing import Literal
 from fastmcp import Context
 from loguru import logger
 
-from basic_memory.config import ConfigManager, has_cloud_credentials
+from basic_memory.config import (
+    BasicMemoryConfig,
+    ConfigManager,
+    ProjectEntry,
+    has_cloud_credentials,
+)
 from basic_memory.mcp.async_client import (
     _explicit_routing,
     _force_local_mode,
@@ -131,9 +136,66 @@ def _merge_projects(
     return merged
 
 
+def _workspace_entry_priority(entry: WorkspaceProjectEntry) -> tuple[bool, int, str, str]:
+    """Prefer default/personal workspaces when duplicate project permalinks exist."""
+    workspace_type_rank = 0 if entry.workspace.workspace_type == "personal" else 1
+    return (
+        # False sorts before True, so the cloud/default workspace comes first.
+        not entry.workspace.is_default,
+        workspace_type_rank,
+        entry.workspace.name.casefold(),
+        entry.workspace.tenant_id,
+    )
+
+
+def _select_attached_cloud_entry(
+    cloud_entries: tuple[WorkspaceProjectEntry, ...],
+    *,
+    config_entry: ProjectEntry | None,
+    config: BasicMemoryConfig | None,
+) -> WorkspaceProjectEntry | None:
+    """Choose the single cloud row that should inherit local project state."""
+    if not cloud_entries:
+        return None
+
+    preferred_workspace_ids: list[str] = []
+    if config_entry and config_entry.workspace_id:
+        preferred_workspace_ids.append(config_entry.workspace_id)
+    if (
+        config
+        and config.default_workspace
+        and config.default_workspace not in preferred_workspace_ids
+    ):
+        preferred_workspace_ids.append(config.default_workspace)
+
+    # The configured default workspace can differ from the cloud-side default.
+    # Use the cloud default only after explicit local config preferences.
+    default_workspace_entry = next(
+        (entry for entry in cloud_entries if entry.workspace.is_default),
+        None,
+    )
+    if (
+        default_workspace_entry is not None
+        and default_workspace_entry.workspace.tenant_id not in preferred_workspace_ids
+    ):
+        preferred_workspace_ids.append(default_workspace_entry.workspace.tenant_id)
+
+    for workspace_id in preferred_workspace_ids:
+        for entry in cloud_entries:
+            if entry.workspace.tenant_id == workspace_id:
+                return entry
+
+    if len(cloud_entries) == 1:
+        return cloud_entries[0]
+
+    return sorted(cloud_entries, key=_workspace_entry_priority)[0]
+
+
 def _merge_workspace_projects(
     local_list: ProjectList | None,
     cloud_entries: tuple[WorkspaceProjectEntry, ...],
+    *,
+    config: BasicMemoryConfig | None = None,
 ) -> list[dict]:
     """Merge local projects with cloud projects from every accessible workspace."""
     local_by_permalink: dict[str, ProjectItem] = {}
@@ -141,20 +203,40 @@ def _merge_workspace_projects(
         for project in local_list.projects:
             local_by_permalink[project.permalink] = project
 
+    config_by_permalink: dict[str, ProjectEntry] = {}
+    if config:
+        config_by_permalink = {
+            generate_permalink(project_name): entry
+            for project_name, entry in config.projects.items()
+        }
+
+    cloud_entries_by_permalink: dict[str, list[WorkspaceProjectEntry]] = {}
+    for entry in cloud_entries:
+        cloud_entries_by_permalink.setdefault(entry.project.permalink, []).append(entry)
+
+    attached_entry_by_permalink: dict[str, WorkspaceProjectEntry | None] = {}
+    for permalink in local_by_permalink:
+        attached_entry_by_permalink[permalink] = _select_attached_cloud_entry(
+            tuple(cloud_entries_by_permalink.get(permalink, ())),
+            config_entry=config_by_permalink.get(permalink),
+            config=config,
+        )
+
     cloud_permalinks = {entry.project.permalink for entry in cloud_entries}
     merged: list[dict] = []
 
     for entry in sorted(
         cloud_entries,
-        key=lambda item: (
-            not item.workspace.is_default,
-            item.workspace.workspace_type != "personal",
-            item.workspace.name.casefold(),
-            item.project.permalink,
-        ),
+        key=lambda item: (*_workspace_entry_priority(item), item.project.permalink),
     ):
         permalink = entry.project.permalink
-        local_proj = local_by_permalink.get(permalink)
+        local_proj = (
+            local_by_permalink.get(permalink)
+            # WorkspaceProjectEntry is a frozen dataclass containing Pydantic
+            # models, so value equality is the intended comparison here.
+            if attached_entry_by_permalink.get(permalink) == entry
+            else None
+        )
         cloud_proj = entry.project
         source = "local+cloud" if local_proj else "cloud"
         local_path = local_proj.path if local_proj else None
@@ -339,7 +421,7 @@ async def list_memory_projects(
                 )
 
     if cloud_entries:
-        merged = _merge_workspace_projects(local_list, cloud_entries)
+        merged = _merge_workspace_projects(local_list, cloud_entries, config=config)
     else:
         merged = _merge_projects(
             local_list,

src/basic_memory/mcp/tools/write_note.py

@@ -12,7 +12,13 @@ from basic_memory.mcp.project_context import get_project_client, add_project_met
 from basic_memory.mcp.server import mcp
 from fastmcp import Context
 from basic_memory.schemas.base import Entity
-from basic_memory.utils import coerce_dict, parse_tags, validate_project_path
+from basic_memory.utils import (
+    build_qualified_permalink_reference,
+    coerce_dict,
+    parse_tags,
+    validate_project_path,
+)
+from basic_memory.workspace_context import current_workspace_permalink_context
 
 # Define TagType as a Union that can accept either a string or a list of strings or None
 TagType = Union[List[str], str, None]
@@ -35,11 +41,7 @@ async def write_note(
     tags: list[str] | str | None = None,
     note_type: str = "note",
     metadata: Annotated[dict | None, BeforeValidator(coerce_dict)] = None,
-    # Force/replace are the file-write idioms models default to.
-    overwrite: Annotated[
-        bool | None,
-        Field(default=None, validation_alias=AliasChoices("overwrite", "force", "replace")),
-    ] = None,
+    overwrite: bool | None = None,
     output_format: Literal["text", "json"] = "text",
     context: Context | None = None,
 ) -> str | dict:
@@ -278,11 +280,20 @@ async def write_note(
                 else:
                     # Re-raise if it's not a conflict error
                     raise  # pragma: no cover
+            response_permalink = result.permalink
+            workspace_context = current_workspace_permalink_context()
+            if response_permalink and workspace_context is not None:
+                response_permalink = build_qualified_permalink_reference(
+                    active_project.permalink,
+                    response_permalink,
+                    workspace_permalink=workspace_context.workspace_slug,
+                )
+
             summary = [
                 f"# {action} note",
                 f"project: {active_project.name}",
                 f"file_path: {result.file_path}",
-                f"permalink: {result.permalink}",
+                f"permalink: {response_permalink}",
                 f"checksum: {result.checksum[:8] if result.checksum else 'unknown'}",
             ]
 
@@ -319,12 +330,12 @@ async def write_note(
 
             # Log the response with structured data
             logger.info(
-                f"MCP tool response: tool=write_note project={active_project.name} action={action} permalink={result.permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved}"
+                f"MCP tool response: tool=write_note project={active_project.name} action={action} permalink={response_permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved}"
             )
             if output_format == "json":
                 return {
                     "title": result.title,
-                    "permalink": result.permalink,
+                    "permalink": response_permalink,
                     "file_path": result.file_path,
                     "checksum": result.checksum,
                     "action": action.lower(),

src/basic_memory/repository/sqlite_search_repository.py

@@ -620,6 +620,25 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             )
             await session.commit()
 
+    async def drop_vector_tables(self) -> None:
+        """Drop SQLite vector tables on a sqlite-vec-enabled connection."""
+        async with db.scoped_session(self.session_maker) as session:
+            vector_sql_result = await session.execute(
+                text(
+                    "SELECT sql FROM sqlite_master "
+                    "WHERE type = 'table' AND name = 'search_vector_embeddings'"
+                )
+            )
+            vector_sql = vector_sql_result.scalar()
+            if vector_sql and "using vec0" in vector_sql.lower():
+                await self._ensure_sqlite_vec_loaded(session)
+
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_embeddings"))
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_chunks"))
+            await session.execute(text("DROP TABLE IF EXISTS search_vector_index"))
+            await session.commit()
+        self._vector_tables_initialized = False
+
     async def delete_stale_vector_rows(self) -> None:
         """Delete vector rows whose source entities no longer exist."""
         await self._ensure_vector_tables()

src/basic_memory/services/link_resolver.py

@@ -40,17 +40,29 @@ async def detect_project_from_workspace_identifier_prefix(
         return None
 
     from basic_memory.mcp.project_context import (
-        _cloud_workspace_discovery_available,
+        _workspace_identifier_discovery_available,
         resolve_workspace_qualified_identifier,
     )
 
-    if not _cloud_workspace_discovery_available(config):
+    if not _workspace_identifier_discovery_available(identifier, config):
         return None
 
-    workspace_resolution = await resolve_workspace_qualified_identifier(
-        identifier,
-        context=context,
+    workspace_discovery_fallback_errors = (
+        "not found",
+        "no accessible workspaces",
+        "unable to discover",
     )
+    try:
+        workspace_resolution = await resolve_workspace_qualified_identifier(
+            identifier,
+            context=context,
+        )
+    except ValueError as exc:
+        message = str(exc).lower()
+        if any(error in message for error in workspace_discovery_fallback_errors):
+            return None
+        raise
+
     if workspace_resolution is None:
         return None
     return workspace_resolution.project_identifier

src/basic_memory/services/search_service.py

@@ -126,19 +126,23 @@ class SearchService:
 
     async def reindex_all(self, background_tasks: Optional[BackgroundTasks] = None) -> None:
         """Reindex all content from database."""
+        from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
 
         logger.info("Starting full reindex")
         # Clear and recreate search index
         await self.repository.execute_query(text("DROP TABLE IF EXISTS search_index"), params={})
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_embeddings"), params={}
-        )
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_chunks"), params={}
-        )
-        await self.repository.execute_query(
-            text("DROP TABLE IF EXISTS search_vector_index"), params={}
-        )
+        if isinstance(self.repository, SQLiteSearchRepository):
+            await self.repository.drop_vector_tables()
+        else:
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_embeddings"), params={}
+            )
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_chunks"), params={}
+            )
+            await self.repository.execute_query(
+                text("DROP TABLE IF EXISTS search_vector_index"), params={}
+            )
         await self.init_search_index()
 
         # Reindex all entities

src/basic_memory/sync/sync_service.py

@@ -1447,7 +1447,16 @@ class SyncService:
                 f"to_name={relation.to_name}"
             )
 
-            resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
+            # Use strict=True: deferred resolution should only fill in to_id when an
+            # exact permalink/title/file_path match exists. The fuzzy fallback (search-based
+            # token match) would silently resolve ambiguous links like
+            # `[[overview (state-management/session-execution)]]` to whichever entity shares
+            # the most tokens, polluting the graph with confidently-wrong edges that no
+            # audit catches. Leaving such relations unresolved keeps to_id=NULL so they
+            # surface as forward references and can be fixed by the producer.
+            resolved_entity = await self.entity_service.link_resolver.resolve_link(
+                relation.to_name, strict=True
+            )
 
             # ignore reference to self
             if resolved_entity and resolved_entity.id != relation.from_id:

test-int/conftest.py

@@ -149,6 +149,22 @@ def clean_routing_env(monkeypatch) -> None:
     monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
 
 
+@pytest.fixture(autouse=True)
+def isolate_data_dir_env(monkeypatch) -> None:
+    """Keep host data-dir env vars from leaking into integration tests.
+
+    Why: GitHub Actions Ubuntu runners set ``XDG_CONFIG_HOME=/home/runner/.config``,
+    and ``resolve_data_dir()`` honors it ahead of ``Path.home() / ".basic-memory"``.
+    Without clearing it, the MCP tool process reads config.json from the host XDG
+    path instead of the tmp dir the ``config_manager`` fixture wrote to — so
+    ``test-project`` is missing from ``config.projects``, ``get_project_mode``
+    falls through to its CLOUD default (#837), and every tool call fails with
+    "Cloud routing requested but no credentials found."
+    """
+    monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)
+    monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
+
+
 POSTGRES_EPHEMERAL_TABLES = [
     "search_vector_embeddings",
     "search_vector_chunks",

test-int/mcp/test_param_aliases_integration.py

@@ -310,31 +310,40 @@ async def test_write_note_accepts_directory_aliases(mcp_server, app, test_projec
 
 
 @pytest.mark.asyncio
-async def test_write_note_accepts_overwrite_aliases(mcp_server, app, test_project):
-    """`force`/`replace` should map to `overwrite`."""
+async def test_write_note_overwrite_canonical_via_mcp(mcp_server, app, test_project):
+    """Canonical overwrite=True must reach the handler (#818 regression)."""
     async with Client(mcp_server) as client:
-        # First create
         await client.call_tool(
             "write_note",
             {
                 "project": test_project.name,
-                "title": "Overwrite Alias Note",
+                "title": "Overwrite Canonical Note",
                 "directory": "overwrite-test",
                 "content": "v1",
             },
         )
-        # Overwrite using `force` alias
+        blocked = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Overwrite Canonical Note",
+                "directory": "overwrite-test",
+                "content": "v2",
+            },
+        )
+        assert "# Error: Note already exists" in blocked.content[0].text
+
         result = await client.call_tool(
             "write_note",
             {
                 "project": test_project.name,
-                "title": "Overwrite Alias Note",
+                "title": "Overwrite Canonical Note",
                 "directory": "overwrite-test",
                 "content": "v2",
-                "force": True,  # alias for overwrite
+                "overwrite": True,
             },
         )
-        assert "Updated note" in result.content[0].text or "Created note" in result.content[0].text
+        assert "# Updated note" in result.content[0].text
 
 
 # --- move_note aliases ---
@@ -538,3 +547,19 @@ async def test_aliases_not_advertised_in_schema(mcp_server, app):
                 assert canonical in props, f"{tool_name}: canonical '{canonical}' missing"
             for alias in must_not_have:
                 assert alias not in props, f"{tool_name}: alias '{alias}' leaked into schema"
+
+        # #818: AliasChoices on optional bool broke external-client JSON schema (null-only).
+        overwrite_schema = tools["write_note"].inputSchema["properties"]["overwrite"]
+        schema_types: set[str] = set()
+        if "type" in overwrite_schema:
+            raw = overwrite_schema["type"]
+            if isinstance(raw, str):
+                schema_types.add(raw)
+            else:
+                schema_types.update(raw)
+        for option in overwrite_schema.get("anyOf", ()):
+            if "type" in option:
+                schema_types.add(option["type"])
+        assert "boolean" in schema_types, (
+            f"write_note overwrite must expose boolean in schema, got {overwrite_schema}"
+        )

test-int/mcp/test_workspace_permalink_integration.py

@@ -70,8 +70,12 @@ def team_workspace() -> WorkspaceInfo:
 @pytest.fixture
 def route_workspaces(app):
     @contextmanager
-    def route(*workspaces: WorkspaceInfo):
-        with _workspace_routing(app, workspaces):
+    def route(*workspaces: WorkspaceInfo, forward_permalink_headers: bool = True):
+        with _workspace_routing(
+            app,
+            workspaces,
+            forward_permalink_headers=forward_permalink_headers,
+        ):
             yield
 
     return route
@@ -117,7 +121,12 @@ def _save_permalink_config(
 
 
 @contextmanager
-def _workspace_routing(app, workspaces: Iterable[WorkspaceInfo]):
+def _workspace_routing(
+    app,
+    workspaces: Iterable[WorkspaceInfo],
+    *,
+    forward_permalink_headers: bool = True,
+):
     """Route MCP tool HTTP calls through an ASGI-backed cloud workspace seam."""
     workspace_list = tuple(workspaces)
     workspace_ids = {workspace.tenant_id for workspace in workspace_list}
@@ -128,10 +137,11 @@ def _workspace_routing(app, workspaces: Iterable[WorkspaceInfo]):
     @asynccontextmanager
     async def factory(workspace: str | None = None):
         assert workspace is None or workspace in workspace_ids
+        headers = workspace_permalink_headers() if forward_permalink_headers else {}
         async with HttpxAsyncClient(
             transport=ASGITransport(app=app),
             base_url="http://test",
-            headers=workspace_permalink_headers(),
+            headers=headers,
         ) as inner:
             yield inner
 
@@ -461,3 +471,39 @@ async def test_team_workspace_permalink_routes_to_specific_workspace(
 
         workspace_read = await _read_json(mcp_server, identifier=expected_permalink)
         assert workspace_read["permalink"] == expected_permalink
+
+
+@pytest.mark.asyncio
+async def test_write_note_by_project_id_qualifies_permalink_when_headers_not_forwarded(
+    mcp_server,
+    test_project,
+    app_config,
+    team_workspace,
+    route_workspaces,
+):
+    """MCP writes should return self-routing IDs even if the API omits slug headers."""
+    _save_permalink_config(app_config, include_project=True, default_project=None)
+
+    title = "Project Id Workspace Permalink"
+    short_permalink = "permalink-suite/project-id-workspace-permalink"
+    expected_permalink = f"{team_workspace.slug}/{test_project.name}/{short_permalink}"
+
+    with route_workspaces(team_workspace, forward_permalink_headers=False):
+        write_payload = await _call_json(
+            mcp_server,
+            "write_note",
+            {
+                "project_id": test_project.external_id,
+                "title": title,
+                "directory": "permalink-suite",
+                "content": f"# {title}\n\nProject ID workspace body.",
+                "output_format": "json",
+            },
+        )
+        assert write_payload["permalink"] == expected_permalink
+
+        workspace_read = await _read_json(
+            mcp_server,
+            identifier=f"memory://{write_payload['permalink']}",
+        )
+        assert workspace_read["title"] == title

tests/README.md

@@ -10,6 +10,8 @@ pytest
 
 # Run tests against Postgres only (requires docker-compose)
 docker-compose -f docker-compose-postgres.yml up -d
+BASIC_MEMORY_TEST_POSTGRES=1 \
+POSTGRES_TEST_URL=postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory \
 pytest -m postgres
 
 # Run tests against BOTH backends
@@ -54,7 +56,7 @@ database_url = None  # Uses default SQLite path
 
 # Postgres config
 database_backend = DatabaseBackend.POSTGRES
-database_url = "postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory_test"
+database_url = "postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory"
 ```
 
 ## Running Postgres Tests
@@ -66,18 +68,22 @@ docker-compose -f docker-compose-postgres.yml up -d
 ```
 
 This starts:
-- Postgres 17 on port **5433** (not 5432 to avoid conflicts)
-- Test database: `basic_memory_test`
+- Postgres 17 with **pgvector** (`pgvector/pgvector:pg17`) on port **5433** (not 5432 to avoid conflicts)
+- Database: `basic_memory`
 - Credentials: `basic_memory_user` / `dev_password`
 
 ### 2. Run Postgres Tests
 
 ```bash
 # Run only Postgres tests
+BASIC_MEMORY_TEST_POSTGRES=1 \
+POSTGRES_TEST_URL=postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory \
 pytest -m postgres
 
 # Run specific test with Postgres
-pytest tests/test_entity_repository.py::test_create -m postgres
+BASIC_MEMORY_TEST_POSTGRES=1 \
+POSTGRES_TEST_URL=postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory \
+pytest tests/repository/test_entity_repository.py::test_create -m postgres
 
 # Skip Postgres tests (default behavior)
 pytest -m "not postgres"
@@ -121,7 +127,7 @@ jobs:
     # Postgres service container
     services:
       postgres:
-        image: postgres:17
+        image: pgvector/pgvector:pg17
         env:
           POSTGRES_DB: basic_memory_test
           POSTGRES_USER: basic_memory_user
@@ -169,4 +175,4 @@ docker-compose -f docker-compose-postgres.yml exec postgres pg_isready -U basic_
 
 - [ ] Add `--run-all-backends` CLI flag to run both backends in sequence
 - [ ] Implement test fixtures for backend-specific features (e.g., Postgres full-text search vs SQLite FTS5)
-- [ ] Add performance comparison benchmarks between backends
+- [ ] Add performance comparison benchmarks between backends

tests/conftest.py

@@ -225,6 +225,19 @@ def isolate_routing_env(monkeypatch) -> None:
     monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
 
 
+@pytest.fixture(autouse=True)
+def isolate_data_dir_env(monkeypatch) -> None:
+    """Keep host data-dir env vars from leaking into tests.
+
+    Why: GitHub Actions Ubuntu runners set ``XDG_CONFIG_HOME=/home/runner/.config``,
+    and ``resolve_data_dir()`` honors it ahead of ``Path.home() / ".basic-memory"``.
+    Without clearing it, tests that monkeypatch HOME still see the host XDG path
+    and assertions against the tmp home directory fail.
+    """
+    monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)
+    monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
+
+
 @pytest_asyncio.fixture(autouse=True)
 async def cleanup_global_db_after_test() -> AsyncGenerator[None, None]:
     """Close any module-level DB engine created outside fixture ownership."""

tests/deps/test_task_failure_callback.py

@@ -0,0 +1,38 @@
+"""Tests for background task done-callback error handling."""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from basic_memory.deps.services import _log_task_failure
+
+
+@pytest.mark.asyncio
+async def test_log_task_failure_ignores_cancelled_task():
+    async def slow():
+        await asyncio.sleep(10)
+
+    task = asyncio.create_task(slow())
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    with patch("basic_memory.deps.services.logger.exception") as mock_exc:
+        _log_task_failure(task)
+        mock_exc.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_log_task_failure_logs_real_exception():
+    async def boom():
+        raise ValueError("sync failed")
+
+    task = asyncio.create_task(boom())
+    with pytest.raises(ValueError):
+        await task
+
+    with patch("basic_memory.deps.services.logger.exception") as mock_exc:
+        _log_task_failure(task)
+        mock_exc.assert_called_once()
+        assert "sync failed" in str(mock_exc.call_args)

tests/indexing/test_batch_indexer.py

@@ -688,6 +688,78 @@ async def test_batch_indexer_index_markdown_file_can_defer_relation_resolution(
     assert source.outgoing_relations[0].to_name == "Deferred Target"
 
 
+@pytest.mark.asyncio
+async def test_batch_indexer_uses_strict_link_resolution_for_deferred_relations(
+    app_config,
+    entity_service,
+    entity_repository,
+    relation_repository,
+    search_service,
+    file_service,
+    project_config,
+    monkeypatch,
+):
+    """Regression: batch indexer's deferred relation resolution must call
+    resolve_link with strict=True.
+
+    Mirror of sync_service.resolve_forward_references. Fuzzy fallback in the
+    deferred path silently fills in to_id from BM25/ts_rank results, polluting
+    the graph with confidently-wrong edges. Entity-creation already uses
+    strict=True; this is the other deferred path.
+    """
+    path = "notes/source.md"
+    await _create_file(
+        project_config.home / path,
+        dedent(
+            """
+            ---
+            title: Source
+            type: note
+            ---
+
+            # Source
+
+            - links_to [[never-resolves-target]]
+            """
+        ),
+    )
+
+    batch_indexer = _make_batch_indexer(
+        app_config,
+        entity_service,
+        entity_repository,
+        relation_repository,
+        search_service,
+        file_service,
+    )
+
+    original_resolve_link = entity_service.link_resolver.resolve_link
+    seen_strict: list[object] = []
+
+    async def spy_resolve_link(*args, **kwargs):
+        seen_strict.append(kwargs.get("strict", False))
+        return await original_resolve_link(*args, **kwargs)
+
+    monkeypatch.setattr(entity_service.link_resolver, "resolve_link", spy_resolve_link)
+
+    await batch_indexer.index_files(
+        {path: await _load_input(file_service, path)},
+        max_concurrent=1,
+    )
+
+    assert seen_strict, "batch indexer did not invoke link_resolver.resolve_link"
+    assert all(strict is True for strict in seen_strict), (
+        f"Deferred resolution must call resolve_link(strict=True). Observed: {seen_strict!r}"
+    )
+
+    # The unresolvable relation stayed unresolved.
+    source = await entity_repository.get_by_file_path(path)
+    assert source is not None
+    assert len(source.outgoing_relations) == 1
+    assert source.outgoing_relations[0].to_id is None
+    assert source.outgoing_relations[0].to_name == "never-resolves-target"
+
+
 @pytest.mark.asyncio
 async def test_batch_indexer_strips_frontmatter_from_search_content_when_body_is_empty(
     app_config,

tests/mcp/conftest.py

@@ -20,6 +20,31 @@ def mcp() -> FastMCP:
     return cast(Any, mcp_server)
 
 
+class ContextState:
+    """Minimal FastMCP context-state stub for MCP tests."""
+
+    def __init__(self):
+        self._state: dict[str, object] = {}
+
+    async def get_state(self, key: str):
+        return self._state.get(key)
+
+    async def set_state(self, key: str, value: object, **kwargs) -> None:
+        self._state[key] = value
+
+    async def info(self, message: str) -> None:
+        self._state["info_message"] = message
+
+
+def ctx(context: ContextState) -> Any:
+    return cast(Any, context)
+
+
+@pytest.fixture
+def context_state() -> ContextState:
+    return ContextState()
+
+
 @pytest.fixture(scope="function")
 def app(
     app_config, project_config, engine_factory, config_manager

tests/mcp/test_async_client_modes.py

@@ -1,4 +1,5 @@
 from contextlib import asynccontextmanager
+from typing import AsyncIterator
 
 import httpx
 import pytest
@@ -16,11 +17,15 @@ from basic_memory.mcp.async_client import (
 @pytest.fixture(autouse=True)
 def _reset_async_client_state(monkeypatch):
     async_client_module._client_factory = None
+    async_client_module._prepared_local_asgi_databases.clear()
+    async_client_module._prepared_local_asgi_database_prepare_locks.clear()
     monkeypatch.delenv("BASIC_MEMORY_FORCE_LOCAL", raising=False)
     monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)
     monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
     yield
     async_client_module._client_factory = None
+    async_client_module._prepared_local_asgi_databases.clear()
+    async_client_module._prepared_local_asgi_database_prepare_locks.clear()
 
 
 @pytest.mark.asyncio
@@ -51,6 +56,233 @@ async def test_get_client_default_uses_local_asgi_transport(config_manager):
         assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
 
 
+@pytest.mark.asyncio
+async def test_get_client_preinitializes_local_asgi_database(config_manager, monkeypatch):
+    """Local ASGI routing initializes DB state before request handling."""
+    from basic_memory import db
+    from basic_memory.api.app import app as fastapi_app
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    async def fake_get_or_create_db(db_path):
+        calls.append(db_path)
+        return engine, session_maker
+
+    monkeypatch.setattr(db, "get_or_create_db", fake_get_or_create_db)
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == [cfg.database_path]
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.parametrize("async_override", [False, True])
+@pytest.mark.asyncio
+async def test_get_client_uses_existing_local_asgi_database_override(
+    config_manager,
+    async_override,
+):
+    """Local ASGI routing honors FastAPI test dependency overrides."""
+    from basic_memory.api.app import app as fastapi_app
+    from basic_memory.deps import get_engine_factory
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_overrides = dict(fastapi_app.dependency_overrides)
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    if async_override:
+
+        async def override_engine_factory():
+            calls.append("override")
+            return engine, session_maker
+
+    else:
+
+        def override_engine_factory():
+            calls.append("override")
+            return engine, session_maker
+
+    fastapi_app.dependency_overrides[get_engine_factory] = override_engine_factory
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == ["override"]
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        fastapi_app.dependency_overrides = previous_overrides
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.asyncio
+async def test_get_client_resolves_local_asgi_database_override_with_fastapi_di(
+    config_manager,
+):
+    """Local ASGI DB pre-init honors FastAPI dependency injection and cleanup."""
+    from fastapi import Request
+
+    from basic_memory.api.app import app as fastapi_app
+    from basic_memory.deps import get_engine_factory
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_overrides = dict(fastapi_app.dependency_overrides)
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+    cleanup = []
+
+    async def override_engine_factory(
+        request: Request,
+    ) -> AsyncIterator[tuple[object, object]]:
+        calls.append(request.app)
+        try:
+            yield engine, session_maker
+        finally:
+            cleanup.append("closed")
+
+    fastapi_app.dependency_overrides[get_engine_factory] = override_engine_factory
+
+    try:
+        async with get_client() as client:
+            assert isinstance(client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+            assert calls == [fastapi_app]
+            assert cleanup == []
+            assert fastapi_app.state.engine is engine
+            assert fastapi_app.state.session_maker is session_maker
+        assert cleanup == ["closed"]
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        fastapi_app.dependency_overrides = previous_overrides
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
+@pytest.mark.asyncio
+async def test_get_client_keeps_local_asgi_database_during_overlapping_contexts(
+    config_manager,
+    monkeypatch,
+):
+    """Local ASGI database state stays installed until the last overlapping client exits."""
+    from basic_memory import db
+    from basic_memory.api.app import app as fastapi_app
+
+    cfg = config_manager.load_config()
+    config_manager.save_config(cfg)
+
+    previous_engine = getattr(fastapi_app.state, "engine", None)
+    previous_session_maker = getattr(fastapi_app.state, "session_maker", None)
+    fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+    fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+
+    engine = object()
+    session_maker = object()
+    calls = []
+
+    async def fake_get_or_create_db(db_path):
+        calls.append(db_path)
+        return engine, session_maker
+
+    monkeypatch.setattr(db, "get_or_create_db", fake_get_or_create_db)
+
+    first_context = get_client()
+    second_context = get_client()
+    first_entered = False
+    second_entered = False
+    first_exited = False
+
+    try:
+        first_client = await first_context.__aenter__()
+        first_entered = True
+        second_client = await second_context.__aenter__()
+        second_entered = True
+
+        assert isinstance(first_client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(second_client._transport, httpx.ASGITransport)  # pyright: ignore[reportPrivateUsage]
+        assert calls == [cfg.database_path]
+
+        await first_context.__aexit__(None, None, None)
+        first_exited = True
+
+        assert fastapi_app.state.engine is engine
+        assert fastapi_app.state.session_maker is session_maker
+
+        await second_context.__aexit__(None, None, None)
+        second_entered = False
+
+        assert not hasattr(fastapi_app.state, "engine")
+        assert not hasattr(fastapi_app.state, "session_maker")
+    finally:
+        if second_entered:
+            await second_context.__aexit__(None, None, None)
+        if first_entered and not first_exited:
+            await first_context.__aexit__(None, None, None)
+
+        if previous_engine is None:
+            fastapi_app.state._state.pop("engine", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.engine = previous_engine
+        if previous_session_maker is None:
+            fastapi_app.state._state.pop("session_maker", None)  # pyright: ignore[reportPrivateUsage]
+        else:
+            fastapi_app.state.session_maker = previous_session_maker
+
+
 @pytest.mark.asyncio
 async def test_get_client_explicit_cloud_uses_api_key(config_manager, monkeypatch):
     cfg = config_manager.load_config()

tests/mcp/test_project_context.py

@@ -11,25 +11,7 @@ from typing import Any, AsyncIterator, cast
 
 import pytest
 
-
-class _ContextState:
-    """Minimal FastMCP context-state stub for unit tests."""
-
-    def __init__(self):
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-    async def info(self, message: str) -> None:
-        self._state["info_message"] = message
-
-
-def _ctx(context: _ContextState) -> Any:
-    return cast(Any, context)
+from tests.mcp.conftest import ContextState, ctx
 
 
 def _workspace(
@@ -218,7 +200,7 @@ async def test_env_constraint_overrides_default(config_manager, config_home, mon
 async def test_workspace_auto_selects_single_and_caches(monkeypatch):
     from basic_memory.mcp.project_context import resolve_workspace_parameter
 
-    context = _ContextState()
+    context = ContextState()
     only_workspace = _workspace(
         tenant_id="11111111-1111-1111-1111-111111111111",
         workspace_type="personal",
@@ -236,7 +218,7 @@ async def test_workspace_auto_selects_single_and_caches(monkeypatch):
         fake_get_available_workspaces,
     )
 
-    resolved = await resolve_workspace_parameter(context=_ctx(context))
+    resolved = await resolve_workspace_parameter(context=ctx(context))
     assert resolved.tenant_id == only_workspace.tenant_id
     assert await context.get_state("active_workspace") == only_workspace.model_dump()
 
@@ -272,7 +254,7 @@ async def test_workspace_requires_user_choice_when_multiple(monkeypatch):
     )
 
     with pytest.raises(ValueError, match="Multiple workspaces are available"):
-        await resolve_workspace_parameter(context=_ctx(_ContextState()))
+        await resolve_workspace_parameter(context=ctx(ContextState()))
 
 
 @pytest.mark.asyncio
@@ -416,7 +398,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
             role="owner",
         ),
     ]
-    context = _ContextState()
+    context = ContextState()
     await context.set_state("active_workspace", cached_workspace.model_dump())
     fetches = 0
 
@@ -431,7 +413,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
     )
 
     with pytest.raises(ValueError) as exc_info:
-        await resolve_workspace_parameter(workspace="organization", context=_ctx(context))
+        await resolve_workspace_parameter(workspace="organization", context=ctx(context))
 
     message = str(exc_info.value)
     assert fetches == 1
@@ -452,7 +434,7 @@ async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
         role="owner",
         is_default=True,
     )
-    context = _ContextState()
+    context = ContextState()
     await context.set_state("active_workspace", cached_workspace.model_dump())
 
     async def fail_if_called(context=None):  # pragma: no cover
@@ -463,7 +445,7 @@ async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
         fail_if_called,
     )
 
-    resolved = await resolve_workspace_parameter(context=_ctx(context))
+    resolved = await resolve_workspace_parameter(context=ctx(context))
     assert resolved.tenant_id == cached_workspace.tenant_id
 
 
@@ -476,7 +458,7 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
         invalidate_workspace_project_index,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -513,8 +495,8 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
         fake_fetch_workspace_project_entries,
     )
 
-    first = await _ensure_workspace_project_index(context=_ctx(context))
-    second = await _ensure_workspace_project_index(context=_ctx(context))
+    first = await _ensure_workspace_project_index(context=ctx(context))
+    second = await _ensure_workspace_project_index(context=ctx(context))
 
     assert [entry.qualified_name for entry in first.entries] == [
         "personal/personal-notes",
@@ -523,8 +505,8 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
     assert second.entries == first.entries
     assert calls == ["personal", "acme"]
 
-    await invalidate_workspace_project_index(_ctx(context))
-    await _ensure_workspace_project_index(context=_ctx(context))
+    await invalidate_workspace_project_index(ctx(context))
+    await _ensure_workspace_project_index(context=ctx(context))
     assert calls == ["personal", "acme", "personal", "acme"]
 
 
@@ -539,7 +521,7 @@ async def test_workspace_project_index_keeps_successes_when_workspace_fetch_fail
         resolve_workspace_project_identifier,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -572,21 +554,21 @@ async def test_workspace_project_index_keeps_successes_when_workspace_fetch_fail
         fake_fetch_workspace_project_entries,
     )
 
-    index = await _ensure_workspace_project_index(context=_ctx(context))
+    index = await _ensure_workspace_project_index(context=ctx(context))
 
     assert [entry.qualified_name for entry in index.entries] == ["personal/meeting-notes"]
     assert [workspace.slug for workspace in index.failed_workspaces] == ["acme"]
 
     resolved = await resolve_workspace_project_identifier(
         "personal/meeting-notes",
-        context=_ctx(context),
+        context=ctx(context),
     )
     assert resolved.project.external_id == "personal-meeting-notes"
 
     with pytest.raises(ValueError, match="Use 'personal/meeting-notes'"):
         await resolve_workspace_project_identifier(
             "meeting-notes",
-            context=_ctx(context),
+            context=ctx(context),
         )
 
 
@@ -800,8 +782,11 @@ async def test_detect_project_from_memory_url_prefix_skips_workspace_discovery_f
 
     monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_if_called)
 
+    # Keep this at two path segments. Three-segment memory URLs are valid
+    # workspace-qualified candidates in mixed local+cloud mode, so they should
+    # attempt workspace discovery even when a local project is configured.
     resolved = await detect_project_from_memory_url_prefix(
-        "memory://notes/foo/bar",
+        "memory://notes/foo",
         BasicMemoryConfig(
             projects={"main": ProjectEntry(path="/tmp/main")},
             cloud_api_key="bmc_test123",
@@ -811,6 +796,101 @@ async def test_detect_project_from_memory_url_prefix_skips_workspace_discovery_f
     assert resolved is None
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("identifier", "message"),
+    [
+        ("memory://docs/topic/note", "No accessible workspaces found for this account."),
+        (
+            "docs/topic/note",
+            "Unable to discover projects in any accessible workspace. Failed workspaces: personal",
+        ),
+    ],
+)
+async def test_detect_project_from_identifier_prefix_ignores_workspace_discovery_failures(
+    monkeypatch,
+    identifier,
+    message,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import BasicMemoryConfig, ProjectEntry
+    from basic_memory.mcp.project_context import detect_project_from_identifier_prefix
+
+    async def fail_workspace_index(context=None):
+        raise ValueError(message)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_workspace_index)
+
+    resolved = await detect_project_from_identifier_prefix(
+        identifier,
+        BasicMemoryConfig(
+            projects={"main": ProjectEntry(path="/tmp/main")},
+            cloud_api_key="bmc_test123",
+        ),
+    )
+
+    assert resolved is None
+
+
+@pytest.mark.asyncio
+async def test_detect_project_from_identifier_prefix_resolves_workspace_with_local_config(
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import BasicMemoryConfig, ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+        detect_project_from_identifier_prefix,
+    )
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (
+            WorkspaceProjectEntry(
+                workspace=personal,
+                project=_project("main", id=1, external_id="personal-main-id"),
+            ),
+        ),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+
+    config = BasicMemoryConfig(
+        projects={"hermes-memory": ProjectEntry(path="/tmp/hermes-memory")},
+        cloud_api_key="bmc_test123",
+    )
+
+    assert (
+        await detect_project_from_identifier_prefix(
+            "memory://personal/main/main-to-do-list",
+            config,
+        )
+        == "personal/main"
+    )
+    assert (
+        await detect_project_from_identifier_prefix(
+            "personal/main/main-to-do-list",
+            config,
+        )
+        == "personal/main"
+    )
+
+
 @pytest.mark.asyncio
 async def test_resolve_workspace_qualified_memory_url_ignores_workspace_project_miss(
     monkeypatch,
@@ -1045,7 +1125,7 @@ async def test_resolve_workspace_project_identifier_uses_active_workspace_for_du
         resolve_workspace_project_identifier,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -1081,7 +1161,7 @@ async def test_resolve_workspace_project_identifier_uses_active_workspace_for_du
 
     resolved = await resolve_workspace_project_identifier(
         "meeting-notes",
-        context=_ctx(context),
+        context=ctx(context),
     )
     assert resolved.workspace.slug == "acme"
     assert resolved.project.external_id == "acme-project-id"
@@ -1232,6 +1312,318 @@ async def test_get_project_client_with_project_id_routes_locally_without_cloud(
     assert captured["validated_project"] == canonical_uuid
 
 
+@pytest.mark.asyncio
+async def test_get_project_client_with_local_project_id_routes_locally_with_cloud_credentials(
+    config_manager, monkeypatch
+):
+    """A local-only project_id should resolve locally before cloud workspace discovery."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("local project_id should not require cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project("Hermes Memory", id=99, external_id=project)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    local_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(project_id=local_uuid) as (_, active):
+        assert active.external_id == local_uuid
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == local_uuid
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_local_project_id_clears_cached_workspace(
+    config_manager, monkeypatch
+):
+    """Local project_id routing must not inherit a previous cloud workspace context."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("local project_id should not require cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project("Hermes Memory", id=99, external_id=project)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    local_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(
+        project_id=local_uuid,
+        context=ctx(context),
+    ) as (_, active):
+        assert active.external_id == local_uuid
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == local_uuid
+    assert await context.get_state("active_workspace") is None
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_project_id_respects_env_constraint(
+    config_manager, monkeypatch
+):
+    """BASIC_MEMORY_MCP_PROJECT must remain authoritative when project_id is supplied."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["env-project"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "env-project")
+    )
+    config.projects["other-project"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "other-project")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("env-constrained local project should not use cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project(str(project), id=99, external_id="env-project-id")
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    requested_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(project_id=requested_uuid) as (_, active):
+        assert active.name == "env-project"
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == "env-project"
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_cloud_project_id_routes_to_workspace_with_local_config(
+    config_manager, monkeypatch
+):
+    """Cloud project_id routing falls through to the workspace index after a local miss."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from mcp.server.fastmcp.exceptions import ToolError
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    cloud_uuid = "22222222-2222-2222-2222-222222222222"
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    cloud_project = _project("main", id=2, external_id=cloud_uuid)
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=cloud_project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    get_client_calls: list[dict[str, str | None]] = []
+    validated_projects: list[str] = []
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        get_client_calls.append({"project_name": project_name, "workspace": workspace})
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        validated_projects.append(project)
+        if project == cloud_uuid:
+            raise ToolError("project not found")
+        return _project(project, id=cloud_project.id, external_id=cloud_project.external_id)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    async with project_context.get_project_client(project_id=cloud_uuid) as (_, active):
+        assert active.external_id == cloud_uuid
+
+    assert get_client_calls == [
+        {"project_name": None, "workspace": None},
+        {"project_name": "main", "workspace": "personal-tenant"},
+    ]
+    assert validated_projects == [cloud_uuid, "main"]
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_clears_stale_cached_project_for_workspace_route(
+    config_manager, monkeypatch
+):
+    """Workspace routes must not reuse a same-name project with a different UUID."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import WorkspaceProjectEntry
+    from basic_memory.schemas.project_info import ProjectItem
+
+    config = config_manager.load_config()
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    expected_uuid = "22222222-2222-2222-2222-222222222222"
+    expected_project = _project("main", id=2, external_id=expected_uuid)
+    stale_project = ProjectItem(
+        id=99,
+        external_id="33333333-3333-3333-3333-333333333333",
+        name="main",
+        path="/tmp/stale-main",
+        is_default=False,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+    await context.set_state("active_project", stale_project.model_dump())
+
+    async def fake_resolve_workspace_project_identifier(project_name, context=None):
+        assert project_name == "personal/main"
+        return WorkspaceProjectEntry(workspace=personal, project=expected_project)
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        assert project_name == "main"
+        assert workspace == "personal-tenant"
+        yield object()
+
+    class FakeResponse:
+        def json(self):
+            return {
+                "external_id": expected_uuid,
+                "project_id": expected_project.id,
+                "name": expected_project.name,
+                "permalink": expected_project.permalink,
+                "path": expected_project.path,
+                "is_active": True,
+                "is_default": False,
+                "resolution_method": "permalink",
+            }
+
+    calls = {"count": 0}
+
+    async def fake_call_post(*args, **kwargs):
+        calls["count"] += 1
+        return FakeResponse()
+
+    monkeypatch.setattr(
+        project_context,
+        "resolve_workspace_project_identifier",
+        fake_resolve_workspace_project_identifier,
+    )
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+
+    async with project_context.get_project_client(
+        project="personal/main",
+        context=ctx(context),
+    ) as (_, active):
+        assert active.external_id == expected_uuid
+
+    cached_project = await context.get_state("active_project")
+    assert isinstance(cached_project, dict)
+    assert cached_project["external_id"] == expected_uuid
+    assert calls["count"] == 1
+
+
 @pytest.mark.asyncio
 async def test_get_project_client_prefers_project_id_over_project_name(monkeypatch):
     """When both project and project_id are passed, the UUID takes precedence."""
@@ -1270,7 +1662,7 @@ async def test_resolve_project_parameter_uses_cached_active_project_before_api_d
     config.default_project = None
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1288,7 +1680,7 @@ async def test_resolve_project_parameter_uses_cached_active_project_before_api_d
         fail_if_called,
     )
 
-    resolved = await resolve_project_parameter(project=None, context=_ctx(context))
+    resolved = await resolve_project_parameter(project=None, context=ctx(context))
     assert resolved == cached_project.name
 
 
@@ -1302,7 +1694,7 @@ async def test_resolve_project_parameter_caches_api_default_project_name(
     config.default_project = None
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     api_calls = {"count": 0}
 
     async def fake_default_lookup():
@@ -1314,8 +1706,8 @@ async def test_resolve_project_parameter_caches_api_default_project_name(
         fake_default_lookup,
     )
 
-    first = await resolve_project_parameter(project=None, context=_ctx(context))
-    second = await resolve_project_parameter(project=None, context=_ctx(context))
+    first = await resolve_project_parameter(project=None, context=ctx(context))
+    second = await resolve_project_parameter(project=None, context=ctx(context))
 
     assert first == "cloud-default"
     assert second == "cloud-default"
@@ -1327,7 +1719,7 @@ async def test_get_active_project_uses_cached_project_before_resolution(monkeypa
     from basic_memory.mcp.project_context import get_active_project
     from basic_memory.schemas.project_info import ProjectItem
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1345,7 +1737,7 @@ async def test_get_active_project_uses_cached_project_before_resolution(monkeypa
         fail_if_called,
     )
 
-    resolved = await get_active_project(client=cast(Any, None), context=_ctx(context))
+    resolved = await get_active_project(client=cast(Any, None), context=ctx(context))
     assert resolved == cached_project
 
 
@@ -1354,7 +1746,7 @@ async def test_get_active_project_uses_cached_project_for_explicit_permalink(mon
     from basic_memory.mcp.project_context import get_active_project
     from basic_memory.schemas.project_info import ProjectItem
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1375,7 +1767,7 @@ async def test_get_active_project_uses_cached_project_for_explicit_permalink(mon
     )
 
     resolved = await get_active_project(
-        client=cast(Any, None), project="my-research", context=_ctx(context)
+        client=cast(Any, None), project="my-research", context=ctx(context)
     )
     assert resolved == cached_project
 
@@ -1391,7 +1783,7 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
     config.permalinks_include_project = False
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1416,7 +1808,7 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://my-research/notes/roadmap.md",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1438,7 +1830,7 @@ async def test_resolve_project_and_path_keeps_workspace_qualified_canonical_path
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1464,7 +1856,7 @@ async def test_resolve_project_and_path_keeps_workspace_qualified_canonical_path
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://team-paul/main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1486,7 +1878,7 @@ async def test_resolve_project_and_path_preserves_personal_workspace_prefix(
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1513,7 +1905,7 @@ async def test_resolve_project_and_path_preserves_personal_workspace_prefix(
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://personal/main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1532,7 +1924,7 @@ async def test_resolve_project_and_path_preserves_existing_project_prefixed_memo
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1545,7 +1937,7 @@ async def test_resolve_project_and_path_preserves_existing_project_prefixed_memo
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1568,7 +1960,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1599,7 +1991,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
         client=cast(Any, None),
         identifier="memory://notes/foo",
         project="main",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1610,7 +2002,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
         client=cast(Any, None),
         identifier="memory://main",
         project="main",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1650,7 +2042,7 @@ async def test_resolve_project_and_path_uses_workspace_context_for_project_root(
             client=cast(Any, None),
             identifier="memory://main",
             project="main",
-            context=_ctx(_ContextState()),
+            context=ctx(ContextState()),
         )
 
     assert active_project == active
@@ -1671,7 +2063,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_cached_project
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1705,7 +2097,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_cached_project
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1725,7 +2117,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_resolved_proje
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     team_workspace = _workspace(
         tenant_id="team-tenant",
         workspace_type="organization",
@@ -1764,7 +2156,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_resolved_proje
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://research/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project.name == "Research"
@@ -1895,7 +2287,7 @@ class TestGetProjectClientRoutingOrder:
         )
         config_manager.save_config(config)
 
-        context = _ContextState()
+        context = ContextState()
         stale_workspace = _workspace(
             tenant_id="team-tenant",
             workspace_type="organization",
@@ -1942,13 +2334,13 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="local-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (client, active_project):
             _, resolved_path, is_memory_url = await resolve_project_and_path(
                 client=cast(Any, client),
                 identifier="memory://notes/foo",
                 project="local-proj",
-                context=_ctx(context),
+                context=ctx(context),
             )
 
         assert active_project == active
@@ -1990,7 +2382,7 @@ class TestGetProjectClientRoutingOrder:
             name="Team Paul",
             role="editor",
         )
-        context = _ContextState()
+        context = ContextState()
         await context.set_state("active_workspace", workspace.model_dump())
 
         async def fail_resolve_workspace_parameter(workspace=None, context=None):
@@ -2034,7 +2426,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 
@@ -2154,7 +2546,7 @@ class TestGetProjectClientRoutingOrder:
             name="Team Paul",
             role="editor",
         )
-        context = _ContextState()
+        context = ContextState()
         await context.set_state("available_workspaces", ["ignored", workspace.model_dump()])
 
         async def fail_workspace_provider():  # pragma: no cover
@@ -2193,7 +2585,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 
@@ -2335,7 +2727,7 @@ class TestGetProjectClientRoutingOrder:
         config.cloud_api_key = "bmc_test123"
         config_manager.save_config(config)
 
-        context = _ContextState()
+        context = ContextState()
         stale_workspace = _workspace(
             tenant_id="other-tenant-id",
             workspace_type="organization",
@@ -2376,7 +2768,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 

tests/mcp/test_project_context_telemetry.py

@@ -4,30 +4,17 @@ from __future__ import annotations
 
 import importlib
 from contextlib import contextmanager
-from typing import Any, cast
 
 import logfire
 import pytest
 
 from basic_memory.config import ProjectEntry
 from basic_memory.schemas.cloud import WorkspaceInfo
+from tests.mcp.conftest import ContextState, ctx
 
 project_context = importlib.import_module("basic_memory.mcp.project_context")
 
 
-class _ContextState:
-    """Minimal FastMCP context-state stub for unit tests."""
-
-    def __init__(self) -> None:
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-
 def _capture_spans():
     spans: list[tuple[str, dict]] = []
 
@@ -42,7 +29,7 @@ def _capture_spans():
 @pytest.mark.asyncio
 async def test_resolve_workspace_parameter_emits_routing_span(monkeypatch) -> None:
     spans, fake_span = _capture_spans()
-    context = _ContextState()
+    context = ContextState()
     workspace = WorkspaceInfo(
         tenant_id="11111111-1111-1111-1111-111111111111",
         workspace_type="personal",
@@ -58,7 +45,7 @@ async def test_resolve_workspace_parameter_emits_routing_span(monkeypatch) -> No
     monkeypatch.setattr(logfire, "span", fake_span)
     monkeypatch.setattr(project_context, "get_available_workspaces", fake_get_available_workspaces)
 
-    resolved = await project_context.resolve_workspace_parameter(context=cast(Any, context))
+    resolved = await project_context.resolve_workspace_parameter(context=ctx(context))
 
     assert resolved.tenant_id == workspace.tenant_id
     assert spans == [

tests/mcp/test_tool_edit_note.py

@@ -914,8 +914,8 @@ async def test_edit_note_workspace_qualified_plain_permalink_requires_explicit_r
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -958,8 +958,8 @@ async def test_edit_note_workspace_qualified_plain_permalink_json_error(
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -1001,8 +1001,8 @@ async def test_edit_note_ambiguous_namespace_identifier_returns_guidance(
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -1054,6 +1054,83 @@ async def test_edit_note_workspace_project_args_compose_explicit_route(monkeypat
     assert captured_routes == [("docs/setup", None)]
 
 
+@pytest.mark.asyncio
+async def test_edit_note_plain_workspace_route_returns_guidance_with_local_config(
+    monkeypatch,
+    config_manager,
+    test_project,
+):
+    """Mixed local+cloud configs should still stop ambiguous plain write routes."""
+    from contextlib import asynccontextmanager
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+
+    edit_note_module = importlib.import_module("basic_memory.mcp.tools.edit_note")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (
+            WorkspaceProjectEntry(
+                workspace=personal,
+                project=ProjectItem(
+                    id=1,
+                    external_id="11111111-1111-1111-1111-111111111111",
+                    name="main",
+                    path="/tmp/main",
+                    is_default=False,
+                ),
+            ),
+        ),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    @asynccontextmanager
+    async def fail_if_called(*args, **kwargs):
+        raise AssertionError("ambiguous plain identifiers should not select a project client")
+        yield
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(edit_note_module, "get_project_client", fail_if_called)
+
+    result = await edit_note(
+        identifier="personal/main/team/plain-edit-note",
+        operation="append",
+        content="\nAppended via plain workspace-qualified permalink.",
+        project=None,
+    )
+
+    assert isinstance(result, str)
+    assert "# Edit Failed - Ambiguous Identifier" in result
+    assert 'project="personal/main"' in result
+
+
 @pytest.mark.asyncio
 async def test_edit_note_three_segment_plain_path_stays_local_without_workspace_discovery(
     monkeypatch,
@@ -1077,8 +1154,8 @@ async def test_edit_note_three_segment_plain_path_stays_local_without_workspace_
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: False,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: False,
     )
     monkeypatch.setattr(
         edit_note_module,

tests/mcp/test_tool_project_management.py

@@ -9,7 +9,8 @@ from sqlalchemy import select
 
 from basic_memory import db
 from basic_memory.mcp.tools import list_memory_projects, create_memory_project, delete_project
-from basic_memory.mcp.tools.project_management import _merge_projects
+from basic_memory.config import BasicMemoryConfig, ProjectEntry
+from basic_memory.mcp.tools.project_management import _merge_projects, _merge_workspace_projects
 from basic_memory.models.project import Project
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 
@@ -909,6 +910,220 @@ def test_merge_projects_overlap():
     assert merged[0]["workspace_tenant_id"] == "org-456"
 
 
+def test_merge_workspace_projects_attaches_local_state_to_one_duplicate_workspace(tmp_path):
+    """A same-name team workspace project should stay cloud-only (#848)."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "local+cloud"
+    assert personal_project["local_path"] == local_path
+    assert personal_project["path"] == local_path
+    assert team_project["source"] == "cloud"
+    assert team_project["local_path"] is None
+    assert team_project["path"] == "/cloud/team-main"
+
+
+def test_merge_workspace_projects_uses_configured_workspace_for_local_state(tmp_path):
+    """Per-project workspace_id should select the attached duplicate row."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(
+        projects={
+            "main": ProjectEntry(
+                path=local_path,
+                workspace_id="team-tenant",
+            )
+        }
+    )
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "cloud"
+    assert personal_project["local_path"] is None
+    assert personal_project["path"] == "/cloud/personal-main"
+    assert team_project["source"] == "local+cloud"
+    assert team_project["local_path"] == local_path
+    assert team_project["path"] == local_path
+
+
+def test_merge_workspace_projects_uses_default_workspace_for_local_state(tmp_path):
+    """Global default_workspace should attach local state before cloud default fallback."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=True,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (personal_ws, [personal_main]),
+            (team_ws, [team_main]),
+        ]
+    )
+    config = BasicMemoryConfig(
+        projects={"main": ProjectEntry(path=local_path)},
+        default_workspace="team-tenant",
+    )
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "cloud"
+    assert personal_project["local_path"] is None
+    assert personal_project["path"] == "/cloud/personal-main"
+    assert team_project["source"] == "local+cloud"
+    assert team_project["local_path"] == local_path
+    assert team_project["path"] == local_path
+
+
+def test_merge_workspace_projects_sorted_fallback_attaches_personal_workspace(tmp_path):
+    """When config has no preference and no cloud default exists, use stable priority."""
+    local_path = str(tmp_path / "main")
+    local_main = _make_project("main", local_path, is_default=True)
+    local_list = _make_list([local_main], default="main")
+    personal_main = _make_project(
+        "main",
+        "/cloud/personal-main",
+        id=10,
+        external_id="personal-main-uuid",
+    )
+    team_main = _make_project(
+        "main",
+        "/cloud/team-main",
+        id=11,
+        external_id="team-main-uuid",
+    )
+    personal_ws = _make_workspace(
+        "personal-tenant",
+        "Personal",
+        slug="personal",
+        is_default=False,
+    )
+    team_ws = _make_workspace(
+        "team-tenant",
+        "Team",
+        workspace_type="organization",
+        slug="team",
+    )
+    workspace_index = _make_workspace_index(
+        [
+            (team_ws, [team_main]),
+            (personal_ws, [personal_main]),
+        ]
+    )
+    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
+
+    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
+
+    by_qualified_name = {project["qualified_name"]: project for project in merged}
+    personal_project = by_qualified_name["personal/main"]
+    team_project = by_qualified_name["team/main"]
+
+    assert personal_project["source"] == "local+cloud"
+    assert personal_project["local_path"] == local_path
+    assert personal_project["path"] == local_path
+    assert team_project["source"] == "cloud"
+    assert team_project["local_path"] is None
+    assert team_project["path"] == "/cloud/team-main"
+
+
 # --- Workspace passthrough tests ---
 
 

tests/mcp/test_tool_read_content.py

@@ -6,6 +6,10 @@ We keep these tests focused on path boundary/security checks, and rely on
 
 from __future__ import annotations
 
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+
 import pytest
 from mcp.server.fastmcp.exceptions import ToolError
 
@@ -157,6 +161,109 @@ async def test_read_content_allows_safe_path_integration(client, test_project):
     assert "safe note" in result["text"]
 
 
+@pytest.mark.asyncio
+async def test_read_content_workspace_memory_url_routes_with_local_config(
+    monkeypatch,
+    config_manager,
+):
+    """Workspace-qualified memory URLs should route even when local projects exist."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+
+    read_content_module = importlib.import_module("basic_memory.mcp.tools.read_content")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    @asynccontextmanager
+    async def fake_get_project_client(project=None, context=None, project_id=None):
+        assert project == "personal/main"
+        assert project_id is None
+        yield (
+            object(),
+            SimpleNamespace(
+                name="main",
+                external_id="11111111-1111-1111-1111-111111111111",
+                home=Path("/tmp/main"),
+            ),
+        )
+
+    async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
+        assert identifier == "memory://personal/main/docs/report"
+        assert project == "main"
+        return None, "personal/main/docs/report", True
+
+    async def fake_resolve_entity_id(client, project_id, url):
+        assert project_id == "11111111-1111-1111-1111-111111111111"
+        assert url == "personal/main/docs/report"
+        return "entity-1"
+
+    class FakeResponse:
+        headers = {"content-type": "text/markdown", "content-length": "17"}
+        text = "# Routed Content"
+        content = b"# Routed Content"
+
+    async def fake_call_get(client, path, **kwargs):
+        assert path == "/v2/projects/11111111-1111-1111-1111-111111111111/resource/entity-1"
+        return FakeResponse()
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(read_content_module, "get_project_client", fake_get_project_client)
+    monkeypatch.setattr(
+        read_content_module,
+        "resolve_project_and_path",
+        fake_resolve_project_and_path,
+    )
+    monkeypatch.setattr(read_content_module, "resolve_entity_id", fake_resolve_entity_id)
+    monkeypatch.setattr(read_content_module, "call_get", fake_call_get)
+
+    result = await read_content(path="memory://personal/main/docs/report")
+
+    assert result == {
+        "type": "text",
+        "text": "# Routed Content",
+        "content_type": "text/markdown",
+        "encoding": "utf-8",
+    }
+
+
 @pytest.mark.asyncio
 async def test_read_content_empty_path_does_not_trigger_security_error(client, test_project):
     try:

tests/mcp/test_tool_read_note.py

@@ -1,5 +1,7 @@
 """Tests for note tools that exercise the full stack with SQLite."""
 
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
 from textwrap import dedent
 
 import pytest
@@ -7,6 +9,7 @@ import pytest
 from basic_memory.mcp.tools import write_note, read_note
 from basic_memory.mcp.tools.read_note import _parse_opening_frontmatter
 from basic_memory.utils import normalize_newlines
+from tests.mcp.conftest import ContextState, ctx
 
 
 def test_parse_opening_frontmatter_handles_crlf():
@@ -166,6 +169,130 @@ async def test_read_note_title_fallback_requires_exact_title_match(monkeypatch,
     assert "existing note content" not in result
 
 
+@pytest.mark.asyncio
+async def test_read_note_explicit_workspace_project_ignores_stale_cached_project(
+    monkeypatch,
+    config_manager,
+):
+    """Explicit workspace routing should use the resolved cloud UUID, not stale cache."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import WorkspaceProjectEntry
+    from basic_memory.schemas.project_info import ProjectItem
+
+    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
+    clients_mod = importlib.import_module("basic_memory.mcp.clients")
+
+    config = config_manager.load_config()
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = project_context.WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    expected_uuid = "22222222-2222-2222-2222-222222222222"
+    expected_project = ProjectItem(
+        id=2,
+        external_id=expected_uuid,
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    stale_project = ProjectItem(
+        id=99,
+        external_id="33333333-3333-3333-3333-333333333333",
+        name="main",
+        path="/tmp/stale-main",
+        is_default=False,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+    await context.set_state("active_project", stale_project.model_dump())
+
+    async def fake_resolve_workspace_project_identifier(project_name, context=None):
+        assert project_name == "personal/main"
+        return WorkspaceProjectEntry(workspace=personal, project=expected_project)
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        assert project_name == "main"
+        assert workspace == "personal-tenant"
+        yield object()
+
+    class FakeProjectResolveResponse:
+        def json(self):
+            return {
+                "external_id": expected_uuid,
+                "project_id": expected_project.id,
+                "name": expected_project.name,
+                "permalink": expected_project.permalink,
+                "path": expected_project.path,
+                "is_active": True,
+                "is_default": False,
+                "resolution_method": "permalink",
+            }
+
+    async def fake_call_post(*args, **kwargs):
+        return FakeProjectResolveResponse()
+
+    class FakeKnowledgeClient:
+        def __init__(self, client, project_id):
+            assert project_id == expected_uuid
+
+        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
+            assert identifier == "personal/main/todo"
+            return "entity-1"
+
+        async def get_entity(self, entity_id: str):
+            assert entity_id == "entity-1"
+            return SimpleNamespace(
+                title="TODO",
+                permalink="personal/main/todo",
+                file_path="TODO.md",
+            )
+
+    class FakeResourceClient:
+        def __init__(self, client, project_id):
+            assert project_id == expected_uuid
+
+        async def read(self, entity_id: str):
+            assert entity_id == "entity-1"
+            return SimpleNamespace(
+                status_code=200,
+                text="---\ntitle: TODO\n---\n\n# TODO - Priorities & Tasks\n",
+            )
+
+    monkeypatch.setattr(
+        project_context,
+        "resolve_workspace_project_identifier",
+        fake_resolve_workspace_project_identifier,
+    )
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+    monkeypatch.setattr(clients_mod, "KnowledgeClient", FakeKnowledgeClient)
+    monkeypatch.setattr(clients_mod, "ResourceClient", FakeResourceClient)
+
+    result = await read_note_module.read_note(
+        "memory://todo",
+        project="personal/main",
+        output_format="json",
+        context=ctx(context),
+    )
+
+    assert isinstance(result, dict)
+    assert result["permalink"] == "personal/main/todo"
+    assert result["content"].strip() == "# TODO - Priorities & Tasks"
+
+
 @pytest.mark.asyncio
 async def test_note_unicode_content(app, test_project):
     """Test handling of unicode content in"""

tests/mcp/test_tool_search.py

@@ -3,6 +3,8 @@
 import pytest
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta
+from types import SimpleNamespace
+from typing import cast
 
 from basic_memory.mcp.tools import write_note
 from basic_memory.mcp.tools.search import (
@@ -171,6 +173,109 @@ async def test_search_memory_url_with_project_prefix(client, test_project):
         pytest.fail(f"Search failed with error: {response}")
 
 
+@pytest.mark.asyncio
+async def test_search_workspace_memory_url_routes_with_local_config(monkeypatch, config_manager):
+    """Workspace-qualified memory URL searches should self-route in mixed mode."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+    from basic_memory.schemas.search import SearchItemType, SearchResult
+
+    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
+    clients_mod = importlib.import_module("basic_memory.mcp.clients")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project_item = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=project_item),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_project_client(project=None, context=None, project_id=None):
+        captured["project"] = project
+        captured["project_id"] = project_id
+        yield object(), SimpleNamespace(name="main", external_id=project_item.external_id)
+
+    async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
+        assert identifier == "memory://personal/main/tests/search-note"
+        assert project == "main"
+        return None, "personal/main/tests/search-note", True
+
+    class FakeSearchClient:
+        def __init__(self, client, project_id):
+            captured["search_project_id"] = project_id
+
+        async def search(self, payload, *, page, page_size):
+            captured["payload"] = payload
+            return SearchResponse(
+                results=[
+                    SearchResult(
+                        title="Search Note",
+                        type=SearchItemType.ENTITY,
+                        score=1.0,
+                        permalink="personal/main/tests/search-note",
+                        file_path="tests/Search Note.md",
+                    )
+                ],
+                current_page=page,
+                page_size=page_size,
+                total=1,
+            )
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
+    monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
+    monkeypatch.setattr(clients_mod, "SearchClient", FakeSearchClient)
+
+    response = await search_notes(
+        query="memory://personal/main/tests/search-note",
+        output_format="json",
+    )
+
+    assert captured["project"] == "personal/main"
+    assert captured["project_id"] is None
+    assert captured["search_project_id"] == "11111111-1111-1111-1111-111111111111"
+    payload = cast(dict[str, object], captured["payload"])
+    assert payload["permalink"] == "personal/main/tests/search-note"
+    assert isinstance(response, dict)
+    assert response["results"][0]["permalink"] == "personal/main/tests/search-note"
+
+
 @pytest.mark.asyncio
 async def test_search_pagination(client, test_project):
     """Test basic search functionality."""

tests/mcp/test_tool_workspace_management.py

@@ -1,12 +1,11 @@
 """Tests for workspace MCP tools."""
 
-from typing import Any, cast
-
 import pytest
 
 from basic_memory.mcp.project_context import get_available_workspaces, set_workspace_provider
 from basic_memory.mcp.tools.workspaces import list_workspaces
 from basic_memory.schemas.cloud import WorkspaceInfo
+from tests.mcp.conftest import ContextState, ctx
 
 
 def _workspace(
@@ -28,17 +27,6 @@ def _workspace(
     )
 
 
-class _ContextState:
-    def __init__(self):
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-
 @pytest.mark.asyncio
 async def test_list_workspaces_formats_workspace_rows(monkeypatch):
     async def fake_get_available_workspaces(context=None):
@@ -145,7 +133,7 @@ async def test_list_workspaces_oauth_error_bubbles_up(monkeypatch):
 
 @pytest.mark.asyncio
 async def test_list_workspaces_uses_context_cache_path(monkeypatch):
-    context = _ContextState()
+    context = ContextState()
     call_count = {"fetches": 0}
     workspace = _workspace(
         tenant_id="33333333-3333-3333-3333-333333333333",
@@ -169,8 +157,8 @@ async def test_list_workspaces_uses_context_cache_path(monkeypatch):
         fake_get_available_workspaces,
     )
 
-    first = await list_workspaces(context=cast(Any, context))
-    second = await list_workspaces(context=cast(Any, context))
+    first = await list_workspaces(context=ctx(context))
+    second = await list_workspaces(context=ctx(context))
 
     assert "# Available Workspaces (1)" in first
     assert "# Available Workspaces (1)" in second
@@ -254,14 +242,14 @@ async def test_get_available_workspaces_provider_caches_in_context():
         return [workspace]
 
     set_workspace_provider(counting_provider)
-    context = _ContextState()
+    context = ContextState()
 
     # First call: provider is invoked, result cached
-    first = await get_available_workspaces(context=cast(Any, context))
+    first = await get_available_workspaces(context=ctx(context))
     assert len(first) == 1
     assert call_count["provider"] == 1
 
     # Second call: served from context cache, provider not called again
-    second = await get_available_workspaces(context=cast(Any, context))
+    second = await get_available_workspaces(context=ctx(context))
     assert len(second) == 1
     assert call_count["provider"] == 1

tests/services/test_link_resolver.py

@@ -139,11 +139,11 @@ async def test_workspace_identifier_project_detection_requires_workspace_shape(
     """Two-segment project-relative paths should not trigger workspace discovery."""
     from basic_memory.services import link_resolver
 
-    def fail_if_called(config):
+    def fail_if_called(identifier, config):
         raise AssertionError("plain project-relative paths should skip cloud discovery")
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
         fail_if_called,
     )
 
@@ -164,8 +164,8 @@ async def test_workspace_identifier_project_detection_skips_without_discovery(
     from basic_memory.services import link_resolver
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
-        lambda config: False,
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: False,
     )
 
     detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
@@ -189,8 +189,8 @@ async def test_workspace_identifier_project_detection_returns_project(
         return SimpleNamespace(project_identifier="team-acme/research")
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
-        lambda config: True,
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         "basic_memory.mcp.project_context.resolve_workspace_qualified_identifier",
@@ -205,6 +205,104 @@ async def test_workspace_identifier_project_detection_returns_project(
     assert detected == "team-acme/research"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Workspace 'team-acme' was not found.",
+        "No accessible workspaces found for this account.",
+        "Unable to discover projects in any accessible workspace. Failed workspaces: team-acme",
+    ],
+)
+async def test_workspace_identifier_project_detection_ignores_discovery_failures(
+    monkeypatch,
+    config_manager,
+    message,
+):
+    """Workspace detection should fall back when cloud discovery is unavailable."""
+    from basic_memory.services import link_resolver
+
+    async def fail_workspace_identifier(identifier, context=None):
+        raise ValueError(message)
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: True,
+    )
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.resolve_workspace_qualified_identifier",
+        fail_workspace_identifier,
+    )
+
+    detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
+        "team-acme/research/note",
+        config_manager.config,
+    )
+
+    assert detected is None
+
+
+@pytest.mark.asyncio
+async def test_workspace_identifier_project_detection_allows_mixed_local_cloud_config(
+    monkeypatch,
+    config_manager,
+):
+    """Plain workspace routes should be discoverable even with local projects configured."""
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+    from basic_memory.services import link_resolver
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    workspace = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (workspace,),
+        (WorkspaceProjectEntry(workspace=workspace, project=project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._ensure_workspace_project_index",
+        fake_index,
+    )
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+
+    detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
+        "personal/main/todo",
+        config_manager.config,
+    )
+
+    assert detected == "personal/main"
+
+
 @pytest.mark.asyncio
 async def test_exact_permalink_match(link_resolver, test_entities, project_prefix):
     """Test resolving a link that exactly matches a permalink."""

tests/services/test_semantic_search.py

@@ -5,7 +5,9 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
+from sqlalchemy import text
 
+from basic_memory import db
 from basic_memory.repository import EntityRepository
 from basic_memory.repository.search_repository_base import VectorSyncBatchResult
 from basic_memory.repository.semantic_errors import (
@@ -256,6 +258,98 @@ async def test_reindex_vectors_respects_embed_opt_out(search_service, monkeypatc
     }
 
 
+@pytest.mark.asyncio
+async def test_reindex_vectors_purges_sqlite_vectors_before_sync(search_service, monkeypatch):
+    """Regression for #829: stale vec0 rows must be purged before batch sync."""
+    repository = _sqlite_repo(search_service)
+    repository._semantic_enabled = True
+    calls: list[str] = []
+
+    monkeypatch.setattr(
+        search_service.entity_repository,
+        "find_all",
+        AsyncMock(return_value=[SimpleNamespace(id=42, entity_metadata={})]),
+    )
+
+    async def delete_stale_vector_rows():
+        calls.append("purge")
+
+    async def sync_entity_vectors_batch(entity_ids, progress_callback=None):
+        assert entity_ids == [42]
+        assert progress_callback is None
+        calls.append("sync")
+        return VectorSyncBatchResult(entities_total=1, entities_synced=1, entities_failed=0)
+
+    monkeypatch.setattr(repository, "delete_stale_vector_rows", delete_stale_vector_rows)
+    monkeypatch.setattr(search_service, "sync_entity_vectors_batch", sync_entity_vectors_batch)
+
+    stats = await search_service.reindex_vectors()
+
+    assert calls == ["purge", "sync"]
+    assert stats == {
+        "total_entities": 1,
+        "embedded": 1,
+        "skipped": 0,
+        "errors": 0,
+    }
+
+
+@pytest.mark.asyncio
+async def test_reindex_all_uses_sqlite_vec_aware_drop(search_service, monkeypatch):
+    """Full service reindex should not drop vec0 tables through a raw connection."""
+    repository = _sqlite_repo(search_service)
+    executed_sql: list[str] = []
+    calls: list[str] = []
+
+    async def execute_query(query, params=None):
+        executed_sql.append(str(query))
+
+    async def drop_vector_tables():
+        calls.append("drop_vector_tables")
+
+    monkeypatch.setattr(repository, "execute_query", execute_query)
+    monkeypatch.setattr(repository, "drop_vector_tables", drop_vector_tables)
+    monkeypatch.setattr(search_service, "init_search_index", AsyncMock())
+    monkeypatch.setattr(search_service.entity_repository, "find_all", AsyncMock(return_value=[]))
+
+    await search_service.reindex_all()
+
+    assert calls == ["drop_vector_tables"]
+    assert all("search_vector_embeddings" not in sql for sql in executed_sql)
+
+
+@pytest.mark.asyncio
+async def test_drop_vector_tables_skips_sqlite_vec_load_for_plain_table(
+    search_service,
+    monkeypatch,
+):
+    """Plain compatibility tables should not require sqlite-vec to be loaded."""
+    repository = _sqlite_repo(search_service)
+    await repository.drop_vector_tables()
+
+    async with db.scoped_session(repository.session_maker) as session:
+        await session.execute(
+            text("CREATE TABLE search_vector_embeddings (rowid INTEGER PRIMARY KEY)")
+        )
+        await session.commit()
+
+    async def fail_if_loaded(_session):
+        raise AssertionError("plain tables should not load sqlite-vec")
+
+    monkeypatch.setattr(repository, "_ensure_sqlite_vec_loaded", fail_if_loaded)
+
+    await repository.drop_vector_tables()
+
+    async with db.scoped_session(repository.session_maker) as session:
+        result = await session.execute(
+            text(
+                "SELECT name FROM sqlite_master "
+                "WHERE type = 'table' AND name = 'search_vector_embeddings'"
+            )
+        )
+        assert result.scalar() is None
+
+
 @pytest.mark.asyncio
 async def test_reindex_vectors_force_full_clears_project_vectors_before_resync(
     search_service, monkeypatch

tests/sync/test_sync_service.py

@@ -108,6 +108,79 @@ Target content
     assert source.relations[0].to_name == target.title
 
 
+@pytest.mark.asyncio
+async def test_resolve_relations_uses_strict_link_resolution(
+    sync_service: SyncService,
+    project_config: ProjectConfig,
+    entity_service: EntityService,
+    monkeypatch,
+):
+    """Regression: deferred forward-reference resolution must call resolve_link
+    with strict=True.
+
+    Producers sometimes emit disambiguator-style links like
+    `[[overview (state-management/session-execution)]]` whose exact text does
+    not match any entity's permalink, title, or file_path. The previous
+    behavior fell through to BM25/ts_rank fuzzy search in
+    LinkResolver._resolve_in_project and silently picked whichever entity
+    shared the most tokens — polluting the graph with confidently-wrong edges
+    that no audit catches.
+
+    Entity-creation already resolves relations with strict=True (see
+    entity_service.update_entity_relations). The deferred sync path must use
+    the same contract; otherwise unresolved relations get silently filled
+    later by fuzzy search.
+    """
+    # Create a source file with a forward reference. The target doesn't exist,
+    # so resolution will fail — which is exactly when fuzzy fallback would
+    # previously silently pick a wrong target.
+    source_content = dedent("""
+        ---
+        type: knowledge
+        ---
+        # Source
+
+        ## Relations
+        - part_of [[never-resolves-target]]
+    """)
+    await create_test_file(project_config.home / "source.md", source_content)
+    await sync_service.sync(project_config.home)
+
+    project_prefix = generate_permalink(project_config.name)
+    source = await entity_service.get_by_permalink(f"{project_prefix}/source")
+    assert len(source.relations) == 1
+    assert source.relations[0].to_id is None  # initial creation already strict
+
+    # Spy on resolve_link to capture the strict flag the deferred resolver uses.
+    original_resolve_link = sync_service.entity_service.link_resolver.resolve_link
+    seen_strict: list[Any] = []
+
+    async def spy_resolve_link(*args, **kwargs):
+        seen_strict.append(kwargs.get("strict", False))
+        return await original_resolve_link(*args, **kwargs)
+
+    monkeypatch.setattr(
+        sync_service.entity_service.link_resolver,
+        "resolve_link",
+        spy_resolve_link,
+    )
+
+    await sync_service.resolve_relations()
+
+    # Deferred resolution invoked resolve_link, and every call passed strict=True.
+    assert seen_strict, "resolve_relations did not invoke link_resolver.resolve_link"
+    assert all(strict is True for strict in seen_strict), (
+        f"Deferred resolution must call resolve_link(strict=True) to avoid silent "
+        f"fuzzy matching. Observed strict values: {seen_strict!r}"
+    )
+
+    # Sanity check: the unresolvable relation stayed unresolved — no silent
+    # fuzzy match polluted it.
+    source = await entity_service.get_by_permalink(f"{project_prefix}/source")
+    assert source.relations[0].to_id is None
+    assert source.relations[0].to_name == "never-resolves-target"
+
+
 @pytest.mark.asyncio
 async def test_resolve_relations_deletes_duplicate_unresolved_relation(
     sync_service: SyncService,

tests/test_config.py

@@ -308,8 +308,9 @@ class TestDataDirHelpers:
     """Module-level helpers that resolve the Basic Memory data directory."""
 
     def test_resolve_data_dir_defaults_to_home_dot_basic_memory(self, config_home, monkeypatch):
-        """Without BASIC_MEMORY_CONFIG_DIR, resolver returns ~/.basic-memory."""
+        """Without BASIC_MEMORY_CONFIG_DIR and XDG_CONFIG_HOME, resolver returns ~/.basic-memory."""
         monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)
+        monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
 
         assert resolve_data_dir() == config_home / ".basic-memory"
 
@@ -320,6 +321,23 @@ class TestDataDirHelpers:
 
         assert resolve_data_dir() == custom
 
+    def test_resolve_data_dir_honors_xdg_config_home(self, tmp_path, monkeypatch):
+        """XDG_CONFIG_HOME is honored when BASIC_MEMORY_CONFIG_DIR is not set."""
+        monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)
+        xdg_config = tmp_path / "xdg-config"
+        monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config))
+
+        assert resolve_data_dir() == xdg_config / "basic-memory"
+
+    def test_basic_memory_config_dir_takes_precedence_over_xdg(self, tmp_path, monkeypatch):
+        """BASIC_MEMORY_CONFIG_DIR takes precedence over XDG_CONFIG_HOME."""
+        xdg_config = tmp_path / "xdg-config"
+        custom = tmp_path / "custom-config"
+        monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config))
+        monkeypatch.setenv("BASIC_MEMORY_CONFIG_DIR", str(custom))
+
+        assert resolve_data_dir() == custom
+
     def test_default_fastembed_cache_dir_uses_data_dir(self, config_home, monkeypatch):
         """Default cache path is a subdir of the Basic Memory data dir."""
         monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)