Compare commits

..

7 Commits

Author SHA1 Message Date
phernandez 21d49023e7 fix(mcp): route multi-project search by name only
The previous revision gated project_id forwarding on the
`cloud_available` composite (factory_mode OR explicit_cloud OR
has_cloud_credentials), mirroring get_project_client. But
`has_cloud_credentials` returns True any time OAuth tokens linger
from a past `bm cloud login`, even when the user is back to working
purely locally. So on a typical dev box the fan-out still forwarded
project_id, hit get_project_client's UUID branch (which treats unknown
identifiers as cloud since local config doesn't key by UUID), and 401d
silently — leaving merged results empty.

Route by name only: project_refs already carry the
workspace/project qualified_name, which is just as unambiguous as the
external_id for both backends. project_id stays in the call signature
purely as a fallback for refs that unexpectedly have no name.

Confirmed live in a restarted MCP — the previous fix did not actually
deliver multi-project results when the dev environment had any cloud
credentials at all.

Tests now drop the routing-mode stubs; the cloud-style tests just key
their MockSearchClient on the workspace-qualified name passed via the
project parameter.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 16:55:21 -05:00
phernandez e8c11b843d fix(core): load sqlite-vec before embeddings purge; add coverage
Three follow-ups from PR review:

**Codex P2 — Load sqlite-vec on the delete session.** The previous
revision swallowed every `vec0` OperationalError as "no embeddings
exist", but sqlite-vec is loaded **per connection**: a pooled
connection that hosts ProjectRepository.delete may not have vec0
loaded even when another connection successfully wrote embeddings.
That would silently leave orphan vectors behind.

The new `_load_sqlite_vec_on_session` helper mirrors
SQLiteSearchRepository._ensure_sqlite_vec_loaded as a free function
and tries to load the extension on the current session. Only when the
load itself fails — because the Python build lacks
enable_load_extension, or the sqlite_vec package isn't installed — do
we skip the embeddings DELETE. Every connection in the pool shares the
same interpreter, so in that case no embeddings could have been
written from any connection and skipping is safe.

**Claude review — Missing logger.debug calls.** The override now logs
at entry, when the project id isn't found, and after the ORM delete,
matching the base Repository.delete contract.

**Claude review — NoResultFound branch uncovered.** New test
`test_delete_returns_false_for_missing_project_id` asserts the False
return for a nonexistent project id.

Verified locally: 32 passed / 1 skipped (SQLite), 27 passed / 1
skipped (Postgres via testcontainers).

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 16:07:19 -05:00
phernandez 43c50ea683 fix(core): tolerate missing vec0 module during project delete
On Windows SQLite CI the embeddings cleanup hits

  sqlite3.OperationalError: no such module: vec0
  [SQL: DELETE FROM search_vector_embeddings WHERE rowid IN (...)]

