docs(cli): document personal workspace sync

Signed-off-by: Drew Cain <groksrc@gmail.com>
chore: update version to 0.21.5 for v0.21.5 release
2026-06-21 13:47:35 +00:00 · 2026-05-26 13:10:21 -05:00 · 2026-05-26 10:58:09 -05:00 · 2026-05-26 10:57:55 -05:00 · 2026-05-26 10:54:44 -05:00 · 2026-05-26 00:33:21 -05:00
78 changed files with 6650 additions and 2281 deletions
@@ -10,6 +10,9 @@ on:
    #   - "src/**/*.js"
    #   - "src/**/*.jsx"

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  claude-review:
    # Only run for organization members and collaborators
@@ -27,7 +30,7 @@ jobs:

    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          fetch-depth: 1

@@ -4,6 +4,9 @@ on:
  issues:
    types: [opened]

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  triage:
    runs-on: ubuntu-latest
@@ -12,7 +15,7 @@ jobs:
      id-token: write
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          fetch-depth: 1

@@ -68,4 +71,4 @@ jobs:

            Read the issue carefully and provide helpful triage with appropriate labels.

-          claude_args: '--allowed-tools "Bash(gh issue:*),Bash(gh search:*),Read"'
+          claude_args: '--allowed-tools "Bash(gh issue:*),Bash(gh search:*),Read"'
@@ -12,6 +12,9 @@ on:
  pull_request_target:
    types: [opened, synchronize]

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  claude:
    if: |
@@ -41,7 +44,7 @@ jobs:
      actions: read # Required for Claude to read CI results on PRs
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          # For pull_request_target, checkout the PR head to review the actual changes
          ref: ${{ github.event_name == 'pull_request_target' && github.event.pull_request.head.sha || github.sha }}
@@ -65,4 +68,3 @@ jobs:
          # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
          # or https://docs.claude.com/en/docs/claude-code/sdk#command-line for available options
          # claude_args: '--model claude-opus-4-1-20250805 --allowed-tools Bash(gh pr:*)'
-
@@ -5,6 +5,9 @@ on:
    branches: [main]
  workflow_dispatch:  # Allow manual triggering

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  dev-release:
    runs-on: ubuntu-latest
@@ -13,12 +16,12 @@ jobs:
      contents: write

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0

      - name: Set up Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
        with:
          python-version: "3.12"

@@ -50,4 +53,4 @@ jobs:
        uses: pypa/gh-action-pypi-publish@release/v1
        with:
          password: ${{ secrets.PYPI_TOKEN }}
-          skip-existing: true  # Don't fail if version already exists
+          skip-existing: true  # Don't fail if version already exists
@@ -9,6 +9,7 @@ on:
 env:
  REGISTRY: ghcr.io
  IMAGE_NAME: basicmachines-co/basic-memory
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"

 jobs:
  docker:
@@ -19,17 +20,17 @@ jobs:

    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          fetch-depth: 0

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
        with:
          platforms: linux/amd64,linux/arm64

      - name: Log in to GitHub Container Registry
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
@@ -37,7 +38,7 @@ jobs:

      - name: Extract metadata
        id: meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v6
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
@@ -48,7 +49,7 @@ jobs:
            type=raw,value=latest,enable={{is_default_branch}}

      - name: Build and push Docker image
-        uses: docker/build-push-action@v5
+        uses: docker/build-push-action@v7
        with:
          context: .
          file: ./Dockerfile
@@ -58,4 +59,3 @@ jobs:
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
-
@@ -7,11 +7,14 @@ on:
      - edited
      - synchronize

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  main:
    runs-on: ubuntu-latest
    steps:
-      - uses: amannn/action-semantic-pull-request@v5
+      - uses: amannn/action-semantic-pull-request@v6
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        with:
@@ -38,4 +41,4 @@ jobs:
            deps
            installer
          # Allow breaking changes (needs "!" after type/scope)
-          requireScopeForBreakingChange: true
+          requireScopeForBreakingChange: true
@@ -5,6 +5,9 @@ on:
    tags:
      - 'v*'  # Trigger on version tags like v1.0.0, v0.13.0, etc.

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  release:
    runs-on: ubuntu-latest
@@ -13,12 +16,12 @@ jobs:
      contents: write

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0

      - name: Set up Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
        with:
          python-version: "3.12"

@@ -39,7 +42,7 @@ jobs:
          echo "Build completed successfully"

      - name: Create GitHub Release
