docs(cli): document personal workspace sync

Signed-off-by: Drew Cain <groksrc@gmail.com>

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/cloud-cli.md

@@ -1,6 +1,6 @@
 # Basic Memory Cloud CLI Guide
 
-The Basic Memory Cloud CLI provides seamless integration between local and cloud knowledge bases using **project-scoped synchronization**. Each project can optionally sync with the cloud, giving you fine-grained control over what syncs and where.
+The Basic Memory Cloud CLI provides seamless integration between local and cloud knowledge bases using **project-scoped synchronization**. Personal workspaces can optionally use local rclone mirrors, giving you fine-grained control over what syncs and where.
 
 ## Overview
 
@@ -8,9 +8,13 @@ The cloud CLI enables you to:
 - **Authenticate cloud access** - OAuth/API key credentials are stored locally for cloud operations
 - **Project-scoped sync** - Each project independently manages its sync configuration
 - **Explicit operations** - Sync only what you want, when you want
-- **Bidirectional sync** - Keep local and cloud in sync with rclone bisync
+- **Bidirectional sync** - Keep Personal workspace local mirrors and cloud in sync with rclone bisync
 - **Offline access** - Work locally, sync when ready
 
+Team workspaces are accessed through the cloud API/MCP and do not support local
+multi-user rclone sync/bisync. Use `bm project list --workspace <workspace>` to
+inspect Team projects.
+
 ## Prerequisites
 
 Before using Basic Memory Cloud, you need:
@@ -39,7 +43,7 @@ If you attempt to log in without an active subscription, you'll receive a "Subsc
 **Projects can exist in three states:**
 
 1. **Cloud-only** - Project exists on cloud, no local copy
-2. **Cloud + Local (synced)** - Project has a local working directory that syncs
+2. **Cloud + Local (synced)** - Personal workspace project has a local working directory that syncs
 3. **Local-only** - Project exists locally and is not routed to cloud
 
 **Example:**
@@ -55,8 +59,8 @@ bm project add work --cloud --local-path ~/work-notes
 bm project add temp --cloud  # No local sync
 
 # Now you can sync individually (after initial --resync):
-bm project bisync --name research
-bm project bisync --name work
+bm cloud bisync --name research
+bm cloud bisync --name work
 # temp stays cloud-only
 ```
 
@@ -87,7 +91,7 @@ Apply OSS discount code `{{OSS_DISCOUNT_CODE}}` during checkout to receive 20% o
 
 ### 2. Set Up Sync
 
-Install rclone and configure credentials:
+Install rclone and configure Personal workspace sync credentials:
 
 ```bash
 bm cloud setup
@@ -99,7 +103,7 @@ bm cloud setup
 3. Generates scoped S3 credentials for sync
 4. Configures single rclone remote: `basic-memory-cloud`
 
-**Result:** You're ready to sync projects. No sync directories created yet - those come with project setup.
+**Result:** You're ready to sync Personal workspace projects. No sync directories created yet - those come with project setup.
 
 Rclone setup uses package managers such as Homebrew, MacPorts, apt, dnf, yum, pacman,
 zypper, snap, winget, Chocolatey, or Scoop when available. It does not run remote
@@ -108,7 +112,7 @@ manual install instructions.
 
 ### 3. Add Projects with Sync
 
-Create projects with optional local sync paths:
+Create Personal workspace projects with optional local sync paths:
 
 ```bash
 # Create cloud project without local sync
@@ -137,10 +141,10 @@ Establish the initial sync baseline. **Best practice:** Always preview with `--d
 
 ```bash
 # Step 1: Preview the initial sync (recommended)
-bm project bisync --name research --resync --dry-run
+bm cloud bisync --name research --resync --dry-run
 
 # Step 2: If all looks good, run the actual sync
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 **What happens under the covers:**
@@ -167,7 +171,7 @@ This will effectively make both Path1 and Path2 filesystems contain a matching s
 After the first sync, just run bisync without `--resync`:
 
 ```bash
-bm project bisync --name research
+bm cloud bisync --name research
 ```
 
 **What happens:**
@@ -235,7 +239,7 @@ bm project add research --cloud --local-path ~/Documents/research
 - Stores sync config in `~/.basic-memory/config.json`
 - Prepares for bisync (but doesn't sync yet)
 
-**Result:** Project ready to sync. Run `bm project bisync --name research --resync` to establish baseline.
+**Result:** Project ready to sync. Run `bm cloud bisync --name research --resync` to establish baseline.
 
 **Use case 3: Add sync to existing cloud project**
 
@@ -292,20 +296,24 @@ For MCP stdio, routing is always local.
 
 ## File Synchronization
 
+Local rclone sync/bisync is supported only for Personal workspaces. Team
+workspaces are cloud-only for local CLI usage; access them through cloud API/MCP
+routing instead of a local multi-user rclone mirror.
+
 ### Understanding the Sync Commands
 
 **There are three sync-related commands:**
 
-1. `bm project sync` - One-way: local → cloud (make cloud match local)
-2. `bm project bisync` - Two-way: local ↔ cloud (recommended)
-3. `bm project check` - Verify files match (no changes)
+1. `bm cloud sync` - One-way: local → cloud (make cloud match local)
+2. `bm cloud bisync` - Two-way: local ↔ cloud (recommended)
+3. `bm cloud check` - Verify files match (no changes)
 
 ### One-Way Sync: Local → Cloud
 
 **Use case:** You made changes locally and want to push to cloud (overwrite cloud).
 
 ```bash