because the sqlite-vec extension isn't loaded into the connection
(some Windows Python builds don't expose enable_load_extension on
sqlite3.Connection — see #711). The vec0 virtual table is registered
in sqlite_master, so the table-existence check passes, but any access
to it fails until the module is loaded.

If vec0 isn't loadable, semantic search was never able to write
embeddings, so there's nothing to clean up. Wrap the embeddings DELETE
in a try/except that swallows OperationalError for vec0 and logs a
debug line — the chunk DELETE below still runs.

The KeyError from test_mcp_sse_forces_local on 3.14 SQLite is an
unrelated transient — same code passes on 3.12/3.13 SQLite, Postgres
3.14, and locally on Python 3.14. Will let CI re-run after this push.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 15:53:49 -05:00
phernandez d3619f97f9 fix(core): dialect-gate sqlite-vec embeddings purge
The previous fix issued an unconditional
`DELETE FROM search_vector_embeddings WHERE rowid IN (...)` during project
deletion, which 500'd on Postgres:

  column "rowid" does not exist

Postgres uses `chunk_id` (a real column with an FK to
search_vector_chunks.id ON DELETE CASCADE), so it doesn't need or accept
the SQLite query at all. The cascade picture by backend:

  - search_index → project: Postgres has FK CASCADE; SQLite FTS5 virtual
    table can't carry FKs and needs explicit cleanup.
  - search_vector_chunks → project: neither backend has an FK, so both
    need an explicit DELETE.
  - search_vector_embeddings → search_vector_chunks: Postgres has FK
    CASCADE on chunk_id; SQLite vec0 virtual table is keyed by rowid
    with no cascade, so embeddings must be purged before the chunks.

Branch is now: chunks DELETE runs on both backends; search_index and
the vec0 embeddings DELETE only run on SQLite.

Verified locally against both backends (SQLite: 31 passed, Postgres
via testcontainers: 26 passed) for tests/services/test_project_removal_bug.py
and the CLI/MCP project-management integration suites.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 15:28:41 -05:00
phernandez 2bd6552e84 fix(mcp): address PR review feedback (P2 + multi-project search)
Two follow-up fixes on the same branch:

1. **Purge sqlite-vec embeddings during project delete** (Codex P2)
   sqlite-vec stores vectors in a vec0 virtual table keyed by chunk rowid
   with no cascade. The previous purge removed search_vector_chunks but
   left the embeddings behind; `_run_vector_query` then keeps returning
   stale vectors that crowd live results.

   ProjectRepository.delete now deletes embeddings first (using the same
   rowid-IN-chunks pattern as SQLiteSearchRepository.delete_project_vector_rows),
   then the chunk rows. Both deletes are skipped if the underlying table
   is absent on a given install. New test test_remove_project_purges_vector_embeddings
   covers the happy path and skips cleanly when the embeddings table
   isn't initialized.

2. **Fix search_all_projects=True on local installs**
   `_search_all_projects` recurses into search_notes with both project=
   and project_id= set. project_id (external UUID) routes through the
   cloud v2 API path, which 401s on local installs because there's no
   JWT to present — so the inner calls silently failed and the merged
   result list stayed empty.

   The fan-out now mirrors get_project_client's cloud_available composite
   (factory mode OR explicit --cloud OR has_cloud_credentials). When that
   composite is false we forward project= only and take the name-routed
   local-ASGI path. Cloud disambiguation still works because the project
   name in project_ref is already the workspace/project qualified_name.

   The existing cloud-style fan-out tests now go through a cloud_routing
   fixture that pins the three signals; a new local_routing test confirms
   project_id is dropped when no cloud route is available.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 15:11:39 -05:00
phernandez 1768de8571 fix(core): skip absent search tables in project delete
search_index is created lazily by SearchRepository.init_search_index and
search_vector_chunks only materializes once semantic search initializes,
so either table may be absent on minimal test DBs. The previous version
of ProjectRepository.delete unconditionally issued DELETEs against both
and crashed CLI integration tests on Postgres and SQLite where
search_vector_chunks hadn't been created yet:

  relation "search_vector_chunks" does not exist
  no such table: search_vector_chunks

Inspect the connection's tables once per call and only delete from
whichever derived tables are present. Idempotent on every backend.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 14:25:07 -05:00
phernandez a85767ad92 fix(core): purge SQLite search_index on project delete
SQLite stores search_index as an FTS5 virtual table, which can't carry a
foreign key, so the ON DELETE CASCADE from search_index.project_id to
project.id only applies on Postgres. On SQLite, deleting a project left
its FTS rows behind — and when auto-increment handed the same id to a
new project, the leftover rows masqueraded as the new tenant's data and
leaked into searches scoped to that project.

- ProjectRepository.delete now explicitly purges search_index and
  search_vector_chunks for the project id in the same session before
  the ORM delete. Idempotent on Postgres (the cascade FK still runs).
- One-time cleanup migration sweeps two leftover shapes: rows whose
  project_id is gone, and rows whose entity_id is gone (the larger
  class from id reuse). Guarded by table-existence checks so fresh
  SQLite installs — where search_index is created at runtime by
  init_search_index, not by Alembic — don't fail the upgrade.
- Regression test seeds both derived tables, calls remove_project,
  and asserts both come out clean. Verified red on pre-fix code.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-05-16 11:43:16 -05:00
27 changed files with 214 additions and 1487 deletions
+19 -57
View File
@@ -63,63 +63,25 @@ 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: read
contents: write
actions: 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@v4.1
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:
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::"
# Personal Access Token with repo scope for homebrew-basic-memory repo
COMMITTER_TOKEN: ${{ secrets.HOMEBREW_TOKEN }}
-9
View File
@@ -56,15 +56,6 @@ 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)
+1 -122
View File
@@ -1,24 +1,6 @@
# CHANGELOG
## 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.
## Unreleased
### Breaking Changes
@@ -29,109 +11,6 @@ and a handful of long-standing parsing and routing footguns are gone.
- 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)
+2 -1
View File
@@ -123,7 +123,8 @@ phone.
time — same files, same format, same wikilinks. Cancel anytime, your data
stays yours.
Built on WorkOS AuthKit, Neon Postgres, and Tigris S3.
Built on WorkOS AuthKit, Neon Postgres, and Tigris S3. Platform source at
[basic-memory-cloud](https://github.com/basicmachines-co/basic-memory-cloud).
### Pricing
+2 -5
View File
@@ -1,8 +1,5 @@
# Docker Compose configuration for Basic Memory with PostgreSQL
# 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.
# Use this for local development and testing with Postgres backend
#
# Usage:
# docker-compose -f docker-compose-postgres.yml up -d
@@ -10,7 +7,7 @@
services:
postgres:
image: pgvector/pgvector:pg17
image: postgres:17
container_name: basic-memory-postgres
environment:
# Local development/test credentials - NOT for production
-1
View File
@@ -263,7 +263,6 @@ 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
+2 -2
View File
@@ -6,12 +6,12 @@
"url": "https://github.com/basicmachines-co/basic-memory.git",
"source": "github"
},
"version": "0.21.3",
"version": "0.20.3",
"packages": [
{
"registryType": "pypi",
"identifier": "basic-memory",
"version": "0.21.3",
"version": "0.20.3",
"runtimeHint": "uvx",
"runtimeArguments": [
{"type": "positional", "value": "basic-memory"},
+1 -1
View File
@@ -1,7 +1,7 @@
"""basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
# Package version - updated by release automation
__version__ = "0.21.3"
__version__ = "0.20.3"
# API version for FastAPI - independent of package version
__api_version__ = "v0"
+5 -8
View File
@@ -21,7 +21,7 @@ from basic_memory.utils import setup_logging, generate_permalink
DATABASE_NAME = "memory.db"
APP_DATABASE_NAME = "memory.db" # Using the same name but in the app directory
DATA_DIR_NAME = "basic-memory"
DATA_DIR_NAME = ".basic-memory"
CONFIG_FILE_NAME = "config.json"
WATCH_STATUS_JSON = "watch-status.json"
CONFIG_DIR_MODE = 0o700
@@ -69,18 +69,15 @@ 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 then to ``XDG_CONFIG_HOME``.
and database state; otherwise falls back to ``<user home>/.basic-memory``.
Cross-platform: ``Path.home()`` reads ``$HOME`` on POSIX and
``%USERPROFILE%`` on Windows, so there's no need to check ``$HOME``
explicitly here.
"""
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)
if config_dir := os.getenv("BASIC_MEMORY_CONFIG_DIR"):
return Path(config_dir)
return Path.home() / DATA_DIR_NAME
def default_fastembed_cache_dir() -> str:
+7 -101
View File
@@ -142,15 +142,6 @@ 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:
@@ -176,7 +167,8 @@ 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 _clear_cached_active_project(context)
await context.set_state("active_project", None)
await context.set_state("default_project_name", None)
await context.set_state("active_workspace", active_workspace.model_dump())
@@ -533,28 +525,6 @@ 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,
@@ -1356,23 +1326,11 @@ async def detect_project_from_identifier_prefix(
# Outcome: keep unqualified search/title input on the active/default project route.
return None
if _workspace_identifier_discovery_available(identifier, config):
workspace_discovery_fallback_errors = (
"not found",
"no accessible workspaces",
"unable to discover",
if _cloud_workspace_discovery_available(config):
workspace_resolution = await resolve_workspace_qualified_identifier(
identifier,
context=context,
)
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
@@ -1387,7 +1345,7 @@ async def detect_project_from_identifier_prefix(
)
except ValueError as exc:
message = str(exc).lower()
if any(error in message for error in workspace_discovery_fallback_errors):
if "not found" in message or "no accessible workspaces" in message:
return None
raise
@@ -1504,54 +1462,9 @@ 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)
@@ -1574,13 +1487,6 @@ 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,
+3 -4
View File
@@ -9,7 +9,7 @@ from pydantic import AliasChoices, Field
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import (
_workspace_identifier_discovery_available,
_cloud_workspace_discovery_available,
detect_project_from_memory_url_prefix,
get_project_client,
add_project_metadata,
@@ -368,9 +368,8 @@ async def edit_note(
config,
context=context,
)
elif _workspace_identifier_discovery_available(
identifier,
config,
elif _cloud_workspace_discovery_available(
config
) and is_workspace_qualified_plain_identifier(identifier):
detected = await detect_project_from_workspace_identifier_prefix(
identifier,
+12 -24
View File
@@ -10,13 +10,8 @@ from loguru import logger
from fastmcp import Context
from pydantic import AliasChoices, BeforeValidator, Field
from basic_memory.config import ConfigManager, has_cloud_credentials
from basic_memory.config import ConfigManager
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,
@@ -485,28 +480,21 @@ async def _search_all_projects(
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)
)
# Why: forwarding project_id (external UUID) routes the per-project search
# through get_project_client's UUID branch, which treats unknown
# identifiers as cloud and 401s when local OAuth tokens linger from a
# past `bm cloud login` even though the project itself is local.
# Outcome: route by `project` name (qualified_name like "workspace/project"
# in cloud refs, plain name in local refs) — both backends resolve
# that without UUID lookup. project_id is only used as a fallback when
# a ref unexpectedly has no name to route by.
for project_ref in project_refs:
recursive_project_id = project_ref["project_id"] if use_cloud_routing else None
project_name = project_ref["project"]
recursive_project_id = None if project_name else project_ref["project_id"]
try:
results = await search_notes(
query=query,
project=project_ref["project"],
project=project_name,
project_id=recursive_project_id,
page=1,
page_size=per_project_page_size,
+5 -17
View File
@@ -40,29 +40,17 @@ async def detect_project_from_workspace_identifier_prefix(
return None
from basic_memory.mcp.project_context import (
_workspace_identifier_discovery_available,
_cloud_workspace_discovery_available,
resolve_workspace_qualified_identifier,
)
if not _workspace_identifier_discovery_available(identifier, config):
if not _cloud_workspace_discovery_available(config):
return None
workspace_discovery_fallback_errors = (
"not found",
"no accessible workspaces",
"unable to discover",
workspace_resolution = await resolve_workspace_qualified_identifier(
identifier,
context=context,
)
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
-16
View File
@@ -149,22 +149,6 @@ 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",
+6 -12
View File
@@ -10,8 +10,6 @@ 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
@@ -56,7 +54,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"
database_url = "postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory_test"
```
## Running Postgres Tests
@@ -68,22 +66,18 @@ docker-compose -f docker-compose-postgres.yml up -d
```
This starts:
- Postgres 17 with **pgvector** (`pgvector/pgvector:pg17`) on port **5433** (not 5432 to avoid conflicts)
- Database: `basic_memory`
- Postgres 17 on port **5433** (not 5432 to avoid conflicts)
- Test database: `basic_memory_test`
- 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
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
pytest tests/test_entity_repository.py::test_create -m postgres
# Skip Postgres tests (default behavior)
pytest -m "not postgres"
@@ -127,7 +121,7 @@ jobs:
# Postgres service container
services:
postgres:
image: pgvector/pgvector:pg17
image: postgres:17
env:
POSTGRES_DB: basic_memory_test
POSTGRES_USER: basic_memory_user
@@ -175,4 +169,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
-13
View File
@@ -225,19 +225,6 @@ 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."""
-25
View File
@@ -20,31 +20,6 @@ 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
+72 -464
View File
@@ -11,7 +11,25 @@ from typing import Any, AsyncIterator, cast
import pytest
from tests.mcp.conftest import ContextState, ctx
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)
def _workspace(
@@ -200,7 +218,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",
@@ -218,7 +236,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()
@@ -254,7 +272,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
@@ -398,7 +416,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
role="owner",
),
]
context = ContextState()
context = _ContextState()
await context.set_state("active_workspace", cached_workspace.model_dump())
fetches = 0
@@ -413,7 +431,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
)
with pytest.raises(ValueError) as exc_info:
await resolve_workspace_parameter(workspace="organization", context=ctx(context))
await resolve_workspace_parameter(workspace="organization", context=_ctx(context))
message = str(exc_info.value)
assert fetches == 1
@@ -434,7 +452,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
@@ -445,7 +463,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
@@ -458,7 +476,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",
@@ -495,8 +513,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",
@@ -505,8 +523,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"]
@@ -521,7 +539,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",
@@ -554,21 +572,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),
)
@@ -782,11 +800,8 @@ 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",
"memory://notes/foo/bar",
BasicMemoryConfig(
projects={"main": ProjectEntry(path="/tmp/main")},
cloud_api_key="bmc_test123",
@@ -796,101 +811,6 @@ 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,
@@ -1125,7 +1045,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",
@@ -1161,7 +1081,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"
@@ -1312,318 +1232,6 @@ 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."""
@@ -1662,7 +1270,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",
@@ -1680,7 +1288,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
@@ -1694,7 +1302,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():
@@ -1706,8 +1314,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"
@@ -1719,7 +1327,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",
@@ -1737,7 +1345,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
@@ -1746,7 +1354,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",
@@ -1767,7 +1375,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
@@ -1783,7 +1391,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",
@@ -1808,7 +1416,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
@@ -1830,7 +1438,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",
@@ -1856,7 +1464,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
@@ -1878,7 +1486,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",
@@ -1905,7 +1513,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
@@ -1924,7 +1532,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",
@@ -1937,7 +1545,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
@@ -1960,7 +1568,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",
@@ -1991,7 +1599,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
@@ -2002,7 +1610,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
@@ -2042,7 +1650,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
@@ -2063,7 +1671,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",
@@ -2097,7 +1705,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
@@ -2117,7 +1725,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",
@@ -2156,7 +1764,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"
@@ -2287,7 +1895,7 @@ class TestGetProjectClientRoutingOrder:
)
config_manager.save_config(config)
context = ContextState()
context = _ContextState()
stale_workspace = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
@@ -2334,13 +1942,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
@@ -2382,7 +1990,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):
@@ -2426,7 +2034,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"
@@ -2546,7 +2154,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
@@ -2585,7 +2193,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"
@@ -2727,7 +2335,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",
@@ -2768,7 +2376,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"
+16 -3
View File
@@ -4,17 +4,30 @@ 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]] = []
@@ -29,7 +42,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",
@@ -45,7 +58,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=ctx(context))
resolved = await project_context.resolve_workspace_parameter(context=cast(Any, context))
assert resolved.tenant_id == workspace.tenant_id
assert spans == [
+8 -85
View File
@@ -914,8 +914,8 @@ async def test_edit_note_workspace_qualified_plain_permalink_requires_explicit_r
monkeypatch.setattr(
edit_note_module,
"_workspace_identifier_discovery_available",
lambda identifier, config: True,
"_cloud_workspace_discovery_available",
lambda 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,
"_workspace_identifier_discovery_available",
lambda identifier, config: True,
"_cloud_workspace_discovery_available",
lambda 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,
"_workspace_identifier_discovery_available",
lambda identifier, config: True,
"_cloud_workspace_discovery_available",
lambda config: True,
)
monkeypatch.setattr(
edit_note_module,
@@ -1054,83 +1054,6 @@ 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,
@@ -1154,8 +1077,8 @@ async def test_edit_note_three_segment_plain_path_stays_local_without_workspace_
monkeypatch.setattr(
edit_note_module,
"_workspace_identifier_discovery_available",
lambda identifier, config: False,
"_cloud_workspace_discovery_available",
lambda config: False,
)
monkeypatch.setattr(
edit_note_module,
-107
View File
@@ -6,10 +6,6 @@ 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
@@ -161,109 +157,6 @@ 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:
-127
View File
@@ -1,7 +1,5 @@
"""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
@@ -9,7 +7,6 @@ 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():
@@ -169,130 +166,6 @@ 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"""
-105
View File
@@ -3,8 +3,6 @@
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 (
@@ -173,109 +171,6 @@ 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."""
+19 -7
View File
@@ -1,11 +1,12 @@
"""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(
@@ -27,6 +28,17 @@ 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):
@@ -133,7 +145,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",
@@ -157,8 +169,8 @@ async def test_list_workspaces_uses_context_cache_path(monkeypatch):
fake_get_available_workspaces,
)
first = await list_workspaces(context=ctx(context))
second = await list_workspaces(context=ctx(context))
first = await list_workspaces(context=cast(Any, context))
second = await list_workspaces(context=cast(Any, context))
assert "# Available Workspaces (1)" in first
assert "# Available Workspaces (1)" in second
@@ -242,14 +254,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=ctx(context))
first = await get_available_workspaces(context=cast(Any, 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=ctx(context))
second = await get_available_workspaces(context=cast(Any, context))
assert len(second) == 1
assert call_count["provider"] == 1
@@ -8,38 +8,8 @@ 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, cloud_routing
):
async def test_search_notes_search_all_projects_qualifies_result_permalinks(monkeypatch):
"""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")
@@ -62,7 +32,10 @@ async def test_search_notes_search_all_projects_qualifies_result_permalinks(
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"
# When the fan-out routes by name only, project_id is None — keep
# the test stable by falling back to the name so the downstream
# SearchClient still has a stable per-project identifier.
self.external_id = external_id or self.name
@asynccontextmanager
async def fake_get_project_client(project=None, context=None, project_id=None):
@@ -74,10 +47,12 @@ async def test_search_notes_search_all_projects_qualifies_result_permalinks(
class MockSearchClient:
def __init__(self, client, project_id):
# The fake project supplies the workspace-qualified name as the
# external_id, so each per-project search keys off that.
self.project_id = project_id
async def search(self, payload, page, page_size):
if self.project_id == "11111111-1111-1111-1111-111111111111":
if self.project_id == "personal/main":
title = "Personal MCP Test Note"
score = 0.5
else:
@@ -111,9 +86,13 @@ async def test_search_notes_search_all_projects_qualifies_result_permalinks(
)
assert isinstance(result, dict)
# Fan-out now routes by qualified_name only; project_id is omitted because
# the workspace-qualified name is already unambiguous and UUID routing
# hits get_project_client's cloud branch even for local projects when
# local OAuth credentials are present.
assert searched_projects == [
("personal/main", "11111111-1111-1111-1111-111111111111"),
("team-paul/main", "22222222-2222-2222-2222-222222222222"),
("personal/main", None),
("team-paul/main", None),
]
assert [item["permalink"] for item in result["results"]] == [
"team-paul/main/tests/mcp-test-note",
@@ -198,9 +177,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, cloud_routing
):
async def test_search_notes_search_all_projects_continues_after_project_failure(monkeypatch):
"""One failing project should not discard successful all-project search results."""
clients_mod = importlib.import_module("basic_memory.mcp.clients")
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
@@ -223,7 +200,7 @@ async def test_search_notes_search_all_projects_continues_after_project_failure(
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"
self.external_id = external_id or self.name
@asynccontextmanager
async def fake_get_project_client(project=None, context=None, project_id=None):
@@ -244,10 +221,12 @@ async def test_search_notes_search_all_projects_continues_after_project_failure(
class MockSearchClient:
def __init__(self, client, project_id):
# Fan-out routes by name now, so the stub project reflects the
# workspace-qualified name back as the SearchClient's project_id.
self.project_id = project_id
async def search(self, payload, page, page_size):
if self.project_id == "22222222-2222-2222-2222-222222222222":
if self.project_id == "team-paul/main":
raise RuntimeError("team index unavailable")
return SearchResponse(
results=[
@@ -287,15 +266,15 @@ async def test_search_notes_search_all_projects_continues_after_project_failure(
@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.
async def test_search_notes_search_all_projects_omits_project_id(monkeypatch):
"""Fan-out must address each project by name, never by external UUID.
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.
project_id routes through get_project_client's UUID branch, which treats
any unknown identifier as cloud so when a local install still has OAuth
tokens from a past `bm cloud login`, every per-project recursive call 401s
and the merged result list silently stays empty. Routing by name avoids
that path on both backends; cloud refs disambiguate via the
workspace/project qualified_name already baked into project_ref["project"].
"""
clients_mod = importlib.import_module("basic_memory.mcp.clients")
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
+6 -104
View File
@@ -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(identifier, config):
def fail_if_called(config):
raise AssertionError("plain project-relative paths should skip cloud discovery")
monkeypatch.setattr(
"basic_memory.mcp.project_context._workspace_identifier_discovery_available",
"basic_memory.mcp.project_context._cloud_workspace_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._workspace_identifier_discovery_available",
lambda identifier, config: False,
"basic_memory.mcp.project_context._cloud_workspace_discovery_available",
lambda 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._workspace_identifier_discovery_available",
lambda identifier, config: True,
"basic_memory.mcp.project_context._cloud_workspace_discovery_available",
lambda config: True,
)
monkeypatch.setattr(
"basic_memory.mcp.project_context.resolve_workspace_qualified_identifier",
@@ -205,104 +205,6 @@ 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."""
+1 -19
View File
@@ -308,9 +308,8 @@ 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 and XDG_CONFIG_HOME, resolver returns ~/.basic-memory."""
"""Without BASIC_MEMORY_CONFIG_DIR, 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"
@@ -321,23 +320,6 @@ 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)