-        uses: softprops/action-gh-release@v2
+        uses: softprops/action-gh-release@v3
        with:
          files: |
            dist/*.whl
@@ -60,26 +63,63 @@ jobs:
    # Only run for stable releases (not dev, beta, or rc versions)
    if: ${{ !contains(github.ref_name, 'dev') && !contains(github.ref_name, 'b') && !contains(github.ref_name, 'rc') }}
    permissions:
-      contents: write
-      actions: read
+      contents: read
    steps:
+      # Inline bump replaces mislav/bump-homebrew-formula-action@v4.x.
+      # The action does a HEAD request to api.github.com /repos/.../tarball/<ref>
+      # with the bearer token and expects a 302 redirect. GitHub now returns
+      # 303 on that endpoint when authenticated, which the action treats as a
+      # fatal error. Re-implementing the bump as plain git+sed keeps the same
+      # contract (update url + sha256, commit, push) with no third-party action.
      - name: Update Homebrew formula
-        uses: mislav/bump-homebrew-formula-action@v3
-        with:
-          # Formula name in homebrew-basic-memory repo
-          formula-name: basic-memory
-          # The tap repository
-          homebrew-tap: basicmachines-co/homebrew-basic-memory
-          # Base branch of the tap repository
-          base-branch: main
-          # Download URL will be automatically constructed from the tag
-          download-url: https://github.com/basicmachines-co/basic-memory/archive/refs/tags/${{ github.ref_name }}.tar.gz
-          # Commit message for the formula update
-          commit-message: |
-            {{formulaName}} {{version}}
-
-            Created by https://github.com/basicmachines-co/basic-memory/actions/runs/${{ github.run_id }}
        env:
-          # Personal Access Token with repo scope for homebrew-basic-memory repo
-          COMMITTER_TOKEN: ${{ secrets.HOMEBREW_TOKEN }}
+          HOMEBREW_TOKEN: ${{ secrets.HOMEBREW_TOKEN }}
+          REF: ${{ github.ref_name }}
+          REPO: ${{ github.repository }}
+          RUN_URL: https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}
+        run: |
+          set -euo pipefail

+          VERSION="${REF#v}"
+          ARCHIVE_URL="https://github.com/${REPO}/archive/refs/tags/${REF}.tar.gz"
+
+          echo "::group::Compute tarball sha256"
+          SHA256="$(curl --fail --silent --location "$ARCHIVE_URL" | sha256sum | awk '{print $1}')"
+          test -n "$SHA256"
+          echo "sha256: $SHA256"
+          echo "::endgroup::"
+
+          echo "::group::Clone tap"
+          git clone \
+            --depth 1 \
+            "https://x-access-token:${HOMEBREW_TOKEN}@github.com/basicmachines-co/homebrew-basic-memory.git" \
+            tap
+          cd tap
+          git config user.name "github-actions[bot]"
+          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
+          echo "::endgroup::"
+
+          echo "::group::Patch Formula/basic-memory.rb"
+          # Pipe-delimited sed because the URL contains slashes. The Formula
+          # only has one `url` and one `sha256` directive, so a first-match
+          # replacement is unambiguous. POSIX character classes ([[:space:]])
+          # keep this portable across BSD and GNU sed.
+          sed -i -E \
+            -e "s|^([[:space:]]*url[[:space:]]+)\"[^\"]+\"|\1\"${ARCHIVE_URL}\"|" \
+            -e "s|^([[:space:]]*sha256[[:space:]]+)\"[^\"]+\"|\1\"${SHA256}\"|" \
+            Formula/basic-memory.rb
+          git --no-pager diff Formula/basic-memory.rb
+          echo "::endgroup::"
+
+          if git diff --quiet Formula/basic-memory.rb; then
+            echo "Formula already at ${REF}; nothing to do."
+            exit 0
+          fi
+
+          echo "::group::Commit & push"
+          git add Formula/basic-memory.rb
+          git commit -m "basic-memory ${VERSION}
+
+          Created by ${RUN_URL}"
+          git push origin HEAD:main
+          echo "::endgroup::"
@@ -11,6 +11,9 @@ on:
  # Outcome: each branch push runs the test suite once, including PR updates.
  push:

+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+
 jobs:
  static-checks:
    name: Static Checks (Python 3.12)
@@ -18,12 +21,12 @@ jobs:
    runs-on: ubuntu-latest

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python 3.12
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: "3.12"
          cache: "pip"
@@ -32,7 +35,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -52,7 +55,7 @@ jobs:

  test-sqlite-unit:
    name: Test SQLite Unit (${{ matrix.os }}, Python ${{ matrix.python-version }})
-    timeout-minutes: 30
+    timeout-minutes: 45
    strategy:
      fail-fast: false
      matrix:
@@ -68,12 +71,12 @@ jobs:
    runs-on: ${{ matrix.os }}

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: ${{ matrix.python-version }}
          cache: 'pip'
@@ -82,7 +85,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -114,12 +117,12 @@ jobs:
    runs-on: ${{ matrix.os }}

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: ${{ matrix.python-version }}
          cache: 'pip'
@@ -128,7 +131,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -144,7 +147,7 @@ jobs:

  test-postgres-unit:
    name: Test Postgres Unit (Python ${{ matrix.python-version }})
-    timeout-minutes: 30
+    timeout-minutes: 60
    strategy:
      fail-fast: false
      matrix:
@@ -171,12 +174,12 @@ jobs:
      BASIC_MEMORY_TEST_POSTGRES_URL: postgresql://basic_memory_user:dev_password@127.0.0.1:5432/basic_memory_test

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: ${{ matrix.python-version }}
          cache: 'pip'
@@ -185,7 +188,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -228,12 +231,12 @@ jobs:
      BASIC_MEMORY_TEST_POSTGRES_URL: postgresql://basic_memory_user:dev_password@127.0.0.1:5432/basic_memory_test

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: ${{ matrix.python-version }}
          cache: 'pip'
@@ -242,7 +245,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -262,12 +265,12 @@ jobs:
    runs-on: ubuntu-latest

    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true

      - name: Set up Python 3.12
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
        with:
          python-version: "3.12"
          cache: "pip"
@@ -276,7 +279,7 @@ jobs:
        run: |
          pip install uv

-      - uses: extractions/setup-just@v3
+      - uses: extractions/setup-just@v4

      - name: Create virtual env
        run: |
@@ -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
@@ -1,6 +1,153 @@
 # CHANGELOG

-## Unreleased
+## 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.
+
+### Maintenance
+
+- **#833**: Replace `mislav/bump-homebrew-formula-action` with an inline
+  bash bump step. The action's `resolveRedirect` HEAD-requests
+  `api.github.com /repos/.../tarball/<ref>` expecting HTTP 302; GitHub now
+  returns 303 for authenticated requests on that endpoint, which broke
+  the v0.21.0 Homebrew job and required a manual tap bump. The inline
+  step does the same work (curl + sha256sum, clone tap, sed-bump
+  `url`/`sha256`, commit, push) without any third-party dependency.
+
+## v0.21.0 (2026-05-16)
+
+Workspace-aware everywhere: every MCP tool and CLI command now routes through the
+same workspace/project model, the search and sync paths are noticeably faster,
+and a handful of long-standing parsing and routing footguns are gone.
+
+### Breaking Changes
+
+- Relation parsing no longer treats unquoted multi-word text before a wikilink as a
+  custom relation type. Use single-token relation types like `relates_to [[Target]]`,
+  or quote multi-word relation types like `"relates to" [[Target]]` or
+  `'relates to' [[Target]]`.
+  - Bare list wikilinks like `- [[Target]]` now index as `links_to`.
+  - Prose list items like `- some other thing [[Target]]` now index as `links_to`.
+  - To preserve existing multi-word relation types on re-sync, quote them before upgrading.
+  - See **#824**.
+
+### Features
+
+- **#816**: `bm orphan` CLI command surfaces entities whose underlying markdown
+  files are gone, with a flag to clean them out.
+- **#789**: Create projects directly by cloud workspace slug from MCP
+  (`create_memory_project(workspace=...)`).
+- **#757**: Discover projects across every accessible cloud workspace in MCP's
+  project list — no more per-workspace blind spots.
+- **#766**: MCP tools accept training-data-friendly parameter aliases
+  (`q`/`search`/`text` for `query`, etc.) so models reach for them naturally.
+- **#776**: `bm db reset` refuses to run while a `basic-memory mcp` process is
+  alive, so resets can no longer corrupt an open session.
+- **#791**: Search responses include result totals so pagers can stop guessing.
+- **#719**: Cloud `note_content` tenant schema primitive lands on the backend.
+- **#715**: `bm project add` accepts a `--visibility` flag for cloud projects.
+
+### Bug Fixes
+
+#### Search and recent activity
+- **#832**: SQLite project deletion now sweeps `search_index`,
+  `search_vector_chunks`, and `search_vector_embeddings`, so a project that
+  reuses a recycled auto-increment id can't inherit the previous tenant's content.
+- **#812**: `recent_activity` orders and filters by `updated_at`, so edits bubble
+  to the top instead of staying pinned to creation time.
+- **#807**: Multi-project `search_notes` is opt-in (`search_all_projects=True`);
+  default search stays scoped to the resolved project.
+- **#785**: `recent_activity` caps responses and emits an explicit truncation
+  footer instead of silently dropping rows.
+- **#713**: Eliminated an N+1 query in search result hydration.
+
+#### Workspace / project routing
+- **#822**: `bm project list` now includes projects from every workspace, not
+  just the current one.
+- **#813**, **#808**, **#806**, **#803**, **#801**, **#795**, **#790**, **#778**,
+  **#777**, **#722**, **#712**, **#704**: Workspace-qualified permalink routing
+  is centralized and applied consistently across `edit_note`, `delete_project`,
+  `build_context`, `memory://` URLs, factory-mode project listing, cloud uploads,
+  and the API client.
+
+#### Sync
+- **#827**: `rclone bisync` filters from `.bmignore` are preserved across syncs.
+- **#815**: Watch service ignores hidden paths relative to the watched project,
+  not just relative to `$HOME`.
+- **#814**: `scan` subprocesses no longer go through the shell, avoiding quoting
+  issues with paths that contain special characters.
+- **#759**: Watch service stays inside `--project` scope.
+- **#746**: Canonical markdown is preserved during single-file sync.
+
+#### Parsing
+- **#796**: Picoschema modifier descriptions (`field?(modifier, description)`)
+  parse correctly.
+- **#769**: Obsidian callout blocks are skipped by the observation parser
+  instead of being mis-extracted as observations.
+
+#### CLI
+- **#775**: `bm project set-cloud` / `set-local` cleans up local DB state for
+  the affected project.
+- **#773**: `bm cloud logout` clears `default_workspace`.
+- **#780**: `bm cloud setup` hint points at `bm cloud sync-setup`.
+- **#734**: `bm project info` shows cloud index freshness.
+- **#718**: Private cloud projects display under their `display_name` instead of
+  raw UUID.
+- **#768**: `read_note` / `view_note` drop no-op pagination params from their
+  signatures.
+
+#### Stability
+- **#774**: `sqlite-vec` failures during init degrade gracefully to keyword-only
+  search instead of crashing startup.
+- **#733**: Delete-vector and cloud-sync cleanup is now consistent.
+- **#702**: Race conditions in concurrent `delete_entity` are resolved.
+- **#724**: `external_id` is preserved when entities are re-upserted during a
+  re-index.
+- **#728**: Vector init no longer issues runtime `ALTER TABLE`.
+- **#744**: `BASIC_MEMORY_CONFIG_DIR` is honored across remaining call sites.
+- **#743**: FastEmbed cache lives under the data dir instead of `/tmp`.
+- **#752**: Cloud projects report `source=cloud` in factory mode.
+- Stripped null bytes from markdown content before DB insert.
+- Allowed long `relation_type` values in API responses.
+
+#### Installer
+- **#772**: Docker-compose config volume mounts under the `appuser` home.
+- **#695**: Bumped `brew outdated` timeout from 15s to 60s.
+
+### Performance
+
+- **#828**: CLI startup no longer pulls in the local ASGI FastAPI app when it
+  isn't needed.
+- **#751**, **#726**, **#717**, **#714**, single-file/batch indexing: marked
+  speedups on the sync hot path; unchanged markdown is skipped entirely.
+- **#731**, **#723**: Vector sync is faster on both backends, with tuned
+  fastembed defaults.
+
+### Maintenance
+
+- **#825**: Dependency refresh + security hardening.
+- Updated to `fastmcp` 3.3.1.
+- **#754**: Removed in-house telemetry wrappers in favor of direct `logfire`
+  usage.
+- **#736**: `ty` is now the default typechecker.
+- **#771**, **#770**, **#716**: New regression guards for vector-row cleanup,
+  long relation types, and recent activity hydration.

 ## v0.20.3 (2026-03-26)

@@ -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:
@@ -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
@@ -106,7 +106,7 @@ The parser excludes these list item patterns:
 |---------|---------|--------|
 | Checkboxes | `- [ ] Todo item`, `- [x] Done`, `- [-] Cancelled` | Task list syntax |
 | Markdown links | `- [text](url)` | URL link syntax |
-| Bare wiki links | `- [[Target]]` | Treated as a relation instead |
+| Bare wiki links | `- [[Target]]` | Treated as a `links_to` relation instead |

 A list item with `#tags` but no `[category]` is still parsed — the tags are extracted and the category defaults to `Note`.

@@ -116,47 +116,65 @@ Relations connect documents to form the knowledge graph. There are two kinds.

 ### Explicit Relations

-Written as list items with a relation type and a `[[wiki link]]` target.
+Written as list items with a relation type and a `[[wiki link]]` target. Unquoted
+relation types are single tokens. Quote relation types that contain spaces.

 **Syntax:**

 ```
 - relation_type [[Target Entity]] (context)
+- "multi word relation type" [[Target Entity]] (context)
+- 'multi word relation type' [[Target Entity]] (context)
 ```

 | Part | Required | Description |
 |------|----------|-------------|
-| `relation_type` | No | Text before `[[`. Defaults to `relates_to` if omitted. |
+| `relation_type` | Yes | Single unquoted token before `[[`, or quoted text for multi-word labels. |
 | `[[Target]]` | Yes | Wiki link to the target entity. Matched by title or permalink. |
 | `(context)` | No | Parenthesized text after `]]`. Supporting details. |

 ### Examples

+Explicit relations:
+
 ```markdown
 - implements [[Search Design]]
 - depends_on [[Database Schema]]
 - works_at [[Y Combinator]] (co-founder)
- [[Some Entity]]
+- "based on" [[Customer Interview]]
+- 'in response to' [[Incident Review]]
 ```

-The last example — a bare `[[wiki link]]` in a list item — gets relation type `relates_to`.
+Bare wiki links and prose list items create implicit `links_to` relations:
+
+```markdown
+- [[Some Entity]]
+- some other thing [[Some Entity]]
+```
+
+Both examples above create `links_to [[Some Entity]]`. Use quotes when the words before
+`[[` are meant to be a multi-word relation type.

 Common relation types:
 - `implements`, `depends_on`, `relates_to`, `inspired_by`
 - `extends`, `part_of`, `contains`, `pairs_with`
 - `works_at`, `authored`, `collaborated_with`

-Any text works as a relation type. These are conventions, not a fixed set.
+Any single-token text or quoted text works as a relation type. These are conventions,
+not a fixed set.

 ### Inline References

-Wiki links appearing in regular prose (not as list items) create implicit `links_to` relations.
+Wiki links appearing in regular prose create implicit `links_to` relations. This includes
+list items that do not match the explicit relation grammar above.

 ```markdown
 This builds on [[Core Design]] and uses [[Utility Functions]].
+- We should revisit [[Search Design]] after the API changes.
 ```

-This creates two relations: `links_to [[Core Design]]` and `links_to [[Utility Functions]]`.
+This creates three relations: `links_to [[Core Design]]`, `links_to [[Utility Functions]]`,
+and `links_to [[Search Design]]`.

 ### Forward References

@@ -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,23 +91,28 @@ 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
 ```

 **What this does:**
-1. Installs rclone automatically (if needed)
+1. Installs rclone with a supported package manager (if needed)
 2. Fetches your tenant information from cloud
 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
+install scripts with `sudo`; if no supported package manager is found, the CLI prints
+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
@@ -132,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:**
@@ -162,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:**
@@ -230,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**

@@ -287,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:**
@@ -322,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:**
@@ -344,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
@@ -361,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:**
@@ -373,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)
@@ -381,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:**
@@ -427,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
@@ -457,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
 ```
@@ -485,6 +498,9 @@ bm cloud create-key "my-laptop"    # Creates key and saves it locally
 ```

 The API key is account-level — it grants access to all your cloud projects. It's stored in `~/.basic-memory/config.json` as `cloud_api_key`.
+On POSIX systems, Basic Memory writes `~/.basic-memory/` as user-private (`0700`) and
+`config.json` as user-read/write only (`0600`). Treat this config file as a credential
+file when an API key is saved.

 ### Setting Project Modes

@@ -574,33 +590,62 @@ bm cloud logout
 **Default patterns:**

 ```gitignore
-# Version control
-.git/**
-
-# Python
-__pycache__/**
-*.pyc
-.venv/**
-venv/**
-
-# Node.js
-node_modules/**
+# Hidden files and directories
+.*

 # Basic Memory internals
-memory.db/**
-memory.db-shm/**
-memory.db-wal/**
-config.json/**
-watch-status.json/**
-.bmignore.rclone/**
+*.db
+*.db-shm
+*.db-wal
+config.json
+
+# Version control
+.git
+.svn
+
+# Python
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.pytest_cache
+.coverage
+*.egg-info
+.tox
+.mypy_cache
+.ruff_cache
+
+# Virtual environments
+.venv
+venv
+env
+.env
+
+# Node.js
+node_modules
+
+# Build artifacts
+build
+dist
+.cache
+
+# IDE
+.idea
+.vscode

 # OS files
-.DS_Store/**
-Thumbs.db/**
+.DS_Store
+Thumbs.db
+desktop.ini

-# Environment files
-.env/**
-.env.local/**
+# Obsidian
+.obsidian
+
+# Temporary files
+*.tmp
+*.swp
+*.swo
+*~
 ```

 **How it works:**
@@ -609,6 +654,11 @@ Thumbs.db/**
 3. Rclone uses filters during sync
 4. Same patterns used by all projects

+During conversion, file patterns exclude the direct match and recursive contents.
+For example, `config.json` becomes both `- config.json` and `- config.json/**`,
+while `.*` becomes both `- .*` and `- .*/**`. Directory-only patterns keep
+their trailing slash, so `cache/` becomes `- cache/` and `- cache/**`.
+
 **Customizing:**

 ```bash
@@ -616,14 +666,41 @@ Thumbs.db/**
 code ~/.basic-memory/.bmignore

 # Add custom patterns
-echo "*.tmp/**" >> ~/.basic-memory/.bmignore
+echo "*.tmp" >> ~/.basic-memory/.bmignore

 # Next sync uses updated patterns
-bm project bisync --name research
+bm cloud bisync --name research
 ```

 ## Troubleshooting

+### Rclone Setup Cannot Install Automatically
+
+**Problem:** `bm cloud setup` cannot find a supported package manager, or package-manager
+installation fails.
+
+**Explanation:** The CLI avoids remote privileged install scripts. It only invokes known
+package managers and otherwise asks you to install rclone manually.
+
+**Solution:** Install rclone with your OS package manager, then rerun setup:
+
+```bash
+# macOS
+brew install rclone
+
+# Debian/Ubuntu
+sudo apt install rclone
+
+# Fedora
+sudo dnf install rclone
+
+# Arch
+sudo pacman -S rclone
+
+# After rclone is on PATH
+bm cloud setup
+```
+
 ### Authentication Issues

 **Problem:** "Authentication failed" or "Invalid token"
@@ -655,7 +732,7 @@ bm cloud login
 **Solution:**

 ```bash
-bm project bisync --name research --resync
+bm cloud bisync --name research --resync
 ```

 **What this does:**
@@ -678,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.
@@ -695,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:**
@@ -718,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
@@ -740,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
@@ -759,8 +836,10 @@ If instance is down, wait a few minutes and retry.

 - **Authentication**: OAuth 2.1 with PKCE flow
 - **Tokens**: Stored securely in `~/.basic-memory/basic-memory-cloud.json`
+- **API keys**: Stored in `~/.basic-memory/config.json`, which is written with private file permissions on POSIX systems
 - **Transport**: All data encrypted in transit (HTTPS)
 - **Credentials**: Scoped S3 credentials (read-write to your tenant only)
+- **Rclone setup**: Uses package managers or manual instructions; no remote privileged install-script fallback
 - **Isolation**: Your data isolated from other tenants
 - **Ignore patterns**: Sensitive files automatically excluded via `.bmignore`

@@ -785,7 +864,7 @@ bm cloud create-key <name>  # Create API key via cloud API (requires OAuth login
 ### Setup

 ```bash
-bm cloud setup              # Install rclone and configure credentials
+bm cloud setup              # Install rclone via package manager and configure credentials
 ```

 ### Project Management
@@ -810,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
@@ -838,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)
@@ -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
@@ -25,11 +25,12 @@ dependencies = [
    "unidecode>=1.3.8",
    "dateparser>=1.2.0",
    "watchfiles>=1.0.4",
-    "fastapi[standard]>=0.115.8",
+    "fastapi[standard]>=0.136.1",
    "alembic>=1.14.1",
    "pillow>=11.1.0",
    "pybars3>=0.9.7",
-    "fastmcp>=3.0.1,<4",
+    # Keep FastMCP pinned until each minor upgrade passes the MCP transport matrix.
+    "fastmcp==3.3.1",
    "pyjwt>=2.10.1",
    "python-dotenv>=1.1.0",
    "pytest-aio>=1.9.0",
@@ -6,12 +6,12 @@
    "url": "https://github.com/basicmachines-co/basic-memory.git",
    "source": "github"
  },
-  "version": "0.20.3",
+  "version": "0.21.5",
  "packages": [
    {
      "registryType": "pypi",
      "identifier": "basic-memory",
-      "version": "0.20.3",
+      "version": "0.21.5",
      "runtimeHint": "uvx",
      "runtimeArguments": [
        {"type": "positional", "value": "basic-memory"},
@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""

 # Package version - updated by release automation
-__version__ = "0.20.3"
+__version__ = "0.21.5"

 # API version for FastAPI - independent of package version
 __api_version__ = "v0"
@@ -0,0 +1,86 @@
+"""Remove orphaned search rows whose project was already deleted.
+
+Revision ID: n7i8j9k0l1m2
+Revises: m6h7i8j9k0l1
+Create Date: 2026-05-15 18:30:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision: str = "n7i8j9k0l1m2"
+down_revision: Union[str, None] = "m6h7i8j9k0l1"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _table_exists(connection, table_name: str) -> bool:
+    """Inspector-based table check, dialect agnostic.
+
+    Trigger: SQLite creates search_index as an FTS5 virtual table at runtime
+    via SearchRepository.init_search_index, not through Alembic, so fresh
+    installs hit this migration before the table exists.
+    Why: a blind DELETE against a missing table fails the whole upgrade.
+    Outcome: callers skip the sweep when the table isn't present yet — the
+    runtime-created table on a fresh DB has no orphans to clean.
+    """
+    return table_name in inspect(connection).get_table_names()
+
+
+def upgrade() -> None:
+    """Purge orphaned search rows left over from prior project deletions.
+
+    Trigger: project deletion on SQLite never removed the derived FTS rows,
+    because the FTS5 virtual table can't carry a foreign key. The leak shows
+    up in two shapes:
+    1. project_id no longer exists in `project` (deleted project, id never
+       reused).
+    2. project_id still exists but `entity_id` no longer exists in `entity`
+       — auto-increment handed the id to a brand-new project and the FTS
+       rows from the deleted predecessor masquerade as the new tenant's data.
+    Why: search_index.project_id is the only scope predicate the search
+    repository applies, so leftover rows surface under the wrong project on
+    every search.
+    Outcome: a one-time sweep deletes both shapes, from the FTS index and
+    from search_vector_chunks. Postgres already cascaded on FK delete, so
+    these statements are no-ops there.
+    """
+    connection = op.get_bind()
+
+    if _table_exists(connection, "search_index"):
+        op.execute(
+            """
+            DELETE FROM search_index
+            WHERE project_id NOT IN (SELECT id FROM project)
+            """
+        )
+        op.execute(
+            """
+            DELETE FROM search_index
+            WHERE entity_id IS NOT NULL
+              AND entity_id NOT IN (SELECT id FROM entity)
+            """
+        )
+
+    if _table_exists(connection, "search_vector_chunks"):
+        op.execute(
+            """
+            DELETE FROM search_vector_chunks
+            WHERE project_id NOT IN (SELECT id FROM project)
+            """
+        )
+        op.execute(
+            """
+            DELETE FROM search_vector_chunks
+            WHERE entity_id NOT IN (SELECT id FROM entity)
+            """
+        )
+
+
+def downgrade() -> None:
+    """No-op: orphan rows cannot be reconstructed."""
+    pass
@@ -10,6 +10,7 @@ import logging
 from fastapi import APIRouter, Form, HTTPException, UploadFile, status, Path

 from basic_memory.deps import (
+    AppConfigDep,
    ChatGPTImporterV2ExternalDep,
    ClaudeConversationsImporterV2ExternalDep,
    ClaudeProjectsImporterV2ExternalDep,
@@ -27,9 +28,21 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/import", tags=["import-v2"])


+async def read_import_upload(file: UploadFile, max_bytes: int) -> bytes:
+    """Read an import upload with a hard cap before JSON parsing."""
+    content = await file.read(max_bytes + 1)
+    if len(content) > max_bytes:
+        raise HTTPException(
+            status_code=status.HTTP_413_CONTENT_TOO_LARGE,
+            detail=f"Import file exceeds maximum size of {max_bytes} bytes.",
+        )
+    return content
+
+
@router.post("/chatgpt", response_model=ChatImportResult)
 async def import_chatgpt(
    importer: ChatGPTImporterV2ExternalDep,
+    config: AppConfigDep,
    file: UploadFile,
    project_id: str = Path(..., description="Project external UUID"),
    directory: str = Form("conversations"),
@@ -49,12 +62,13 @@ async def import_chatgpt(
        HTTPException: If import fails.
    """
    logger.info(f"V2 Importing ChatGPT conversations for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)


@router.post("/claude/conversations", response_model=ChatImportResult)
 async def import_claude_conversations(
    importer: ClaudeConversationsImporterV2ExternalDep,
+    config: AppConfigDep,
    file: UploadFile,
    project_id: str = Path(..., description="Project external UUID"),
    directory: str = Form("conversations"),
@@ -74,12 +88,13 @@ async def import_claude_conversations(
        HTTPException: If import fails.
    """
    logger.info(f"V2 Importing Claude conversations for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)


@router.post("/claude/projects", response_model=ProjectImportResult)
 async def import_claude_projects(
    importer: ClaudeProjectsImporterV2ExternalDep,
+    config: AppConfigDep,
    file: UploadFile,
    project_id: str = Path(..., description="Project external UUID"),
    directory: str = Form("projects"),
@@ -99,12 +114,13 @@ async def import_claude_projects(
        HTTPException: If import fails.
    """
    logger.info(f"V2 Importing Claude projects for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)


@router.post("/memory-json", response_model=EntityImportResult)
 async def import_memory_json(
    importer: MemoryJsonImporterV2ExternalDep,
+    config: AppConfigDep,
    file: UploadFile,
    project_id: str = Path(..., description="Project external UUID"),
    directory: str = Form("conversations"),
@@ -126,7 +142,7 @@ async def import_memory_json(
    logger.info(f"V2 Importing memory.json for project {project_id}")
    try:
        file_data = []
-        file_bytes = await file.read()
+        file_bytes = await read_import_upload(file, config.import_upload_max_bytes)
        file_str = file_bytes.decode("utf-8")
        for line in file_str.splitlines():
            json_data = json.loads(line)
@@ -138,6 +154,8 @@ async def import_memory_json(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=result.error_message or "Import failed",
            )
+    except HTTPException:
+        raise
    except Exception as e:
        logger.exception("V2 Import failed")
        raise HTTPException(
@@ -147,13 +165,16 @@ async def import_memory_json(
    return result


-async def import_file(importer: Importer, file: UploadFile, destination_directory: str):
+async def import_file(
+    importer: Importer, file: UploadFile, destination_directory: str, max_bytes: int
+):
    """Helper function to import a file using an importer instance.

    Args:
        importer: The importer instance to use
        file: The file to import
        destination_directory: Destination directory for imported content
+        max_bytes: Maximum upload size in bytes; raises HTTP 413 if exceeded

    Returns:
        Import result from the importer
@@ -163,7 +184,8 @@ async def import_file(importer: Importer, file: UploadFile, destination_director
    """
    try:
        # Process file
-        json_data = json.load(file.file)
+        upload_bytes = await read_import_upload(file, max_bytes)
+        json_data = json.loads(upload_bytes)
        result = await importer.import_data(json_data, destination_directory)
        if not result.success:  # pragma: no cover
            raise HTTPException(
@@ -173,6 +195,8 @@ async def import_file(importer: Importer, file: UploadFile, destination_director

        return result

+    except HTTPException:
+        raise
    except Exception as e:
        logger.exception("V2 Import failed")
        raise HTTPException(
@@ -14,6 +14,24 @@ class BisyncError(Exception):
    pass


+def _rclone_exclude_filters(pattern: str) -> list[str]:
+    """Return rclone exclude filters for a gitignore-style pattern."""
+    if pattern.endswith("/"):
+        # Trigger: gitignore-style patterns ending in / are directory-only rules.
+        # Why: stripping the slash would also exclude a same-named file.
+        # Outcome: rclone keeps the directory rule and excludes recursive contents.
+        return [f"- {pattern}", f"- {pattern}**"]
+
+    path_pattern = pattern.removesuffix("/**")
+
+    # Trigger: rclone treats a directory contents filter separately from the
+    # directory/file path itself.
+    # Why: files like config.json and directory markers like .obsidian must both
+    # be excluded, along with anything below matching directories.
+    # Outcome: every ignore pattern excludes the direct match and recursive children.
+    return [f"- {path_pattern}", f"- {path_pattern}/**"]
+
+
 async def get_mount_info() -> TenantMountInfo:
    """Get current tenant information from cloud API."""
    try:
@@ -78,19 +96,11 @@ def convert_bmignore_to_rclone_filters() -> Path:
                    patterns.append(line)
                    continue

-                # Convert gitignore pattern to rclone filter syntax
-                # gitignore: node_modules  → rclone: - node_modules/**
-                # gitignore: *.pyc        → rclone: - *.pyc
-                if "*" in line:
-                    # Pattern already has wildcard, just add exclude prefix
-                    patterns.append(f"- {line}")
-                else:
-                    # Directory pattern - add /** for recursive exclude
-                    patterns.append(f"- {line}/**")
+                patterns.extend(_rclone_exclude_filters(line))

    except Exception:
        # If we can't read the file, create a minimal filter
-        patterns = ["# Error reading .bmignore, using minimal filters", "- .git/**"]
+        patterns = ["# Error reading .bmignore, using minimal filters", "- .git", "- .git/**"]

    # Write rclone filter file
    rclone_filter_path.write_text("\n".join(patterns) + "\n")
@@ -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
@@ -4,7 +4,7 @@ import os
 import platform
 import shutil
 import subprocess
-from typing import Optional
+from typing import Any, Optional, cast

 from rich.console import Console

@@ -53,7 +53,9 @@ def run_command(command: list[str], check: bool = True) -> subprocess.CompletedP


 def install_rclone_macos() -> None:
-    """Install rclone on macOS using Homebrew or official script."""
+    """Install rclone on macOS using package managers."""
+    install_errors: list[str] = []
+
    # Try Homebrew first
    if shutil.which("brew"):
        try:
@@ -61,35 +63,37 @@ def install_rclone_macos() -> None:
            run_command(["brew", "install", "rclone"])
            console.print("[green]rclone installed via Homebrew[/green]")
            return
-        except RcloneInstallError:
-            console.print(
-                "[yellow]Homebrew installation failed, trying official script...[/yellow]"
-            )
+        except RcloneInstallError as exc:
+            install_errors.append(f"Homebrew failed: {exc}")
+            console.print("[yellow]Homebrew installation failed, trying MacPorts...[/yellow]")

-    # Fallback to official script
-    console.print("[blue]Installing rclone via official script...[/blue]")
-    try:
-        run_command(["sh", "-c", "curl https://rclone.org/install.sh | sudo bash"])
-        console.print("[green]rclone installed via official script[/green]")
-    except RcloneInstallError:
-        raise RcloneInstallError(
-            "Failed to install rclone. Please install manually: brew install rclone"
-        )
+    if shutil.which("port"):
+        try:
+            console.print("[blue]Installing rclone via MacPorts...[/blue]")
+            run_command(["sudo", "port", "install", "rclone"])
+            console.print("[green]rclone installed via MacPorts[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"MacPorts failed: {exc}")
+            console.print("[yellow]MacPorts installation failed[/yellow]")
+
+    details = "\n".join(f"- {error}" for error in install_errors)
+    if details:
+        details = f"\n\nAttempts:\n{details}"
+    raise RcloneInstallError(
+        "Could not install rclone automatically with an available package manager.\n\n"
+        "Install rclone manually with one of:\n"
+        "  brew install rclone\n"
+        "  sudo port install rclone\n"
+        "  Download from https://rclone.org/downloads/ and add rclone to PATH"
+        f"{details}"
+    )


 def install_rclone_linux() -> None:
-    """Install rclone on Linux using package managers or official script."""
-    # Try snap first (most universal)
-    if shutil.which("snap"):
-        try:
-            console.print("[blue]Installing rclone via snap...[/blue]")
-            run_command(["sudo", "snap", "install", "rclone"])
-            console.print("[green]rclone installed via snap[/green]")
-            return
-        except RcloneInstallError:
-            console.print("[yellow]Snap installation failed, trying apt...[/yellow]")
+    """Install rclone on Linux using package managers."""
+    install_errors: list[str] = []

-    # Try apt (Debian/Ubuntu)
    if shutil.which("apt"):
        try:
            console.print("[blue]Installing rclone via apt...[/blue]")
@@ -97,18 +101,75 @@ def install_rclone_linux() -> None:
            run_command(["sudo", "apt", "install", "-y", "rclone"])
            console.print("[green]rclone installed via apt[/green]")
            return
-        except RcloneInstallError:
-            console.print("[yellow]apt installation failed, trying official script...[/yellow]")
+        except RcloneInstallError as exc:
+            install_errors.append(f"apt failed: {exc}")
+            console.print("[yellow]apt installation failed, trying dnf...[/yellow]")

-    # Fallback to official script
-    console.print("[blue]Installing rclone via official script...[/blue]")
-    try:
-        run_command(["sh", "-c", "curl https://rclone.org/install.sh | sudo bash"])
-        console.print("[green]rclone installed via official script[/green]")
-    except RcloneInstallError:
-        raise RcloneInstallError(
-            "Failed to install rclone. Please install manually: sudo snap install rclone"
-        )
+    if shutil.which("dnf"):
+        try:
+            console.print("[blue]Installing rclone via dnf...[/blue]")
+            run_command(["sudo", "dnf", "install", "-y", "rclone"])
+            console.print("[green]rclone installed via dnf[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"dnf failed: {exc}")
+            console.print("[yellow]dnf installation failed, trying yum...[/yellow]")
+
+    if shutil.which("yum"):
+        try:
+            console.print("[blue]Installing rclone via yum...[/blue]")
+            run_command(["sudo", "yum", "install", "-y", "rclone"])
+            console.print("[green]rclone installed via yum[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"yum failed: {exc}")
+            console.print("[yellow]yum installation failed, trying pacman...[/yellow]")
+
+    if shutil.which("pacman"):
+        try:
+            console.print("[blue]Installing rclone via pacman...[/blue]")
+            run_command(["sudo", "pacman", "-S", "--noconfirm", "rclone"])
+            console.print("[green]rclone installed via pacman[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"pacman failed: {exc}")
+            console.print("[yellow]pacman installation failed, trying zypper...[/yellow]")
+
+    if shutil.which("zypper"):
+        try:
+            console.print("[blue]Installing rclone via zypper...[/blue]")
+            run_command(["sudo", "zypper", "--non-interactive", "install", "rclone"])
+            console.print("[green]rclone installed via zypper[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"zypper failed: {exc}")
+            console.print("[yellow]zypper installation failed, trying snap...[/yellow]")
+
+    if shutil.which("snap"):
+        try:
+            console.print("[blue]Installing rclone via snap...[/blue]")
+            run_command(["sudo", "snap", "install", "rclone"])
+            console.print("[green]rclone installed via snap[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"snap failed: {exc}")
+            console.print("[yellow]snap installation failed[/yellow]")
+
+    details = "\n".join(f"- {error}" for error in install_errors)
+    if details:
+        details = f"\n\nAttempts:\n{details}"
+    raise RcloneInstallError(
+        "Could not install rclone automatically with an available package manager.\n\n"
+        "Install rclone manually with one of your OS package managers, for example:\n"
+        "  sudo apt install rclone\n"
+        "  sudo dnf install rclone\n"
+        "  sudo yum install rclone\n"
+        "  sudo pacman -S rclone\n"
+        "  sudo zypper install rclone\n"
+        "  sudo snap install rclone\n"
+        "Or download from https://rclone.org/downloads/ and add rclone to PATH"
+        f"{details}"
+    )


 def install_rclone_windows() -> None:
@@ -209,27 +270,38 @@ def refresh_windows_path() -> None:
    if platform.system().lower() != "windows":
        return

-    # Importing here after performing platform detection. Also note that we have to ignore pylance/pyright
-    # warnings about winreg attributes so that "errors" don't appear on non-Windows platforms.
+    # Importing here after performing platform detection. Non-Windows type checkers may still
+    # resolve a stub without registry members, so keep this platform-only module dynamic here.
    import winreg

+    winreg_module = cast(Any, winreg)
    user_key_path = r"Environment"
    system_key_path = r"System\CurrentControlSet\Control\Session Manager\Environment"
    new_path = ""

    # Read user PATH
    try:
-        reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, user_key_path, 0, winreg.KEY_READ)  # type: ignore[reportAttributeAccessIssue]
-        user_path, _ = winreg.QueryValueEx(reg_key, "PATH")  # type: ignore[reportAttributeAccessIssue]
-        winreg.CloseKey(reg_key)  # type: ignore[reportAttributeAccessIssue]
+        reg_key = winreg_module.OpenKey(
+            winreg_module.HKEY_CURRENT_USER,
+            user_key_path,
+            0,
+            winreg_module.KEY_READ,
+        )
+        user_path, _ = winreg_module.QueryValueEx(reg_key, "PATH")
+        winreg_module.CloseKey(reg_key)
    except Exception:
        user_path = ""

    # Read system PATH
    try:
-        reg_key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, system_key_path, 0, winreg.KEY_READ)  # type: ignore[reportAttributeAccessIssue]
-        system_path, _ = winreg.QueryValueEx(reg_key, "PATH")  # type: ignore[reportAttributeAccessIssue]
-        winreg.CloseKey(reg_key)  # type: ignore[reportAttributeAccessIssue]
+        reg_key = winreg_module.OpenKey(
+            winreg_module.HKEY_LOCAL_MACHINE,
+            system_key_path,
+            0,
+            winreg_module.KEY_READ,
+        )
+        system_path, _ = winreg_module.QueryValueEx(reg_key, "PATH")
+        winreg_module.CloseKey(reg_key)
    except Exception:
        system_path = ""

@@ -6,10 +6,11 @@ from rich.table import Table

 from basic_memory.cli.commands.command_utils import run_with_cleanup
 from basic_memory.config import ConfigManager
-from basic_memory.mcp.project_context import (
-    _workspace_choices,
-    _workspace_matches_identifier,
-    get_available_workspaces,
+from basic_memory.mcp.project_context import get_available_workspaces
+from basic_memory.schemas.cloud import (
+    format_workspace_choices,
+    format_workspace_selection_choices,
+    workspace_matches_identifier,
 )

 console = Console()
@@ -62,7 +63,10 @@ def list_workspaces() -> None:

@workspace_app.command("set-default")
 def set_default_workspace(
-    identifier: str = typer.Argument(..., help="Workspace name or tenant_id to set as default"),
+    identifier: str = typer.Argument(
+        ...,
+        help="Workspace name, slug, type, or tenant_id to set as default",
+    ),
 ) -> None:
    """Set the default cloud workspace.

@@ -71,6 +75,7 @@ def set_default_workspace(

    Examples:
      bm cloud workspace set-default Personal
+      bm cloud workspace set-default organization
      bm cloud workspace set-default 11111111-1111-1111-1111-111111111111
    """

@@ -87,19 +92,19 @@ def set_default_workspace(
        console.print("[yellow]No accessible workspaces found.[/yellow]")
        raise typer.Exit(1)

-    matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, identifier)]
+    matches = [ws for ws in workspaces if workspace_matches_identifier(ws, identifier)]

    if not matches:
        console.print(f"[red]Error: Workspace '{identifier}' not found[/red]")
-        console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+        console.print(f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]")
        raise typer.Exit(1)

    if len(matches) > 1:
+        console.print(f"[red]Error: Workspace '{identifier}' matches multiple workspaces.[/red]")
        console.print(
-            f"[red]Error: Workspace name '{identifier}' matches multiple workspaces. "
-            f"Use tenant_id instead.[/red]"
+            "[dim]Choose one of these matching workspaces by slug:\n"
+            f"{format_workspace_selection_choices(matches)}[/dim]"
        )
-        console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
        raise typer.Exit(1)

    selected = matches[0]
@@ -7,6 +7,7 @@ from pathlib import Path
 from typing import cast

 import typer
+from loguru import logger
 from rich.console import Console, Group
 from rich.panel import Panel
 from rich.table import Table
@@ -33,6 +34,10 @@ from basic_memory.schemas.cloud import (
    CloudProjectIndexStatus,
    CloudTenantIndexStatusResponse,
    ProjectVisibility,
+    WorkspaceInfo,
+    format_workspace_choices,
+    format_workspace_selection_choices,
+    workspace_matches_identifier,
 )
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.utils import generate_permalink, normalize_project_path
@@ -281,27 +286,25 @@ def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:


 def _resolve_workspace_id(config, workspace: str | None) -> str | None:
-    """Resolve a workspace name or tenant_id to a tenant_id."""
+    """Resolve a workspace name, slug, type, or tenant_id to a tenant_id."""
    from basic_memory.mcp.project_context import (
-        _workspace_choices,
-        _workspace_matches_identifier,
        get_available_workspaces,
    )

    if workspace is not None:
        workspaces = run_with_cleanup(get_available_workspaces())
-        matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
+        matches = [ws for ws in workspaces if workspace_matches_identifier(ws, workspace)]
        if not matches:
            console.print(f"[red]Error: Workspace '{workspace}' not found[/red]")
            if workspaces:
-                console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+                console.print(f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]")
            raise typer.Exit(1)
        if len(matches) > 1:
+            console.print(f"[red]Error: Workspace '{workspace}' matches multiple workspaces.[/red]")
            console.print(
-                f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. "
-                f"Use tenant_id instead.[/red]"
+                "[dim]Choose one of these matching workspaces by slug:\n"
+                f"{format_workspace_selection_choices(matches)}[/dim]"
            )
-            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
            raise typer.Exit(1)
        return matches[0].tenant_id

@@ -312,9 +315,9 @@ def _resolve_workspace_id(config, workspace: str | None) -> str | None:
        workspaces = run_with_cleanup(get_available_workspaces())
        if len(workspaces) == 1:
            return workspaces[0].tenant_id
-    except Exception:
+    except Exception as exc:
        # Workspace resolution is optional until a command needs a specific tenant.
-        pass
+        logger.debug("Workspace resolution failed: {}", exc)

    return None

@@ -323,7 +326,11 @@ def _resolve_workspace_id(config, workspace: str | None) -> str | None:
 def list_projects(
    local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
-    workspace: str = typer.Option(None, "--workspace", help="Cloud workspace name or tenant_id"),
+    workspace: str = typer.Option(
+        None,
+        "--workspace",
+        help="Cloud workspace name, slug, type, or tenant_id",
+    ),
    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
 ) -> None:
    """List Basic Memory projects from local and (when available) cloud."""
@@ -339,17 +346,90 @@ def list_projects(

    try:
        config = ConfigManager().config
-        # Use explicit workspace, fall back to config default
-        effective_workspace = workspace or config.default_workspace
+        workspace_filter = workspace
+        workspace_filter_requested = workspace_filter is not None

        local_result: ProjectList | None = None
-        cloud_result: ProjectList | None = None
+        cloud_results: list[tuple[WorkspaceInfo | None, ProjectList]] = []
+        available_cloud_workspaces: list[WorkspaceInfo] = []
        cloud_error: Exception | None = None
+        cloud_workspace_error: Exception | None = None
+        failed_cloud_workspaces: list[tuple[WorkspaceInfo, Exception]] = []
+
+        def _fetch_cloud_workspace_results() -> tuple[
+            list[tuple[WorkspaceInfo | None, ProjectList]],
+            list[WorkspaceInfo],
+            Exception | None,
+            list[tuple[WorkspaceInfo, Exception]],
+        ]:
+            from basic_memory.mcp.project_context import (
+                get_available_workspaces,
+            )
+
+            try:
+                workspaces = run_with_cleanup(get_available_workspaces())
+            except Exception as exc:
+                fallback_workspace = workspace_filter or config.default_workspace
+                return (
+                    [(None, run_with_cleanup(_list_projects(fallback_workspace)))],
+                    [],
+                    exc,
+                    [],
+                )
+
+            selected_workspaces = workspaces
+            if workspace_filter is not None:
+                matches = [
+                    ws for ws in workspaces if workspace_matches_identifier(ws, workspace_filter)
+                ]
+                if not matches:
+                    console.print(f"[red]Error: Workspace '{workspace_filter}' not found[/red]")
+                    if workspaces:
+                        console.print(
+                            f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]"
+                        )
+                    raise typer.Exit(1)
+                if len(matches) > 1:
+                    console.print(
+                        f"[red]Error: Workspace '{workspace_filter}' matches multiple workspaces.[/red]"
+                    )
+                    console.print(
+                        "[dim]Choose one of these matching workspaces by slug:\n"
+                        f"{format_workspace_selection_choices(matches)}[/dim]"
+                    )
+                    raise typer.Exit(1)
+                selected_workspaces = matches
+
+            if not selected_workspaces:
+                return [], workspaces, None, []
+
+            results: list[tuple[WorkspaceInfo | None, ProjectList]] = []
+            failed_workspaces: list[tuple[WorkspaceInfo, Exception]] = []
+            for cloud_workspace in selected_workspaces:
+                try:
+                    results.append(
+                        (
+                            cloud_workspace,
+                            run_with_cleanup(_list_projects(cloud_workspace.tenant_id)),
+                        )
+                    )
+                except Exception as exc:
+                    failed_workspaces.append((cloud_workspace, exc))
+
+            if not results and failed_workspaces:
+                raise failed_workspaces[0][1]
+
+            return results, workspaces, None, failed_workspaces

        if cloud:
            with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                with force_routing(cloud=True):
-                    cloud_result = run_with_cleanup(_list_projects(effective_workspace))
+                    (
+                        cloud_results,
+                        available_cloud_workspaces,
+                        cloud_workspace_error,
+                        failed_cloud_workspaces,
+                    ) = _fetch_cloud_workspace_results()
        elif local:
            with force_routing(local=True):
                local_result = run_with_cleanup(_list_projects())
@@ -362,29 +442,17 @@ def list_projects(
                try:
                    with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                        with force_routing(cloud=True):
-                            cloud_result = run_with_cleanup(_list_projects(effective_workspace))
+                            (
+                                cloud_results,
+                                available_cloud_workspaces,
+                                cloud_workspace_error,
+                                failed_cloud_workspaces,
+                            ) = _fetch_cloud_workspace_results()
+                except typer.Exit:
+                    raise
                except Exception as exc:  # pragma: no cover
                    cloud_error = exc

-        # Resolve workspace name for cloud projects (best-effort)
-        cloud_ws_name: str | None = None
-        cloud_ws_type: str | None = None
-        if cloud_result and effective_workspace:
-            try:
-                from basic_memory.mcp.project_context import get_available_workspaces
-
-                with console.status("[bold blue]Resolving workspace...", spinner="dots"):
-                    workspaces = run_with_cleanup(get_available_workspaces())
-                matched = next(
-                    (ws for ws in workspaces if ws.tenant_id == effective_workspace),
-                    None,
-                )
-                if matched:
-                    cloud_ws_name = matched.name
-                    cloud_ws_type = matched.workspace_type
-            except Exception:
-                pass
-
        table = Table(title="Basic Memory Projects")
        table.add_column("Name", style="cyan")
        table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
@@ -395,29 +463,126 @@ def list_projects(
        table.add_column("Sync", style="green")
        table.add_column("Default", style="magenta")

-        project_names_by_permalink: dict[str, str] = {}
+        row_names_by_key: dict[tuple[str | None, str], str] = {}
        local_projects_by_permalink: dict[str, ProjectItem] = {}
-        cloud_projects_by_permalink: dict[str, ProjectItem] = {}
+        cloud_projects_by_key: dict[tuple[str | None, str], ProjectItem] = {}
+        cloud_workspaces_by_key: dict[tuple[str | None, str], WorkspaceInfo | None] = {}

        if local_result:
            for project in local_result.projects:
                permalink = generate_permalink(project.name)
-                project_names_by_permalink[permalink] = project.name
                local_projects_by_permalink[permalink] = project

-        if cloud_result:
+        for cloud_workspace, cloud_result in cloud_results:
+            workspace_key = cloud_workspace.tenant_id if cloud_workspace else None
            for project in cloud_result.projects:
                permalink = generate_permalink(project.name)
-                project_names_by_permalink[permalink] = project.name
-                cloud_projects_by_permalink[permalink] = project
+                row_key = (workspace_key, permalink)
+                row_names_by_key[row_key] = project.name
+                cloud_projects_by_key[row_key] = project
+                cloud_workspaces_by_key[row_key] = cloud_workspace
+
+        cloud_permalinks = {permalink for _, permalink in cloud_projects_by_key}
+        for permalink, project in local_projects_by_permalink.items():
+            if permalink not in cloud_permalinks:
+                row_names_by_key[(None, permalink)] = project.name
+
+        cloud_keys_by_permalink: dict[str, list[tuple[str | None, str]]] = {}
+        for row_key in cloud_projects_by_key:
+            cloud_keys_by_permalink.setdefault(row_key[1], []).append(row_key)
+
+        configured_names_by_permalink = {
+            generate_permalink(project_name): project_name for project_name in config.projects
+        }
+
+        def _workspace_priority(row_key: tuple[str | None, str]) -> tuple[bool, int, str, str]:
+            """Prefer the user's default/personal workspace when a project is duplicated."""
+            workspace = cloud_workspaces_by_key.get(row_key)
+            if workspace is None:
+                return (True, 2, "", row_key[0] or "")
+            workspace_type_rank = 0 if workspace.workspace_type == "personal" else 1
+            return (
+                not workspace.is_default,
+                workspace_type_rank,
+                workspace.name.casefold(),
+                row_key[0] or "",
+            )
+
+        def _select_attached_row_key(
+            permalink: str, entry: ProjectEntry | None
+        ) -> tuple[str | None, str] | None:
+            """Choose the single row that owns local config/default/sync state."""
+            cloud_keys = cloud_keys_by_permalink.get(permalink, [])
+            if not cloud_keys:
+                return (None, permalink)
+
+            preferred_workspace_ids: list[str] = []
+            if entry and entry.workspace_id:
+                preferred_workspace_ids.append(entry.workspace_id)
+            if config.default_workspace and config.default_workspace not in preferred_workspace_ids:
+                preferred_workspace_ids.append(config.default_workspace)
+            default_cloud_workspace = next(
+                (item for item in available_cloud_workspaces if item.is_default),
+                None,
+            )
+            if (
+                default_cloud_workspace
+                and default_cloud_workspace.tenant_id not in preferred_workspace_ids
+            ):
+                preferred_workspace_ids.append(default_cloud_workspace.tenant_id)
+
+            for workspace_id in preferred_workspace_ids:
+                for row_key in cloud_keys:
+                    if row_key[0] == workspace_id:
+                        return row_key
+
+            if workspace_filter_requested and preferred_workspace_ids:
+                # A filtered list can exclude the workspace that owns local config state.
+                # In that case, do not attach local/default/sync state to another workspace row.
+                return None
+
+            default_workspace_keys = [
+                row_key
+                for row_key in cloud_keys
+                if (row_workspace := cloud_workspaces_by_key.get(row_key)) is not None
+                and row_workspace.is_default
+            ]
+            if len(default_workspace_keys) == 1:
+                return default_workspace_keys[0]
+
+            if len(cloud_keys) == 1:
+                return cloud_keys[0]
+
+            return sorted(cloud_keys, key=_workspace_priority)[0]
+
+        attached_row_by_permalink: dict[str, tuple[str | None, str] | None] = {}
+        for permalink in set(local_projects_by_permalink) | set(configured_names_by_permalink):
+            configured_name = configured_names_by_permalink.get(permalink)
+            local_project = local_projects_by_permalink.get(permalink)
+            entry_name = configured_name or (local_project.name if local_project else None)
+            entry = config.projects.get(entry_name) if entry_name else None
+            attached_row_by_permalink[permalink] = _select_attached_row_key(permalink, entry)

        # --- Build unified project list ---
        project_rows: list[dict] = []
-        for permalink in sorted(project_names_by_permalink):
-            project_name = project_names_by_permalink[permalink]
-            local_project = local_projects_by_permalink.get(permalink)
-            cloud_project = cloud_projects_by_permalink.get(permalink)
-            entry = config.projects.get(project_name)
+        sorted_row_keys = sorted(
+            row_names_by_key,
+            key=lambda key: (row_names_by_key[key], key[0] or ""),
+        )
+        for row_key in sorted_row_keys:
+            _, permalink = row_key
+            project_name = row_names_by_key[row_key]
+            is_attached_row = attached_row_by_permalink.get(permalink) == row_key
+            local_project = local_projects_by_permalink.get(permalink) if is_attached_row else None
+            cloud_project = cloud_projects_by_key.get(row_key)
+            cloud_workspace = cloud_workspaces_by_key.get(row_key)
+            configured_name = configured_names_by_permalink.get(permalink)
+            configured_entry = (
+                config.projects.get(configured_name)
+                if configured_name
+                else config.projects.get(project_name)
+            )
+            entry = configured_entry if is_attached_row else None

            local_path = ""
            if local_project is not None:
@@ -447,9 +612,13 @@ def list_projects(
            else:
                cli_route = ProjectMode.LOCAL.value

-            is_default = config.default_project == project_name
+            default_permalink = (
+                generate_permalink(config.default_project) if config.default_project else None
+            )
+            is_default = bool(is_attached_row and permalink == default_permalink)

-            has_sync = bool(entry and entry.local_sync_path)
+            sync_supported = cloud_workspace is None or cloud_workspace.workspace_type == "personal"
+            has_sync = bool(is_attached_row and entry and entry.local_sync_path and sync_supported)
            # Determine MCP transport based on project routing mode
            if entry and entry.mode == ProjectMode.CLOUD:
                mcp_transport = "https"
@@ -459,9 +628,8 @@ def list_projects(
                mcp_transport = "stdio"

            # Show workspace name (type) for cloud-sourced projects
-            ws_label = ""
-            if cloud_project is not None and cloud_ws_name:
-                ws_label = f"{cloud_ws_name} ({cloud_ws_type})" if cloud_ws_type else cloud_ws_name
+            cloud_ws_name = cloud_workspace.name if cloud_workspace else None
+            cloud_ws_type = cloud_workspace.workspace_type if cloud_workspace else None

            # display_name is a human label for private UUID-named projects (e.g., "My Project").
            # Keep "name" as the canonical identifier for scripting/JSON consumers;
@@ -481,8 +649,8 @@ def list_projects(
            }
            if display_name:
                row_data["display_name"] = display_name
-            if ws_label:
-                row_data["workspace"] = cloud_ws_name or ""
+            if cloud_project is not None and cloud_ws_name:
+                row_data["workspace"] = cloud_ws_name
                if cloud_ws_type:
                    row_data["workspace_type"] = cloud_ws_type

@@ -514,6 +682,20 @@ def list_projects(
                "[dim]Showing local projects only. "
                "Run 'bm cloud login' or 'bm cloud api-key save <key>' if this is a credentials issue.[/dim]"
            )
+        if cloud_workspace_error is not None:
+            console.print(
+                f"[yellow]Cloud workspace discovery failed: {cloud_workspace_error}[/yellow]"
+            )
+            console.print(
+                "[dim]Showing cloud projects from the configured/default workspace only.[/dim]"
+            )
+        for failed_workspace, error in failed_cloud_workspaces:
+            console.print(
+                f"[yellow]Cloud project discovery failed for workspace "
+                f"{failed_workspace.name}: {error}[/yellow]"
+            )
+    except typer.Exit:
+        raise
    except Exception as e:
        console.print(f"[red]Error listing projects: {str(e)}[/red]")
        raise typer.Exit(1)
@@ -531,7 +713,7 @@ def add_project(
    workspace: str = typer.Option(
        None,
        "--workspace",
-        help="Cloud workspace name or tenant_id (cloud mode only)",
+        help="Cloud workspace name, slug, type, or tenant_id (cloud mode only)",
    ),
    visibility: str = typer.Option(
        None,
@@ -914,7 +1096,7 @@ def set_cloud(
    workspace: str = typer.Option(
        None,
        "--workspace",
-        help="Cloud workspace name or tenant_id to associate with this project",
+        help="Cloud workspace name, slug, type, or tenant_id to associate with this project",
    ),
 ) -> None:
    """Set a project to cloud mode (route through cloud API).
@@ -21,13 +21,27 @@ 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
+CONFIG_FILE_MODE = 0o600

 Environment = Literal["test", "dev", "user"]


+def _secure_config_dir(path: Path) -> None:
+    """Restrict config directory permissions on platforms with POSIX modes."""
+    if os.name != "nt":
+        path.chmod(CONFIG_DIR_MODE)
+
+
+def _secure_config_file(path: Path) -> None:
+    """Restrict config file permissions because config can contain cloud credentials."""
+    if os.name != "nt":
+        path.chmod(CONFIG_FILE_MODE)
+
+
 class ProjectMode(str, Enum):
    """Per-project routing mode."""

@@ -55,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:
@@ -168,13 +185,15 @@ class BasicMemoryConfig(BaseSettings):
    env: Environment = Field(default="dev", description="Environment name")

    projects: Dict[str, ProjectEntry] = Field(
-        default_factory=lambda: {
-            "main": ProjectEntry(
-                path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
-            )
-        }
-        if os.getenv("BASIC_MEMORY_HOME")
-        else {},
+        default_factory=lambda: (
+            {
+                "main": ProjectEntry(
+                    path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
+                )
+            }
+            if os.getenv("BASIC_MEMORY_HOME")
+            else {}
+        ),
        description="Mapping of project names to their ProjectEntry configuration",
    )
    default_project: Optional[str] = Field(
@@ -278,6 +297,11 @@ class BasicMemoryConfig(BaseSettings):
        description="Optional FastEmbed embed() parallelism override.",
        gt=0,
    )
+    import_upload_max_bytes: int = Field(
+        default=100 * 1024 * 1024,
+        description="Maximum uploaded JSON export size accepted by API import endpoints.",
+        gt=0,
+    )
    semantic_vector_k: int = Field(
        default=100,
        description="Vector candidate count for vector and hybrid retrieval.",
@@ -776,6 +800,7 @@ class ConfigManager:

        # Ensure config directory exists
        self.config_dir.mkdir(parents=True, exist_ok=True)
+        _secure_config_dir(self.config_dir)

    @property
    def config(self) -> BasicMemoryConfig:
@@ -888,6 +913,7 @@ class ConfigManager:
                    # Create backup before overwriting so users can revert if needed
                    backup_path = self.config_file.with_suffix(".json.bak")
                    shutil.copy2(self.config_file, backup_path)
+                    _secure_config_file(backup_path)
                    logger.info(f"Migrating config to current format (backup: {backup_path})")
                    save_basic_memory_config(self.config_file, _CONFIG_CACHE)

@@ -1043,9 +1069,12 @@ def has_cloud_credentials(config: BasicMemoryConfig) -> bool:
 def save_basic_memory_config(file_path: Path, config: BasicMemoryConfig) -> None:
    """Save configuration to file."""
    try:
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        _secure_config_dir(file_path.parent)
        # Use model_dump with mode='json' to serialize datetime objects properly
        config_dict = config.model_dump(mode="json")
        file_path.write_text(json.dumps(config_dict, indent=2))
+        _secure_config_file(file_path)
    except Exception as e:  # pragma: no cover
        logger.error(f"Failed to save config: {e}")

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

@@ -85,6 +85,28 @@ def parse_observation(token: Token) -> Dict[str, Any]:


 # Relation handling functions
+def parse_relation_type(content: str) -> str | None:
+    """Return the explicit relation label before the first wikilink, if any."""
+    before_link = content.partition("[[")[0].strip()
+    if not before_link:
+        return None
+
+    # Trigger: relation labels that need spaces must be quoted.
+    # Why: unquoted multi-word prefixes are indistinguishable from prose
+    # containing a wikilink.
+    # Outcome: `some_type [[Target]]`, `"some type" [[Target]]`, and
+    # `'some type' [[Target]]` are explicit; `some other thing [[Target]]`
+    # falls back to inline `links_to` handling.
+    quote = before_link[0]
+    if quote in {"'", '"'} and before_link.endswith(quote):
+        quoted_label = before_link[1:-1].strip()
+        return quoted_label or None
+
+    if any(char.isspace() for char in before_link):
+        return None
+    return before_link
+
+
 def is_explicit_relation(token: Token) -> bool:
    """Check if token looks like our relation format."""
    if token.type != "inline":  # pragma: no cover
@@ -92,7 +114,7 @@ def is_explicit_relation(token: Token) -> bool:

    # Use token.tag which contains the actual content for test tokens, fallback to content
    content = (token.tag or token.content).strip()
-    return "[[" in content and "]]" in content
+    return "[[" in content and "]]" in content and parse_relation_type(content) is not None


 def parse_relation(token: Token) -> Dict[str, Any] | None:
@@ -101,20 +123,18 @@ def parse_relation(token: Token) -> Dict[str, Any] | None:
    # Use token.tag which contains the actual content for test tokens, fallback to content
    content = (token.tag or token.content).strip()

+    rel_type = parse_relation_type(content)
+    if rel_type is None:
+        return None
+
    # Extract [[target]]
    target = None
-    rel_type = "relates_to"  # default
    context = None

    start = content.find("[[")
-    end = content.find("]]")
+    end = content.find("]]", start + 2)

    if start != -1 and end != -1:
-        # Get text before link as relation type
-        before = content[:start].strip()
-        if before:
-            rel_type = before
-
        # Get target
        target = normalize_project_reference(content[start + 2 : end].strip())

@@ -217,6 +237,8 @@ def relation_plugin(md: MarkdownIt) -> None:

    Explicit relations:
    - relation_type [[target]] (context)
+    - "multi word relation type" [[target]] (context)
+    - 'multi word relation type' [[target]] (context)

    Implicit relations (links in content):
    Some text with [[target]] reference
@@ -1,14 +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.api.app import app as fastapi_app
 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."""
@@ -35,12 +57,12 @@ def _build_timeout() -> Timeout:
    )


-def _asgi_client(timeout: Timeout) -> AsyncClient:
-    """Create a local ASGI client."""
+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
@@ -49,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(
@@ -258,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
@@ -11,7 +11,7 @@ compatibility with existing MCP tools.
 import asyncio
 from contextlib import asynccontextmanager, nullcontext
 from dataclasses import dataclass, field
-from typing import AsyncIterator, Awaitable, Callable, Optional, List, Tuple, cast
+from typing import AsyncIterator, Awaitable, Callable, List, Optional, Sequence, Tuple, cast
 from uuid import UUID

 from httpx import AsyncClient
@@ -25,7 +25,14 @@ from mcp.server.fastmcp.exceptions import ToolError
 import logfire
 from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectMode, has_cloud_credentials
 from basic_memory.project_resolver import ProjectResolver
-from basic_memory.schemas.cloud import WorkspaceInfo, WorkspaceListResponse
+from basic_memory.schemas.cloud import (
+    WorkspaceInfo,
+    WorkspaceListResponse,
+    format_workspace_choices,
+    format_workspace_selection_choices,
+    workspace_matches_exact_identifier,
+    workspace_matches_identifier,
+)
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.schemas.v2 import ProjectResolveResponse
 from basic_memory.schemas.memory import memory_url_path
@@ -135,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:
@@ -160,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())

@@ -299,29 +314,6 @@ async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = N
    return [project.name for project in project_list.projects]


-def _workspace_matches_identifier(workspace: WorkspaceInfo, identifier: str) -> bool:
-    """Return True when identifier matches workspace tenant_id, slug, or name."""
-    if workspace.tenant_id == identifier:
-        return True
-    if workspace.slug.casefold() == identifier.casefold():
-        return True
-    return workspace.name.lower() == identifier.lower()
-
-
-def _workspace_choices(workspaces: list[WorkspaceInfo]) -> str:
-    """Format deterministic workspace choices for prompt-style errors."""
-    return "\n".join(
-        [
-            (
-                f"- {item.name} "
-                f"(slug={item.slug}, type={item.workspace_type}, "
-                f"role={item.role}, tenant_id={item.tenant_id})"
-            )
-            for item in workspaces
-        ]
-    )
-
-
 def _workspace_project_index_from_state(raw: object) -> WorkspaceProjectIndex | None:
    """Deserialize a cached workspace project index from MCP context state."""
    if not isinstance(raw, dict):
@@ -541,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,
@@ -624,7 +638,7 @@ async def _resolve_workspace_segments(
    return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path)


-def _format_qualified_choices(entries: tuple[WorkspaceProjectEntry, ...]) -> str:
+def _format_qualified_choices(entries: Sequence[WorkspaceProjectEntry]) -> str:
    """Format qualified project choices for collision errors."""
    return " or ".join(entry.qualified_name for entry in entries)

@@ -750,7 +764,7 @@ async def _ensure_workspace_project_index(
                )
            continue

-        workspace_entries = cast(tuple[WorkspaceProjectEntry, ...], result)
+        workspace_entries = result
        successful_fetches += 1
        entries_list.extend(workspace_entries)

@@ -853,7 +867,7 @@ async def resolve_workspace_project_identifier(
            )
        return matches[0]

-    matches = index.entries_by_permalink.get(project_permalink, ())
+    matches = list(index.entries_by_permalink.get(project_permalink, ()))
    if not matches:
        failed_note = ""
        if index.failed_workspaces:
@@ -987,7 +1001,9 @@ async def resolve_workspace_parameter(
            cached_raw = await context.get_state("active_workspace")
            if isinstance(cached_raw, dict):
                cached_workspace = WorkspaceInfo.model_validate(cached_raw)
-                if workspace is None or _workspace_matches_identifier(cached_workspace, workspace):
+                if workspace is None or workspace_matches_exact_identifier(
+                    cached_workspace, workspace
+                ):
                    logger.debug(
                        f"Using cached workspace from context: {cached_workspace.tenant_id}"
                    )
@@ -1003,19 +1019,17 @@ async def resolve_workspace_parameter(
        selected_workspace: WorkspaceInfo | None = None

        if workspace:
-            matches = [
-                item for item in workspaces if _workspace_matches_identifier(item, workspace)
-            ]
+            matches = [item for item in workspaces if workspace_matches_identifier(item, workspace)]
            if not matches:
                raise ValueError(
                    f"Workspace '{workspace}' was not found.\n"
-                    f"Available workspaces:\n{_workspace_choices(workspaces)}"
+                    f"Available workspaces:\n{format_workspace_choices(workspaces)}"
                )
            if len(matches) > 1:
                raise ValueError(
-                    f"Workspace name '{workspace}' matches multiple workspaces. "
-                    "Use tenant_id instead.\n"
-                    f"Available workspaces:\n{_workspace_choices(workspaces)}"
+                    f"Workspace '{workspace}' matches multiple workspaces. "
+                    "Choose one of these matching workspaces by slug or tenant_id:\n"
+                    f"{format_workspace_selection_choices(matches)}"
                )
            selected_workspace = matches[0]
        elif len(workspaces) == 1:
@@ -1023,8 +1037,8 @@ async def resolve_workspace_parameter(
        else:
            raise ValueError(
                "Multiple workspaces are available. Ask the user which workspace to use, then retry "
-                "with the 'workspace' argument set to the tenant_id or unique name.\n"
-                f"Available workspaces:\n{_workspace_choices(workspaces)}"
+                "with the 'workspace' argument set to the tenant_id or unique name/slug/type.\n"
+                f"Available workspaces:\n{format_workspace_choices(workspaces)}"
            )

        await _set_cached_active_workspace(context, selected_workspace)
@@ -1342,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

@@ -1361,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

@@ -1478,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)

@@ -1503,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,
@@ -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,
@@ -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,
@@ -10,8 +10,13 @@ from loguru import logger
 from fastmcp import Context
 from pydantic import AliasChoices, BeforeValidator, Field

-from basic_memory.config import ConfigManager
+from basic_memory.config import ConfigManager, has_cloud_credentials
 from basic_memory.utils import build_canonical_permalink, coerce_dict, coerce_list
+from basic_memory.mcp.async_client import (
+    _explicit_routing,
+    _force_local_mode,
+    is_factory_mode,
+)
 from basic_memory.mcp.container import get_container
 from basic_memory.mcp.project_context import (
    detect_project_from_identifier_prefix,
@@ -479,12 +484,30 @@ async def _search_all_projects(
    total = 0
    any_project_has_more = False

+    # Trigger: caller asked for an account-wide search.
+    # Why: project_id (external UUID) routes through the cloud v2 API path,
+    #      which 401s on local installs because there's no JWT to present.
+    #      Project names route through the local-ASGI path and work for both
+    #      backends — cloud disambiguates names via the workspace/project
+    #      qualified_name already baked into project_ref["project"].
+    # Outcome: forward project_id only when the same signals get_project_client
+    #          uses to pick a cloud route are present. Mirrors the cloud_available
+    #          composite in project_context.get_project_client (single source of
+    #          truth for "can we route to cloud?").
+    config = ConfigManager().config
+    use_cloud_routing = (
+        is_factory_mode()
+        or (_explicit_routing() and not _force_local_mode())
+        or has_cloud_credentials(config)
+    )
+
    for project_ref in project_refs:
+        recursive_project_id = project_ref["project_id"] if use_cloud_routing else None
        try:
            results = await search_notes(
                query=query,
                project=project_ref["project"],
-                project_id=project_ref["project_id"],
+                project_id=recursive_project_id,
                page=1,
                page_size=per_project_page_size,
                search_type=search_type,
@@ -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:
@@ -66,10 +68,14 @@ async def write_note(

    Relations format:
        - Explicit: `- relation_type [[Entity]] (optional context)`
-        - Inline: Any `[[Entity]]` reference creates a relation
+        - Quoted: `- "multi word relation type" [[Entity]] (optional context)`
+        - Quoted: `- 'multi word relation type' [[Entity]] (optional context)`
+        - Inline: Any other `[[Entity]]` reference creates a `links_to` relation

        Examples:
        `- depends_on [[Content Parser]] (Need for semantic extraction)`
+        `- "based on" [[Design Notes]]`
+        `- 'in response to' [[Incident Review]]`
        `- implements [[Search Spec]] (Initial implementation)`
        `- This feature extends [[Base Design]] and uses [[Core Utils]]`

@@ -274,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'}",
            ]

@@ -315,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(),
@@ -5,11 +5,17 @@ from __future__ import annotations
 from dataclasses import dataclass
 from datetime import date, datetime
 import re
-from typing import Any, Iterable, List
+from typing import Any, Iterable, List, cast


 _KEY_RE = re.compile(r"^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$")
 _NUMERIC_RE = re.compile(r"^-?\d+(\.\d+)?$")
+_COMPARISON_OPERATORS = {
+    "$gt": "gt",
+    "$gte": "gte",
+    "$lt": "lt",
+    "$lte": "lte",
+}


@dataclass(frozen=True)
@@ -48,6 +54,11 @@ def _normalize_scalar(value: Any) -> Any:
    return value


+def _normalize_numeric(value: object) -> float:
+    """Normalize a value already proven numeric by _is_numeric_value."""
+    return float(cast(str | int | float, value))
+
+
 def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter]:
    """Parse metadata filters into normalized clauses.

@@ -73,7 +84,12 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
        if isinstance(raw_value, dict):
            if len(raw_value) != 1:
                raise ValueError(f"Invalid metadata filter for '{raw_key}': {raw_value}")
-            op, value = next(iter(raw_value.items()))
+            raw_op, value = next(iter(raw_value.items()))
+            if not isinstance(raw_op, str):
+                raise ValueError(
+                    f"Unsupported operator '{raw_op}' in metadata filter for '{raw_key}'"
+                )
+            op = raw_op

            if op == "$in":
                if not isinstance(value, list) or not value:
@@ -83,15 +99,20 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
                )
                continue

-            if op in {"$gt", "$gte", "$lt", "$lte"}:
+            if op in _COMPARISON_OPERATORS:
                if _is_numeric_value(value):
-                    normalized = float(value)
+                    normalized = _normalize_numeric(value)
                    comparison = "numeric"
                else:
                    normalized = _normalize_scalar(value)
                    comparison = "text"
                parsed.append(
-                    ParsedMetadataFilter(path_parts, op.lstrip("$"), normalized, comparison)
+                    ParsedMetadataFilter(
+                        path_parts,
+                        _COMPARISON_OPERATORS[op],
+                        normalized,
+                        comparison,
+                    )
                )
                continue

@@ -99,7 +120,7 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
                if not isinstance(value, list) or len(value) != 2:
                    raise ValueError(f"$between requires [min, max] for '{raw_key}'")
                if _is_numeric_collection(value):
-                    normalized = [float(v) for v in value]
+                    normalized = [_normalize_numeric(v) for v in value]
                    comparison = "numeric"
                else:
                    normalized = [_normalize_scalar(v) for v in value]
@@ -4,7 +4,9 @@ from pathlib import Path
 from typing import Optional, Sequence, Union


-from sqlalchemy import text
+from loguru import logger
+from sqlalchemy import inspect as sa_inspect, select, text
+from sqlalchemy.exc import NoResultFound, OperationalError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

 from basic_memory import db
@@ -12,6 +14,65 @@ from basic_memory.models.project import Project
 from basic_memory.repository.repository import Repository


+async def _load_sqlite_vec_on_session(session) -> bool:
+    """Ensure the sqlite-vec extension is loaded on this session's connection.
+
+    Returns True when vec0 is available after the call. Returns False when the
+    extension can't be loaded on this Python build (e.g., python.org macOS or
+    Windows interpreters without `enable_load_extension`) — every connection in
+    the pool shares the same interpreter, so a False here also means no
+    embedding row could ever have been written, and skipping the embeddings
+    purge is safe.
+
+    Mirrors SQLiteSearchRepository._ensure_sqlite_vec_loaded but as a free
+    function: we don't have a SearchRepository instance during project delete,
+    and the per-connection nature of extension loading means a pooled connection
+    routed to this session might not have vec loaded even when other
+    connections wrote embeddings.
+    """
+    try:
+        await session.execute(text("SELECT vec_version()"))
+        return True
+    except OperationalError:
+        pass
+
+    try:
+        import sqlite_vec  # type: ignore[import-not-found]
+    except ImportError:
+        logger.debug("sqlite-vec package not installed; skipping vec purge")
+        return False
+
+    async_connection = await session.connection()
+    raw_connection = await async_connection.get_raw_connection()
+    driver_connection = raw_connection.driver_connection
+
+    if not hasattr(driver_connection, "enable_load_extension"):
+        # Trigger: CPython build without sqlite extension support (#711).
+        # Why: load_extension is unavailable, so no connection in this pool
+        #      can host vec0. No embeddings exist anywhere.
+        # Outcome: skip the embeddings purge entirely.
+        logger.debug(
+            "Skipping search_vector_embeddings purge: this Python build does "
+            "not support SQLite extension loading"
+        )
+        return False
+
+    try:
+        await driver_connection.enable_load_extension(True)
+        await driver_connection.load_extension(sqlite_vec.loadable_path())
+        await driver_connection.enable_load_extension(False)
+        await session.execute(text("SELECT vec_version()"))
+    except Exception as exc:
+        logger.warning(
+            "Failed to load sqlite-vec for project delete cleanup; "
+            "skipping embeddings purge: {}",
+            exc,
+        )
+        return False
+
+    return True
+
+
 class ProjectRepository(Repository[Project]):
    """Repository for Project model.

@@ -121,6 +182,83 @@ class ProjectRepository(Repository[Project]):
                return target_project
            return None  # pragma: no cover

+    async def delete(self, entity_id: int) -> bool:
+        """Delete a project and its derived search rows in one transaction.
+
+        The cascade picture differs by backend:
+
+        - search_index → project: Postgres has ON DELETE CASCADE FK; SQLite
+          stores search_index as an FTS5 virtual table and can't carry FKs,
+          so it needs explicit cleanup.
+        - search_vector_chunks → project: neither backend has an FK here, so
+          both need an explicit DELETE.
+        - search_vector_embeddings → search_vector_chunks: Postgres has an FK
+          (chunk_id REFERENCES … ON DELETE CASCADE); SQLite stores embeddings
+          in a vec0 virtual table keyed by rowid with no cascade. On SQLite
+          the embeddings must be purged before the chunk rows, otherwise
+          `_run_vector_query` keeps returning stale vectors that crowd out
+          live results.
+
+        Each derived table is created lazily (search_index by
+        SearchRepository.init_search_index, the vector tables once semantic
+        search initializes), so any of them may be absent on minimal test DBs.
+        Inspect the connection once and skip whichever is missing.
+        """
+        logger.debug(f"Deleting Project and search rows for project_id: {entity_id}")
+        async with db.scoped_session(self.session_maker) as session:
+            try:
+                result = await session.execute(
+                    select(self.Model).filter(self.primary_key == entity_id)
+                )
+                project = result.scalars().one()
+            except NoResultFound:
+                logger.debug(f"No Project found to delete: {entity_id}")
+                return False
+
+            dialect_name = session.bind.dialect.name if session.bind else "sqlite"
+            is_sqlite = dialect_name == "sqlite"
+
+            existing_tables = await session.run_sync(
+                lambda sync_session: set(sa_inspect(sync_session.connection()).get_table_names())
+            )
+
+            # search_index: SQLite has no FK on the FTS5 virtual table; Postgres
+            # cascades from the project FK, so the explicit DELETE is redundant.
+            if is_sqlite and "search_index" in existing_tables:
+                await session.execute(
+                    text("DELETE FROM search_index WHERE project_id = :project_id"),
+                    {"project_id": entity_id},
+                )
+
+            # search_vector_chunks: no FK to project on either backend, so both
+            # backends need this. SQLite must purge vec0 embeddings first
+            # (rowid pseudocolumn — Postgres uses chunk_id and would 500 here);
+            # Postgres' chunk_id FK CASCADE handles its embeddings cleanup when
+            # we delete the chunk rows below.
+            if "search_vector_chunks" in existing_tables:
+                if is_sqlite and "search_vector_embeddings" in existing_tables:
+                    # Extension loading is per-connection. We must load vec0 on
+                    # *this* session before the DELETE; otherwise a different
+                    # pooled connection might have written embeddings that we'd
+                    # silently leave behind.
+                    if await _load_sqlite_vec_on_session(session):
+                        await session.execute(
+                            text(
+                                "DELETE FROM search_vector_embeddings WHERE rowid IN ("
+                                "SELECT id FROM search_vector_chunks "
+                                "WHERE project_id = :project_id)"
+                            ),
+                            {"project_id": entity_id},
+                        )
+                await session.execute(
+                    text("DELETE FROM search_vector_chunks WHERE project_id = :project_id"),
+                    {"project_id": entity_id},
+                )
+
+            await session.delete(project)
+            logger.debug(f"Deleted Project and search rows for project_id: {entity_id}")
+            return True
+
    async def update_path(self, project_id: int, new_path: str) -> Optional[Project]:
        """Update project path.

@@ -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()
@@ -88,6 +88,51 @@ class WorkspaceListResponse(BaseModel):
    )


+def workspace_matches_exact_identifier(workspace: WorkspaceInfo, identifier: str) -> bool:
+    """Return True when identifier matches workspace tenant_id, slug, or name."""
+    if workspace.tenant_id == identifier:
+        return True
+    if workspace.slug.casefold() == identifier.casefold():
+        return True
+    return workspace.name.casefold() == identifier.casefold()
+
+
+def workspace_matches_identifier(workspace: WorkspaceInfo, identifier: str) -> bool:
+    """Return True when identifier matches workspace tenant_id, slug, name, or type."""
+    return (
+        workspace_matches_exact_identifier(workspace, identifier)
+        or workspace.workspace_type.casefold() == identifier.casefold()
+    )
+
+
+def format_workspace_choices(workspaces: list[WorkspaceInfo]) -> str:
+    """Format deterministic workspace choices for prompt-style errors."""
+    return "\n".join(
+        [
+            (
+                f"- {item.name} "
+                f"(slug={item.slug}, type={item.workspace_type}, "
+                f"role={item.role}, tenant_id={item.tenant_id})"
+            )
+            for item in workspaces
+        ]
+    )
+
+
+def format_workspace_selection_choices(workspaces: list[WorkspaceInfo]) -> str:
+    """Format matching workspaces with copyable unique identifiers first."""
+    return "\n".join(
+        [
+            (
+                f"- {item.name} ({item.workspace_type}, role={item.role})\n"
+                f"  workspace: {item.slug}\n"
+                f"  tenant_id: {item.tenant_id}"
+            )
+            for item in workspaces
+        ]
+    )
+
+
 class CloudProjectIndexStatus(BaseModel):
    """Index freshness summary for one cloud project."""

@@ -4,7 +4,7 @@ from __future__ import annotations

 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from typing import List, Optional, Tuple, TYPE_CHECKING
+from typing import Any, List, Optional, Tuple, TYPE_CHECKING


 from loguru import logger
@@ -307,7 +307,7 @@ class ContextService:
        entity_id_values = ", ".join([str(i) for i in entity_ids])

        # Parameters for bindings - include project_id for security filtering
-        params = {
+        params: dict[str, Any] = {
            "max_depth": max_depth,
            "max_results": max_results,
            "project_id": self.search_repository.project_id,
@@ -322,9 +322,9 @@ class ContextService:
                since_utc = (
                    since.astimezone(timezone.utc) if since.tzinfo else since
                )  # pragma: no cover
-                params["since_date"] = since_utc.replace(tzinfo=None)  # pyright: ignore  # pragma: no cover
+                params["since_date"] = since_utc.replace(tzinfo=None)  # pragma: no cover
            else:
-                params["since_date"] = since.isoformat()  # pyright: ignore
+                params["since_date"] = since.isoformat()
            date_filter = "AND e.created_at >= :since_date"
            relation_date_filter = "AND e_from.created_at >= :since_date"
            timeframe_condition = "AND eg.relation_date >= :since_date"
@@ -1042,9 +1042,9 @@ class EntityService(BaseService[EntityModel]):
            for rel, resolved in zip(markdown.relations, resolved_entities):
                # Handle exceptions from gather and None results
                target_entity: Optional[Entity] = None
-                if not isinstance(resolved, Exception):
-                    # Type narrowing: resolved is Optional[Entity] here, not Exception
-                    target_entity = resolved  # pyright: ignore [reportAssignmentType]
+                if not isinstance(resolved, BaseException):
+                    # asyncio.gather(..., return_exceptions=True) can return any BaseException.
+                    target_entity = resolved

                if target_entity is None and not resolve_targets:
                    target_entity = await self._resolve_deferred_self_relation(rel.target, entity)
@@ -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
@@ -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
@@ -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",
@@ -1,34 +1,30 @@
-"""
-Integration test for long relation_type values (regression guard for issue #721).
+"""Integration test for long prose before inline wikilinks.

-When a markdown bullet contains an inline `[[wikilink]]` preceded by long prose,
-`parse_relation()` extracts ALL of that prose as the `relation_type`. Previously
-the response model `RelationType` had a `MaxLen(200)` constraint that caused
-edit_note (which round-trips through the response model when re-indexing) to
-fail with:
+Issue #721 was originally triggered by markdown bullets that contained inline
+`[[wikilinks]]` preceded by long prose. The parser treated all prose before the
+wikilink as `relation_type`, and the response model's former `MaxLen(200)`
+constraint caused edit_note to fail with:

    1 validation error for EntityResponseV2
    relations.0.relation_type
      String should have at most 200 characters

-Commit 01cbad1d removed the cap from `RelationType`. This test locks in that
-fix so a future contributor reintroducing `MaxLen` will see the test fail
-before shipping it.
-
-Out of scope: improving `parse_relation()` to fall back to a default relation
-type when the prose-before-link looks like a sentence rather than a label.
-That's a knowledge-graph-quality improvement, not a correctness fix, and is
-not required to keep edit_note working.
+The relation grammar now fixes the root ambiguity too: unquoted multi-word
+prefixes are prose, so this shape should index as a generic `links_to`
+relation rather than preserving prose as a custom relation type.
 """

 import pytest
 from fastmcp import Client

+from basic_memory.repository.relation_repository import RelationRepository
+

@pytest.mark.asyncio
-async def test_edit_note_handles_long_prose_around_wikilink(mcp_server, app, test_project):
-    """edit_note must succeed on notes whose inline wikilinks have >200 chars
-    of prose preceding them — that prose becomes the parsed relation_type."""
+async def test_edit_note_handles_long_prose_around_wikilink(
+    mcp_server, app, test_project, engine_factory
+):
+    """Long prose before an inline wikilink should not become a relation type."""
    long_prose = (
        "**Lorem ipsum dolor sit amet** — consectetur adipiscing elit, sed do eiusmod "
        "tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, "
@@ -75,3 +71,11 @@ async def test_edit_note_handles_long_prose_around_wikilink(mcp_server, app, tes
        )
        assert len(edit_result.content) == 1
        assert "Edited note (append)" in edit_result.content[0].text
+
+    _, session_maker = engine_factory
+    relation_repository = RelationRepository(session_maker, project_id=test_project.id)
+    links_to_relations = await relation_repository.find_by_type("links_to")
+    prose_type_relations = await relation_repository.find_by_type(long_prose)
+
+    assert any(relation.to_name == "Some Note Title" for relation in links_to_relations)
+    assert not prose_type_relations
@@ -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}"
+        )
@@ -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
@@ -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
@@ -482,6 +482,23 @@ async def test_import_missing_file(client: AsyncClient, v2_project_url: str):
    assert response.status_code in [400, 422]  # Either bad request or unprocessable entity


+@pytest.mark.asyncio
+async def test_import_rejects_oversized_file(
+    client: AsyncClient, tmp_path, app_config, v2_project_url: str
+):
+    """Import endpoints should reject files before parsing unbounded JSON."""
+    app_config.import_upload_max_bytes = 8
+    file_path = tmp_path / "large.json"
+    file_path.write_text(json.dumps([{"message": "too large"}]), encoding="utf-8")
+
+    with open(file_path, "rb") as f:
+        files = {"file": ("large.json", f, "application/json")}
+        response = await client.post(f"{v2_project_url}/import/chatgpt", files=files)
+
+    assert response.status_code == 413
+    assert "maximum size" in response.json()["detail"]
+
+
@pytest.mark.asyncio
 async def test_import_empty_file(client: AsyncClient, tmp_path, v2_project_url: str):
    """Test importing an empty file via v2 endpoint."""
@@ -302,7 +302,7 @@ async def test_create_entity_with_observations_and_relations(

 ## Observations
 - [note] This is a test observation #tag1 (context)
- related to [[OtherEntity]]
+- "related to" [[OtherEntity]]
 """,
    }

@@ -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",
    [
@@ -32,13 +32,48 @@ def test_convert_bmignore_to_rclone_filters_creates_and_converts(config_home):
    # Comments/empties preserved
    assert "# comment" in content
    assert "" in content
-    # Directory pattern becomes recursive exclude
+    # Plain and wildcard patterns exclude direct matches and recursive contents.
+    assert "- node_modules" in content
    assert "- node_modules/**" in content
-    # Wildcard pattern becomes simple exclude
    assert "- *.pyc" in content
+    assert "- *.pyc/**" in content
+    assert "- .git" in content
    assert "- .git/**" in content


+def test_convert_bmignore_to_rclone_filters_excludes_files_and_hidden_directory_contents(
+    config_home,
+):
+    bmignore = get_bmignore_path()
+    bmignore.parent.mkdir(parents=True, exist_ok=True)
+    bmignore.write_text("config.json\n.*\nnode_modules/**\n", encoding="utf-8")
+
+    rclone_filter = convert_bmignore_to_rclone_filters()
+    content = rclone_filter.read_text(encoding="utf-8").splitlines()
+
+    assert "- config.json" in content
+    assert "- config.json/**" in content
+    assert "- .*" in content
+    assert "- .*/**" in content
+    assert "- node_modules" in content
+    assert "- node_modules/**" in content
+
+
+def test_convert_bmignore_to_rclone_filters_preserves_directory_only_patterns(config_home):
+    bmignore = get_bmignore_path()
+    bmignore.parent.mkdir(parents=True, exist_ok=True)
+    bmignore.write_text("cache/\nconfig.json/**\n", encoding="utf-8")
+
+    rclone_filter = convert_bmignore_to_rclone_filters()
+    content = rclone_filter.read_text(encoding="utf-8").splitlines()
+
+    assert "- cache/" in content
+    assert "- cache/**" in content
+    assert "- cache" not in content
+    assert "- config.json" in content
+    assert "- config.json/**" in content
+
+
 def test_convert_bmignore_to_rclone_filters_is_cached_when_up_to_date(config_home):
    bmignore = get_bmignore_path()
    bmignore.parent.mkdir(parents=True, exist_ok=True)