-bm project sync --name research
+bm cloud sync --name research
 ```
 
 **What happens:**
@@ -327,10 +335,10 @@ bm project sync --name research
 
 ```bash
 # First time - establish baseline
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 
 # Subsequent syncs
-bm project bisync --name research
+bm cloud bisync --name research
 ```
 
 **What happens:**
@@ -349,7 +357,7 @@ echo "Local change" > ~/Documents/research/notes.md
 # Cloud now has: "Cloud change"
 
 # Run bisync
-bm project bisync --name research
+bm cloud bisync --name research
 
 # Result: Newer file wins (based on modification time)
 # If cloud was more recent, cloud version kept
@@ -366,7 +374,7 @@ bm project bisync --name research
 **Use case:** Check if local and cloud match without making changes.
 
 ```bash
-bm project check --name research
+bm cloud check --name research
 ```
 
 **What happens:**
@@ -378,7 +386,7 @@ bm project check --name research
 
 ```bash
 # One-way check (faster)
-bm project check --name research --one-way
+bm cloud check --name research --one-way
 ```
 
 ### Preview Changes (Dry Run)
@@ -386,7 +394,7 @@ bm project check --name research --one-way
 **Use case:** See what would change without actually syncing.
 
 ```bash
-bm project bisync --name research --dry-run
+bm cloud bisync --name research --dry-run
 ```
 
 **What happens:**
@@ -432,20 +440,20 @@ bm project add work --cloud --local-path ~/work-notes
 bm project add personal --cloud --local-path ~/personal
 
 # Establish baselines
-bm project bisync --name research --resync
-bm project bisync --name work --resync
-bm project bisync --name personal --resync
+bm cloud bisync --name research --resync
+bm cloud bisync --name work --resync
+bm cloud bisync --name personal --resync
 
 # Daily workflow: sync everything
-bm project bisync --name research
-bm project bisync --name work
-bm project bisync --name personal
+bm cloud bisync --name research
+bm cloud bisync --name work
+bm cloud bisync --name personal
 ```
 
 **Future:** `--all` flag will sync all configured projects:
 
 ```bash
-bm project bisync --all  # Coming soon
+bm cloud bisync --all  # Coming soon
 ```
 
 ### Mixed Usage
@@ -462,8 +470,8 @@ bm project add archive --cloud
 bm project add temp-notes --cloud
 
 # Sync only the configured ones
-bm project bisync --name research
-bm project bisync --name work
+bm cloud bisync --name research
+bm cloud bisync --name work
 
 # Archive and temp-notes stay cloud-only
 ```
@@ -661,7 +669,7 @@ code ~/.basic-memory/.bmignore
 echo "*.tmp" >> ~/.basic-memory/.bmignore
 
 # Next sync uses updated patterns
-bm project bisync --name research
+bm cloud bisync --name research
 ```
 
 ## Troubleshooting
@@ -724,7 +732,7 @@ bm cloud login
 **Solution:**
 
 ```bash
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 **What this does:**
@@ -747,7 +755,7 @@ bm project bisync --name research --resync
 echo "# Research Notes" > ~/Documents/research/README.md
 
 # Now run bisync
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 **Why this happens:** Bisync creates listing files that track the state of each side. When both directories are completely empty, these listing files are considered invalid by rclone.
@@ -764,10 +772,10 @@ bm project bisync --name research --resync
 
 ```bash
 # Clear bisync state
-bm project bisync-reset research
+bm cloud bisync-reset research
 
 # Re-establish baseline
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 **What this does:**
@@ -787,16 +795,16 @@ bm project bisync --name research --resync
 
 ```bash
 # Check what would be deleted
-bm project bisync --name research --dry-run
+bm cloud bisync --name research --dry-run
 
 # If correct, establish new baseline
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 **Solution 2:** Use one-way sync if you know local is correct:
 
 ```bash
-bm project sync --name research
+bm cloud sync --name research
 ```
 
 ### Project Not Configured for Sync
@@ -809,7 +817,7 @@ bm project sync --name research
 
 ```bash
 bm cloud sync-setup research ~/Documents/research
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```
 
 ### Connection Issues
@@ -881,19 +889,19 @@ bm project set-local <name>  # Revert project to local mode
 
 ```bash
 # One-way sync (local → cloud)
