admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mcp): route workspace-qualified memory urls (#790)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
Paul Hernandez
2026-05-03 18:18:10 -05:00
committed by GitHub
parent 0a72d81bb3
commit 05adda1502
20 changed files with 1931 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/api/app.py
+37 -1
View File
@@ -4,6 +4,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
from fastapi.exception_handlers import http_exception_handler
from fastapi.responses import JSONResponse
from fastapi.routing import APIRouter
from loguru import logger
@@ -29,6 +30,12 @@ import logfire
from basic_memory.config import init_api_logging
from basic_memory.services.exceptions import EntityAlreadyExistsError
from basic_memory.services.initialization import initialize_app
from basic_memory.workspace_context import (
WORKSPACE_SLUG_HEADER,
WORKSPACE_TYPE_HEADER,
workspace_permalink_context_validation_error,
workspace_permalink_context,
)
@asynccontextmanager
@@ -87,6 +94,32 @@ app = FastAPI(
lifespan=lifespan,
)
@app.middleware("http")
async def workspace_permalink_context_middleware(request: Request, call_next):
"""Populate workspace permalink context from request headers."""
workspace_slug = request.headers.get(WORKSPACE_SLUG_HEADER)
workspace_type = request.headers.get(WORKSPACE_TYPE_HEADER)
validation_error = workspace_permalink_context_validation_error(workspace_slug, workspace_type)
if validation_error is not None:
return JSONResponse(
status_code=400,
content={"detail": validation_error},
)
if not workspace_slug:
return await call_next(request)
# ContextVar state remains active across the awaited downstream handler while
# this context manager is open, so entity creation can see request metadata.
with workspace_permalink_context(
workspace_slug=workspace_slug,
workspace_type=workspace_type,
):
return await call_next(request)
# Include v2 routers FIRST (more specific paths must match before /{project} catch-all)
app.include_router(v2_knowledge, prefix="/v2/projects/{project_id}")
app.include_router(v2_memory, prefix="/v2/projects/{project_id}")
@@ -146,4 +179,7 @@ async def exception_handler(request, exc): # pragma: no cover
error_type=type(exc).__name__,
error=str(exc),
)
return await http_exception_handler(request, HTTPException(status_code=500, detail=str(exc)))
return await http_exception_handler(
request,
HTTPException(status_code=500, detail="Internal server error"),
)
src/basic_memory/mcp/async_client.py
+3
View File
@@ -94,9 +94,12 @@ async def _cloud_client(
workspace: Optional[str] = None,
) -> AsyncIterator[AsyncClient]:
"""Create a cloud proxy client with resolved credentials."""
from basic_memory.workspace_context import workspace_permalink_headers
token = await _resolve_cloud_token(config)
proxy_base_url = f"{config.cloud_host}/proxy"
headers = {"Authorization": f"Bearer {token}"}
headers.update(workspace_permalink_headers())
if workspace:
headers["X-Workspace-ID"] = workspace
logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")
src/basic_memory/mcp/project_context.py
+282 -8
View File
@@ -9,7 +9,7 @@ compatibility with existing MCP tools.
"""
import asyncio
from contextlib import asynccontextmanager
from contextlib import asynccontextmanager, nullcontext
from dataclasses import dataclass, field
from typing import AsyncIterator, Awaitable, Callable, Optional, List, Tuple, cast
from uuid import UUID
@@ -30,6 +30,10 @@ from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.schemas.v2 import ProjectResolveResponse
from basic_memory.schemas.memory import memory_url_path
from basic_memory.utils import generate_permalink, normalize_project_reference
from basic_memory.workspace_context import (
current_workspace_permalink_context,
workspace_permalink_context,
)
# --- Workspace provider injection ---
# Mirrors the set_client_factory() pattern in async_client.py.
@@ -62,6 +66,18 @@ class WorkspaceProjectIndex:
failed_workspaces: tuple[WorkspaceInfo, ...] = ()
@dataclass(frozen=True)
class WorkspaceMemoryUrlResolution:
"""Resolved workspace/project route for a workspace-qualified memory URL."""
entry: WorkspaceProjectEntry
canonical_path: str
@property
def project_identifier(self) -> str:
return self.entry.qualified_name
def set_workspace_provider(provider: Callable[[], Awaitable[list[WorkspaceInfo]]]) -> None:
"""Override workspace discovery (for cloud app, testing, etc)."""
global _workspace_provider
@@ -390,6 +406,130 @@ def _unqualified_project_identifier(identifier: str) -> str:
return project_identifier
def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str] | None:
"""Split ``memory://<workspace>/<project>/<path>`` into route segments."""
if not identifier.strip().startswith("memory://"):
return None
normalized = normalize_project_reference(memory_url_path(identifier))
parts = normalized.split("/", 2)
if len(parts) != 3:
return None
workspace_slug, project_identifier, remainder = parts
if not workspace_slug or not project_identifier or not remainder:
return None
return workspace_slug, project_identifier, remainder
def _canonical_memory_path_for_workspace(
*,
workspace_slug: str,
workspace_type: str,
project_permalink: str,
remainder: str,
include_project: bool,
) -> str:
"""Return the stored canonical path for a workspace-qualified memory URL."""
normalized_remainder = remainder.strip("/")
if workspace_type == "organization":
prefix = f"{generate_permalink(workspace_slug)}/{project_permalink}"
elif workspace_type == "personal":
prefix = project_permalink if include_project else ""
else:
raise ValueError(f"Unsupported workspace_type for memory URL routing: {workspace_type}")
if not prefix:
return normalized_remainder
if not normalized_remainder:
return prefix
return f"{prefix}/{normalized_remainder}"
def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool:
"""Return True when workspace discovery can be used without forcing local routing."""
from basic_memory.mcp.async_client import (
_explicit_routing,
_force_local_mode,
is_factory_mode,
)
if _explicit_routing() and _force_local_mode():
return False
# Trigger: local project config is present even though cloud credentials are saved.
# Why: existing local `memory://...` URLs must not depend on workspace discovery.
# Outcome: only factory, explicit cloud, or cloud-only sessions attempt discovery here.
return (
is_factory_mode()
or (_explicit_routing() and not _force_local_mode())
or (not config.projects and has_cloud_credentials(config))
)
async def resolve_workspace_qualified_memory_url(
identifier: str,
context: Optional[Context] = None,
) -> WorkspaceMemoryUrlResolution | None:
"""Resolve a workspace-qualified memory URL against accessible workspaces."""
segments = _split_workspace_memory_url_segments(identifier)
if segments is None:
return None
workspace_slug, project_identifier, remainder = segments
index = await _ensure_workspace_project_index(context=context)
workspace = next(
(item for item in index.workspaces if item.slug.casefold() == workspace_slug.casefold()),
None,
)
if workspace is None:
return None
project_permalink = generate_permalink(project_identifier)
matches = [
entry
for entry in index.entries_by_permalink.get(project_permalink, ())
if entry.workspace.tenant_id == workspace.tenant_id
]
if not matches:
if any(
failed_workspace.tenant_id == workspace.tenant_id
for failed_workspace in index.failed_workspaces
):
raise ValueError(
f"Projects for workspace '{workspace.name}' ({workspace.slug}) "
"could not be loaded. Retry after workspace discovery recovers."
)
# Trigger: first segment matches a workspace slug but the second does not
# match a project in that workspace.
# Why: workspace-qualified URLs require both route segments to match; otherwise
# existing project-prefixed URLs like `memory://main/notes/foo` can collide
# with a workspace slug named `main`.
# Outcome: treat this as not workspace-qualified and let the caller use
# the existing project-prefix/default-project resolver.
return None
if len(matches) > 1:
details = ", ".join(
f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches
)
raise ValueError(
f"Project '{project_identifier}' matched multiple projects in workspace "
f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. "
f"Matches: {details}"
)
entry = matches[0]
canonical_path = _canonical_memory_path_for_workspace(
workspace_slug=entry.workspace.slug,
workspace_type=entry.workspace.workspace_type,
project_permalink=entry.project.permalink,
remainder=remainder,
include_project=ConfigManager().config.permalinks_include_project,
)
return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path)
def _format_qualified_choices(entries: tuple[WorkspaceProjectEntry, ...]) -> str:
"""Format qualified project choices for collision errors."""
return " or ".join(entry.qualified_name for entry in entries)
@@ -608,6 +748,15 @@ async def resolve_workspace_project_identifier(
f"Project '{project_identifier}' was not found in workspace "
f"'{workspace.name}' ({workspace.slug}). Available projects: {available}"
)
if len(matches) > 1:
details = ", ".join(
f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches
)
raise ValueError(
f"Project '{project_identifier}' matched multiple projects in workspace "
f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. "
f"Matches: {details}"
)
return matches[0]
matches = index.entries_by_permalink.get(project_permalink, ())
@@ -680,6 +829,56 @@ async def _default_workspace_project_entry(
return default_entries[0] if default_entries else None
async def _workspace_metadata_by_tenant_id(
tenant_id: str,
context: Optional[Context] = None,
) -> WorkspaceInfo | None:
"""Return non-index workspace metadata for a configured tenant id."""
cached_workspace = await _get_cached_active_workspace(context)
if cached_workspace and cached_workspace.tenant_id == tenant_id:
return cached_workspace
if cached_workspace and context:
# Trigger: the configured workspace_id differs from cached workspace metadata.
# Why: tenant_id routes the request, but stale workspace slug/type would corrupt
# memory URL normalization and canonical permalink headers.
# Outcome: drop stale metadata and route without permalink decoration.
await context.set_state("active_workspace", None)
if context:
cached_raw = await context.get_state("available_workspaces")
if isinstance(cached_raw, list):
for item in cached_raw:
if not isinstance(item, dict):
continue
workspace = WorkspaceInfo.model_validate(item)
if workspace.tenant_id == tenant_id:
return workspace
if _workspace_provider is not None:
# Trigger: the hosting runtime can provide workspace metadata directly.
# Why: configured workspace_id is already sufficient for tenant routing, but
# canonical organization permalinks also need slug/type context.
# Outcome: use the injected runtime seam without loading the workspace project index.
workspace = next(
(
workspace
for workspace in await get_available_workspaces(context=context)
if workspace.tenant_id == tenant_id
),
None,
)
if workspace is None:
raise ValueError(
f"Configured workspace_id '{tenant_id}' was not returned by the workspace "
"metadata provider. Reconfigure the project workspace or retry after "
"workspace metadata recovers."
)
return workspace
return None
async def resolve_workspace_parameter(
workspace: Optional[str] = None,
context: Optional[Context] = None,
@@ -856,13 +1055,57 @@ async def resolve_project_and_path(
return active_project, identifier, False
normalized_path = normalize_project_reference(memory_url_path(identifier))
cached_project = await _get_cached_active_project(context)
cached_workspace = await _get_cached_active_workspace(context)
if cached_project and cached_workspace:
workspace_prefix = generate_permalink(cached_workspace.slug)
qualified_prefix = f"{workspace_prefix}/{cached_project.permalink}"
if normalized_path == qualified_prefix or normalized_path.startswith(
f"{qualified_prefix}/"
):
remainder = (
""
if normalized_path == qualified_prefix
else normalized_path.removeprefix(f"{qualified_prefix}/")
)
resolved_path = _canonical_memory_path_for_workspace(
workspace_slug=cached_workspace.slug,
workspace_type=cached_workspace.workspace_type,
project_permalink=cached_project.permalink,
remainder=remainder,
include_project=bool(include_project),
)
return cached_project, resolved_path, True
workspace_context = current_workspace_permalink_context()
if workspace_context and project:
workspace_prefix = generate_permalink(workspace_context.workspace_slug)
project_permalink = generate_permalink(_unqualified_project_identifier(project))
qualified_prefix = f"{workspace_prefix}/{project_permalink}"
if normalized_path == qualified_prefix or normalized_path.startswith(
f"{qualified_prefix}/"
):
active_project = await get_active_project(client, project, context, headers)
remainder = (
""
if normalized_path == qualified_prefix
else normalized_path.removeprefix(f"{qualified_prefix}/")
)
resolved_path = _canonical_memory_path_for_workspace(
workspace_slug=workspace_context.workspace_slug,
workspace_type=workspace_context.workspace_type,
project_permalink=project_permalink,
remainder=remainder,
include_project=bool(include_project),
)
return active_project, resolved_path, True
project_prefix, remainder = _split_project_prefix(normalized_path)
include_project = config.permalinks_include_project
# Trigger: memory URL begins with a potential project segment
# Why: allow project-scoped memory URLs without requiring a separate project parameter
# Outcome: attempt to resolve the prefix as a project and route to it
if project_prefix:
cached_project = await _get_cached_active_project(context)
if cached_project and _project_matches_identifier(cached_project, project_prefix):
resolved_project = await resolve_project_parameter(project_prefix, context=context)
if resolved_project and generate_permalink(resolved_project) != generate_permalink(
@@ -975,6 +1218,30 @@ def detect_project_from_url_prefix(identifier: str, config: BasicMemoryConfig) -
return None
async def detect_project_from_memory_url_prefix(
identifier: str,
config: BasicMemoryConfig,
context: Optional[Context] = None,
) -> Optional[str]:
"""Resolve a project from a memory URL prefix, including workspace-qualified URLs."""
if not identifier.strip().startswith("memory://"):
return None
local_project = detect_project_from_url_prefix(identifier, config)
if local_project is not None:
return local_project
if _cloud_workspace_discovery_available(config):
resolution = await resolve_workspace_qualified_memory_url(
identifier,
context=context,
)
if resolution is not None:
return resolution.project_identifier
return None
@asynccontextmanager
async def get_project_client(
project: Optional[str] = None,
@@ -1091,6 +1358,7 @@ async def get_project_client(
if project_entry and project_entry.workspace_id:
# Per-project config stores the cloud tenant id directly
workspace_id = project_entry.workspace_id
active_ws = await _workspace_metadata_by_tenant_id(workspace_id, context=context)
else:
resolved_entry = cloud_default_entry
if resolved_entry is None or not _project_matches_identifier(
@@ -1113,12 +1381,18 @@ async def get_project_client(
workspace_id=workspace_id,
):
logger.debug("Using resolved workspace for cloud project routing")
async with get_client(
project_name=project_for_api,
workspace=workspace_id,
) as client:
active_project = await get_active_project(client, project_for_api, context)
yield client, active_project
permalink_context = (
workspace_permalink_context(active_ws.slug, active_ws.workspace_type)
if active_ws is not None
else nullcontext()
)
with permalink_context:
async with get_client(
project_name=project_for_api,
workspace=workspace_id,
) as client:
active_project = await get_active_project(client, project_for_api, context)
yield client, active_project
return
# Step 4: Local routing (default)
src/basic_memory/mcp/tools/build_context.py
+9 -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 (
detect_project_from_url_prefix,
detect_project_from_memory_url_prefix,
get_project_client,
resolve_project_and_path,
)
@@ -209,9 +209,14 @@ async def build_context(
Raises:
ToolError: If project doesn't exist or depth parameter is invalid
"""
# Detect project from memory URL prefix before routing
if project is None:
detected = detect_project_from_url_prefix(url, ConfigManager().config)
# Detect project from memory URL prefix before routing.
# project_id routes by external UUID, so it bypasses URL discovery entirely.
if project is None and project_id is None:
detected = await detect_project_from_memory_url_prefix(
url,
ConfigManager().config,
context=context,
)
if detected:
project = detected
src/basic_memory/mcp/tools/delete_note.py
+62 -6
View File
@@ -7,8 +7,15 @@ from mcp.server.fastmcp.exceptions import ToolError
from pydantic import AliasChoices, Field
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import detect_project_from_url_prefix, get_project_client
from basic_memory.mcp.project_context import (
detect_project_from_memory_url_prefix,
get_project_client,
resolve_project_and_path,
)
from basic_memory.mcp.server import mcp
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.utils import generate_permalink, normalize_project_reference
from basic_memory.workspace_context import current_workspace_permalink_context
def _format_delete_error_response(project: str, error_message: str, identifier: str) -> str:
@@ -148,6 +155,32 @@ delete_note("{project}", "correct-identifier-from-search")
If the note should be deleted but the operation keeps failing, send a message to support@basicmemory.com."""
def _directory_path_for_delete(
target_identifier: str,
active_project: ProjectItem,
*,
include_project_prefix: bool,
) -> str:
"""Return the project-relative directory path expected by the delete API."""
directory = normalize_project_reference(target_identifier).strip("/")
project_permalink = active_project.permalink
route_prefixes: list[str] = []
workspace_context = current_workspace_permalink_context()
if workspace_context and workspace_context.should_prefix_permalinks:
route_prefixes.append(
f"{generate_permalink(workspace_context.workspace_slug)}/{project_permalink}"
)
if include_project_prefix:
route_prefixes.append(project_permalink)
for route_prefix in route_prefixes:
if directory.startswith(f"{route_prefix}/"):
return directory.removeprefix(f"{route_prefix}/")
return directory
@mcp.tool(
description="Delete a note or directory by title, permalink, or path",
annotations={"destructiveHint": True, "openWorldHint": False},
@@ -231,12 +264,16 @@ async def delete_note(
commands and alternative formats to try.
"""
# Detect project from memory URL prefix before routing
# Trigger: identifier starts with memory:// and no explicit project was provided
# Trigger: identifier starts with memory:// and no explicit project/project_id was provided
# Why: only gate on memory:// to avoid misrouting plain paths like "research/note"
# where "research" is a directory, not a project name
# Outcome: project is set from the URL prefix, routing goes to the correct project
if project is None and identifier.strip().startswith("memory://"):
detected = detect_project_from_url_prefix(identifier, ConfigManager().config)
if project is None and project_id is None and identifier.strip().startswith("memory://"):
detected = await detect_project_from_memory_url_prefix(
identifier,
ConfigManager().config,
context=context,
)
if detected:
project = detected
@@ -253,11 +290,30 @@ async def delete_note(
# Use typed KnowledgeClient for API calls
knowledge_client = KnowledgeClient(client, active_project.external_id)
_, target_identifier, is_memory_url = await resolve_project_and_path(
client,
identifier,
active_project.name,
context,
)
# Handle directory deletes
if is_directory:
try:
result = await knowledge_client.delete_directory(identifier)
# Trigger: directory input was routed from a memory:// URL.
# Why: resolve_project_and_path returns canonical permalinks, while
# delete_directory filters by project-relative file_path prefixes.
# Outcome: strip only the route prefix before calling the delete API.
directory_identifier = (
_directory_path_for_delete(
target_identifier,
active_project,
include_project_prefix=ConfigManager().config.permalinks_include_project,
)
if is_memory_url
else target_identifier
)
result = await knowledge_client.delete_directory(directory_identifier)
if output_format == "json":
return {
"deleted": result.failed_deletes == 0,
@@ -339,7 +395,7 @@ delete_note("path/to/file.md")
note_file_path = None
try:
# Resolve identifier to entity ID
entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
entity_id = await knowledge_client.resolve_entity(target_identifier, strict=True)
if output_format == "json":
entity = await knowledge_client.get_entity(entity_id)
note_title = entity.title
src/basic_memory/mcp/tools/edit_note.py
+19 -5
View File
@@ -9,9 +9,10 @@ from pydantic import AliasChoices, Field
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import (
detect_project_from_url_prefix,
detect_project_from_memory_url_prefix,
get_project_client,
add_project_metadata,
resolve_project_and_path,
)
from basic_memory.mcp.server import mcp
from basic_memory.schemas.base import Entity
@@ -288,12 +289,16 @@ async def edit_note(
effective_replacements = expected_replacements if expected_replacements is not None else 1
# Detect project from memory URL prefix before routing
# Trigger: identifier starts with memory:// and no explicit project was provided
# Trigger: identifier starts with memory:// and no explicit project/project_id was provided
# Why: only gate on memory:// to avoid misrouting plain paths like "research/note"
# where "research" is a directory, not a project name
# Outcome: project is set from the URL prefix, routing goes to the correct project
if project is None and identifier.strip().startswith("memory://"):
detected = detect_project_from_url_prefix(identifier, ConfigManager().config)
if project is None and project_id is None and identifier.strip().startswith("memory://"):
detected = await detect_project_from_memory_url_prefix(
identifier,
ConfigManager().config,
context=context,
)
if detected:
project = detected
@@ -346,6 +351,12 @@ async def edit_note(
# Use typed KnowledgeClient for API calls
knowledge_client = KnowledgeClient(client, active_project.external_id)
_, entity_identifier, _ = await resolve_project_and_path(
client,
identifier,
active_project.name,
context,
)
file_created = False
entity_id = ""
@@ -353,7 +364,10 @@ async def edit_note(
# Try to resolve the entity; for append/prepend, create it if not found
try:
entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
entity_id = await knowledge_client.resolve_entity(
entity_identifier,
strict=True,
)
except Exception as resolve_error:
# Trigger: entity does not exist yet
# Why: append/prepend can meaningfully create a new note from the content,
src/basic_memory/mcp/tools/read_content.py
+9 -4
View File
@@ -18,7 +18,7 @@ from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import (
detect_project_from_url_prefix,
detect_project_from_memory_url_prefix,
get_project_client,
resolve_project_and_path,
)
@@ -217,9 +217,14 @@ async def read_content(
HTTPError: If project doesn't exist or is inaccessible
SecurityError: If path attempts path traversal
"""
# Detect project from memory URL prefix before routing
if project is None:
detected = detect_project_from_url_prefix(path, ConfigManager().config)
# Detect project from memory URL prefix before routing.
# project_id routes by external UUID, so it bypasses URL discovery entirely.
if project is None and project_id is None:
detected = await detect_project_from_memory_url_prefix(
path,
ConfigManager().config,
context=context,
)
if detected:
project = detected
src/basic_memory/mcp/tools/read_note.py
+11 -6
View File
@@ -11,7 +11,7 @@ from fastmcp import Context
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import (
detect_project_from_url_prefix,
detect_project_from_memory_url_prefix,
get_project_client,
resolve_project_and_path,
)
@@ -33,10 +33,10 @@ def _parse_opening_frontmatter(content: str) -> tuple[str, dict | None]:
If parsing fails or frontmatter is not a mapping, returns body unchanged and None.
"""
original_content = content
if not content.startswith("---\n"):
lines = content.splitlines(keepends=True)
if not lines or lines[0].strip() != "---":
return original_content, None
lines = content.splitlines(keepends=True)
closing_index = None
for i in range(1, len(lines)):
if lines[i].strip() == "---":
@@ -130,9 +130,14 @@ async def read_note(
If the exact note isn't found, this tool provides helpful suggestions
including related notes, search commands, and note creation templates.
"""
# Detect project from memory URL prefix before routing
if project is None:
detected = detect_project_from_url_prefix(identifier, ConfigManager().config)
# Detect project from memory URL prefix before routing.
# project_id routes by external UUID, so it bypasses URL discovery entirely.
if project is None and project_id is None:
detected = await detect_project_from_memory_url_prefix(
identifier,
ConfigManager().config,
context=context,
)
if detected:
project = detected
src/basic_memory/mcp/tools/search.py
+9 -4
View File
@@ -13,7 +13,7 @@ from basic_memory.config import ConfigManager
from basic_memory.utils import coerce_dict, coerce_list
from basic_memory.mcp.container import get_container
from basic_memory.mcp.project_context import (
detect_project_from_url_prefix,
detect_project_from_memory_url_prefix,
get_project_client,
resolve_project_and_path,
)
@@ -551,9 +551,14 @@ async def search_notes(
remainder = re.sub(r"\b(AND|OR|NOT)\b", "", remainder).strip()
query = remainder or None
# Detect project from memory URL prefix before routing
if project is None and query is not None:
detected = detect_project_from_url_prefix(query, ConfigManager().config)
# Detect project from memory URL prefix before routing.
# project_id routes by external UUID, so it bypasses URL discovery entirely.
if project is None and project_id is None and query is not None:
detected = await detect_project_from_memory_url_prefix(
query,
ConfigManager().config,
context=context,
)
if detected:
project = detected
src/basic_memory/services/entity_service.py
+11 -5
View File
@@ -48,6 +48,7 @@ from basic_memory.services.exceptions import (
from basic_memory.services.link_resolver import LinkResolver
from basic_memory.services.search_service import SearchService
from basic_memory.utils import build_canonical_permalink
from basic_memory.workspace_context import workspace_slug_for_canonical_permalinks
@dataclass(frozen=True)
@@ -205,15 +206,20 @@ class EntityService(BaseService[EntityModel]):
if self.app_config:
include_project = self.app_config.permalinks_include_project
workspace_permalink = workspace_slug_for_canonical_permalinks()
project_permalink = None
# Trigger: project-prefixed permalinks are enabled
# Why: we need the project slug to build the canonical permalink
# Outcome: fetch and cache the project's permalink
if include_project:
# Trigger: project-prefixed permalinks are enabled, or organization workspace
# context requires a complete workspace/project canonical permalink.
# Why: project slug is the stable middle segment for globally addressable links.
# Outcome: fetch and cache the project's permalink before building the canonical URL.
if include_project or workspace_permalink:
project_permalink = await self._get_project_permalink()
desired_permalink = build_canonical_permalink(
project_permalink, file_path_str, include_project=include_project
project_permalink,
file_path_str,
include_project=include_project,
workspace_permalink=workspace_permalink,
)
# Make unique if needed - enhanced to handle character conflicts
src/basic_memory/utils.py
+30 -3
View File
@@ -224,18 +224,43 @@ def build_canonical_permalink(
project_permalink: Optional[str],
file_path: Union[Path, str, PathLike],
include_project: bool = True,
*,
workspace_permalink: Optional[str] = None,
) -> str:
"""Build a canonical permalink, optionally prefixed with project slug.
"""Build a canonical permalink, optionally prefixed with workspace/project slugs.
Args:
project_permalink: URL-friendly project identifier (slug). If None, no prefix is added.
file_path: Original file path or permalink-like string.
include_project: When True, prefix with project slug.
workspace_permalink: Optional URL-friendly workspace identifier. When provided,
prefix the project-qualified permalink with this workspace slug.
Returns:
Canonical permalink string.
"""
normalized_path = generate_permalink(file_path)
normalized_workspace = generate_permalink(workspace_permalink) if workspace_permalink else None
if normalized_workspace:
if not project_permalink:
raise ValueError("workspace_permalink requires project_permalink")
normalized_project = generate_permalink(project_permalink)
workspace_project_prefix = f"{normalized_workspace}/{normalized_project}"
if normalized_path == workspace_project_prefix or normalized_path.startswith(
f"{workspace_project_prefix}/"
):
return normalized_path
if normalized_path == normalized_project or normalized_path.startswith(
f"{normalized_project}/"
):
project_path = normalized_path
else:
project_path = f"{normalized_project}/{normalized_path}"
return f"{normalized_workspace}/{project_path}"
if not include_project or not project_permalink:
return normalized_path
@@ -244,9 +269,11 @@ def build_canonical_permalink(
if normalized_path == normalized_project or normalized_path.startswith(
f"{normalized_project}/"
):
return normalized_path
project_path = normalized_path
else:
project_path = f"{normalized_project}/{normalized_path}"
return f"{normalized_project}/{normalized_path}"
return project_path
def setup_logging(
src/basic_memory/workspace_context.py
+114
View File
@@ -0,0 +1,114 @@
"""Request-local workspace context for canonical permalink generation."""
import re
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from typing import Iterator
WORKSPACE_SLUG_HEADER = "X-Basic-Memory-Workspace-Slug"
WORKSPACE_TYPE_HEADER = "X-Basic-Memory-Workspace-Type"
_WORKSPACE_SLUG_PATTERN = re.compile(r"^[a-z0-9_-]+$")
_WORKSPACE_TYPES = {"personal", "organization"}
@dataclass(frozen=True)
class WorkspacePermalinkContext:
"""Workspace metadata needed to build canonical organization permalinks."""
workspace_slug: str
workspace_type: str
@property
def should_prefix_permalinks(self) -> bool:
return self.workspace_type == "organization" and bool(self.workspace_slug)
_workspace_permalink_context: ContextVar[WorkspacePermalinkContext | None] = ContextVar(
"basic_memory_workspace_permalink_context",
default=None,
)
def current_workspace_permalink_context() -> WorkspacePermalinkContext | None:
"""Return the active workspace permalink context, when one is set."""
return _workspace_permalink_context.get()
def validate_workspace_permalink_context_values(
workspace_slug: str | None,
workspace_type: str | None,
) -> None:
"""Validate workspace permalink metadata before it can affect stored permalinks."""
validation_error = workspace_permalink_context_validation_error(workspace_slug, workspace_type)
if validation_error is not None:
raise ValueError(validation_error)
def workspace_permalink_context_validation_error(
workspace_slug: str | None,
workspace_type: str | None,
) -> str | None:
"""Return the validation error for workspace permalink metadata, if any."""
if bool(workspace_slug) != bool(workspace_type):
return "workspace_slug and workspace_type must be provided together"
if not workspace_slug or not workspace_type:
return None
if _WORKSPACE_SLUG_PATTERN.fullmatch(workspace_slug) is None:
return f"{WORKSPACE_SLUG_HEADER} must match [a-z0-9_-]+"
if workspace_type not in _WORKSPACE_TYPES:
allowed = ", ".join(sorted(_WORKSPACE_TYPES))
return f"{WORKSPACE_TYPE_HEADER} must be one of: {allowed}"
return None
@contextmanager
def workspace_permalink_context(
workspace_slug: str | None,
workspace_type: str | None,
) -> Iterator[None]:
"""Set request-local workspace permalink metadata.
Cloud can populate this per request without storing workspace metadata in
local project config. The slug/type pair is all permalink generation needs.
"""
validate_workspace_permalink_context_values(workspace_slug, workspace_type)
if not workspace_slug or not workspace_type:
yield
return
token = _workspace_permalink_context.set(
WorkspacePermalinkContext(
workspace_slug=workspace_slug,
workspace_type=workspace_type,
)
)
try:
yield
finally:
_workspace_permalink_context.reset(token)
def workspace_permalink_headers() -> dict[str, str]:
"""Return HTTP headers for forwarding workspace permalink context."""
context = current_workspace_permalink_context()
if context is None:
return {}
return {
WORKSPACE_SLUG_HEADER: context.workspace_slug,
WORKSPACE_TYPE_HEADER: context.workspace_type,
}
def workspace_slug_for_canonical_permalinks() -> str | None:
"""Return the workspace slug when new permalinks should include it."""
context = current_workspace_permalink_context()
if context and context.should_prefix_permalinks:
return context.workspace_slug
return None
tests/api/v2/test_workspace_permalink_headers.py
+41
View File
@@ -0,0 +1,41 @@
"""Tests for workspace permalink context headers."""
import pytest
from httpx import AsyncClient
from basic_memory.workspace_context import WORKSPACE_SLUG_HEADER, WORKSPACE_TYPE_HEADER
@pytest.mark.asyncio
@pytest.mark.parametrize(
"headers, expected_detail",
[
(
{WORKSPACE_SLUG_HEADER: "team-paul"},
"workspace_slug and workspace_type must be provided together",
),
(
{
WORKSPACE_SLUG_HEADER: "../team-paul",
WORKSPACE_TYPE_HEADER: "organization",
},
f"{WORKSPACE_SLUG_HEADER} must match [a-z0-9_-]+",
),
(
{
WORKSPACE_SLUG_HEADER: "team-paul",
WORKSPACE_TYPE_HEADER: "enterprise",
},
f"{WORKSPACE_TYPE_HEADER} must be one of: organization, personal",
),
],
)
async def test_workspace_permalink_headers_fail_fast(
client: AsyncClient,
headers: dict[str, str],
expected_detail: str,
):
response = await client.get("/v2/projects/", headers=headers)
assert response.status_code == 400
assert response.json()["detail"] == expected_detail
tests/conftest.py
+8
View File
@@ -217,6 +217,14 @@ def suppress_logfire_no_config_warning(monkeypatch) -> None:
monkeypatch.setenv("LOGFIRE_IGNORE_NO_CONFIG", "1")
@pytest.fixture(autouse=True)
def isolate_routing_env(monkeypatch) -> None:
"""Prevent command-routing env flags from leaking across tests."""
monkeypatch.delenv("BASIC_MEMORY_FORCE_LOCAL", raising=False)
monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)
monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
@pytest_asyncio.fixture(autouse=True)
async def cleanup_global_db_after_test() -> AsyncGenerator[None, None]:
"""Close any module-level DB engine created outside fixture ownership."""
tests/mcp/test_project_context.py
+844 -7
View File
@@ -612,6 +612,328 @@ async def test_resolve_workspace_project_identifier_handles_qualified_and_collis
assert resolved.project.external_id == "personal-project-id"
@pytest.mark.asyncio
async def test_detect_project_from_memory_url_prefix_resolves_workspace_slug(monkeypatch):
import basic_memory.mcp.project_context as project_context
from basic_memory.config import BasicMemoryConfig
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
detect_project_from_memory_url_prefix,
)
personal = _workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
team = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=personal,
project=_project("main", id=1, external_id="personal-main-id"),
),
WorkspaceProjectEntry(
workspace=team,
project=_project("main", id=2, external_id="team-main-id"),
),
)
index = _build_workspace_project_index((personal, team), entries)
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: True)
resolved = await detect_project_from_memory_url_prefix(
"memory://team-paul/main/notes/foo",
BasicMemoryConfig(projects={}),
)
assert resolved == "team-paul/main"
@pytest.mark.asyncio
async def test_detect_project_from_memory_url_prefix_prefers_local_project_prefix(
monkeypatch,
):
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_memory_url_prefix
async def fail_if_called(context=None): # pragma: no cover
raise AssertionError("Local project-prefixed memory URLs must not discover workspaces")
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_if_called)
resolved = await detect_project_from_memory_url_prefix(
"memory://main/notes/foo",
BasicMemoryConfig(
projects={"main": ProjectEntry(path="/tmp/main")},
cloud_api_key="bmc_test123",
),
)
assert resolved == "main"
@pytest.mark.asyncio
async def test_detect_project_from_memory_url_prefix_skips_workspace_discovery_for_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 detect_project_from_memory_url_prefix
async def fail_if_called(context=None): # pragma: no cover
raise AssertionError("Saved cloud credentials must not force local workspace discovery")
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_if_called)
resolved = await detect_project_from_memory_url_prefix(
"memory://notes/foo/bar",
BasicMemoryConfig(
projects={"main": ProjectEntry(path="/tmp/main")},
cloud_api_key="bmc_test123",
),
)
assert resolved is None
@pytest.mark.asyncio
async def test_resolve_workspace_qualified_memory_url_ignores_workspace_project_miss(
monkeypatch,
):
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_qualified_memory_url,
)
workspace = _workspace(
tenant_id="main-tenant",
workspace_type="organization",
slug="main",
name="Main Workspace",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=workspace,
project=_project("research", id=1, external_id="research-project-id"),
),
)
index = _build_workspace_project_index((workspace,), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
resolved = await resolve_workspace_qualified_memory_url("memory://main/notes/foo")
assert resolved is None
@pytest.mark.asyncio
async def test_resolve_workspace_qualified_memory_url_fails_on_duplicate_project_permalink(
monkeypatch,
):
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_qualified_memory_url,
)
team = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=team,
project=_project("main", id=1, external_id="team-main-id-1"),
),
WorkspaceProjectEntry(
workspace=team,
project=_project("Main", id=2, external_id="team-main-id-2"),
),
)
index = _build_workspace_project_index((team,), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
with pytest.raises(ValueError, match="matched multiple projects"):
await resolve_workspace_qualified_memory_url("memory://team-paul/main/notes/foo")
@pytest.mark.asyncio
async def test_resolve_workspace_qualified_memory_url_uses_personal_canonical_path(
config_manager,
monkeypatch,
):
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_qualified_memory_url,
)
config = config_manager.load_config()
config.permalinks_include_project = True
config_manager.save_config(config)
personal = _workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
entries = (
WorkspaceProjectEntry(
workspace=personal,
project=_project("main", id=1, external_id="personal-main-id"),
),
)
index = _build_workspace_project_index((personal,), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
resolved = await resolve_workspace_qualified_memory_url("memory://personal/main/notes/foo")
assert resolved is not None
assert resolved.canonical_path == "main/notes/foo"
@pytest.mark.asyncio
async def test_resolve_workspace_qualified_memory_url_keeps_org_canonical_path_without_project_prefix(
config_manager,
monkeypatch,
):
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_qualified_memory_url,
)
config = config_manager.load_config()
config.permalinks_include_project = False
config_manager.save_config(config)
team = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=team,
project=_project("main", id=1, external_id="team-main-id"),
),
)
index = _build_workspace_project_index((team,), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
resolved = await resolve_workspace_qualified_memory_url("memory://team-paul/main/notes/foo")
assert resolved is not None
assert resolved.canonical_path == "team-paul/main/notes/foo"
@pytest.mark.asyncio
async def test_get_project_client_routes_duplicate_project_through_workspace_slug(
config_manager,
monkeypatch,
):
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
get_project_client,
)
config = config_manager.load_config()
config.projects = {}
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
_workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
team = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
team_project = _project("main", id=2, external_id="team-main-id")
seen: dict[str, object] = {}
async def fake_resolve_workspace_project_identifier(project_name, context=None):
assert project_name == "team-paul/main"
return WorkspaceProjectEntry(workspace=team, project=team_project)
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
return _project(project_name, id=team_project.id, external_id=team_project.external_id)
monkeypatch.setattr(
"basic_memory.mcp.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: True)
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_active_project",
fake_get_active_project,
)
async with get_project_client(project="team-paul/main") as (_client, active_project):
assert active_project.external_id == "team-main-id"
assert seen == {"project_name": "main", "workspace": "team-tenant"}
@pytest.mark.asyncio
async def test_resolve_workspace_project_identifier_uses_active_workspace_for_duplicate(
monkeypatch,
@@ -713,9 +1035,7 @@ async def test_resolve_workspace_project_identifier_resolves_by_external_id(monk
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
resolved = await resolve_workspace_project_identifier(
"22222222-2222-2222-2222-222222222222"
)
resolved = await resolve_workspace_project_identifier("22222222-2222-2222-2222-222222222222")
assert resolved.workspace.slug == "acme"
assert resolved.project.external_id == "22222222-2222-2222-2222-222222222222"
@@ -1004,6 +1324,135 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
assert is_memory_url is True
@pytest.mark.asyncio
async def test_resolve_project_and_path_keeps_workspace_qualified_canonical_path(
config_manager,
monkeypatch,
):
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import resolve_project_and_path
from basic_memory.schemas.project_info import ProjectItem
config = config_manager.load_config()
config.permalinks_include_project = True
config_manager.save_config(config)
context = _ContextState()
cached_project = ProjectItem(
id=1,
external_id="11111111-1111-1111-1111-111111111111",
name="main",
path="/tmp/main",
is_default=False,
)
team_workspace = _workspace(
tenant_id="team-tenant",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
await context.set_state("active_project", cached_project.model_dump())
await context.set_state("active_workspace", team_workspace.model_dump())
async def fake_call_post(*args, **kwargs):
raise ToolError("project not found")
monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
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),
)
assert active_project == cached_project
assert resolved_path == "team-paul/main/notes/foo"
assert is_memory_url is True
@pytest.mark.asyncio
async def test_resolve_project_and_path_strips_personal_workspace_prefix(
config_manager,
monkeypatch,
):
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import resolve_project_and_path
from basic_memory.schemas.project_info import ProjectItem
config = config_manager.load_config()
config.permalinks_include_project = True
config_manager.save_config(config)
context = _ContextState()
cached_project = ProjectItem(
id=1,
external_id="11111111-1111-1111-1111-111111111111",
name="main",
path="/tmp/main",
is_default=False,
)
personal_workspace = _workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
await context.set_state("active_project", cached_project.model_dump())
await context.set_state("active_workspace", personal_workspace.model_dump())
async def fake_call_post(*args, **kwargs):
raise ToolError("project not found")
monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
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),
)
assert active_project == cached_project
assert resolved_path == "main/notes/foo"
assert is_memory_url is True
@pytest.mark.asyncio
async def test_resolve_project_and_path_preserves_existing_project_prefixed_memory_url(
config_manager,
):
from basic_memory.mcp.project_context import resolve_project_and_path
from basic_memory.schemas.project_info import ProjectItem
config = config_manager.load_config()
config.permalinks_include_project = True
config_manager.save_config(config)
context = _ContextState()
cached_project = ProjectItem(
id=1,
external_id="11111111-1111-1111-1111-111111111111",
name="main",
path="/tmp/main",
is_default=False,
)
await context.set_state("active_project", cached_project.model_dump())
active_project, resolved_path, is_memory_url = await resolve_project_and_path(
client=cast(Any, None),
identifier="memory://main/notes/foo",
context=_ctx(context),
)
assert active_project == cached_project
assert resolved_path == "main/notes/foo"
assert is_memory_url is True
class TestDetectProjectFromUrlPrefix:
"""Test detect_project_from_url_prefix for URL-based project detection."""
@@ -1058,6 +1507,25 @@ class TestDetectProjectFromUrlPrefix:
assert result == "My Research"
@pytest.mark.asyncio
async def test_detect_project_from_memory_url_prefix_ignores_plain_paths(monkeypatch):
import basic_memory.mcp.project_context as project_context
from basic_memory.config import BasicMemoryConfig
from basic_memory.mcp.project_context import detect_project_from_memory_url_prefix
async def fail_if_called(context=None): # pragma: no cover
raise AssertionError("Plain paths must not trigger workspace discovery")
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_if_called)
resolved = await detect_project_from_memory_url_prefix(
"team-paul/main/notes/foo",
BasicMemoryConfig(projects={}),
)
assert resolved is None
class TestGetProjectClientRoutingOrder:
"""Test that get_project_client respects explicit routing before workspace resolution."""
@@ -1090,9 +1558,13 @@ class TestGetProjectClientRoutingOrder:
@pytest.mark.asyncio
async def test_cloud_project_uses_per_project_workspace_id(self, config_manager, monkeypatch):
"""Cloud project with workspace_id in config should use it without network lookup."""
"""Cloud project with workspace_id uses cached workspace permalink context."""
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.workspace_context import (
WorkspacePermalinkContext,
current_workspace_permalink_context,
)
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
@@ -1103,23 +1575,41 @@ class TestGetProjectClientRoutingOrder:
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
from contextlib import asynccontextmanager
from basic_memory.schemas.project_info import ProjectItem
seen: dict[str, object] = {}
workspace = _workspace(
tenant_id="per-project-tenant-id",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
context = _ContextState()
await context.set_state("active_workspace", workspace.model_dump())
async def fail_resolve_workspace_parameter(workspace=None, context=None):
raise AssertionError("Configured workspace_id should route without workspace discovery")
raise AssertionError(
"Configured workspace_id should not prompt for workspace selection"
)
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require workspace discovery")
monkeypatch.setattr(
"basic_memory.mcp.project_context.resolve_workspace_parameter",
fail_resolve_workspace_parameter,
)
monkeypatch.setattr(
"basic_memory.mcp.project_context._ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
seen["permalink_context"] = current_workspace_permalink_context()
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
@@ -1137,10 +1627,357 @@ class TestGetProjectClientRoutingOrder:
fake_get_active_project,
)
async with get_project_client(
project="cloud-proj",
context=_ctx(context),
) as (_client, active_project):
assert active_project.external_id == "cloud-project-id"
assert seen["project_name"] == "cloud-proj"
assert seen["workspace"] == "per-project-tenant-id"
permalink_context = cast(WorkspacePermalinkContext | None, seen["permalink_context"])
assert permalink_context is not None
assert permalink_context.workspace_slug == "team-paul"
assert permalink_context.workspace_type == "organization"
@pytest.mark.asyncio
async def test_cloud_project_workspace_id_uses_workspace_provider_for_permalink_context(
self, config_manager, monkeypatch
):
"""Injected workspace metadata supplies slug/type without project index discovery."""
from contextlib import asynccontextmanager
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.workspace_context import (
WorkspacePermalinkContext,
current_workspace_permalink_context,
)
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "cloud-proj"),
mode=ProjectMode.CLOUD,
workspace_id="per-project-tenant-id",
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
workspace = _workspace(
tenant_id="per-project-tenant-id",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
async def fake_workspace_provider():
return [workspace]
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require project discovery")
monkeypatch.setattr(project_context, "_workspace_provider", fake_workspace_provider)
monkeypatch.setattr(
project_context,
"_ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
seen: dict[str, object] = {}
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
seen["permalink_context"] = current_workspace_permalink_context()
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
return ProjectItem(
id=1,
external_id="cloud-project-id",
name=project_name,
path="/cloud-proj",
is_default=False,
)
monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
async with get_project_client(project="cloud-proj") as (_client, active_project):
assert active_project.external_id == "cloud-project-id"
assert seen == {"project_name": "cloud-proj", "workspace": "per-project-tenant-id"}
assert seen["project_name"] == "cloud-proj"
assert seen["workspace"] == "per-project-tenant-id"
permalink_context = cast(WorkspacePermalinkContext | None, seen["permalink_context"])
assert permalink_context is not None
assert permalink_context.workspace_slug == "team-paul"
assert permalink_context.workspace_type == "organization"
@pytest.mark.asyncio
async def test_cloud_project_workspace_id_uses_cached_available_workspaces_for_permalink_context(
self, config_manager, monkeypatch
):
"""Cached workspace metadata supplies slug/type before the provider is consulted."""
from contextlib import asynccontextmanager
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.workspace_context import (
WorkspacePermalinkContext,
current_workspace_permalink_context,
)
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "cloud-proj"),
mode=ProjectMode.CLOUD,
workspace_id="per-project-tenant-id",
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
workspace = _workspace(
tenant_id="per-project-tenant-id",
workspace_type="organization",
slug="team-paul",
name="Team Paul",
role="editor",
)
context = _ContextState()
await context.set_state("available_workspaces", ["ignored", workspace.model_dump()])
async def fail_workspace_provider(): # pragma: no cover
raise AssertionError("Cached workspace metadata should be used before provider")
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require project discovery")
monkeypatch.setattr(project_context, "_workspace_provider", fail_workspace_provider)
monkeypatch.setattr(
project_context,
"_ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
seen: dict[str, object] = {}
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
seen["permalink_context"] = current_workspace_permalink_context()
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
return ProjectItem(
id=1,
external_id="cloud-project-id",
name=project_name,
path="/cloud-proj",
is_default=False,
)
monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
async with get_project_client(
project="cloud-proj",
context=_ctx(context),
) as (_client, active_project):
assert active_project.external_id == "cloud-project-id"
assert seen["project_name"] == "cloud-proj"
assert seen["workspace"] == "per-project-tenant-id"
permalink_context = cast(WorkspacePermalinkContext | None, seen["permalink_context"])
assert permalink_context is not None
assert permalink_context.workspace_slug == "team-paul"
assert permalink_context.workspace_type == "organization"
@pytest.mark.asyncio
async def test_cloud_project_workspace_id_errors_when_provider_omits_workspace(
self, config_manager, monkeypatch
):
"""Provider metadata must include the configured tenant before slug/type forwarding."""
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "cloud-proj"),
mode=ProjectMode.CLOUD,
workspace_id="per-project-tenant-id",
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
async def fake_workspace_provider():
return [
_workspace(
tenant_id="other-tenant-id",
workspace_type="organization",
slug="other-team",
name="Other Team",
role="editor",
)
]
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require project discovery")
monkeypatch.setattr(project_context, "_workspace_provider", fake_workspace_provider)
monkeypatch.setattr(
project_context,
"_ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
with pytest.raises(
ValueError,
match="Configured workspace_id 'per-project-tenant-id' was not returned",
):
async with get_project_client(project="cloud-proj"):
pass
@pytest.mark.asyncio
async def test_cloud_project_workspace_id_routes_without_discovery(
self, config_manager, monkeypatch
):
"""Configured workspace_id should route even when workspace metadata is unavailable."""
from contextlib import asynccontextmanager
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.workspace_context import current_workspace_permalink_context
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "cloud-proj"),
mode=ProjectMode.CLOUD,
workspace_id="per-project-tenant-id",
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require workspace discovery")
monkeypatch.setattr(
project_context,
"_ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
seen: dict[str, object] = {}
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
seen["permalink_context"] = current_workspace_permalink_context()
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
return ProjectItem(
id=1,
external_id="cloud-project-id",
name=project_name,
path="/cloud-proj",
is_default=False,
)
monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
monkeypatch.setattr(
project_context,
"get_active_project",
fake_get_active_project,
)
async with get_project_client(project="cloud-proj") as (_client, active_project):
assert active_project.external_id == "cloud-project-id"
assert seen["project_name"] == "cloud-proj"
assert seen["workspace"] == "per-project-tenant-id"
assert seen["permalink_context"] is None
@pytest.mark.asyncio
async def test_cloud_project_workspace_id_clears_stale_cached_workspace(
self, config_manager, monkeypatch
):
"""Configured workspace_id must not reuse slug/type from another tenant."""
from contextlib import asynccontextmanager
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.workspace_context import current_workspace_permalink_context
config = config_manager.load_config()
config.projects["cloud-proj"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "cloud-proj"),
mode=ProjectMode.CLOUD,
workspace_id="per-project-tenant-id",
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
context = _ContextState()
stale_workspace = _workspace(
tenant_id="other-tenant-id",
workspace_type="organization",
slug="other-team",
name="Other Team",
role="editor",
)
await context.set_state("active_workspace", stale_workspace.model_dump())
async def fail_ensure_workspace_project_index(context=None): # pragma: no cover
raise AssertionError("Configured workspace_id should not require workspace discovery")
monkeypatch.setattr(
project_context,
"_ensure_workspace_project_index",
fail_ensure_workspace_project_index,
)
seen: dict[str, object] = {}
@asynccontextmanager
async def fake_get_client(project_name=None, workspace=None):
seen["workspace"] = workspace
seen["permalink_context"] = current_workspace_permalink_context()
yield object()
async def fake_get_active_project(client, project_name, context=None, headers=None):
return ProjectItem(
id=1,
external_id="cloud-project-id",
name=project_name,
path="/cloud-proj",
is_default=False,
)
monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
async with get_project_client(
project="cloud-proj",
context=_ctx(context),
) as (_client, active_project):
assert active_project.external_id == "cloud-project-id"
assert seen["workspace"] == "per-project-tenant-id"
assert seen["permalink_context"] is None
assert await context.get_state("active_workspace") is None
@pytest.mark.asyncio
async def test_cloud_project_uses_workspace_project_index(self, config_manager, monkeypatch):
tests/mcp/test_tool_delete_note.py
+103 -3
View File
@@ -150,12 +150,14 @@ async def test_delete_note_detects_project_from_memory_url(client, test_project)
@pytest.mark.asyncio
async def test_delete_note_skips_detection_for_plain_path(client, test_project):
"""delete_note should NOT call detect_project_from_url_prefix for plain path identifiers.
"""delete_note should NOT call detect_project_from_memory_url_prefix for plain paths.
A plain path like 'research/note' should not be misrouted to a project
named 'research' — the 'research' segment is a directory, not a project.
"""
with patch("basic_memory.mcp.tools.delete_note.detect_project_from_url_prefix") as mock_detect:
with patch(
"basic_memory.mcp.tools.delete_note.detect_project_from_memory_url_prefix"
) as mock_detect:
# Use a plain path (no memory:// prefix) — detection should not be called
await delete_note(
identifier="test/nonexistent-note",
@@ -168,10 +170,108 @@ async def test_delete_note_skips_detection_for_plain_path(client, test_project):
@pytest.mark.asyncio
async def test_delete_note_skips_detection_when_project_provided(client, test_project):
"""delete_note should skip URL detection when project is explicitly provided."""
with patch("basic_memory.mcp.tools.delete_note.detect_project_from_url_prefix") as mock_detect:
with patch(
"basic_memory.mcp.tools.delete_note.detect_project_from_memory_url_prefix"
) as mock_detect:
await delete_note(
identifier=f"memory://{test_project.name}/test/some-note",
project=test_project.name,
)
mock_detect.assert_not_called()
@pytest.mark.asyncio
async def test_delete_directory_memory_url_strips_project_prefix(client, test_project):
"""Directory deletes use project-relative file paths after memory:// routing."""
await write_note(
project=test_project.name,
title="Delete Directory Memory URL",
directory="memory-url-dir",
content="# Delete Directory Memory URL\nDelete via project-prefixed memory URL.",
)
result = await delete_note(
identifier=f"memory://{test_project.name}/memory-url-dir",
is_directory=True,
project=test_project.name,
output_format="json",
)
assert isinstance(result, dict)
assert result["deleted"] is True
assert result["total_files"] == 1
assert result["successful_deletes"] == 1
assert result["failed_deletes"] == 0
@pytest.mark.asyncio
async def test_delete_directory_memory_url_keeps_real_project_prefix_when_prefixes_disabled(
client,
test_project,
config_manager,
):
"""When project prefixes are disabled, resolved directory paths are already project-relative."""
config = config_manager.load_config()
config.permalinks_include_project = False
config_manager.save_config(config)
real_directory = f"{test_project.name}/archive"
await write_note(
project=test_project.name,
title="Delete Directory Real Project Prefix",
directory=real_directory,
content="# Delete Directory Real Project Prefix\nDelete target content.",
)
await write_note(
project=test_project.name,
title="Keep Directory Decoy",
directory="archive",
content="# Keep Directory Decoy\nDecoy content.",
)
result = await delete_note(
identifier=f"memory://{test_project.name}/{real_directory}",
is_directory=True,
project=test_project.name,
output_format="json",
)
assert isinstance(result, dict)
assert result["deleted"] is True
assert result["total_files"] == 1
assert result["successful_deletes"] == 1
assert result["failed_deletes"] == 0
deleted = await read_note("Delete Directory Real Project Prefix", project=test_project.name)
assert "Delete target content" not in deleted
decoy = await read_note("Keep Directory Decoy", project=test_project.name)
assert "Decoy content" in decoy
@pytest.mark.asyncio
async def test_delete_directory_workspace_memory_url_strips_route_prefix(client, test_project):
"""Workspace-qualified memory:// directory deletes still hit project-relative files."""
from basic_memory.workspace_context import workspace_permalink_context
directory = "workspace-memory-url-dir"
with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
await write_note(
project=test_project.name,
title="Delete Workspace Directory Memory URL",
directory=directory,
content="# Delete Workspace Directory Memory URL\nDelete via workspace memory URL.",
)
result = await delete_note(
identifier=f"memory://team-paul/{test_project.name}/{directory}",
is_directory=True,
project=test_project.name,
output_format="json",
)
assert isinstance(result, dict)
assert result["deleted"] is True
assert result["total_files"] == 1
assert result["successful_deletes"] == 1
assert result["failed_deletes"] == 0
tests/mcp/test_tool_edit_note.py
+69 -2
View File
@@ -798,6 +798,35 @@ async def test_edit_note_detects_project_from_memory_url(client, test_project):
assert f"project: {test_project.name}" in result
@pytest.mark.asyncio
async def test_edit_note_workspace_qualified_memory_url_keeps_complete_permalink(
client,
test_project,
):
from basic_memory.workspace_context import workspace_permalink_context
expected_permalink = f"team-paul/{test_project.name}/team/tool-edit-note"
with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
await write_note(
project=test_project.name,
title="Tool Edit Note",
directory="team",
content="# Tool Edit Note\nOriginal content.",
)
result = await edit_note(
project=test_project.name,
identifier=f"memory://{expected_permalink}",
operation="append",
content="\nAppended in team workspace.",
)
assert isinstance(result, str)
assert "Edited note (append)" in result
assert f"permalink: {expected_permalink}" in result
@pytest.mark.asyncio
async def test_edit_note_skips_detection_for_plain_path(client, test_project):
"""edit_note should NOT call detect_project_from_url_prefix for plain path identifiers.
@@ -805,7 +834,9 @@ async def test_edit_note_skips_detection_for_plain_path(client, test_project):
A plain path like 'research/note' should not be misrouted to a project
named 'research' — the 'research' segment is a directory, not a project.
"""
with patch("basic_memory.mcp.tools.edit_note.detect_project_from_url_prefix") as mock_detect:
with patch(
"basic_memory.mcp.tools.edit_note.detect_project_from_memory_url_prefix"
) as mock_detect:
# Use a plain path (no memory:// prefix) — detection should not be called
await edit_note(
identifier="test/some-note",
@@ -820,7 +851,9 @@ async def test_edit_note_skips_detection_for_plain_path(client, test_project):
@pytest.mark.asyncio
async def test_edit_note_skips_detection_when_project_provided(client, test_project):
"""edit_note should skip URL detection when project is explicitly provided."""
with patch("basic_memory.mcp.tools.edit_note.detect_project_from_url_prefix") as mock_detect:
with patch(
"basic_memory.mcp.tools.edit_note.detect_project_from_memory_url_prefix"
) as mock_detect:
await edit_note(
identifier=f"memory://{test_project.name}/test/some-note",
operation="append",
@@ -829,3 +862,37 @@ async def test_edit_note_skips_detection_when_project_provided(client, test_proj
)
mock_detect.assert_not_called()
@pytest.mark.asyncio
async def test_edit_note_skips_detection_when_project_id_provided(
monkeypatch,
client,
test_project,
):
"""project_id is authoritative, so memory URL discovery must not run first."""
await write_note(
project=test_project.name,
title="Project ID Memory URL Edit",
directory="test",
content="# Project ID Memory URL Edit\nOriginal content.",
)
async def fail_if_called(*args, **kwargs):
raise AssertionError("project_id routing should bypass URL discovery")
import importlib
edit_note_module = importlib.import_module("basic_memory.mcp.tools.edit_note")
monkeypatch.setattr(edit_note_module, "detect_project_from_memory_url_prefix", fail_if_called)
result = await edit_note(
identifier=f"memory://{test_project.name}/test/project-id-memory-url-edit",
project_id=test_project.external_id,
operation="append",
content="\nAppended via project_id.",
)
assert isinstance(result, str)
assert "Edited note (append)" in result
assert f"project: {test_project.name}" in result
tests/mcp/test_tool_read_note.py
+171
View File
@@ -5,9 +5,20 @@ from textwrap import dedent
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
def test_parse_opening_frontmatter_handles_crlf():
"""JSON read_note output should strip frontmatter from Windows-written markdown."""
body, frontmatter = _parse_opening_frontmatter(
"---\r\ntitle: Windows Note\r\ntype: note\r\n---\r\n\r\nBody text\r\n"
)
assert frontmatter == {"title": "Windows Note", "type": "note"}
assert body.strip() == "Body text"
@pytest.mark.asyncio
async def test_read_note_by_title(app, test_project):
"""Test reading a note by its title."""
@@ -261,6 +272,166 @@ async def test_read_note_memory_url_with_project_prefix(app, test_project):
assert "Testing memory:// URL handling with project prefix" in content
@pytest.mark.asyncio
async def test_read_note_skips_url_detection_when_project_id_provided(
monkeypatch,
app,
test_project,
):
"""project_id is authoritative, so memory URL discovery must not run first."""
await write_note(
project=test_project.name,
title="Project ID Memory URL Read",
directory="test",
content="Read by project id without discovery",
)
async def fail_if_called(*args, **kwargs):
raise AssertionError("project_id routing should bypass URL discovery")
import importlib
read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
monkeypatch.setattr(read_note_module, "detect_project_from_memory_url_prefix", fail_if_called)
result = await read_note(
f"memory://{test_project.name}/test/project-id-memory-url-read",
project_id=test_project.external_id,
output_format="json",
)
assert isinstance(result, dict)
assert result["content"].strip() == "Read by project id without discovery"
@pytest.mark.asyncio
async def test_team_workspace_write_stores_complete_canonical_permalink(
app,
test_project,
entity_repository,
):
from basic_memory.workspace_context import workspace_permalink_context
expected_permalink = f"team-paul/{test_project.name}/team/team-workspace-note"
with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
write_result = await write_note(
project=test_project.name,
title="Team Workspace Note",
directory="team",
content="Team canonical content",
)
assert f"permalink: {expected_permalink}" in write_result
stored = await entity_repository.get_by_permalink(expected_permalink)
assert stored is not None
assert stored.permalink == expected_permalink
read_result = await read_note(
expected_permalink,
project=test_project.name,
output_format="json",
)
assert isinstance(read_result, dict)
assert read_result["permalink"] == expected_permalink
assert read_result["content"].strip() == "Team canonical content"
@pytest.mark.asyncio
async def test_team_workspace_write_stores_complete_permalink_when_project_prefixes_disabled(
app,
test_project,
entity_repository,
config_manager,
):
from basic_memory.workspace_context import workspace_permalink_context
config = config_manager.load_config()
config.permalinks_include_project = False
config_manager.save_config(config)
expected_permalink = f"team-paul/{test_project.name}/team/team-no-project-prefix-note"
with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
write_result = await write_note(
project=test_project.name,
title="Team No Project Prefix Note",
directory="team",
content="Team canonical content without project-prefix config",
)
assert f"permalink: {expected_permalink}" in write_result
stored = await entity_repository.get_by_permalink(expected_permalink)
assert stored is not None
assert stored.permalink == expected_permalink
read_result = await read_note(
f"memory://{expected_permalink}",
project=test_project.name,
output_format="json",
)
assert isinstance(read_result, dict)
assert read_result["permalink"] == expected_permalink
assert read_result["content"].strip() == "Team canonical content without project-prefix config"
@pytest.mark.asyncio
async def test_personal_workspace_write_keeps_project_scoped_permalink(
app,
test_project,
entity_repository,
):
from basic_memory.workspace_context import workspace_permalink_context
expected_permalink = f"{test_project.name}/personal/personal-workspace-note"
with workspace_permalink_context(workspace_slug="personal", workspace_type="personal"):
write_result = await write_note(
project=test_project.name,
title="Personal Workspace Note",
directory="personal",
content="Personal canonical content",
)
assert f"permalink: {expected_permalink}" in write_result
stored = await entity_repository.get_by_permalink(expected_permalink)
assert stored is not None
assert stored.permalink == expected_permalink
@pytest.mark.asyncio
async def test_read_note_workspace_qualified_memory_url_uses_complete_permalink(
app,
test_project,
):
from basic_memory.workspace_context import workspace_permalink_context
expected_permalink = f"team-paul/{test_project.name}/team/tool-read-note"
with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
await write_note(
project=test_project.name,
title="Tool Read Note",
directory="team",
content="Workspace URL read content",
)
result = await read_note(
f"memory://{expected_permalink}",
project=test_project.name,
output_format="json",
)
assert isinstance(result, dict)
assert result["permalink"] == expected_permalink
assert result["content"].strip() == "Workspace URL read content"
@pytest.mark.asyncio
async def test_read_note_memory_url_fallback_uses_search_tool_normalization(
monkeypatch, app, test_project
tests/utils/test_permalink_formatting.py
+55 -1
View File
@@ -7,7 +7,8 @@ import pytest
from basic_memory.config import ProjectConfig
from basic_memory.services import EntityService
from basic_memory.sync.sync_service import SyncService
from basic_memory.utils import generate_permalink
from basic_memory.utils import build_canonical_permalink, generate_permalink
from basic_memory.workspace_context import workspace_permalink_context
async def create_test_file(path: Path, content: str = "test content") -> None:
@@ -124,3 +125,56 @@ def test_chinese_character_preservation(input_path, expected):
def test_mixed_character_sets(input_path, expected):
"""Test handling of mixed character sets and edge cases."""
assert generate_permalink(input_path) == expected
def test_build_canonical_permalink_prefixes_same_workspace_and_project_slug():
"""Workspace and project slugs may be equal but remain distinct permalink segments."""
assert (
build_canonical_permalink(
"acme",
"notes/foo.md",
workspace_permalink="acme",
)
== "acme/acme/notes/foo"
)
def test_build_canonical_permalink_preserves_complete_workspace_prefix():
"""Already workspace-qualified canonical paths should not gain duplicate prefixes."""
assert (
build_canonical_permalink(
"main",
"team-paul/main/notes/foo.md",
workspace_permalink="team-paul",
)
== "team-paul/main/notes/foo"
)
def test_build_canonical_permalink_workspace_prefix_ignores_project_prefix_flag():
"""Organization workspace canonical permalinks always include workspace and project."""
assert (
build_canonical_permalink(
"main",
"notes/foo.md",
include_project=False,
workspace_permalink="team-paul",
)
== "team-paul/main/notes/foo"
)
@pytest.mark.asyncio
async def test_entity_service_workspace_permalink_uses_project_when_prefixes_disabled(
entity_service: EntityService,
project_config: ProjectConfig,
):
"""Workspace note creation uses complete canonical shape even without local project prefixes."""
assert entity_service.app_config is not None
entity_service.app_config.permalinks_include_project = False
with workspace_permalink_context("team-paul", "organization"):
permalink = await entity_service.resolve_permalink("team/no-project-prefix-service.md")
project_permalink = generate_permalink(project_config.name)
assert permalink == f"team-paul/{project_permalink}/team/no-project-prefix-service"
tests/utils/test_workspace_context.py
+44
View File
@@ -0,0 +1,44 @@
"""Tests for request-local workspace permalink context."""
import pytest
from basic_memory.workspace_context import (
WORKSPACE_SLUG_HEADER,
WORKSPACE_TYPE_HEADER,
current_workspace_permalink_context,
workspace_permalink_context,
workspace_permalink_headers,
)
def test_workspace_permalink_context_requires_slug_and_type_together():
with pytest.raises(ValueError, match="provided together"):
with workspace_permalink_context("team-paul", None):
pass
def test_workspace_permalink_context_rejects_unsafe_slug():
with pytest.raises(ValueError, match=WORKSPACE_SLUG_HEADER):
with workspace_permalink_context("../team-paul", "organization"):
pass
def test_workspace_permalink_context_rejects_unknown_type():
with pytest.raises(ValueError, match=WORKSPACE_TYPE_HEADER):
with workspace_permalink_context("team-paul", "enterprise"):
pass
def test_workspace_permalink_headers_reflect_active_context():
assert workspace_permalink_headers() == {}
with workspace_permalink_context("team-paul", "organization"):
context = current_workspace_permalink_context()
assert context is not None
assert context.workspace_slug == "team-paul"
assert workspace_permalink_headers() == {
WORKSPACE_SLUG_HEADER: "team-paul",
WORKSPACE_TYPE_HEADER: "organization",
}
assert current_workspace_permalink_context() is None