@@ -0,0 +1,54 @@
+"""Tests for secure rclone installer fallbacks."""
+
+import pytest
+
+from basic_memory.cli.commands.cloud import rclone_installer
+
+
+def test_macos_installer_does_not_fallback_to_remote_script(monkeypatch):
+    """Homebrew failure should produce manual guidance, not curl-piped sudo bash."""
+    commands: list[list[str]] = []
+
+    def fake_which(command: str) -> str | None:
+        return "/opt/homebrew/bin/brew" if command == "brew" else None
+
+    def fake_run(command: list[str], check: bool = True):
+        commands.append(command)
+        raise rclone_installer.RcloneInstallError("brew failed")
+
+    monkeypatch.setattr(rclone_installer.shutil, "which", fake_which)
+    monkeypatch.setattr(rclone_installer, "run_command", fake_run)
+
+    with pytest.raises(rclone_installer.RcloneInstallError) as exc_info:
+        rclone_installer.install_rclone_macos()
+
+    assert commands == [["brew", "install", "rclone"]]
+    assert "curl" not in str(exc_info.value)
+    assert "sudo bash" not in str(exc_info.value)
+    assert "brew install rclone" in str(exc_info.value)
+
+
+def test_linux_installer_uses_package_managers_only(monkeypatch):
+    """Linux package-manager failures should not fall through to remote script execution."""
+    commands: list[list[str]] = []
+
+    def fake_which(command: str) -> str | None:
+        return f"/usr/bin/{command}" if command in {"apt", "snap"} else None
+
+    def fake_run(command: list[str], check: bool = True):
+        commands.append(command)
+        raise rclone_installer.RcloneInstallError("install failed")
+
+    monkeypatch.setattr(rclone_installer.shutil, "which", fake_which)
+    monkeypatch.setattr(rclone_installer, "run_command", fake_run)
+
+    with pytest.raises(rclone_installer.RcloneInstallError) as exc_info:
+        rclone_installer.install_rclone_linux()
+
+    assert commands == [
+        ["sudo", "apt", "update"],
+        ["sudo", "snap", "install", "rclone"],
+    ]
+    assert all("curl" not in token for command in commands for token in command)
+    assert "sudo bash" not in str(exc_info.value)
+    assert "sudo apt install rclone" in str(exc_info.value)
@@ -80,3 +80,26 @@ def test_bm_version_does_not_import_heavy_modules():
    assert "CLEAN" in result.stdout, (
        f"Heavy modules loaded during --version: {result.stdout.strip()}"
    )