-bm project sync --name <project>
-bm project sync --name <project> --dry-run
-bm project sync --name <project> --verbose
+bm cloud sync --name <project>
+bm cloud sync --name <project> --dry-run
+bm cloud sync --name <project> --verbose
 
 # Two-way sync (local ↔ cloud) - Recommended
-bm project bisync --name <project>          # After first --resync
-bm project bisync --name <project> --resync # First time / force baseline
-bm project bisync --name <project> --dry-run
-bm project bisync --name <project> --verbose
+bm cloud bisync --name <project>          # After first --resync
+bm cloud bisync --name <project> --resync # First time / force baseline
+bm cloud bisync --name <project> --dry-run
+bm cloud bisync --name <project> --verbose
 
 # Integrity check
-bm project check --name <project>
-bm project check --name <project> --one-way
+bm cloud check --name <project>
+bm cloud check --name <project> --one-way
 
 # List project files by route
 bm project ls --name <project>          # Default target: local
@@ -909,9 +917,9 @@ bm project ls --name <project> --cloud --path <subpath>
 1. **Authenticate cloud access** - `bm cloud login`
 2. **Install rclone** - `bm cloud setup`
 3. **Add projects with sync** - `bm project add research --cloud --local-path ~/Documents/research`
-4. **Preview first sync** - `bm project bisync --name research --resync --dry-run`
-5. **Establish baseline** - `bm project bisync --name research --resync`
-6. **Daily workflow** - `bm project bisync --name research`
+4. **Preview first sync** - `bm cloud bisync --name research --resync --dry-run`
+5. **Establish baseline** - `bm cloud bisync --name research --resync`
+6. **Daily workflow** - `bm cloud bisync --name research`
 
 **Key benefits:**
 - ✅ Each project independently syncs (or doesn't)

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/cli/commands/cloud/project_sync.py

@@ -95,7 +95,10 @@ def sync_project_command(
     dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
     verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
 ) -> None:
-    """One-way sync: local -> cloud (make cloud identical to local).
+    """Personal workspace local mirror only.
+
+    One-way sync: local -> cloud (make cloud identical to local).
+    Not supported for Team workspaces - use cloud API/MCP routing instead.
 
     Example:
       bm cloud sync --name research
@@ -143,7 +146,10 @@ def bisync_project_command(
     resync: bool = typer.Option(False, "--resync", help="Force new baseline"),
     verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
 ) -> None:
-    """Two-way sync: local <-> cloud (bidirectional sync).
+    """Personal workspace local mirror only.
+
+    Two-way sync: local <-> cloud (bidirectional sync).
+    Not supported for Team workspaces - use cloud API/MCP routing instead.
 
     Examples:
       bm cloud bisync --name research --resync  # First time
@@ -203,7 +209,10 @@ def check_project_command(
     name: str = typer.Option(..., "--name", help="Project name to check"),
     one_way: bool = typer.Option(False, "--one-way", help="Check one direction only (faster)"),
 ) -> None:
-    """Verify file integrity between local and cloud.
+    """Personal workspace local mirror only.
+
+    Verify file integrity between local and cloud.
+    Not supported for Team workspaces - use cloud API/MCP routing instead.
 
     Example:
       bm cloud check --name research
@@ -246,7 +255,10 @@ def check_project_command(
 def bisync_reset(
     name: str = typer.Argument(..., help="Project name to reset bisync state for"),
 ) -> None:
-    """Clear bisync state for a project.
+    """Personal workspace local mirror only.
+
+    Clear bisync state for a project.
+    Not supported for Team workspaces - use cloud API/MCP routing instead.
 
     This removes the bisync metadata files, forcing a fresh --resync on next bisync.
     Useful when bisync gets into an inconsistent state or when remote path changes.
@@ -277,7 +289,10 @@ def setup_project_sync(
     name: str = typer.Argument(..., help="Project name"),
     local_path: str = typer.Argument(..., help="Local sync directory"),
 ) -> None:
-    """Configure local sync for an existing cloud project.
+    """Personal workspace local mirror only.
+
+    Configure local sync for an existing cloud project.
+    Not supported for Team workspaces - use cloud API/MCP routing instead.
 
     Example:
       bm cloud sync-setup research ~/Documents/research

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/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

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/cli/cloud/test_project_sync_command.py

@@ -12,6 +12,21 @@ from basic_memory.config import ProjectMode
 runner = CliRunner()
 
 
+@pytest.mark.parametrize(
+    "command",
+    ["sync", "bisync", "check", "bisync-reset", "sync-setup"],
+)
+def test_cloud_sync_command_help_marks_personal_workspace_only(command):
+    """Cloud sync help should explain that local mirrors are Personal-only."""
+    importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
+
+    result = runner.invoke(app, ["cloud", command, "--help"])
+
+    assert result.exit_code == 0, result.output
+    assert "Personal workspace local mirror only" in result.output
+    assert "Not supported for Team workspaces" in result.output
+
+
 @pytest.mark.parametrize(
     "argv",
     [

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/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/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)