+
+
+def test_bm_help_does_not_import_api_app():
+    """Regression test: 'bm --help' must not build the FastAPI app graph."""
+    check_script = (
+        "import sys; "
+        "sys.argv = ['bm', '--help']; "
+        "import basic_memory.cli.main; "
+        "heavy = [m for m in sys.modules "
+        "if m == 'basic_memory.api.app' or m.startswith('basic_memory.api.v2.routers')]; "
+        "print(','.join(heavy) if heavy else 'CLEAN')"
+    )
+    result = subprocess.run(
+        ["uv", "run", "python", "-c", check_script],
+        capture_output=True,
+        text=True,
+        timeout=10,
+        cwd=Path(__file__).parent.parent.parent,
+    )
+    assert result.returncode == 0
+    assert "CLEAN" in result.stdout, (
+        f"API app modules loaded during --help: {result.stdout.strip()}"
+    )
@@ -10,6 +10,7 @@ from typer.testing import CliRunner

 from basic_memory.cli.app import app
 from basic_memory.mcp.clients.project import ProjectClient
+from basic_memory.schemas.cloud import WorkspaceInfo
 from basic_memory.schemas.project_info import ProjectList

 # Importing registers project subcommands on the shared app instance.
@@ -53,6 +54,26 @@ def mock_client(monkeypatch):
    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)


+def _workspace(
+    *,
+    tenant_id: str,
+    slug: str,
+    name: str,
+    workspace_type: str,
+    is_default: bool = False,
+) -> WorkspaceInfo:
+    return WorkspaceInfo(
+        tenant_id=tenant_id,
+        workspace_type=workspace_type,
+        slug=slug,
+        name=name,
+        role="owner",
+        is_default=is_default,
+        organization_id=None,
+        has_active_subscription=True,
+    )
+
+
 def test_project_list_shows_local_cloud_presence_and_routes(
    runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch
 ):
@@ -215,6 +236,680 @@ def test_project_list_shows_display_name_for_private_projects(
    assert private_project["display_name"] == "My Project"


+def test_project_list_cloud_fetches_all_workspaces_and_labels_duplicate_permalinks(
+    runner: CliRunner, write_config, monkeypatch
+):
+    """Cloud project list should include every workspace without collapsing matching names."""
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+    team = _workspace(
+        tenant_id="tenant-team",
+        slug="team",
+        name="Team",
+        workspace_type="organization",
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal, team]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    payloads_by_workspace = {
+        None: {"projects": [], "default_project": None},
+        "tenant-personal": {
+            "projects": [
+                {
+                    "id": 1,
+                    "external_id": "11111111-1111-1111-1111-111111111111",
+                    "name": "shared",
+                    "path": "/personal/shared",
+                    "is_default": True,
+                }
+            ],
+            "default_project": "shared",
+        },
+        "tenant-team": {
+            "projects": [
+                {
+                    "id": 2,
+                    "external_id": "22222222-2222-2222-2222-222222222222",
+                    "name": "shared",
+                    "path": "/team/shared",
+                    "is_default": False,
+                }
+            ],
+            "default_project": None,
+        },
+    }
+    seen_workspaces: list[str | None] = []
+
+    async def fake_list_projects(self):
+        workspace = self.http_client.workspace
+        seen_workspaces.append(workspace)
+        return ProjectList.model_validate(payloads_by_workspace[workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"})
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    # The initial None call is the local project fetch before cloud workspaces are listed.
+    assert seen_workspaces == [None, "tenant-personal", "tenant-team"]
+
+    data = json.loads(result.stdout)
+    shared_projects = [project for project in data["projects"] if project["name"] == "shared"]
+    assert len(shared_projects) == 2
+    assert {project["cloud_path"] for project in shared_projects} == {
+        "/personal/shared",
+        "/team/shared",
+    }
+    assert {project["workspace"] for project in shared_projects} == {"Personal", "Team"}
+    assert {project["workspace_type"] for project in shared_projects} == {
+        "personal",
+        "organization",
+    }
+
+    table_result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"})
+
+    assert table_result.exit_code == 0
+    assert "Personal (personal)" in table_result.stdout
+    assert "Team (organization)" in table_result.stdout
+
+
+def test_project_list_workspace_discovery_failure_warns_and_uses_fallback(
+    runner: CliRunner, write_config, monkeypatch
+):
+    """Workspace discovery failures should fall back and explain the degraded result."""
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "default_workspace": "tenant-default",
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    async def fail_get_available_workspaces():
+        raise RuntimeError("workspace service unavailable")
+
+    payloads_by_workspace = {
+        None: {"projects": [], "default_project": None},
+        "tenant-default": {
+            "projects": [
+                {
+                    "id": 1,
+                    "external_id": "11111111-1111-1111-1111-111111111111",
+                    "name": "fallback-project",
+                    "path": "/fallback-project",
+                    "is_default": False,
+                }
+            ],
+            "default_project": None,
+        },
+    }
+
+    async def fake_list_projects(self):
+        return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fail_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"})
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    assert "fallback-project" in result.stdout
+    assert "Cloud workspace discovery failed: workspace service unavailable" in result.stdout
+    assert "Showing cloud projects from the configured/default workspace only" in result.stdout
+
+
+def test_project_list_partial_workspace_failure_warns_and_keeps_successes(
+    runner: CliRunner, write_config, monkeypatch
+):
+    """A failed workspace fetch should not hide projects from successful workspaces."""
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+    team = _workspace(
+        tenant_id="tenant-team",
+        slug="team",
+        name="Team",
+        workspace_type="organization",
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal, team]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    payloads_by_workspace = {
+        None: {"projects": [], "default_project": None},
+        "tenant-personal": {
+            "projects": [
+                {
+                    "id": 1,
+                    "external_id": "11111111-1111-1111-1111-111111111111",
+                    "name": "personal-project",
+                    "path": "/personal-project",
+                    "is_default": False,
+                }
+            ],
+            "default_project": None,
+        },
+    }
+
+    async def fake_list_projects(self):
+        if self.http_client.workspace == "tenant-team":
+            raise RuntimeError("team unavailable")
+        return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"})
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    assert "personal-project" in result.stdout
+    assert "Cloud project discovery failed for workspace Team: team unavailable" in result.stdout
+
+
+def test_project_list_workspace_type_filter_selects_unique_workspace(
+    runner: CliRunner, write_config, monkeypatch
+):
+    """--workspace can use a workspace type when it resolves to one workspace."""
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+    team = _workspace(
+        tenant_id="tenant-team",
+        slug="team",
+        name="Team",
+        workspace_type="organization",
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal, team]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    payloads_by_workspace = {
+        None: {"projects": [], "default_project": None},
+        "tenant-team": {
+            "projects": [
+                {
+                    "id": 1,
+                    "external_id": "11111111-1111-1111-1111-111111111111",
+                    "name": "team-project",
+                    "path": "/team-project",
+                    "is_default": True,
+                }
+            ],
+            "default_project": "team-project",
+        },
+    }
+    seen_workspaces: list[str | None] = []
+
+    async def fake_list_projects(self):
+        workspace = self.http_client.workspace
+        seen_workspaces.append(workspace)
+        return ProjectList.model_validate(payloads_by_workspace[workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(
+        app,
+        ["project", "list", "--workspace", "organization", "--json"],
+        env={"COLUMNS": "240"},
+    )
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    assert seen_workspaces == [None, "tenant-team"]
+    data = json.loads(result.stdout)
+    assert [project["name"] for project in data["projects"]] == ["team-project"]
+    assert data["projects"][0]["workspace"] == "Team"
+
+
+def test_project_list_workspace_type_filter_lists_ambiguous_matches(
+    runner: CliRunner, write_config, monkeypatch
+):
+    """Ambiguous workspace type filters should list copyable matching slugs."""
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    async def fake_get_available_workspaces():
+        return [
+            _workspace(
+                tenant_id="tenant-personal",
+                slug="personal",
+                name="Personal",
+                workspace_type="personal",
+                is_default=True,
+            ),
+            _workspace(
+                tenant_id="tenant-team-alpha",
+                slug="team-alpha",
+                name="Team Alpha",
+                workspace_type="organization",
+            ),
+            _workspace(
+                tenant_id="tenant-team-beta",
+                slug="team-beta",
+                name="Team Beta",
+                workspace_type="organization",
+            ),
+        ]
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield object()
+
+    async def fake_list_projects(self):
+        return ProjectList.model_validate({"projects": [], "default_project": None})
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(
+        app,
+        ["project", "list", "--workspace", "organization"],
+        env={"COLUMNS": "240"},
+    )
+
+    assert result.exit_code == 1
+    assert "Workspace 'organization' matches multiple workspaces" in result.stdout
+    assert "Choose one of these matching workspaces by slug" in result.stdout
+    assert "workspace: team-alpha" in result.stdout
+    assert "workspace: team-beta" in result.stdout
+    assert "tenant_id: tenant-team-alpha" in result.stdout
+    assert "tenant_id: tenant-team-beta" in result.stdout
+    assert "workspace: personal" not in result.stdout
+
+
+def test_project_list_invalid_workspace_exits_without_local_fallback(
+    runner: CliRunner, write_config, tmp_path, monkeypatch
+):
+    """Invalid explicit workspace filters should stop instead of showing local-only rows."""
+    local_path = (tmp_path / "main").as_posix()
+    write_config(
+        {
+            "env": "dev",
+            "projects": {"main": {"path": local_path, "mode": "local"}},
+            "default_project": "main",
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    async def fake_list_projects(self):
+        return ProjectList.model_validate(
+            {
+                "projects": [
+                    {
+                        "id": 1,
+                        "external_id": "11111111-1111-1111-1111-111111111111",
+                        "name": "main",
+                        "path": local_path,
+                        "is_default": True,
+                    }
+                ],
+                "default_project": "main",
+            }
+        )
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(
+        app,
+        ["project", "list", "--workspace", "missing"],
+        env={"COLUMNS": "240"},
+    )
+
+    assert result.exit_code == 1
+    assert "Workspace 'missing' not found" in result.stdout
+    assert "Basic Memory Projects" not in result.stdout
+    assert "Cloud project discovery failed" not in result.stdout
+
+
+def test_project_list_attaches_local_state_to_one_duplicate_cloud_project(
+    runner: CliRunner, write_config, tmp_path, monkeypatch
+):
+    """Local path/default/sync state should not appear on every matching workspace."""
+    local_path = (tmp_path / "main").as_posix()
+    write_config(
+        {
+            "env": "dev",
+            "projects": {
+                "main": {
+                    "path": local_path,
+                    "mode": "local",
+                    "local_sync_path": local_path,
+                }
+            },
+            "default_project": "main",
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+    team = _workspace(
+        tenant_id="tenant-team",
+        slug="team",
+        name="Team",
+        workspace_type="organization",
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal, team]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    payloads_by_workspace = {
+        None: {
+            "projects": [
+                {
+                    "id": 1,
+                    "external_id": "11111111-1111-1111-1111-111111111111",
+                    "name": "main",
+                    "path": local_path,
+                    "is_default": True,
+                }
+            ],
+            "default_project": "main",
+        },
+        "tenant-personal": {
+            "projects": [
+                {
+                    "id": 2,
+                    "external_id": "22222222-2222-2222-2222-222222222222",
+                    "name": "main",
+                    "path": "/basic-memory",
+                    "is_default": True,
+                }
+            ],
+            "default_project": "main",
+        },
+        "tenant-team": {
+            "projects": [
+                {
+                    "id": 3,
+                    "external_id": "33333333-3333-3333-3333-333333333333",
+                    "name": "main",
+                    "path": "/basic-memory",
+                    "is_default": True,
+                }
+            ],
+            "default_project": "main",
+        },
+    }
+
+    async def fake_list_projects(self):
+        return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"})
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    data = json.loads(result.stdout)
+    main_rows = [project for project in data["projects"] if project["name"] == "main"]
+    assert len(main_rows) == 2
+
+    personal_row = next(project for project in main_rows if project["workspace"] == "Personal")
+    team_row = next(project for project in main_rows if project["workspace"] == "Team")
+
+    assert personal_row["local_path"] == project_cmd.format_path(local_path)
+    assert personal_row["cli_route"] == "local"
+    assert personal_row["mcp_stdio"] == "stdio"
+    assert personal_row["sync"] is True
+    assert personal_row["is_default"] is True
+
+    assert team_row["local_path"] == ""
+    assert team_row["cli_route"] == "cloud"
+    assert team_row["mcp_stdio"] == "https"
+    assert team_row["sync"] is False
+    assert team_row["is_default"] is False
+
+    filtered_result = runner.invoke(
+        app,
+        ["project", "list", "--workspace", "organization", "--json"],
+        env={"COLUMNS": "240"},
+    )
+
+    assert filtered_result.exit_code == 0, (
+        f"Exit code: {filtered_result.exit_code}, output: {filtered_result.stdout}"
+    )
+    filtered_data = json.loads(filtered_result.stdout)
+    assert len(filtered_data["projects"]) == 1
+    filtered_team_row = filtered_data["projects"][0]
+    assert filtered_team_row["workspace"] == "Team"
+    assert filtered_team_row["local_path"] == ""
+    assert filtered_team_row["cli_route"] == "cloud"
+    assert filtered_team_row["mcp_stdio"] == "https"
+    assert filtered_team_row["sync"] is False
+    assert filtered_team_row["is_default"] is False
+
+
+def test_project_list_hides_bisync_flag_for_attached_team_workspace(
+    runner: CliRunner, write_config, tmp_path, monkeypatch
+):
+    """Bisync is only supported for personal workspaces."""
+    local_path = (tmp_path / "team-main").as_posix()
+    write_config(
+        {
+            "env": "dev",
+            "projects": {
+                "main": {
+                    "path": local_path,
+                    "mode": "cloud",
+                    "workspace_id": "tenant-team",
+                    "local_sync_path": local_path,
+                }
+            },
+            "default_project": "main",
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    personal = _workspace(
+        tenant_id="tenant-personal",
+        slug="personal",
+        name="Personal",
+        workspace_type="personal",
+        is_default=True,
+    )
+    team = _workspace(
+        tenant_id="tenant-team",
+        slug="team",
+        name="Team",
+        workspace_type="organization",
+    )
+
+    async def fake_get_available_workspaces():
+        return [personal, team]
+
+    class FakeClient:
+        def __init__(self, workspace: str | None):
+            self.workspace = workspace
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield FakeClient(workspace)
+
+    project_payload = {
+        "projects": [
+            {
+                "id": 1,
+                "external_id": "11111111-1111-1111-1111-111111111111",
+                "name": "main",
+                "path": "/basic-memory",
+                "is_default": True,
+            }
+        ],
+        "default_project": "main",
+    }
+    payloads_by_workspace = {
+        None: {"projects": [], "default_project": None},
+        "tenant-personal": project_payload,
+        "tenant-team": project_payload,
+    }
+
+    async def fake_list_projects(self):
+        return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace])
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects)
+
+    result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"})
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
+    data = json.loads(result.stdout)
+    main_rows = [project for project in data["projects"] if project["name"] == "main"]
+    team_row = next(project for project in main_rows if project["workspace"] == "Team")
+
+    assert team_row["cli_route"] == "cloud"
+    assert team_row["mcp_stdio"] == "https"
+    assert team_row["sync"] is False
+    assert team_row["is_default"] is True
+
+
 def test_project_ls_local_mode_defaults_to_local_route(
    runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch
 ):
@@ -182,6 +182,46 @@ class TestWorkspaceSetDefault:
        assert result.exit_code == 1
        assert "not found" in result.stdout

+    def test_set_default_workspace_ambiguous_type_lists_matching_choices(self, runner, monkeypatch):
+        async def fake_get_available_workspaces(context=None):
+            return [
+                _workspace(
+                    tenant_id="11111111-1111-1111-1111-111111111111",
+                    workspace_type="personal",
+                    slug="personal",
+                    name="Personal",
+                    role="owner",
+                    is_default=True,
+                ),
+                _workspace(
+                    tenant_id="22222222-2222-2222-2222-222222222222",
+                    workspace_type="organization",
+                    slug="team-alpha",
+                    name="Team Alpha",
+                    role="editor",
+                ),
+                _workspace(
+                    tenant_id="33333333-3333-3333-3333-333333333333",
+                    workspace_type="organization",
+                    slug="team-beta",
+                    name="Team Beta",
+                    role="owner",
+                ),
+            ]
+
+        monkeypatch.setattr(
+            workspace_cmd, "get_available_workspaces", fake_get_available_workspaces
+        )
+
+        result = runner.invoke(app, ["cloud", "workspace", "set-default", "organization"])
+
+        assert result.exit_code == 1
+        assert "Workspace 'organization' matches multiple workspaces" in result.stdout
+        assert "Choose one of these matching workspaces by slug" in result.stdout
+        assert "workspace: team-alpha" in result.stdout
+        assert "workspace: team-beta" in result.stdout
+        assert "workspace: personal" not in result.stdout
+
    def test_set_default_workspace_no_workspaces(self, runner, monkeypatch):
        async def fake_get_available_workspaces(context=None):
            return []
@@ -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."""
@@ -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)
@@ -128,3 +128,51 @@ def test_unicode_targets():
    assert relation.type == "type"
    assert relation.target == "Target"
    assert relation.context == "测试"
+
+
+def test_quoted_multi_word_relation_type_parses_as_explicit_relation():
+    """Quoted relation labels allow explicit multi-word relation types."""
+    md = MarkdownIt().use(relation_plugin)
+
+    tokens = md.parse('- "some type" [[Target]] (context)')
+    token = next(t for t in tokens if t.type == "inline")
+
+    assert token.meta["relations"] == [
+        {"type": "some type", "target": "Target", "context": "context"}
+    ]
+    assert parse_relation(token) == {"type": "some type", "target": "Target", "context": "context"}
+
+
+def test_single_quoted_multi_word_relation_type_parses_as_explicit_relation():
+    """Single-quoted relation labels also allow explicit multi-word relation types."""
+    md = MarkdownIt().use(relation_plugin)
+
+    tokens = md.parse("- 'some type' [[Target]] (context)")
+    token = next(t for t in tokens if t.type == "inline")
+
+    assert token.meta["relations"] == [
+        {"type": "some type", "target": "Target", "context": "context"}
+    ]
+    assert parse_relation(token) == {"type": "some type", "target": "Target", "context": "context"}
+
+
+def test_unquoted_multi_word_prefix_is_inline_link_not_relation_type():
+    """Unquoted prose before a wikilink is an inline link, not a relation type."""
+    md = MarkdownIt().use(relation_plugin)
+
+    tokens = md.parse("- some other thing [[Target]]")
+    token = next(t for t in tokens if t.type == "inline")
+
+    assert token.meta["relations"] == [{"type": "links_to", "target": "Target", "context": None}]
+    assert parse_relation(token) is None
+
+
+def test_bare_list_wikilink_is_inline_link_not_default_explicit_relation():
+    """A list item containing only a wikilink is still a generic inline link."""
+    md = MarkdownIt().use(relation_plugin)
+
+    tokens = md.parse("- [[Target]]")
+    token = next(t for t in tokens if t.type == "inline")
+
+    assert token.meta["relations"] == [{"type": "links_to", "target": "Target", "context": None}]
+    assert parse_relation(token) is None
@@ -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
@@ -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()
@@ -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
@@ -340,6 +322,106 @@ async def test_workspace_invalid_selection_lists_choices(monkeypatch):
        await resolve_workspace_parameter(workspace="missing-workspace")


+@pytest.mark.asyncio
+async def test_workspace_ambiguous_type_selection_lists_matching_choices(monkeypatch):
+    from basic_memory.mcp.project_context import resolve_workspace_parameter
+
+    workspaces = [
+        _workspace(
+            tenant_id="11111111-1111-1111-1111-111111111111",
+            workspace_type="personal",
+            slug="personal",
+            name="Personal",
+            role="owner",
+            is_default=True,
+        ),
+        _workspace(
+            tenant_id="22222222-2222-2222-2222-222222222222",
+            workspace_type="organization",
+            slug="team-alpha",
+            name="Team Alpha",
+            role="editor",
+        ),
+        _workspace(
+            tenant_id="33333333-3333-3333-3333-333333333333",
+            workspace_type="organization",
+            slug="team-beta",
+            name="Team Beta",
+            role="owner",
+        ),
+    ]
+
+    async def fake_get_available_workspaces(context=None):
+        return workspaces
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+
+    with pytest.raises(ValueError) as exc_info:
+        await resolve_workspace_parameter(workspace="organization")
+
+    message = str(exc_info.value)
+    assert "Workspace 'organization' matches multiple workspaces" in message
+    assert "workspace: team-alpha" in message
+    assert "workspace: team-beta" in message
+    assert "workspace: personal" not in message
+
+
+@pytest.mark.asyncio
+async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(monkeypatch):
+    from basic_memory.mcp.project_context import resolve_workspace_parameter
+
+    cached_workspace = _workspace(
+        tenant_id="22222222-2222-2222-2222-222222222222",
+        workspace_type="organization",
+        slug="team-alpha",
+        name="Team Alpha",
+        role="editor",
+    )
+    workspaces = [
+        _workspace(
+            tenant_id="11111111-1111-1111-1111-111111111111",
+            workspace_type="personal",
+            slug="personal",
+            name="Personal",
+            role="owner",
+            is_default=True,
+        ),
+        cached_workspace,
+        _workspace(
+            tenant_id="33333333-3333-3333-3333-333333333333",
+            workspace_type="organization",
+            slug="team-beta",
+            name="Team Beta",
+            role="owner",
+        ),
+    ]
+    context = ContextState()
+    await context.set_state("active_workspace", cached_workspace.model_dump())
+    fetches = 0
+
+    async def fake_get_available_workspaces(context=None):
+        nonlocal fetches
+        fetches += 1
+        return workspaces
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        fake_get_available_workspaces,
+    )
+
+    with pytest.raises(ValueError) as exc_info:
+        await resolve_workspace_parameter(workspace="organization", context=ctx(context))
+
+    message = str(exc_info.value)
+    assert fetches == 1
+    assert "Workspace 'organization' matches multiple workspaces" in message
+    assert "workspace: team-alpha" in message
+    assert "workspace: team-beta" in message
+
+
@pytest.mark.asyncio
 async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
@@ -352,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
@@ -363,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


@@ -376,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",
@@ -413,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",
@@ -423,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"]


@@ -439,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",
@@ -472,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),
        )


@@ -700,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",
@@ -711,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,
@@ -945,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",
@@ -981,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"
@@ -1132,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."""
@@ -1170,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",
@@ -1188,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


@@ -1202,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():
@@ -1214,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"
@@ -1227,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",
@@ -1245,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


@@ -1254,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",
@@ -1275,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

@@ -1291,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",
@@ -1316,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
@@ -1338,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",
@@ -1364,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
@@ -1386,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",
@@ -1413,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
@@ -1432,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",
@@ -1445,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
@@ -1468,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",
@@ -1499,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
@@ -1510,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
@@ -1550,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
@@ -1571,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",
@@ -1605,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
@@ -1625,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",
@@ -1664,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"
@@ -1795,7 +2287,7 @@ class TestGetProjectClientRoutingOrder:
        )
        config_manager.save_config(config)

-        context = _ContextState()
+        context = ContextState()
        stale_workspace = _workspace(
            tenant_id="team-tenant",
            workspace_type="organization",
@@ -1842,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
@@ -1890,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):
@@ -1934,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"

@@ -2054,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
@@ -2093,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"

@@ -2235,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",
@@ -2276,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"

@@ -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 == [
@@ -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,
@@ -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 ---


@@ -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:
@@ -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"""
@@ -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."""
@@ -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
@@ -7,6 +7,7 @@ import pytest

 from basic_memory import config as config_module
 from basic_memory.mcp.tools import write_note, read_note, delete_note
+from basic_memory.repository.relation_repository import RelationRepository
 from basic_memory.utils import normalize_newlines


@@ -309,7 +310,7 @@ async def test_write_note_with_tag_array_from_bug_report(app, test_project):


@pytest.mark.asyncio
-async def test_write_note_verbose(app, test_project):
+async def test_write_note_verbose(app, test_project, engine_factory):
    """Test creating a new note.

    Should:
@@ -326,7 +327,7 @@ async def test_write_note_verbose(app, test_project):
 # Test\nThis is a test note

 - [note] First observation
- relates to [[Knowledge]]
+- "relates to" [[Knowledge]]

 """,
        tags=["test", "documentation"],
@@ -343,6 +344,11 @@ async def test_write_note_verbose(app, test_project):
    assert "- test, documentation" in result
    assert f"[Session: Using project '{test_project.name}']" in result

+    _, session_maker = engine_factory
+    relation_repository = RelationRepository(session_maker, project_id=test_project.id)
+    relations = await relation_repository.find_by_type("relates to")
+    assert any(relation.to_name == "Knowledge" for relation in relations)
+

@pytest.mark.asyncio
 async def test_write_note_preserves_custom_metadata(app, project_config, test_project):
@@ -8,8 +8,38 @@ import pytest
 from basic_memory.schemas.search import SearchItemType, SearchResponse, SearchResult


+def _stub_routing_mode(monkeypatch, *, cloud: bool) -> None:
+    """Pin the three cloud-route signals search.py reads.
+
+    `_search_all_projects` only forwards project_id (external UUID) when a
+    cloud route is available. The composite mirrors get_project_client:
+    factory mode OR explicit --cloud OR has_cloud_credentials. Tests stub
+    all three so a dev box with OAuth tokens on disk can't bleed into the
+    local-mode case.
+    """
+    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
+    monkeypatch.setattr(search_mod, "is_factory_mode", lambda: False)
+    monkeypatch.setattr(search_mod, "_explicit_routing", lambda: cloud)
+    monkeypatch.setattr(search_mod, "_force_local_mode", lambda: False)
+    monkeypatch.setattr(search_mod, "has_cloud_credentials", lambda config: cloud)
+
+
+@pytest.fixture
+def cloud_routing(monkeypatch):
+    """Force the cloud-routing path for multi-project search tests."""
+    _stub_routing_mode(monkeypatch, cloud=True)
+
+
+@pytest.fixture
+def local_routing(monkeypatch):
+    """Force the local-routing path for multi-project search tests."""
+    _stub_routing_mode(monkeypatch, cloud=False)
+
+
@pytest.mark.asyncio
-async def test_search_notes_search_all_projects_qualifies_result_permalinks(monkeypatch):
+async def test_search_notes_search_all_projects_qualifies_result_permalinks(
+    monkeypatch, cloud_routing
+):
    """Multi-project search belongs to search_notes and keeps result ids routable."""
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
@@ -169,7 +199,7 @@ async def test_search_notes_search_all_projects_with_no_refs_returns_empty_all_p

@pytest.mark.asyncio
 async def test_search_notes_search_all_projects_continues_after_project_failure(
-    monkeypatch,
+    monkeypatch, cloud_routing
 ):
    """One failing project should not discard successful all-project search results."""
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
@@ -254,3 +284,85 @@ async def test_search_notes_search_all_projects_continues_after_project_failure(
    assert result["total"] == 1
    assert any("team-paul/main" in warning for warning in warnings)
    assert any("team index unavailable" in warning for warning in warnings)
+
+
+@pytest.mark.asyncio
+async def test_search_notes_search_all_projects_local_omits_project_id(
+    monkeypatch, local_routing
+):
+    """Without a cloud route, fan-out must address each project by name only.
+
+    project_id (external UUID) routes through the cloud v2 API path, which
+    returns 401 on local installs because there's no JWT to present. Local
+    fan-out has to fall back to the name-routed path so each per-project
+    search actually returns results instead of silently failing.
+    """
+    clients_mod = importlib.import_module("basic_memory.mcp.clients")
+    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
+
+    project_refs = [
+        {
+            "project": "alpha",
+            "project_id": "11111111-1111-1111-1111-111111111111",
+        },
+        {
+            "project": "beta",
+            "project_id": "22222222-2222-2222-2222-222222222222",
+        },
+    ]
+    searched_projects: list[tuple[str | None, str | None]] = []
+
+    async def fake_load_search_project_refs(context=None):
+        return project_refs
+
+    class StubProject:
+        def __init__(self, name: str | None, external_id: str | None):
+            self.name = name or "main"
+            self.external_id = external_id or "local-main"
+
+    @asynccontextmanager
+    async def fake_get_project_client(project=None, context=None, project_id=None):
+        searched_projects.append((project, project_id))
+        yield object(), StubProject(project, project_id)
+
+    async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
+        return StubProject(project, None), identifier, False
+
+    class MockSearchClient:
+        def __init__(self, client, project_id):
+            self.project_id = project_id
+
+        async def search(self, payload, page, page_size):
+            return SearchResponse(
+                results=[
+                    SearchResult(
+                        title=f"Note in {self.project_id or 'local'}",
+                        permalink="notes/example",
+                        content="",
+                        type=SearchItemType.ENTITY,
+                        score=0.5,
+                        file_path="/notes/example.md",
+                    )
+                ],
+                current_page=page,
+                page_size=page_size,
+                total=1,
+            )
+
+    monkeypatch.setattr(search_mod, "_load_search_project_refs", fake_load_search_project_refs)
+    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", MockSearchClient)
+
+    result = await search_mod.search_notes(
+        query="anything",
+        search_all_projects=True,
+        output_format="json",
+    )
+
+    assert isinstance(result, dict)
+    assert searched_projects == [("alpha", None), ("beta", None)], (
+        "Local fan-out must omit project_id so the recursive search_notes calls "
+        "take the name-routed path."
+    )
+    assert result["total"] == 2
@@ -1006,7 +1006,7 @@ async def test_edit_entity_with_observations_and_relations(
        # Test Note
        
        - [note] This is an observation
-        - links to [[Other Entity]]
+        - "links to" [[Other Entity]]
        
        Original content
        """).strip()
@@ -1028,7 +1028,7 @@ async def test_edit_entity_with_observations_and_relations(
    updated = await entity_service.edit_entity(
        identifier=_permalink(entity),
        operation="append",
-        content="\n- [category] New observation\n- relates to [[New Entity]]",
+        content='\n- [category] New observation\n- "relates to" [[New Entity]]',
    )

    # Verify observations and relations were updated
@@ -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."""
@@ -6,7 +6,9 @@ from datetime import timezone, datetime
 from pathlib import Path

 import pytest
+from sqlalchemy import text

+from basic_memory import db
 from basic_memory.services.project_service import ProjectService


@@ -136,3 +138,215 @@ async def test_remove_project_with_related_entities(project_service: ProjectServ
                    project = await project_service.get_project(test_project_name)
                    if project:
                        await project_service.repository.delete(project.id)
+
+
+async def _table_exists(session_maker, table: str) -> bool:
+    """Return True if the named table is present on the current connection."""
+    from sqlalchemy import inspect as sa_inspect
+
+    async with db.scoped_session(session_maker) as session:
+        return await session.run_sync(
+            lambda sync_session: table in sa_inspect(sync_session.connection()).get_table_names()
+        )
+
+
+@pytest.mark.asyncio
+async def test_remove_project_purges_search_rows(project_service: ProjectService):
+    """Project deletion must sweep the derived search tables.
+
+    SQLite stores search_index as an FTS5 virtual table, which cannot carry a
+    foreign key, so without an explicit purge the FTS rows survive the project
+    and leak into the next project that reuses the same auto-increment id.
+    Postgres has the cascade FK, but we expect the same end-state on either
+    backend. This test fails on the pre-fix code: search_index still holds the
+    project's rows after remove_project completes.
+    """
+    test_project_name = f"test-search-cleanup-{os.urandom(4).hex()}"
+    with tempfile.TemporaryDirectory() as temp_dir:
+        test_project_path = str(Path(temp_dir) / "test-search-cleanup")
+        os.makedirs(test_project_path, exist_ok=True)
+
+        await project_service.add_project(test_project_name, test_project_path)
+        project = await project_service.get_project(test_project_name)
+        assert project is not None
+        project_id = project.id
+
+        # Seed both derived tables directly. The bug is in the cleanup path,
+        # not the indexer, so a synthetic row is enough to prove the sweep.
+        async with db.scoped_session(project_service.repository.session_maker) as session:
+            await session.execute(
+                text(
+                    "INSERT INTO search_index "
+                    "(id, title, content_stems, content_snippet, permalink, "
+                    " file_path, type, project_id) "
+                    "VALUES (:id, :title, :stems, :snippet, :permalink, "
+                    " :file_path, :type, :project_id)"
+                ),
+                {
+                    "id": 999_001,
+                    "title": "leak canary",
+                    "stems": "leak canary",
+                    "snippet": "leak canary",
+                    "permalink": f"leak-canary-{project_id}",
+                    "file_path": "leak-canary.md",
+                    "type": "entity",
+                    "project_id": project_id,
+                },
+            )
+            await session.execute(
+                text(
+                    "INSERT INTO search_vector_chunks "
+                    "(entity_id, project_id, chunk_key, chunk_text, source_hash, "
+                    " entity_fingerprint, embedding_model) "
+                    "VALUES (:entity_id, :project_id, :chunk_key, :chunk_text, "
+                    " :source_hash, :entity_fingerprint, :embedding_model)"
+                ),
+                {
+                    "entity_id": 999_001,
+                    "project_id": project_id,
+                    "chunk_key": "canary",
+                    "chunk_text": "leak canary",
+                    "source_hash": "abc",
+                    "entity_fingerprint": "",
+                    "embedding_model": "",
+                },
+            )
+
+        async with db.scoped_session(project_service.repository.session_maker) as session:
+            pre_index = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_index WHERE project_id = :pid"),
+                    {"pid": project_id},
+                )
+            ).scalar_one()
+            pre_chunks = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :pid"),
+                    {"pid": project_id},
+                )
+            ).scalar_one()
+        assert pre_index >= 1, "seed row should exist before removal"
+        assert pre_chunks >= 1, "seed chunk should exist before removal"
+
+        await project_service.remove_project(test_project_name)
+
+        async with db.scoped_session(project_service.repository.session_maker) as session:
+            post_index = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_index WHERE project_id = :pid"),
+                    {"pid": project_id},
+                )
+            ).scalar_one()
+            post_chunks = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :pid"),
+                    {"pid": project_id},
+                )
+            ).scalar_one()
+
+        assert post_index == 0, (
+            f"search_index still has {post_index} rows for deleted project_id={project_id} "
+            "— project deletion did not sweep the FTS table."
+        )
+        assert post_chunks == 0, (
+            f"search_vector_chunks still has {post_chunks} rows for deleted "
+            f"project_id={project_id}."
+        )
+
+
+@pytest.mark.asyncio
+async def test_delete_returns_false_for_missing_project_id(project_service: ProjectService):
+    """ProjectRepository.delete must return False when the project id is gone.
+
+    The override loses the base Repository.delete contract if the NoResultFound
+    branch isn't covered — a silent True would mislead callers into thinking
+    a non-existent project was removed.
+    """
+    result = await project_service.repository.delete(9_999_999)
+    assert result is False
+
+
+@pytest.mark.asyncio
+async def test_remove_project_purges_vector_embeddings(project_service: ProjectService):
+    """Project deletion must also drop sqlite-vec embeddings keyed by chunk rowid.
+
+    sqlite-vec stores vectors in a vec0 virtual table that has no cascade
+    behavior. If embeddings linger after the chunks they reference are gone,
+    `_run_vector_query` pulls them as top-k candidates and crowds out live
+    results. The test only runs when the embeddings table is present, which
+    matches the install path that exercises semantic search.
+    """
+    test_project_name = f"test-vec-cleanup-{os.urandom(4).hex()}"
+    session_maker = project_service.repository.session_maker
+
+    # The embeddings table only exists once semantic search has initialized.
+    # Skipping when it's absent keeps this test honest on minimal CI DBs.
+    if not await _table_exists(session_maker, "search_vector_embeddings"):
+        pytest.skip("search_vector_embeddings is not present on this connection")
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        test_project_path = str(Path(temp_dir) / "test-vec-cleanup")
+        os.makedirs(test_project_path, exist_ok=True)
+
+        await project_service.add_project(test_project_name, test_project_path)
+        project = await project_service.get_project(test_project_name)
+        assert project is not None
+        project_id = project.id
+
+        async with db.scoped_session(session_maker) as session:
+            await session.execute(
+                text(
+                    "INSERT INTO search_vector_chunks "
+                    "(id, entity_id, project_id, chunk_key, chunk_text, source_hash, "
+                    " entity_fingerprint, embedding_model) "
+                    "VALUES (:id, :entity_id, :project_id, :chunk_key, :chunk_text, "
+                    " :source_hash, :entity_fingerprint, :embedding_model)"
+                ),
+                {
+                    "id": 999_201,
+                    "entity_id": 999_201,
+                    "project_id": project_id,
+                    "chunk_key": "vec-canary",
+                    "chunk_text": "vec canary",
+                    "source_hash": "abc",
+                    "entity_fingerprint": "",
+                    "embedding_model": "",
+                },
+            )
+            # vec0 requires a vector matching the configured dimensions, but the
+            # delete path filters by rowid; a non-existing dimension would block
+            # this seed step. Skip the insert if the embeddings DDL hasn't run.
+            try:
+                await session.execute(
+                    text(
+                        "INSERT INTO search_vector_embeddings (rowid, embedding) "
+                        "VALUES (:rowid, :embedding)"
+                    ),
+                    {"rowid": 999_201, "embedding": "[" + ",".join(["0.0"] * 384) + "]"},
+                )
+            except Exception:
+                pytest.skip("search_vector_embeddings rejected the synthetic seed row")
+
+        async with db.scoped_session(session_maker) as session:
+            pre = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_vector_embeddings WHERE rowid = :rowid"),
+                    {"rowid": 999_201},
+                )
+            ).scalar_one()
+        assert pre >= 1, "seed embedding should exist before removal"
+
+        await project_service.remove_project(test_project_name)
+
+        async with db.scoped_session(session_maker) as session:
+            post = (
+                await session.execute(
+                    text("SELECT COUNT(*) FROM search_vector_embeddings WHERE rowid = :rowid"),
+                    {"rowid": 999_201},
+                )
+            ).scalar_one()
+
+        assert post == 0, (
+            f"search_vector_embeddings still has {post} rows for rowid 999_201 "
+            "— project deletion did not sweep the sqlite-vec embeddings table."
+        )
@@ -235,9 +235,7 @@ async def test_after_date_uses_updated_at(search_service):
    await search_service.repository.index_item(recently_updated_row)
    await search_service.repository.index_item(stale_row)

-    results = await search_service.search(
-        SearchQuery(after_date=cutoff.isoformat())
-    )
+    results = await search_service.search(SearchQuery(after_date=cutoff.isoformat()))

    permalinks = {r.permalink for r in results}
    # recently-updated entity must appear despite old created_at
@@ -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
@@ -1,5 +1,7 @@
 """Test configuration management."""

+import os
+import stat
 import tempfile
 import pytest
 from datetime import datetime
@@ -306,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"

@@ -318,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)
@@ -449,6 +469,23 @@ class TestConfigManager:
        with pytest.raises(ValueError, match="Project 'nonexistent' not found"):
            config_manager.set_default_project("nonexistent")

+    @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits are not portable to Windows")
+    def test_save_config_uses_private_permissions(self, temp_config_manager):
+        """Config can contain cloud credentials, so writes should enforce private modes."""
+        config_manager = temp_config_manager
+        config = config_manager.load_config()
+        config.cloud_api_key = "bmc_test123"
+
+        config_manager.config_dir.chmod(0o777)
+        config_manager.config_file.chmod(0o666)
+        config_manager.save_config(config)
+
+        dir_mode = stat.S_IMODE(config_manager.config_dir.stat().st_mode)
+        file_mode = stat.S_IMODE(config_manager.config_file.stat().st_mode)
+
+        assert dir_mode == 0o700
+        assert file_mode == 0o600
+
    def test_disable_permalinks_flag_default(self):
        """Test that disable_permalinks flag defaults to False."""
        config = BasicMemoryConfig()
Author	SHA1	Message	Date
Drew Cain	41d12cba95	docs(cli): document personal workspace sync Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-26 13:10:21 -05:00
Drew Cain	ff5d872a8c	chore: update version to 0.21.5 for v0.21.5 release	2026-05-26 10:58:09 -05:00
Drew Cain	96ee4eafd2	docs: add v0.21.5 changelog entry Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-26 10:57:55 -05:00
Drew Cain	b109b7337f	fix(mcp): attach local state to one workspace project row (#854 ) Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-26 10:54:44 -05:00
Drew Cain	36b51b676e	fix(mcp): return workspace-qualified write permalinks (#853 ) Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-26 00:33:21 -05:00
Drew Cain	9af320187c	fix(core): load sqlite-vec before vector table cleanup (#852 ) Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-26 00:27:47 -05:00
Paul Hernandez	5a34a420c9	fix(mcp): preinitialize local ASGI database (#838 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-25 15:57:00 -05:00
Drew Cain	a7e2368f9e	chore: update version to 0.21.4 for v0.21.4 release	2026-05-23 14:55:49 -05:00
Sean Campbell	c755127317	fix(cli): ignore CancelledError in background task done callback (#839 ) (#842 ) Signed-off-by: rudi193-cmd <rudi193@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com>	2026-05-23 14:27:59 -05:00
Sean Campbell	94c04ee456	fix(mcp): restore write_note overwrite schema for external clients (#818 ) (#841 ) Signed-off-by: rudi193-cmd <rudi193@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com>	2026-05-23 14:19:43 -05:00
Drew Cain	d4ed02ba74	docs(core): move release process from CONTRIBUTING.md to AGENTS.md (#846 ) Signed-off-by: Drew Cain <groksrc@gmail.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-05-23 14:00:01 -05:00
Drew Cain	40ed7129c8	chore: update version to 0.21.3 for v0.21.3 release	2026-05-23 13:46:48 -05:00
Drew Cain	c4ef7abff5	test(core): isolate XDG_CONFIG_HOME so host env can't leak into tests (#845 ) Signed-off-by: Drew Cain <groksrc@gmail.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-05-23 13:34:41 -05:00
Paul Hernandez	c75f45f3cf	Update README.md update cloud description to remove link to private cloud repo Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>	2026-05-23 12:22:18 -05:00
Drew Cain	5ae8a733ea	fix(mcp): route mixed local/cloud projects correctly (#837 ) Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-23 11:31:13 -05:00
Rafael Madriz	b94ef01f82	feat(core): add XDG_CONFIG_HOME support (#844 ) Signed-off-by: Rafael Madriz <rafa@rafaelmadriz.com>	2026-05-22 15:18:06 -05:00
Sean Campbell	12af7930de	docs(installer): use pgvector image for Postgres compose (#840 ) Signed-off-by: rudi193-cmd <rudi193@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Drew Cain <groksrc@gmail.com>	2026-05-22 08:54:18 -05:00
phernandez	60ec6728de	chore: update version to 0.21.1 for v0.21.1 release	2026-05-16 18:56:38 -05:00
phernandez	a84e77f4af	docs: add v0.21.1 changelog entry CI-only point release to validate #833's inline Homebrew bump end-to-end on a real tag push. Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-16 18:56:30 -05:00
Paul Hernandez	6910620968	ci(installer): inline Homebrew formula bump (#833 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-16 18:55:25 -05:00
phernandez	214a54d740	chore: update version to 0.21.0 for v0.21.0 release	2026-05-16 17:15:53 -05:00
phernandez	5a90c7cedf	docs: add v0.21.0 changelog entry Promotes the Unreleased breaking-change note and adds the full v0.21.0 section covering ~80 commits since v0.20.3: workspace-routing fixes across MCP/CLI/API, recent_activity ordering and search opt-in changes, sync hardening, sqlite-vec graceful degrade, perf wins on CLI startup and sync, and the project-delete cleanup landed in #832. Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-16 17:15:43 -05:00
Paul Hernandez	9e3fe26a83	fix(core): purge SQLite search_index on project delete (#832 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-16 16:47:24 -05:00
Paul Hernandez	47ee982041	perf(cli): defer local ASGI app import (#828 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 17:59:58 -05:00
Paul Hernandez	4d22c398c6	fix(sync): preserve bmignore rclone filters (#827 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 17:48:49 -05:00
phernandez	8ac2d975f9	update README.md Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 17:13:34 -05:00
Paul Hernandez	34830bfad7	Update README.md update agent matrix Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>	2026-05-15 14:55:00 -05:00
Paul Hernandez	f5e0c42047	Update README.md remove sync commands from cli examples Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>	2026-05-15 14:52:27 -05:00
phernandez	3f98da8c67	update README.md Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 14:50:35 -05:00
phernandez	14ff77d1c2	update to fastmcp 3.3.1 Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 12:28:03 -05:00
phernandez	4e4da6128d	ci: run github actions on node 24 Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 12:14:06 -05:00
Paul Hernandez	3bed6d8890	chore(deps): update deps and harden security (#825 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-15 10:45:14 -05:00
Paul Hernandez	4cba7ba01c	fix(core): parse prose wikilinks as inline links (#824 ) Signed-off-by: phernandez <paul@basicmachines.co>	2026-05-14 11:43:11 -05:00
Drew Cain	8eeec64e28	fix: `basic-memory project list` does not list projects from all workspaces (#822 ) Signed-off-by: Drew Cain <groksrc@gmail.com>	2026-05-14 09:46:02 -05:00