Compare commits

..

2 Commits

Author SHA1 Message Date
Paul Hernandez 1915a6b0a3 Revert "feat: Add project-prefixed permalinks and memory URL routing (#544)"
This reverts commit 545804f194.
2026-02-13 17:46:31 -06:00
Paul Hernandez 545804f194 feat: Add project-prefixed permalinks and memory URL routing (#544)
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-13 17:45:50 -06:00
47 changed files with 462 additions and 1142 deletions
+20 -3
View File
@@ -37,7 +37,18 @@ jobs:
run: |
pip install uv
- uses: extractions/setup-just@v3
- name: Install just (Linux)
if: runner.os != 'Windows'
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Install just (Windows)
if: runner.os == 'Windows'
run: |
# Install just using Chocolatey (pre-installed on GitHub Actions Windows runners)
choco install just --yes
shell: pwsh
- name: Create virtual env
run: |
@@ -85,7 +96,10 @@ jobs:
run: |
pip install uv
- uses: extractions/setup-just@v3
- name: Install just
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Create virtual env
run: |
@@ -119,7 +133,10 @@ jobs:
run: |
pip install uv
- uses: extractions/setup-just@v3
- name: Install just
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Create virtual env
run: |
@@ -60,9 +60,7 @@ def import_chatgpt(
console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
# Create importer and run import
importer = ChatGPTImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
importer = ChatGPTImporter(config.home, markdown_processor, file_service)
with conversations_json.open("r", encoding="utf-8") as file:
json_data = json.load(file)
result = run_with_cleanup(importer.import_data(json_data, folder))
@@ -57,9 +57,7 @@ def import_claude(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer
importer = ClaudeConversationsImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
importer = ClaudeConversationsImporter(config.home, markdown_processor, file_service)
# Process the file
base_path = config.home / folder
@@ -56,9 +56,7 @@ def import_projects(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer
importer = ClaudeProjectsImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
importer = ClaudeProjectsImporter(config.home, markdown_processor, file_service)
# Process the file
base_path = config.home / base_folder if base_folder else config.home
@@ -55,9 +55,7 @@ def memory_json(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer
importer = MemoryJsonImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
importer = MemoryJsonImporter(config.home, markdown_processor, file_service)
# Process the file
base_path = config.home if not destination_folder else config.home / destination_folder
-5
View File
@@ -196,11 +196,6 @@ class BasicMemoryConfig(BaseSettings):
description="Disable automatic permalink generation in frontmatter. When enabled, new notes won't have permalinks added and sync won't update permalinks. Existing permalinks will still work for reading.",
)
permalinks_include_project: bool = Field(
default=True,
description="When True, generated permalinks are prefixed with the project slug (e.g., 'specs/search'). Existing permalinks remain unchanged unless explicitly updated.",
)
skip_initialization_sync: bool = Field(
default=False,
description="Skip expensive initialization synchronization. Useful for cloud/stateless deployments where project reconciliation is not needed.",
+12 -72
View File
@@ -41,12 +41,7 @@ async def get_chatgpt_importer(
file_service: FileServiceDep,
) -> ChatGPTImporter:
"""Create ChatGPTImporter with dependencies."""
return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ChatGPTImporter(project_config.home, markdown_processor, file_service)
ChatGPTImporterDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer)]
@@ -58,12 +53,7 @@ async def get_chatgpt_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep,
) -> ChatGPTImporter:
"""Create ChatGPTImporter with v2 dependencies."""
return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ChatGPTImporter(project_config.home, markdown_processor, file_service)
ChatGPTImporterV2Dep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2)]
@@ -75,12 +65,7 @@ async def get_chatgpt_importer_v2_external(
file_service: FileServiceV2ExternalDep,
) -> ChatGPTImporter:
"""Create ChatGPTImporter with v2 external_id dependencies."""
return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ChatGPTImporter(project_config.home, markdown_processor, file_service)
ChatGPTImporterV2ExternalDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2_external)]
@@ -95,12 +80,7 @@ async def get_claude_conversations_importer(
file_service: FileServiceDep,
) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with dependencies."""
return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service)
ClaudeConversationsImporterDep = Annotated[
@@ -114,12 +94,7 @@ async def get_claude_conversations_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep,
) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with v2 dependencies."""
return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service)
ClaudeConversationsImporterV2Dep = Annotated[
@@ -133,12 +108,7 @@ async def get_claude_conversations_importer_v2_external(
file_service: FileServiceV2ExternalDep,
) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with v2 external_id dependencies."""
return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service)
ClaudeConversationsImporterV2ExternalDep = Annotated[
@@ -155,12 +125,7 @@ async def get_claude_projects_importer(
file_service: FileServiceDep,
) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with dependencies."""
return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service)
ClaudeProjectsImporterDep = Annotated[ClaudeProjectsImporter, Depends(get_claude_projects_importer)]
@@ -172,12 +137,7 @@ async def get_claude_projects_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep,
) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with v2 dependencies."""
return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service)
ClaudeProjectsImporterV2Dep = Annotated[
@@ -191,12 +151,7 @@ async def get_claude_projects_importer_v2_external(
file_service: FileServiceV2ExternalDep,
) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with v2 external_id dependencies."""
return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service)
ClaudeProjectsImporterV2ExternalDep = Annotated[
@@ -213,12 +168,7 @@ async def get_memory_json_importer(
file_service: FileServiceDep,
) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with dependencies."""
return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return MemoryJsonImporter(project_config.home, markdown_processor, file_service)
MemoryJsonImporterDep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer)]
@@ -230,12 +180,7 @@ async def get_memory_json_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep,
) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with v2 dependencies."""
return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return MemoryJsonImporter(project_config.home, markdown_processor, file_service)
MemoryJsonImporterV2Dep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer_v2)]
@@ -247,12 +192,7 @@ async def get_memory_json_importer_v2_external(
file_service: FileServiceV2ExternalDep,
) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with v2 external_id dependencies."""
return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
return MemoryJsonImporter(project_config.home, markdown_processor, file_service)
MemoryJsonImporterV2ExternalDep = Annotated[
-24
View File
@@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Any, Optional, TypeVar
from basic_memory.markdown.markdown_processor import MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown
from basic_memory.schemas.importer import ImportResult
from basic_memory.utils import build_canonical_permalink, generate_permalink
if TYPE_CHECKING: # pragma: no cover
from basic_memory.services.file_service import FileService
@@ -30,7 +29,6 @@ class Importer[T: ImportResult]:
base_path: Path,
markdown_processor: MarkdownProcessor,
file_service: "FileService",
project_name: Optional[str] = None,
):
"""Initialize the import service.
@@ -42,8 +40,6 @@ class Importer[T: ImportResult]:
self.base_path = base_path.resolve() # Get absolute path
self.markdown_processor = markdown_processor
self.file_service = file_service
self.project_name = project_name
self.project_permalink = generate_permalink(project_name) if project_name else None
@abstractmethod
async def import_data(self, source_data, destination_folder: str, **kwargs: Any) -> T:
@@ -77,26 +73,6 @@ class Importer[T: ImportResult]:
# FileService.write_file handles directory creation and returns checksum
return await self.file_service.write_file(file_path, content)
def canonical_permalink(self, path: str) -> str:
"""Build a canonical permalink for imported content."""
include_project = True
# Trigger: importer has app config with permalink prefixing flag
# Why: imported notes should align with canonical permalink format
# Outcome: include project prefix when enabled
if self.file_service.app_config is not None:
include_project = self.file_service.app_config.permalinks_include_project
return build_canonical_permalink(
self.project_permalink,
path,
include_project=include_project,
)
def build_import_paths(self, path: str) -> tuple[str, str]:
"""Return (permalink, file_path) for an imported entity."""
permalink = self.canonical_permalink(path)
return permalink, f"{path}.md"
async def ensure_folder_exists(self, folder: str) -> None:
"""Ensure folder exists using FileService.
+8 -13
View File
@@ -51,20 +51,11 @@ class ChatGPTImporter(Importer[ChatImportResult]):
chats_imported = 0
for chat in conversations:
created_at = chat["create_time"]
date_prefix = datetime.fromtimestamp(created_at).astimezone().strftime("%Y%m%d")
clean_title = clean_filename(chat["title"])
relative_path = (
f"{destination_folder}/{date_prefix}-{clean_title}"
if destination_folder
else f"{date_prefix}-{clean_title}"
)
permalink, file_path = self.build_import_paths(relative_path)
# Convert to entity
entity = self._format_chat_content(chat, permalink)
entity = self._format_chat_content(destination_folder, chat)
# Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path)
# Count messages
@@ -92,7 +83,7 @@ class ChatGPTImporter(Importer[ChatImportResult]):
return self.handle_error("Failed to import ChatGPT conversations", e)
def _format_chat_content(
self, conversation: Dict[str, Any], permalink: str
self, folder: str, conversation: Dict[str, Any]
) -> EntityMarkdown: # pragma: no cover
"""Convert chat conversation to Basic Memory entity.
@@ -114,6 +105,10 @@ class ChatGPTImporter(Importer[ChatImportResult]):
root_id = node_id
break
# Generate permalink
date_prefix = datetime.fromtimestamp(created_at).astimezone().strftime("%Y%m%d")
clean_title = clean_filename(conversation["title"])
# Format content
content = self._format_chat_markdown(
title=conversation["title"],
@@ -131,7 +126,7 @@ class ChatGPTImporter(Importer[ChatImportResult]):
"title": conversation["title"],
"created": format_timestamp(created_at),
"modified": format_timestamp(modified_at),
"permalink": permalink,
"permalink": f"{folder}/{date_prefix}-{clean_title}",
}
),
content=content,
@@ -54,27 +54,18 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
for chat in conversations:
# Get name, providing default for unnamed conversations
chat_name = chat.get("name") or f"Conversation {chat.get('uuid', 'untitled')}"
date_prefix = datetime.fromisoformat(chat["created_at"].replace("Z", "+00:00")).strftime(
"%Y%m%d"
)
clean_title = clean_filename(chat_name)
relative_path = (
f"{destination_folder}/{date_prefix}-{clean_title}"
if destination_folder
else f"{date_prefix}-{clean_title}"
)
permalink, file_path = self.build_import_paths(relative_path)
# Convert to entity
entity = self._format_chat_content(
folder=destination_folder,
name=chat_name,
messages=chat["chat_messages"],
created_at=chat["created_at"],
modified_at=chat["updated_at"],
permalink=permalink,
)
# Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path)
chats_imported += 1
@@ -93,11 +84,11 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
def _format_chat_content(
self,
folder: str,
name: str,
messages: List[Dict[str, Any]],
created_at: str,
modified_at: str,
permalink: str,
) -> EntityMarkdown:
"""Convert chat messages to Basic Memory entity format.
@@ -111,6 +102,11 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
Returns:
EntityMarkdown instance representing the conversation.
"""
# Generate permalink using folder name (relative path)
date_prefix = datetime.fromisoformat(created_at.replace("Z", "+00:00")).strftime("%Y%m%d")
clean_title = clean_filename(name)
permalink = f"{folder}/{date_prefix}-{clean_title}"
# Format content
content = self._format_chat_markdown(
name=name,
@@ -63,28 +63,17 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
await self.file_service.ensure_directory(docs_dir)
# Import prompt template if it exists
if project.get("prompt_template"):
prompt_path = (
f"{destination_folder}/{project_dir}/prompt-template"
if destination_folder
else f"{project_dir}/prompt-template"
)
permalink, file_path = self.build_import_paths(prompt_path)
prompt_entity = self._format_prompt_markdown(project, permalink)
if prompt_entity:
await self.write_entity(prompt_entity, file_path)
if prompt_entity := self._format_prompt_markdown(project, destination_folder):
# Write file using relative path - FileService handles base_path
file_path = f"{prompt_entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(prompt_entity, file_path)
prompts_imported += 1
# Import project documents
for doc in project.get("docs", []):
doc_file = clean_filename(doc["filename"])
doc_path = (
f"{destination_folder}/{project_dir}/docs/{doc_file}"
if destination_folder
else f"{project_dir}/docs/{doc_file}"
)
permalink, file_path = self.build_import_paths(doc_path)
entity = self._format_project_markdown(project, doc, permalink)
entity = self._format_project_markdown(project, doc, destination_folder)
# Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path)
docs_imported += 1
@@ -100,7 +89,7 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
return self.handle_error("Failed to import Claude projects", e)
def _format_project_markdown(
self, project: Dict[str, Any], doc: Dict[str, Any], permalink: str
self, project: Dict[str, Any], doc: Dict[str, Any], destination_folder: str = ""
) -> EntityMarkdown:
"""Format a project document as a Basic Memory entity.
@@ -116,6 +105,17 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
created_at = doc.get("created_at") or project["created_at"]
modified_at = project["updated_at"]
# Generate clean names for organization
project_dir = clean_filename(project["name"])
doc_file = clean_filename(doc["filename"])
# Build permalink with optional destination folder prefix
permalink = (
f"{destination_folder}/{project_dir}/docs/{doc_file}"
if destination_folder
else f"{project_dir}/docs/{doc_file}"
)
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
@@ -136,7 +136,7 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
return entity
def _format_prompt_markdown(
self, project: Dict[str, Any], permalink: str
self, project: Dict[str, Any], destination_folder: str = ""
) -> Optional[EntityMarkdown]:
"""Format project prompt template as a Basic Memory entity.
@@ -155,6 +155,16 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
created_at = project["created_at"]
modified_at = project["updated_at"]
# Generate clean project directory name
project_dir = clean_filename(project["name"])
# Build permalink with optional destination folder prefix
permalink = (
f"{destination_folder}/{project_dir}/prompt-template"
if destination_folder
else f"{project_dir}/prompt-template"
)
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
@@ -80,12 +80,11 @@ class MemoryJsonImporter(Importer[EntityImportResult]):
entity_type = entity_data.get("entityType") or entity_data.get("type") or "entity"
# Build permalink with optional destination folder prefix
relative_path = (
permalink = (
f"{destination_folder}/{entity_type}/{name}"
if destination_folder
else f"{entity_type}/{name}"
)
permalink, file_path = self.build_import_paths(relative_path)
# Ensure entity type directory exists using FileService with relative path
entity_type_dir = (
@@ -110,6 +109,7 @@ class MemoryJsonImporter(Importer[EntityImportResult]):
)
# Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path)
entities_created += 1
+2 -4
View File
@@ -1,8 +1,6 @@
"""Markdown-it plugins for Basic Memory markdown parsing."""
from typing import List, Any, Dict
from basic_memory.utils import normalize_project_reference
from markdown_it import MarkdownIt
from markdown_it.token import Token
@@ -116,7 +114,7 @@ def parse_relation(token: Token) -> Dict[str, Any] | None:
rel_type = before
# Get target
target = normalize_project_reference(content[start + 2 : end].strip())
target = content[start + 2 : end].strip()
# Look for context after
after = content[end + 2 :].strip()
@@ -162,7 +160,7 @@ def parse_inline_relations(content: str) -> List[Dict[str, Any]]:
# No matching ]] found
break
target = normalize_project_reference(content[start + 2 : end].strip())
target = content[start + 2 : end].strip()
if target:
relations.append({"type": "links_to", "target": target, "context": None})
-95
View File
@@ -17,14 +17,11 @@ from httpx._types import (
)
from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.config import ConfigManager
from basic_memory.project_resolver import ProjectResolver
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
async def resolve_project_parameter(
@@ -153,98 +150,6 @@ async def get_active_project(
return active_project
def _split_project_prefix(path: str) -> tuple[Optional[str], str]:
"""Split a possible project prefix from a memory URL path."""
if "/" not in path:
return None, path
project_prefix, remainder = path.split("/", 1)
if not project_prefix or not remainder:
return None, path
if "*" in project_prefix:
return None, path
return project_prefix, remainder
async def resolve_project_and_path(
client: AsyncClient,
identifier: str,
project: Optional[str] = None,
context: Optional[Context] = None,
headers: HeaderTypes | None = None,
) -> tuple[ProjectItem, str, bool]:
"""Resolve project and normalized path for memory:// identifiers.
Returns:
Tuple of (active_project, normalized_path, is_memory_url)
"""
is_memory_url = identifier.strip().startswith("memory://")
if not is_memory_url:
active_project = await get_active_project(client, project, context, headers)
return active_project, identifier, False
normalized_path = normalize_project_reference(memory_url_path(identifier))
project_prefix, remainder = _split_project_prefix(normalized_path)
include_project = ConfigManager().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:
try:
from basic_memory.mcp.tools.utils import call_post
response = await call_post(
client,
"/v2/projects/resolve",
json={"identifier": project_prefix},
headers=headers,
)
resolved = ProjectResolveResponse.model_validate(response.json())
except ToolError as exc:
if "project not found" not in str(exc).lower():
raise
else:
resolved_project = await resolve_project_parameter(project_prefix)
if resolved_project and generate_permalink(resolved_project) != generate_permalink(
project_prefix
):
raise ValueError(
f"Project is constrained to '{resolved_project}', cannot use '{project_prefix}'."
)
active_project = ProjectItem(
id=resolved.project_id,
external_id=resolved.external_id,
name=resolved.name,
path=resolved.path,
is_default=resolved.is_default,
)
if context:
context.set_state("active_project", active_project)
resolved_path = (
f"{resolved.permalink}/{remainder}" if include_project else remainder
)
return active_project, resolved_path, True
# Trigger: no resolvable project prefix in the memory URL
# Why: preserve existing memory URL behavior within the active project
# Outcome: use the active project and normalize the path for lookup
active_project = await get_active_project(client, project, context, headers)
resolved_path = normalized_path
if include_project:
# Trigger: project-prefixed permalinks are enabled and the path lacks a prefix
# Why: ensure memory URL lookups align with canonical permalinks
# Outcome: prefix the path with the active project's permalink
project_prefix = active_project.permalink
if resolved_path != project_prefix and not resolved_path.startswith(f"{project_prefix}/"):
resolved_path = f"{project_prefix}/{resolved_path}"
return active_project, resolved_path, True
def add_project_metadata(result: str, project_name: str) -> str:
"""Add project context as metadata footer for assistant session tracking.
+7 -8
View File
@@ -5,10 +5,14 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import GraphContext, MemoryUrl
from basic_memory.schemas.memory import (
GraphContext,
MemoryUrl,
memory_url_path,
)
@mcp.tool(
@@ -96,18 +100,13 @@ async def build_context(
# URL is already validated and normalized by MemoryUrl type annotation
async with get_project_client(project, context) as (client, active_project):
# Resolve memory:// identifier with project-prefix awareness
_, resolved_path, _ = await resolve_project_and_path(
client, url, project, context
)
# Import here to avoid circular import
from basic_memory.mcp.clients import MemoryClient
# Use typed MemoryClient for API calls
memory_client = MemoryClient(client, active_project.external_id)
return await memory_client.build_context(
resolved_path,
memory_url_path(url),
depth=depth or 1,
timeframe=timeframe,
page=page,
+3 -9
View File
@@ -15,7 +15,7 @@ from PIL import Image as PILImage
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get, resolve_entity_id
from basic_memory.schemas.memory import memory_url_path
@@ -202,17 +202,11 @@ async def read_content(
logger.info("Reading file", path=path, project=project)
async with get_project_client(project, context) as (client, active_project):
# Resolve path with project-prefix awareness for memory:// URLs
_, url, _ = await resolve_project_and_path(client, path, project, context)
url = memory_url_path(path)
# Validate path to prevent path traversal attacks
# For memory:// URLs, validate the extracted path (not the raw URL which
# has a scheme prefix that confuses path validation)
raw_path = memory_url_path(path) if path.startswith("memory://") else path
project_path = active_project.home
if not validate_project_path(raw_path, project_path) or not validate_project_path(
url, project_path
):
if not validate_project_path(url, project_path):
logger.warning(
"Attempted path traversal attack blocked",
path=path,
+7 -19
View File
@@ -6,7 +6,7 @@ from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.formatting import format_note_preview_ascii
from basic_memory.mcp.tools.search import search_notes
@@ -82,19 +82,12 @@ async def read_note(
including related notes, search commands, and note creation templates.
"""
async with get_project_client(project, context) as (client, active_project):
# Resolve identifier with project-prefix awareness for memory:// URLs
_, entity_path, _ = await resolve_project_and_path(
client, identifier, project, context
)
# Validate identifier to prevent path traversal attacks
# For memory:// URLs, validate the extracted path (not the raw URL which
# has a scheme prefix that confuses path validation)
raw_path = memory_url_path(identifier) if identifier.startswith("memory://") else identifier
processed_path = entity_path
# We need to check both the raw identifier and the processed path
processed_path = memory_url_path(identifier)
project_path = active_project.home
if not validate_project_path(raw_path, project_path) or not validate_project_path(
if not validate_project_path(identifier, project_path) or not validate_project_path(
processed_path, project_path
):
logger.warning(
@@ -106,6 +99,7 @@ async def read_note(
return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries"
# Get the file via REST API - first try direct identifier resolution
entity_path = memory_url_path(identifier)
logger.info(
f"Attempting to read note from Project: {active_project.name} identifier: {entity_path}"
)
@@ -141,10 +135,7 @@ async def read_note(
# Fallback 1: Try title search via API
logger.info(f"Search title for: {identifier}")
title_results = await search_notes.fn(
query=identifier,
search_type="title",
project=active_project.name,
context=context,
query=identifier, search_type="title", project=project, context=context
)
# Handle both SearchResponse object and error strings
@@ -179,10 +170,7 @@ async def read_note(
# Fallback 2: Text search as a last resort
logger.info(f"Title search failed, trying text search for: {identifier}")
text_results = await search_notes.fn(
query=identifier,
search_type="text",
project=active_project.name,
context=context,
query=identifier, search_type="text", project=project, context=context
)
# We didn't find a direct match, construct a helpful error message
+36 -44
View File
@@ -6,7 +6,7 @@ from typing import List, Optional, Dict, Any, Literal
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.formatting import format_search_results_ascii
from basic_memory.mcp.server import mcp
from basic_memory.schemas.search import (
@@ -399,50 +399,42 @@ async def search_notes(
types = types or []
entity_types = entity_types or []
# Create a SearchQuery object based on the parameters
search_query = SearchQuery()
# Set the appropriate search field based on search_type
if search_type == "text":
search_query.text = query
elif search_type == "vector":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.VECTOR
elif search_type == "hybrid":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.HYBRID
elif search_type == "title":
search_query.title = query
elif search_type == "permalink" and "*" in query:
search_query.permalink_match = query
elif search_type == "permalink":
search_query.permalink = query
else: # pragma: no cover
search_query.text = query # Default to text search
# Add optional filters if provided (empty lists are treated as no filter)
if entity_types:
search_query.entity_types = [SearchItemType(t) for t in entity_types]
if types:
search_query.types = types
if after_date:
search_query.after_date = after_date
if metadata_filters:
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
if status:
search_query.status = status
async with get_project_client(project, context) as (client, active_project):
# Handle memory:// URLs by resolving to permalink search
_, resolved_query, is_memory_url = await resolve_project_and_path(
client, query, project, context
)
if is_memory_url:
query = resolved_query
search_type = "permalink"
# Create a SearchQuery object based on the parameters
search_query = SearchQuery()
# Set the appropriate search field based on search_type
if search_type == "text":
search_query.text = query
elif search_type == "vector":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.VECTOR
elif search_type == "hybrid":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.HYBRID
elif search_type == "title":
search_query.title = query
elif search_type == "permalink" and "*" in query:
search_query.permalink_match = query
elif search_type == "permalink":
search_query.permalink = query
else: # pragma: no cover
search_query.text = query # Default to text search
# Add optional filters if provided (empty lists are treated as no filter)
if entity_types:
search_query.entity_types = [SearchItemType(t) for t in entity_types]
if types:
search_query.types = types
if after_date:
search_query.after_date = after_date
if metadata_filters:
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
if status:
search_query.status = status
logger.info(f"Searching for {search_query} in project {active_project.name}")
try:
@@ -539,7 +539,9 @@ class ContextService:
{relation_date_filter}
{relation_project_filter}
)
LEFT JOIN entity e_to ON (r.to_id = e_to.id)
WHERE eg.depth < :max_depth
AND (r.to_id IS NULL OR e_to.project_id = :project_id)
UNION ALL
+5 -41
View File
@@ -24,7 +24,6 @@ from basic_memory.models import Entity as EntityModel
from basic_memory.models import Observation, Relation
from basic_memory.models.knowledge import Entity
from basic_memory.repository import ObservationRepository, RelationRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.schemas import Entity as EntitySchema
from basic_memory.schemas.base import Permalink
@@ -42,7 +41,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.utils import generate_permalink
class EntityService(BaseService[EntityModel]):
@@ -67,7 +66,6 @@ class EntityService(BaseService[EntityModel]):
self.link_resolver = link_resolver
self.search_service = search_service
self.app_config = app_config
self._project_permalink: Optional[str] = None
async def detect_file_path_conflicts(
self, file_path: str, skip_check: bool = False
@@ -161,23 +159,7 @@ class EntityService(BaseService[EntityModel]):
if markdown and markdown.frontmatter.permalink:
desired_permalink = markdown.frontmatter.permalink
else:
# Trigger: generating a permalink for a new file
# Why: canonical permalinks may require project prefix for global addressing
# Outcome: include project slug when enabled in config
include_project = True
if self.app_config:
include_project = self.app_config.permalinks_include_project
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:
project_permalink = await self._get_project_permalink()
desired_permalink = build_canonical_permalink(
project_permalink, file_path_str, include_project=include_project
)
desired_permalink = generate_permalink(file_path_str)
# Make unique if needed - enhanced to handle character conflicts
# Use lightweight existence check instead of loading full entity
@@ -190,21 +172,6 @@ class EntityService(BaseService[EntityModel]):
return permalink
async def _get_project_permalink(self) -> Optional[str]:
"""Get and cache the current project's permalink."""
if self._project_permalink is not None:
return self._project_permalink
project_id = self.repository.project_id
if project_id is None: # pragma: no cover
return None # pragma: no cover
project_repository = ProjectRepository(self.repository.session_maker)
project = await project_repository.get_by_id(project_id)
if project:
self._project_permalink = project.permalink
return self._project_permalink
def _build_frontmatter_markdown(
self, title: str, entity_type: str, permalink: str
) -> EntityMarkdown:
@@ -346,12 +313,9 @@ class EntityService(BaseService[EntityModel]):
# Merge new metadata with existing metadata
existing_markdown.frontmatter.metadata.update(post.metadata)
# Always ensure the permalink in the metadata is the canonical one from the database.
# The schema_to_markdown call above uses EntitySchema.permalink which computes a
# non-prefixed permalink (e.g., "test/note"). The metadata merge on the previous line
# would overwrite the project-prefixed permalink (e.g., "project/test/note") stored
# in the existing file. Setting it unconditionally preserves the correct value.
existing_markdown.frontmatter.metadata["permalink"] = new_permalink
# Ensure the permalink in the metadata is the resolved one
if new_permalink != entity.permalink:
existing_markdown.frontmatter.metadata["permalink"] = new_permalink
# Create a new post with merged metadata
merged_post = frontmatter.Post(post.content, **existing_markdown.frontmatter.metadata)
+102 -295
View File
@@ -1,21 +1,14 @@
"""Service for resolving markdown links to permalinks."""
from typing import Optional, Tuple, Dict
from typing import Optional, Tuple
from loguru import logger
from basic_memory.config import BasicMemoryConfig, ConfigManager
from basic_memory.models import Entity, Project
from basic_memory.models import Entity
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.search_repository import create_search_repository
from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.services.search_service import SearchService
from basic_memory.utils import (
build_canonical_permalink,
generate_permalink,
normalize_project_reference,
)
class LinkResolver:
@@ -33,12 +26,6 @@ class LinkResolver:
"""Initialize with repositories."""
self.entity_repository = entity_repository
self.search_service = search_service
self._project_repository = ProjectRepository(entity_repository.session_maker)
self._app_config: BasicMemoryConfig = ConfigManager().config
self._project_permalink: Optional[str] = None
self._project_cache_by_identifier: Dict[str, Project] = {}
self._entity_repository_cache: Dict[int, EntityRepository] = {}
self._search_service_cache: Dict[int, SearchService] = {}
async def resolve_link(
self,
@@ -60,69 +47,111 @@ class LinkResolver:
# Clean link text and extract any alias
clean_text, alias = self._normalize_link_text(link_text)
explicit_project_reference = "::" in clean_text
clean_text = normalize_project_reference(clean_text)
# Trigger: link uses project namespace syntax (project::note)
# Why: treat it as an explicit cross-project reference
# Outcome: resolve only within the referenced project scope
if explicit_project_reference:
project_prefix, remainder = self._split_project_prefix(clean_text)
if not project_prefix:
return None
# --- Path Resolution ---
# Note: All paths in Basic Memory are stored as POSIX strings (forward slashes)
# for cross-platform compatibility. See entity_repository.py which normalizes
# paths using Path().as_posix(). This allows consistent path operations here.
project_resources = await self._get_project_resources(project_prefix)
if not project_resources:
return None
# --- Relative Path Resolution ---
# Trigger: source_path is provided AND link contains "/"
# Why: Resolve paths like [[nested/deep-note]] relative to source folder first
# Outcome: [[nested/deep-note]] from testing/link-test.md → testing/nested/deep-note.md
if source_path and "/" in clean_text:
source_folder = source_path.rsplit("/", 1)[0] if "/" in source_path else ""
if source_folder:
# Construct relative path from source folder
relative_path = f"{source_folder}/{clean_text}"
project, entity_repository, search_service = project_resources
return await self._resolve_in_project(
entity_repository=entity_repository,
search_service=search_service,
link_text=remainder,
use_search=use_search,
strict=strict,
source_path=None,
project_permalink=project.permalink,
# Try with .md extension
if not relative_path.endswith(".md"):
relative_path_md = f"{relative_path}.md"
entity = await self.entity_repository.get_by_file_path(relative_path_md)
if entity:
return entity
# Try as-is (already has extension or is a permalink)
entity = await self.entity_repository.get_by_file_path(relative_path)
if entity:
return entity
# When source_path is provided, use context-aware resolution:
# Check both permalink and title matches, prefer closest to source.
# Example: [[testing]] from folder/note.md prefers folder/testing.md
# over a root testing.md with permalink "testing".
if source_path:
# Gather all potential matches
candidates: list[Entity] = []
# Check permalink match
permalink_entity = await self.entity_repository.get_by_permalink(clean_text)
if permalink_entity:
candidates.append(permalink_entity)
# Check title matches
title_entities = await self.entity_repository.get_by_title(clean_text)
for entity in title_entities:
# Avoid duplicates (permalink match might also be in title matches)
if entity.id not in [c.id for c in candidates]:
candidates.append(entity)
if candidates:
if len(candidates) == 1:
return candidates[0]
else:
# Multiple candidates - pick closest to source
return self._find_closest_entity(candidates, source_path)
# Standard resolution (no source context): permalink first, then title
# 1. Try exact permalink match first (most efficient)
entity = await self.entity_repository.get_by_permalink(clean_text)
if entity:
logger.debug(f"Found exact permalink match: {entity.permalink}")
return entity
# 2. Try exact title match
found = await self.entity_repository.get_by_title(clean_text)
if found:
# Return first match (shortest path) if no source context
entity = found[0]
logger.debug(f"Found title match: {entity.title}")
return entity
# 3. Try file path
found_path = await self.entity_repository.get_by_file_path(clean_text)
if found_path:
logger.debug(f"Found entity with path: {found_path.file_path}")
return found_path
# 4. Try file path with .md extension if not already present
if not clean_text.endswith(".md") and "/" in clean_text:
file_path_with_md = f"{clean_text}.md"
found_path_md = await self.entity_repository.get_by_file_path(file_path_with_md)
if found_path_md:
logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
return found_path_md
# In strict mode, don't try fuzzy search - return None if no exact match found
if strict:
return None
# 5. Fall back to search for fuzzy matching (only if not in strict mode)
if use_search and "*" not in clean_text:
results = await self.search_service.search(
query=SearchQuery(text=clean_text, entity_types=[SearchItemType.ENTITY]),
)
current_project_permalink = await self._get_current_project_permalink()
resolved = await self._resolve_in_project(
entity_repository=self.entity_repository,
search_service=self.search_service,
link_text=clean_text,
use_search=use_search,
strict=strict,
source_path=source_path,
project_permalink=current_project_permalink,
)
if resolved:
return resolved
if results:
# Look for best match
best_match = min(results, key=lambda x: x.score) # pyright: ignore
logger.trace(
f"Selected best match from {len(results)} results: {best_match.permalink}"
)
if best_match.permalink:
return await self.entity_repository.get_by_permalink(best_match.permalink)
# Trigger: local resolution failed and identifier looks like project/path
# Why: allow explicit project path references without namespace syntax
# Outcome: attempt resolution in the referenced project if it exists
project_prefix, remainder = self._split_project_prefix(clean_text)
if not project_prefix:
return None
project_resources = await self._get_project_resources(project_prefix)
if not project_resources:
return None
project, entity_repository, search_service = project_resources
if project.id == self.entity_repository.project_id:
return None
return await self._resolve_in_project(
entity_repository=entity_repository,
search_service=search_service,
link_text=remainder,
use_search=use_search,
strict=strict,
source_path=None,
project_permalink=project.permalink,
)
# if we couldn't find anything then return None
return None
def _normalize_link_text(self, link_text: str) -> Tuple[str, Optional[str]]:
"""Normalize link text and extract alias if present.
@@ -152,228 +181,6 @@ class LinkResolver:
return text, alias
async def _resolve_in_project(
self,
*,
entity_repository: EntityRepository,
search_service: SearchService,
link_text: str,
use_search: bool,
strict: bool,
source_path: Optional[str],
project_permalink: Optional[str],
) -> Optional[Entity]:
"""Resolve a link within a specific project scope."""
clean_text = link_text
include_project = self._include_project_permalinks()
canonical_permalink: Optional[str] = None
legacy_permalink: Optional[str] = None
# Trigger: permalinks include project slug and project permalink is known
# Why: support globally addressable permalinks while keeping legacy links resolvable
# Outcome: include canonical and legacy candidates for resolution
if include_project and project_permalink:
canonical_permalink = build_canonical_permalink(
project_permalink, clean_text, include_project=True
)
if clean_text.startswith(f"{project_permalink}/"):
legacy_candidate = clean_text.removeprefix(f"{project_permalink}/")
if legacy_candidate:
legacy_permalink = legacy_candidate
permalink_candidates = []
for candidate in (clean_text, canonical_permalink, legacy_permalink):
if candidate and candidate not in permalink_candidates:
permalink_candidates.append(candidate)
# --- Path Resolution ---
# Note: All paths in Basic Memory are stored as POSIX strings (forward slashes)
# for cross-platform compatibility. See entity_repository.py which normalizes
# paths using Path().as_posix(). This allows consistent path operations here.
# --- Relative Path Resolution ---
# Trigger: source_path is provided AND link contains "/"
# Why: Resolve paths like [[nested/deep-note]] relative to source folder first
# Outcome: [[nested/deep-note]] from testing/link-test.md → testing/nested/deep-note.md
if source_path and "/" in clean_text:
if not (
include_project
and project_permalink
and clean_text.startswith(f"{project_permalink}/")
):
source_folder = source_path.rsplit("/", 1)[0] if "/" in source_path else ""
if source_folder:
# Construct relative path from source folder
relative_path = f"{source_folder}/{clean_text}"
# Try with .md extension
if not relative_path.endswith(".md"):
relative_path_md = f"{relative_path}.md"
entity = await entity_repository.get_by_file_path(relative_path_md)
if entity:
return entity
# Try as-is (already has extension or is a permalink)
entity = await entity_repository.get_by_file_path(relative_path)
if entity:
return entity
# When source_path is provided, use context-aware resolution:
# Check both permalink and title matches, prefer closest to source.
# Example: [[testing]] from folder/note.md prefers folder/testing.md
# over a root testing.md with permalink "testing".
if source_path:
# Gather all potential matches
candidates: list[Entity] = []
# Check permalink match
for candidate_permalink in permalink_candidates:
permalink_entity = await entity_repository.get_by_permalink(candidate_permalink)
if permalink_entity and permalink_entity.id not in [c.id for c in candidates]:
candidates.append(permalink_entity)
# Check title matches
title_entities = await entity_repository.get_by_title(clean_text)
for entity in title_entities:
# Avoid duplicates (permalink match might also be in title matches)
if entity.id not in [c.id for c in candidates]:
candidates.append(entity)
if candidates:
if len(candidates) == 1:
return candidates[0]
else:
# Multiple candidates - pick closest to source
return self._find_closest_entity(candidates, source_path)
# Standard resolution (no source context): permalink first, then title
# 1. Try exact permalink match first (most efficient)
for candidate_permalink in permalink_candidates:
entity = await entity_repository.get_by_permalink(candidate_permalink)
if entity:
logger.debug(f"Found exact permalink match: {entity.permalink}")
return entity
# 2. Try exact title match
found = await entity_repository.get_by_title(clean_text)
if found:
# Return first match (shortest path) if no source context
entity = found[0]
logger.debug(f"Found title match: {entity.title}")
return entity
# 3. Try file path
found_path = await entity_repository.get_by_file_path(clean_text)
if found_path:
logger.debug(f"Found entity with path: {found_path.file_path}")
return found_path
# 4. Try file path with .md extension if not already present
if not clean_text.endswith(".md") and "/" in clean_text:
file_path_with_md = f"{clean_text}.md"
found_path_md = await entity_repository.get_by_file_path(file_path_with_md)
if found_path_md:
logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
return found_path_md
# In strict mode, don't try fuzzy search - return None if no exact match found
if strict:
return None
# 5. Fall back to search for fuzzy matching (only if not in strict mode)
if use_search and "*" not in clean_text:
results = await search_service.search(
query=SearchQuery(text=clean_text, entity_types=[SearchItemType.ENTITY]),
)
if results:
# Look for best match
best_match = min(results, key=lambda x: x.score) # pyright: ignore
logger.trace(
f"Selected best match from {len(results)} results: {best_match.permalink}"
)
if best_match.permalink:
return await entity_repository.get_by_permalink(best_match.permalink)
# if we couldn't find anything then return None
return None
def _include_project_permalinks(self) -> bool:
"""Return True when permalinks should include the project slug."""
return self._app_config.permalinks_include_project
async def _get_current_project_permalink(self) -> Optional[str]:
"""Get and cache the current project's permalink."""
if self._project_permalink is not None:
return self._project_permalink
project_id = self.entity_repository.project_id
if project_id is None: # pragma: no cover
return None # pragma: no cover
project = await self._project_repository.get_by_id(project_id)
if project:
self._project_permalink = project.permalink
return self._project_permalink
async def _get_project_by_identifier(self, identifier: str) -> Optional[Project]:
"""Resolve project by name or permalink."""
cache_key = identifier.strip().lower()
if cache_key in self._project_cache_by_identifier:
return self._project_cache_by_identifier[cache_key]
project = await self._project_repository.get_by_name(identifier)
if not project:
project = await self._project_repository.get_by_name_case_insensitive(identifier)
if not project:
project = await self._project_repository.get_by_permalink(generate_permalink(identifier))
if project:
self._project_cache_by_identifier[cache_key] = project
return project
async def _get_project_resources(
self, project_identifier: str
) -> Optional[Tuple[Project, EntityRepository, SearchService]]:
"""Fetch repositories and services scoped to a project."""
project = await self._get_project_by_identifier(project_identifier)
if not project:
return None
entity_repository = self._entity_repository_cache.get(project.id)
if not entity_repository:
entity_repository = EntityRepository(
self.entity_repository.session_maker, project_id=project.id
)
self._entity_repository_cache[project.id] = entity_repository
search_service = self._search_service_cache.get(project.id)
if not search_service:
search_repository = create_search_repository(
self.entity_repository.session_maker,
project_id=project.id,
database_backend=self._app_config.database_backend,
)
search_service = SearchService(
search_repository,
entity_repository,
self.search_service.file_service,
)
self._search_service_cache[project.id] = search_service
return project, entity_repository, search_service
def _split_project_prefix(self, identifier: str) -> Tuple[Optional[str], str]:
"""Split project prefix from a path-like identifier."""
if "/" not in identifier:
return None, identifier
project_prefix, remainder = identifier.split("/", 1)
if not project_prefix or not remainder:
return None, identifier
return project_prefix, remainder
def _find_closest_entity(self, entities: list[Entity], source_path: str) -> Entity:
"""Find the entity closest to the source file path.
+1 -44
View File
@@ -7,7 +7,7 @@ import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Protocol, Union, runtime_checkable, List, Optional
from typing import Protocol, Union, runtime_checkable, List
from loguru import logger
from unidecode import unidecode
@@ -202,49 +202,6 @@ def generate_permalink(file_path: Union[Path, str, PathLike], split_extension: b
return return_val
def normalize_project_reference(identifier: str) -> str:
"""Normalize project-prefixed references.
Converts project namespace syntax ("project::note") to path syntax ("project/note").
Leaves non-namespaced identifiers unchanged.
"""
if "::" not in identifier:
return identifier
project, remainder = identifier.split("::", 1)
remainder = remainder.lstrip("/")
return f"{project}/{remainder}"
def build_canonical_permalink(
project_permalink: Optional[str],
file_path: Union[Path, str, PathLike],
include_project: bool = True,
) -> str:
"""Build a canonical permalink, optionally prefixed with project slug.
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.
Returns:
Canonical permalink string.
"""
normalized_path = generate_permalink(file_path)
if not include_project or not project_permalink:
return normalized_path
normalized_project = generate_permalink(project_permalink)
if normalized_path == normalized_project or normalized_path.startswith(
f"{normalized_project}/"
):
return normalized_path
return f"{normalized_project}/{normalized_path}"
def setup_logging(
log_level: str = "INFO",
log_to_file: bool = False,
@@ -283,7 +283,7 @@ async def test_read_content_empty_file(mcp_server, app, test_project):
# Should still have frontmatter even with empty content
assert "title: Empty Test" in content
assert f"permalink: {test_project.name}/test/empty-test" in content
assert "permalink: test/empty-test" in content
@pytest.mark.asyncio
+3 -3
View File
@@ -45,7 +45,7 @@ async def test_read_note_after_write(mcp_server, app, test_project):
# Should contain the note content and metadata
assert "# Test Note" in result_text
assert "This is test content." in result_text
assert f"{test_project.name}/test/test-note" in result_text # permalink
assert "test/test-note" in result_text # permalink
@pytest.mark.asyncio
@@ -78,7 +78,7 @@ async def test_read_note_underscored_folder_by_permalink(mcp_server, app, test_p
assert "_archive/articles/Example Note.md" in write_text
# Verify the permalink has underscores stripped (this is the expected behavior)
assert f"{test_project.name}/archive/articles/example-note" in write_text
assert "archive/articles/example-note" in write_text
# Now try to read the note using the permalink (without underscores)
# This is the exact scenario from the bug report - using the permalink
@@ -99,4 +99,4 @@ async def test_read_note_underscored_folder_by_permalink(mcp_server, app, test_p
# Should contain the note content
assert "# Example Note" in result_text
assert "This is a test note in an underscored folder." in result_text
assert f"{test_project.name}/archive/articles/example-note" in result_text # permalink
assert "archive/articles/example-note" in result_text # permalink
+2 -2
View File
@@ -224,7 +224,7 @@ async def test_search_permalink_exact(mcp_server, app, test_project):
"search_notes",
{
"project": test_project.name,
"query": f"{test_project.name}/api/api-documentation",
"query": "api/api-documentation",
"search_type": "permalink",
},
)
@@ -278,7 +278,7 @@ async def test_search_permalink_pattern(mcp_server, app, test_project):
"search_notes",
{
"project": test_project.name,
"query": f"{test_project.name}/meetings/*",
"query": "meetings/*",
"search_type": "permalink",
},
)
+10 -10
View File
@@ -38,7 +38,7 @@ async def test_write_note_basic_creation(mcp_server, app, test_project):
assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text
assert "file_path: basic/Simple Note.md" in response_text
assert f"permalink: {test_project.name}/basic/simple-note" in response_text
assert "permalink: basic/simple-note" in response_text
assert "## Tags" in response_text
assert "- simple, test" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -65,7 +65,7 @@ async def test_write_note_no_tags(mcp_server, app, test_project):
assert "# Created note" in response_text
assert "file_path: test/No Tags Note.md" in response_text
assert f"permalink: {test_project.name}/test/no-tags-note" in response_text
assert "permalink: test/no-tags-note" in response_text
# Should not have tags section when no tags provided
@@ -107,7 +107,7 @@ async def test_write_note_update_existing(mcp_server, app, test_project):
assert "# Updated note" in response_text
assert f"project: {test_project.name}" in response_text
assert "file_path: test/Update Test.md" in response_text
assert f"permalink: {test_project.name}/test/update-test" in response_text
assert "permalink: test/update-test" in response_text
assert "- updated, modified" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -136,7 +136,7 @@ async def test_write_note_tag_array(mcp_server, app, test_project):
assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text
assert "file_path: test/Array Tags Test.md" in response_text
assert f"permalink: {test_project.name}/test/array-tags-test" in response_text
assert "permalink: test/array-tags-test" in response_text
assert "## Tags" in response_text
assert "python" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -206,7 +206,7 @@ async def test_write_note_unicode_content(mcp_server, app, test_project):
assert f"project: {test_project.name}" in response_text
assert "file_path: test/Unicode Test 🌟.md" in response_text
# Permalink should be sanitized
assert f"permalink: {test_project.name}/test/unicode-test" in response_text
assert "permalink: test/unicode-test" in response_text
assert "## Tags" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -256,7 +256,7 @@ async def test_write_note_complex_content_with_observations_relations(
assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text
assert "file_path: knowledge/Complex Knowledge Note.md" in response_text
assert f"permalink: {test_project.name}/knowledge/complex-knowledge-note" in response_text
assert "permalink: knowledge/complex-knowledge-note" in response_text
# Should show observation and relation counts
assert "## Observations" in response_text
@@ -309,7 +309,7 @@ async def test_write_note_preserve_frontmatter(mcp_server, app, test_project):
assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text
assert "file_path: test/Frontmatter Note.md" in response_text
assert f"permalink: {test_project.name}/test/frontmatter-note" in response_text
assert "permalink: test/frontmatter-note" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -338,7 +338,7 @@ async def test_write_note_kebab_filenames_basic(mcp_server, app, test_project, a
# File path and permalink should be kebab-case and sanitized
assert f"project: {test_project.name}" in response_text
assert "file_path: my-folder/my-note-with-invalid-chars.md" in response_text
assert f"permalink: {test_project.name}/my-folder/my-note-with-invalid-chars" in response_text
assert "permalink: my-folder/my-note-with-invalid-chars" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -366,7 +366,7 @@ async def test_write_note_kebab_filenames_repeat_invalid(mcp_server, app, test_p
assert f"project: {test_project.name}" in response_text
assert "file_path: my-folder/crazy-note-name.md" in response_text
assert f"permalink: {test_project.name}/my-folder/crazy-note-name" in response_text
assert "permalink: my-folder/crazy-note-name" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -416,7 +416,7 @@ async def test_write_note_file_path_os_path_join(mcp_server, app, test_project,
assert f"project: {test_project.name}" in response_text
assert f"file_path: {expected_path}" in response_text
assert f"permalink: {test_project.name}/{expected_permalink}" in response_text
assert f"permalink: {expected_permalink}" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text
+1 -1
View File
@@ -171,7 +171,7 @@ async def test_create_entity(client: AsyncClient, file_service, v2_project_url):
assert isinstance(entity.id, int)
assert entity.api_version == "v2"
assert entity.permalink == "test-project/test/test-v2-entity"
assert entity.permalink == "test/test-v2-entity"
assert entity.file_path == "test/TestV2Entity.md"
assert entity.entity_type == data["entity_type"]
+4 -15
View File
@@ -38,9 +38,7 @@ async def test_imported_conversations_have_correct_permalink_and_title(
file_service = FileService(base_path, processor)
# Create importer
importer = ClaudeConversationsImporter(
base_path, processor, file_service, project_name=project_config.name
)
importer = ClaudeConversationsImporter(base_path, processor, file_service)
# Sample conversation data
conversations = [
@@ -82,10 +80,7 @@ async def test_imported_conversations_have_correct_permalink_and_title(
content = conv_path.read_text()
assert "---" in content, "File should have frontmatter markers"
assert "title: My Test Conversation Title" in content, "File should have title in frontmatter"
assert (
f"permalink: {project_config.name}/conversations/20250115-my-test-conversation-title"
in content
), (
assert "permalink: conversations/20250115-My_Test_Conversation_Title" in content, (
"File should have permalink in frontmatter"
)
@@ -102,10 +97,7 @@ async def test_imported_conversations_have_correct_permalink_and_title(
assert entity.title == "My Test Conversation Title", (
f"Title should be from frontmatter, got: {entity.title}"
)
assert (
entity.permalink
== f"{project_config.name}/conversations/20250115-my-test-conversation-title"
), (
assert entity.permalink == "conversations/20250115-My_Test_Conversation_Title", (
f"Permalink should be from frontmatter, got: {entity.permalink}"
)
@@ -119,9 +111,6 @@ async def test_imported_conversations_have_correct_permalink_and_title(
assert search_result.title == "My Test Conversation Title", (
f"Search title should be from frontmatter, got: {search_result.title}"
)
assert (
search_result.permalink
== f"{project_config.name}/conversations/20250115-my-test-conversation-title"
), (
assert search_result.permalink == "conversations/20250115-My_Test_Conversation_Title", (
f"Search permalink should not be null, got: {search_result.permalink}"
)
-8
View File
@@ -176,14 +176,6 @@ def test_relation_plugin():
assert rel["target"] == "Component"
assert rel["context"] == "with context"
# Test project namespace normalization
content = dedent("""
- relates_to [[other-project::Component]]
""")
token = [t for t in md.parse(content) if t.type == "inline"][0]
rel = token.meta["relations"][0]
assert rel["target"] == "other-project/Component"
# Test implicit relations in text
content = "Some text with a [[Link]] and [[Another Link]]"
token = [t for t in md.parse(content) if t.type == "inline"][0]
@@ -22,7 +22,6 @@ from basic_memory.mcp.tools import write_note, read_note
from basic_memory.sync.sync_service import SyncService
from basic_memory.config import ProjectConfig
from basic_memory.services import EntityService
from basic_memory.utils import generate_permalink
async def force_full_scan(sync_service: SyncService) -> None:
@@ -71,7 +70,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_a
assert "file_path: edge-cases/Node A.md" in result_a
assert f"permalink: {test_project.name}/edge-cases/node-a" in result_a
assert "permalink: edge-cases/node-a" in result_a
# Verify Node A content via read
content_a = await read_note.fn("edge-cases/node-a", project=test_project.name)
@@ -88,7 +87,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_b
assert "file_path: edge-cases/Node B.md" in result_b
assert f"permalink: {test_project.name}/edge-cases/node-b" in result_b
assert "permalink: edge-cases/node-b" in result_b
# Step 3: Create third note "Node C" (this is where the bug occurs)
result_c = await write_note.fn(
@@ -100,7 +99,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_c
assert "file_path: edge-cases/Node C.md" in result_c
assert f"permalink: {test_project.name}/edge-cases/node-c" in result_c
assert "permalink: edge-cases/node-c" in result_c
# CRITICAL CHECK: Verify Node A still has its original content
# This is where the bug manifests - Node A.md gets overwritten with Node C content
@@ -259,7 +258,6 @@ async def test_sync_permalink_collision_file_overwrite_bug(
project_dir = project_config.home
edge_cases_dir = project_dir / "edge-cases"
edge_cases_dir.mkdir(parents=True, exist_ok=True)
project_prefix = generate_permalink(project_config.name)
# Step 1: Create Node A file
node_a_content = dedent("""
@@ -286,7 +284,7 @@ async def test_sync_permalink_collision_file_overwrite_bug(
await sync_service.sync(project_dir)
# Verify Node A is in database
node_a = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-a")
node_a = await entity_service.get_by_permalink("edge-cases/node-a")
assert node_a is not None
assert node_a.title == "Node A"
@@ -377,8 +375,8 @@ async def test_sync_permalink_collision_file_overwrite_bug(
assert "Content for Node C" in node_c_after_sync
# Verify database has both entities correctly
node_a_db = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-a")
node_c_db = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-c")
node_a_db = await entity_service.get_by_permalink("edge-cases/node-a")
node_c_db = await entity_service.get_by_permalink("edge-cases/node-c")
assert node_a_db is not None, "Node A should exist in database"
assert node_a_db.title == "Node A", "Node A database entry should have correct title"
+3 -6
View File
@@ -18,11 +18,11 @@ async def test_get_basic_discussion_context(client, test_graph, test_project):
assert isinstance(context, GraphContext)
assert len(context.results) == 1
assert context.results[0].primary_result.permalink == f"{test_project.name}/test/root"
assert context.results[0].primary_result.permalink == "test/root"
assert len(context.results[0].related_results) > 0
# Verify metadata
assert context.metadata.uri == f"{test_project.name}/test/root"
assert context.metadata.uri == "test/root"
assert context.metadata.depth == 1 # default depth
assert context.metadata.timeframe is not None
assert isinstance(context.metadata.generated_at, datetime)
@@ -38,10 +38,7 @@ async def test_get_discussion_context_pattern(client, test_graph, test_project):
assert isinstance(context, GraphContext)
assert len(context.results) > 1 # Should match multiple test/* paths
assert all(
f"{test_project.name}/test/" in item.primary_result.permalink
for item in context.results
) # pyright: ignore [reportOperatorIssue]
assert all("test/" in item.primary_result.permalink for item in context.results) # pyright: ignore [reportOperatorIssue]
assert context.metadata.depth == 1
+4 -4
View File
@@ -29,7 +29,7 @@ async def test_edit_note_append_operation(client, test_project):
assert "Edited note (append)" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
assert "Added 3 lines to end of note" in result
assert f"[Session: Using project '{test_project.name}']" in result
@@ -57,7 +57,7 @@ async def test_edit_note_prepend_operation(client, test_project):
assert "Edited note (prepend)" in result
assert f"project: {test_project.name}" in result
assert "file_path: meetings/Meeting Notes.md" in result
assert f"permalink: {test_project.name}/meetings/meeting-notes" in result
assert "permalink: meetings/meeting-notes" in result
assert "Added 3 lines to beginning of note" in result
assert f"[Session: Using project '{test_project.name}']" in result
@@ -439,7 +439,7 @@ async def test_edit_note_preserves_permalink_when_frontmatter_missing(client, te
)
assert isinstance(first_result, str)
assert f"permalink: {test_project.name}/test/test-note" in first_result
assert "permalink: test/test-note" in first_result
# Perform another edit - this should preserve the permalink even if the
# file doesn't have a permalink in its frontmatter
@@ -453,6 +453,6 @@ async def test_edit_note_preserves_permalink_when_frontmatter_missing(client, te
assert isinstance(second_result, str)
assert "Edited note (append)" in second_result
assert f"project: {test_project.name}" in second_result
assert f"permalink: {test_project.name}/test/test-note" in second_result
assert "permalink: test/test-note" in second_result
assert f"[Session: Using project '{test_project.name}']" in second_result
# The edit should succeed without validation errors
+3 -3
View File
@@ -67,7 +67,7 @@ async def test_move_note_success(app, client, test_project):
content = await read_note.fn("target/moved-note", project=test_project.name)
assert "# Test Note" in content
assert "Original content here" in content
assert f"permalink: {test_project.name}/target/moved-note" in content
assert "permalink: target/moved-note" in content
@pytest.mark.asyncio
@@ -426,7 +426,7 @@ async def test_move_note_rename_only(client, test_project):
content = await read_note.fn("test/new-name", project=test_project.name)
assert "# OriginalName" in content # Title in content remains same
assert "Content to rename" in content
assert f"permalink: {test_project.name}/test/new-name" in content
assert "permalink: test/new-name" in content
@pytest.mark.asyncio
@@ -587,7 +587,7 @@ async def test_move_note_preserves_frontmatter(app, client, test_project):
content = await read_note.fn("target/moved-custom-note", project=test_project.name)
assert "title: Custom Frontmatter Note" in content
assert "type: note" in content
assert f"permalink: {test_project.name}/target/moved-custom-note" in content
assert "permalink: target/moved-custom-note" in content
assert "# Custom Frontmatter Note" in content
assert "Content with custom metadata" in content
+1 -16
View File
@@ -119,7 +119,7 @@ async def test_note_unicode_content(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Unicode Test.md" in result
assert f"permalink: {test_project.name}/test/unicode-test" in result
assert "permalink: test/unicode-test" in result
assert "checksum:" in result # Checksum exists but may be "unknown"
# Read back should preserve unicode
@@ -201,21 +201,6 @@ async def test_read_note_memory_url(app, test_project):
assert "Testing memory:// URL handling" in content
@pytest.mark.asyncio
async def test_read_note_memory_url_with_project_prefix(app, test_project):
"""Test reading a note using a memory:// URL with explicit project prefix."""
await write_note.fn(
project=test_project.name,
title="Project Prefixed Memory URL Test",
directory="test",
content="Testing memory:// URL handling with project prefix",
)
memory_url = f"memory://{test_project.name}/test/project-prefixed-memory-url-test"
content = await read_note.fn(memory_url)
assert "Testing memory:// URL handling with project prefix" in content
class TestReadNoteSecurityValidation:
"""Test read_note security validation features."""
-17
View File
@@ -150,23 +150,6 @@ async def test_read_file_memory_url(app, synced_files, test_project):
assert "Testing memory:// URL handling for resources" in response["text"]
@pytest.mark.asyncio
async def test_read_file_memory_url_with_project_prefix(app, synced_files, test_project):
"""Test reading a resource using a memory:// URL with explicit project prefix."""
await write_note.fn(
project=test_project.name,
title="Project Prefixed Resource URL Test",
directory="test",
content="Testing memory:// URL handling for resources with project prefix",
)
memory_url = f"memory://{test_project.name}/test/project-prefixed-resource-url-test"
response = await read_content.fn(memory_url)
assert response["type"] == "text"
assert "Testing memory:// URL handling for resources with project prefix" in response["text"]
@pytest.mark.asyncio
async def test_image_optimization_functions(app):
"""Test the image optimization helper functions."""
+14 -70
View File
@@ -1,7 +1,6 @@
"""Tests for search MCP tools."""
import pytest
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from basic_memory.mcp.tools import write_note
@@ -29,10 +28,7 @@ async def test_search_text(client, test_project):
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
assert any(r.permalink == "test/test-search-note" for r in response.results)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -63,10 +59,7 @@ async def test_search_title(client, test_project):
else:
# Success case - verify SearchResponse
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
assert any(r.permalink == "test/test-search-note" for r in response.results)
@pytest.mark.asyncio
@@ -84,19 +77,14 @@ async def test_search_permalink(client, test_project):
# Search for it
response = await search_notes.fn(
project=test_project.name,
query=f"{test_project.name}/test/test-search-note",
search_type="permalink",
project=test_project.name, query="test/test-search-note", search_type="permalink"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
assert any(r.permalink == "test/test-search-note" for r in response.results)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -117,49 +105,19 @@ async def test_search_permalink_match(client, test_project):
# Search for it
response = await search_notes.fn(
project=test_project.name,
query=f"{test_project.name}/test/test-search-*",
search_type="permalink",
project=test_project.name, query="test/test-search-*", search_type="permalink"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
assert any(r.permalink == "test/test-search-note" for r in response.results)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_memory_url_with_project_prefix(client, test_project):
"""Test searching with a memory:// URL that includes the project prefix."""
result = await write_note.fn(
project=test_project.name,
title="Memory URL Search Note",
directory="test",
content="# Memory URL Search\nThis note should be found via memory URL search",
)
assert result
response = await search_notes.fn(
query=f"memory://{test_project.name}/test/memory-url-search-note"
)
if isinstance(response, SearchResponse):
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/memory-url-search-note"
for r in response.results
)
else:
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_pagination(client, test_project):
"""Test basic search functionality."""
@@ -182,10 +140,7 @@ async def test_search_pagination(client, test_project):
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) == 1
assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
assert any(r.permalink == "test/test-search-note" for r in response.results)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -360,23 +315,21 @@ class TestSearchToolErrorHandling:
async def test_search_notes_exception_handling(self, monkeypatch):
"""Test exception handling in search_notes."""
import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
project_url = "http://test"
name = "test-project"
id = 1
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise an exception
class MockSearchClient:
def __init__(self, *args, **kwargs):
@@ -386,7 +339,6 @@ class TestSearchToolErrorHandling:
raise Exception("syntax error")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
@@ -398,23 +350,21 @@ class TestSearchToolErrorHandling:
async def test_search_notes_permission_error(self, monkeypatch):
"""Test search_notes with permission error."""
import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
project_url = "http://test"
name = "test-project"
id = 1
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise a permission error
class MockSearchClient:
def __init__(self, *args, **kwargs):
@@ -424,7 +374,6 @@ class TestSearchToolErrorHandling:
raise Exception("permission denied")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
@@ -438,6 +387,7 @@ class TestSearchToolErrorHandling:
async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch, search_type):
"""Vector/hybrid search types should populate retrieval_mode in API payload."""
import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
@@ -452,11 +402,6 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
@@ -468,7 +413,6 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes.fn(
+43 -42
View File
@@ -29,29 +29,31 @@ async def test_write_note(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
assert "## Tags" in result
assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Try reading it back via permalink
content = await read_note.fn("test/test-note", project=test_project.name)
expected = normalize_newlines(
dedent("""
assert (
normalize_newlines(
dedent("""
---
title: Test Note
type: note
permalink: {permalink}
permalink: test/test-note
tags:
- test
- documentation
---
# Test
This is a test note
""").format(permalink=f"{test_project.name}/test/test-note").strip()
""").strip()
)
in content
)
assert expected in content
@pytest.mark.asyncio
@@ -65,22 +67,24 @@ async def test_write_note_no_tags(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Simple Note.md" in result
assert f"permalink: {test_project.name}/test/simple-note" in result
assert "permalink: test/simple-note" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Should be able to read it back
content = await read_note.fn("test/simple-note", project=test_project.name)
expected = normalize_newlines(
dedent("""
assert (
normalize_newlines(
dedent("""
---
title: Simple Note
type: note
permalink: {permalink}
permalink: test/simple-note
---
Just some text
""").format(permalink=f"{test_project.name}/test/simple-note").strip()
""").strip()
)
in content
)
assert expected in content
@pytest.mark.asyncio
@@ -105,7 +109,7 @@ async def test_write_note_update_existing(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
assert "## Tags" in result
assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result
@@ -120,7 +124,7 @@ async def test_write_note_update_existing(app, test_project):
assert "# Updated note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
assert "## Tags" in result
assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result
@@ -134,16 +138,16 @@ async def test_write_note_update_existing(app, test_project):
---
title: Test Note
type: note
permalink: {permalink}
permalink: test/test-note
tags:
- test
- documentation
---
# Test
This is an updated note
"""
).format(permalink=f"{test_project.name}/test/test-note").strip()
).strip()
)
== content
)
@@ -290,7 +294,7 @@ async def test_write_note_with_tag_array_from_bug_report(app, test_project):
assert result
assert f"project: {test_project.name}" in result
assert f"permalink: {test_project.name}/folder/title" in result
assert "permalink: folder/title" in result
assert "Tags" in result
assert "hipporag" in result
assert f"[Session: Using project '{test_project.name}']" in result
@@ -323,7 +327,7 @@ async def test_write_note_verbose(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
assert "## Observations" in result
assert "- note: 1" in result
assert "## Relations" in result
@@ -437,7 +441,7 @@ async def test_write_note_preserves_content_frontmatter(app, test_project):
---
title: Test Note
type: note
permalink: {permalink}
permalink: test/test-note
version: 1.0
author: name
tags:
@@ -449,7 +453,7 @@ async def test_write_note_preserves_content_frontmatter(app, test_project):
This is a test note
"""
).format(permalink=f"{test_project.name}/test/test-note").strip()
).strip()
)
in content
)
@@ -476,7 +480,7 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
)
assert "# Created note" in result1
assert f"project: {test_project.name}" in result1
assert f"permalink: {test_project.name}/test/note-1" in result1
assert "permalink: test/note-1" in result1
# Step 2: Create second note with different title
result2 = await write_note.fn(
@@ -484,7 +488,7 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
)
assert "# Created note" in result2
assert f"project: {test_project.name}" in result2
assert f"permalink: {test_project.name}/test/note-2" in result2
assert "permalink: test/note-2" in result2
# Step 3: Try to create/replace first note again
# This scenario would trigger the UNIQUE constraint failure before the fix
@@ -505,13 +509,10 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
assert "Updated note" in result3 or "Created note" in result3
# The result should contain either the original permalink or a unique one
assert (
f"permalink: {test_project.name}/test/note-1" in result3
or f"permalink: {test_project.name}/test/note-1-1" in result3
)
assert "permalink: test/note-1" in result3 or "permalink: test/note-1-1" in result3
# Verify we can read back the content
if f"permalink: {test_project.name}/test/note-1" in result3:
if "permalink: test/note-1" in result3:
# Updated existing note case
content = await read_note.fn("test/note-1", project=test_project.name)
assert "Replacement content for note 1" in content
@@ -544,19 +545,20 @@ async def test_write_note_with_custom_entity_type(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: guides/Test Guide.md" in result
assert f"permalink: {test_project.name}/guides/test-guide" in result
assert "permalink: guides/test-guide" in result
assert "## Tags" in result
assert "- guide, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter
content = await read_note.fn("guides/test-guide", project=test_project.name)
expected = normalize_newlines(
dedent("""
assert (
normalize_newlines(
dedent("""
---
title: Test Guide
type: guide
permalink: {permalink}
permalink: guides/test-guide
tags:
- guide
- documentation
@@ -564,9 +566,10 @@ async def test_write_note_with_custom_entity_type(app, test_project):
# Guide Content
This is a guide
""").format(permalink=f"{test_project.name}/guides/test-guide").strip()
""").strip()
)
in content
)
assert expected in content
@pytest.mark.asyncio
@@ -585,7 +588,7 @@ async def test_write_note_with_report_entity_type(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: reports/Monthly Report.md" in result
assert f"permalink: {test_project.name}/reports/monthly-report" in result
assert "permalink: reports/monthly-report" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter
@@ -609,7 +612,7 @@ async def test_write_note_with_config_entity_type(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: config/System Config.md" in result
assert f"permalink: {test_project.name}/config/system-config" in result
assert "permalink: config/system-config" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter
@@ -637,7 +640,7 @@ async def test_write_note_entity_type_default_behavior(app, test_project):
assert "# Created note" in result
assert f"project: {test_project.name}" in result
assert "file_path: test/Default Type Test.md" in result
assert f"permalink: {test_project.name}/test/default-type-test" in result
assert "permalink: test/default-type-test" in result
assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type defaults to "note"
@@ -1032,9 +1035,7 @@ class TestWriteNoteSecurityValidation:
assert "paths must stay within project boundaries" not in result
assert "# Created note" in result
assert "file_path: security-tests/Full Feature Security Test.md" in result
assert (
f"permalink: {test_project.name}/security-tests/full-feature-security-test" in result
)
assert "permalink: security-tests/full-feature-security-test" in result
# Should process observations and relations
assert "## Observations" in result
@@ -37,7 +37,7 @@ async def test_write_note_spaces_to_hyphens(app, test_project, app_config):
)
assert "file_path: test/my-awesome-note.md" in result
assert f"permalink: {test_project.name}/test/my-awesome-note" in result
assert "permalink: test/my-awesome-note" in result
@pytest.mark.asyncio
@@ -53,7 +53,7 @@ async def test_write_note_underscores_to_hyphens(app, test_project, app_config):
)
assert "file_path: test/my-note-with-underscores.md" in result
assert f"permalink: {test_project.name}/test/my-note-with-underscores" in result
assert "permalink: test/my-note-with-underscores" in result
@pytest.mark.asyncio
@@ -69,7 +69,7 @@ async def test_write_note_camelcase_to_kebab(app, test_project, app_config):
)
assert "file_path: test/my-awesome-feature.md" in result
assert f"permalink: {test_project.name}/test/my-awesome-feature" in result
assert "permalink: test/my-awesome-feature" in result
@pytest.mark.asyncio
@@ -85,7 +85,7 @@ async def test_write_note_mixed_case_to_lowercase(app, test_project, app_config)
)
assert "file_path: test/mixed-case-example.md" in result
assert f"permalink: {test_project.name}/test/mixed-case-example" in result
assert "permalink: test/mixed-case-example" in result
# =============================================================================
@@ -110,7 +110,7 @@ async def test_write_note_single_period_preserved(app, test_project, app_config)
)
assert "file_path: test/test-3.0-version.md" in result
assert f"permalink: {test_project.name}/test/test-3.0-version" in result
assert "permalink: test/test-3.0-version" in result
@pytest.mark.asyncio
@@ -126,7 +126,7 @@ async def test_write_note_multiple_periods_preserved(app, test_project, app_conf
)
assert "file_path: test/version-1.2.3-release.md" in result
assert f"permalink: {test_project.name}/test/version-1.2.3-release" in result
assert "permalink: test/version-1.2.3-release" in result
# =============================================================================
@@ -147,7 +147,7 @@ async def test_write_note_special_chars_to_hyphens(app, test_project, app_config
)
assert "file_path: test/test-2.0-new-feature.md" in result
assert f"permalink: {test_project.name}/test/test-2.0-new-feature" in result
assert "permalink: test/test-2.0-new-feature" in result
@pytest.mark.asyncio
@@ -163,7 +163,7 @@ async def test_write_note_parentheses_removed(app, test_project, app_config):
)
assert "file_path: test/feature-v2.0-update.md" in result
assert f"permalink: {test_project.name}/test/feature-v2.0-update" in result
assert "permalink: test/feature-v2.0-update" in result
@pytest.mark.asyncio
@@ -179,7 +179,7 @@ async def test_write_note_apostrophes_removed(app, test_project, app_config):
)
assert "file_path: test/users-guide.md" in result
assert f"permalink: {test_project.name}/test/users-guide" in result
assert "permalink: test/users-guide" in result
# =============================================================================
@@ -200,10 +200,7 @@ async def test_write_note_all_transformations_combined(app, test_project, app_co
)
assert "file_path: test/my-project-v3.0-feature-update-draft.md" in result
assert (
f"permalink: {test_project.name}/test/my-project-v3.0-feature-update-draft"
in result
)
assert "permalink: test/my-project-v3.0-feature-update-draft" in result
@pytest.mark.asyncio
@@ -220,7 +217,7 @@ async def test_write_note_consecutive_special_chars_collapsed(app, test_project,
# Multiple underscores/hyphens should collapse to single hyphen
assert "file_path: test/test-multiple-separators.md" in result
assert f"permalink: {test_project.name}/test/test-multiple-separators" in result
assert "permalink: test/test-multiple-separators" in result
# =============================================================================
@@ -241,7 +238,7 @@ async def test_write_note_leading_trailing_hyphens_trimmed(app, test_project, ap
)
assert "file_path: test/test-note.md" in result
assert f"permalink: {test_project.name}/test/test-note" in result
assert "permalink: test/test-note" in result
@pytest.mark.asyncio
@@ -257,7 +254,7 @@ async def test_write_note_all_special_chars_becomes_valid_filename(app, test_pro
)
assert "file_path: test/test.md" in result
assert f"permalink: {test_project.name}/test/test" in result
assert "permalink: test/test" in result
# =============================================================================
@@ -279,7 +276,7 @@ async def test_write_note_folder_path_unaffected(app, test_project, app_config):
# Folder paths should be preserved (sanitized but not kebab-cased)
assert "file_path: My_Folder/Sub Folder/test-note.md" in result
assert f"permalink: {test_project.name}/my-folder/sub-folder/test-note" in result
assert "permalink: my-folder/sub-folder/test-note" in result
@pytest.mark.asyncio
@@ -295,7 +292,7 @@ async def test_write_note_root_folder_with_kebab(app, test_project, app_config):
)
assert "file_path: test-3.0-note.md" in result
assert f"permalink: {test_project.name}/test-3.0-note" in result
assert "permalink: test-3.0-note" in result
# =============================================================================
@@ -318,7 +315,7 @@ async def test_write_note_kebab_disabled_preserves_original(app, test_project, a
# Periods and spaces should be preserved
assert "file_path: test/Test 3.0 Version.md" in result
# Permalinks are ALWAYS kebab-case regardless of setting, and preserve periods
assert f"permalink: {test_project.name}/test/test-3.0-version" in result
assert "permalink: test/test-3.0-version" in result
@pytest.mark.asyncio
@@ -334,7 +331,7 @@ async def test_write_note_kebab_disabled_preserves_underscores(app, test_project
)
assert "file_path: test/my_note_example.md" in result
assert f"permalink: {test_project.name}/test/my-note-example" in result
assert "permalink: test/my-note-example" in result
@pytest.mark.asyncio
@@ -350,7 +347,7 @@ async def test_write_note_kebab_disabled_preserves_case(app, test_project, app_c
)
assert "file_path: test/MyAwesomeNote.md" in result
assert f"permalink: {test_project.name}/test/my-awesome-note" in result
assert "permalink: test/my-awesome-note" in result
# =============================================================================
@@ -378,7 +375,7 @@ async def test_permalinks_always_kebab_case(app, test_project, app_config):
# Filename preserves original, permalink is kebab-case
assert "file_path: test/Test Note 1.md" in result1
assert f"permalink: {test_project.name}/test/test-note-1" in result1
assert "permalink: test/test-note-1" in result1
# Test with kebab enabled
ConfigManager().config.kebab_filenames = True
@@ -392,4 +389,4 @@ async def test_permalinks_always_kebab_case(app, test_project, app_config):
# Both filename and permalink are kebab-case
assert "file_path: test/test-note-2.md" in result2
assert f"permalink: {test_project.name}/test/test-note-2" in result2
assert "permalink: test/test-note-2" in result2
+2 -2
View File
@@ -38,8 +38,8 @@ async def test_search_successful_results(client, test_project):
assert content["query"] == "test content"
# Verify individual result format
assert any(r["id"] == f"{test_project.name}/docs/test-document-1" for r in content["results"])
assert any(r["id"] == f"{test_project.name}/docs/test-document-2" for r in content["results"])
assert any(r["id"] == "docs/test-document-1" for r in content["results"])
assert any(r["id"] == "docs/test-document-2" for r in content["results"])
@pytest.mark.asyncio
+5 -5
View File
@@ -137,7 +137,7 @@ async def test_find_connected_timeframe(
@pytest.mark.asyncio
async def test_build_context(context_service, test_graph):
"""Test exact permalink lookup."""
url = memory_url.validate_strings("memory://test-project/test/root")
url = memory_url.validate_strings("memory://test/root")
context_result = await context_service.build_context(url)
# Check metadata
@@ -156,7 +156,7 @@ async def test_build_context(context_service, test_graph):
assert primary_result.id == test_graph["root"].id
assert primary_result.type == "entity"
assert primary_result.title == "Root"
assert primary_result.permalink == "test-project/test/root"
assert primary_result.permalink == "test/root"
assert primary_result.file_path == "test/Root.md"
assert primary_result.created_at is not None
@@ -185,7 +185,7 @@ async def test_build_context_with_observations(context_service, test_graph):
# Let's use those existing observations
# Build context
url = memory_url.validate_strings("memory://test-project/test/root")
url = memory_url.validate_strings("memory://test/root")
context_result = await context_service.build_context(url, include_observations=True)
# Check the metadata
@@ -220,9 +220,9 @@ async def test_build_context_not_found(context_service):
@pytest.mark.asyncio
async def test_context_metadata(context_service, test_graph):
"""Test metadata is correctly populated."""
context = await context_service.build_context("memory://test-project/test/root", depth=2)
context = await context_service.build_context("memory://test/root", depth=2)
metadata = context.metadata
assert metadata.uri == "test-project/test/root"
assert metadata.uri == "test/root"
assert metadata.depth == 2
assert metadata.generated_at is not None
assert metadata.primary_count > 0
+1 -1
View File
@@ -53,7 +53,7 @@ async def test_directory_tree(directory_service: DirectoryService, test_graph):
assert node_file.entity_id == 1
assert node_file.entity_type == "deeper"
assert node_file.title == "Deeper Entity"
assert node_file.permalink == "test-project/test/deeper-entity"
assert node_file.permalink == "test/deeper-entity"
assert node_file.directory_path == "/test/Deeper Entity.md"
assert node_file.file_path == "test/Deeper Entity.md"
assert node_file.has_children is False
+10 -24
View File
@@ -20,31 +20,27 @@ from basic_memory.utils import generate_permalink
@pytest.mark.asyncio
async def test_create_entity(
entity_service: EntityService, file_service: FileService, project_config: ProjectConfig
):
async def test_create_entity(entity_service: EntityService, file_service: FileService):
"""Test successful entity creation."""
entity_data = EntitySchema(
title="Test Entity",
directory="",
entity_type="test",
)
# Save expected permalink before create_entity mutates entity_data._permalink
expected_permalink = f"{generate_permalink(project_config.name)}/{entity_data.permalink}"
# Act
entity = await entity_service.create_entity(entity_data)
# Assert Entity
assert isinstance(entity, EntityModel)
assert entity.permalink == expected_permalink
assert entity.permalink == entity_data.permalink
assert entity.file_path == entity_data.file_path
assert entity.entity_type == "test"
assert entity.created_at is not None
assert len(entity.relations) == 0
# Verify we can retrieve it using permalink
retrieved = await entity_service.get_by_permalink(entity.permalink)
retrieved = await entity_service.get_by_permalink(entity_data.permalink)
assert retrieved.title == "Test Entity"
assert retrieved.entity_type == "test"
assert retrieved.created_at is not None
@@ -63,9 +59,7 @@ async def test_create_entity(
@pytest.mark.asyncio
async def test_create_entity_file_exists(
entity_service: EntityService, file_service: FileService, project_config: ProjectConfig
):
async def test_create_entity_file_exists(entity_service: EntityService, file_service: FileService):
"""Test successful entity creation."""
entity_data = EntitySchema(
title="Test Entity",
@@ -83,8 +77,7 @@ async def test_create_entity_file_exists(
file_content, _ = await file_service.read_file(file_path)
assert (
f"---\ntitle: Test Entity\ntype: test\npermalink: {generate_permalink(project_config.name)}/test-entity\n---\n\nfirst"
== file_content
"---\ntitle: Test Entity\ntype: test\npermalink: test-entity\n---\n\nfirst" == file_content
)
entity_data = EntitySchema(
@@ -115,9 +108,7 @@ async def test_create_entity_unique_permalink(
entity = await entity_service.create_entity(entity_data)
# default permalink
assert entity.permalink == (
f"{generate_permalink(project_config.name)}/{generate_permalink(entity.file_path)}"
)
assert entity.permalink == generate_permalink(entity.file_path)
# move file
file_path = file_service.get_entity_path(entity)
@@ -545,11 +536,7 @@ async def test_create_with_content(entity_service: EntityService, file_service:
@pytest.mark.asyncio
async def test_update_with_content(
entity_service: EntityService,
file_service: FileService,
project_config: ProjectConfig,
):
async def test_update_with_content(entity_service: EntityService, file_service: FileService):
content = """# Git Workflow Guide"""
# Create test entity
@@ -573,14 +560,13 @@ async def test_update_with_content(
file_content, _ = await file_service.read_file(file_path)
# assert content is in file
project_prefix = generate_permalink(project_config.name)
assert (
dedent(
f"""
"""
---
title: Git Workflow Guide
type: test
permalink: {project_prefix}/test/git-workflow-guide
permalink: test/git-workflow-guide
---
# Git Workflow Guide
@@ -1428,7 +1414,7 @@ async def test_move_entity_success(
app_config = BasicMemoryConfig(update_permalinks_on_move=False)
# Move entity
assert entity.permalink == f"{generate_permalink(project_config.name)}/original/test-note"
assert entity.permalink == "original/test-note"
await entity_service.move_entity(
identifier=entity.permalink,
destination_path="moved/test-note.md",
+34 -82
View File
@@ -6,10 +6,9 @@ import pytest
import pytest_asyncio
from basic_memory.models.knowledge import Entity as EntityModel
from basic_memory.repository import EntityRepository
from basic_memory.schemas.base import Entity as EntitySchema
from basic_memory.services.link_resolver import LinkResolver
from basic_memory.models.knowledge import Entity as EntityModel
@pytest_asyncio.fixture
@@ -111,46 +110,40 @@ async def link_resolver(entity_repository, search_service, test_entities):
return LinkResolver(entity_repository, search_service)
@pytest.fixture
def project_prefix(test_entities) -> str:
"""Project permalink prefix for expected permalinks."""
return test_entities[0].permalink.split("/", 1)[0]
@pytest.mark.asyncio
async def test_exact_permalink_match(link_resolver, test_entities, project_prefix):
async def test_exact_permalink_match(link_resolver, test_entities):
"""Test resolving a link that exactly matches a permalink."""
entity = await link_resolver.resolve_link("components/core-service")
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
@pytest.mark.asyncio
async def test_exact_title_match(link_resolver, test_entities, project_prefix):
async def test_exact_title_match(link_resolver, test_entities):
"""Test resolving a link that matches an entity title."""
entity = await link_resolver.resolve_link("Core Service")
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
@pytest.mark.asyncio
async def test_duplicate_title_match(link_resolver, test_entities, project_prefix):
async def test_duplicate_title_match(link_resolver, test_entities):
"""Test resolving a link that matches an entity title."""
entity = await link_resolver.resolve_link("Core Service")
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
@pytest.mark.asyncio
async def test_fuzzy_title_partial_match(link_resolver, project_prefix):
async def test_fuzzy_title_partial_match(link_resolver):
# Test partial match
result = await link_resolver.resolve_link("Auth Serv")
assert result is not None, "Did not find partial match"
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
@pytest.mark.asyncio
async def test_fuzzy_title_exact_match(link_resolver, project_prefix):
async def test_fuzzy_title_exact_match(link_resolver):
# Test partial match
result = await link_resolver.resolve_link("auth-service")
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
@pytest.mark.asyncio
@@ -190,7 +183,7 @@ async def test_resolve_file(link_resolver):
@pytest.mark.asyncio
async def test_folder_title_pattern_with_md_extension(link_resolver, test_entities, project_prefix):
async def test_folder_title_pattern_with_md_extension(link_resolver, test_entities):
"""Test resolving folder/title patterns that need .md extension added.
This tests the new logic added in step 4 of resolve_link that handles
@@ -200,25 +193,25 @@ async def test_folder_title_pattern_with_md_extension(link_resolver, test_entiti
# "components/Core Service" should resolve to file path "components/Core Service.md"
entity = await link_resolver.resolve_link("components/Core Service")
assert entity is not None
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
assert entity.file_path == "components/Core Service.md"
# Test with different entity
entity = await link_resolver.resolve_link("config/Service Config")
assert entity is not None
assert entity.permalink == f"{project_prefix}/config/service-config"
assert entity.permalink == "config/service-config"
assert entity.file_path == "config/Service Config.md"
# Test with nested folder structure
entity = await link_resolver.resolve_link("specs/subspec/Sub Features 1")
assert entity is not None
assert entity.permalink == f"{project_prefix}/specs/subspec/sub-features-1"
assert entity.permalink == "specs/subspec/sub-features-1"
assert entity.file_path == "specs/subspec/Sub Features 1.md"
# Test that it doesn't try to add .md to things that already have it
entity = await link_resolver.resolve_link("components/Core Service.md")
assert entity is not None
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
# Test that it doesn't try to add .md to single words (no slash)
entity = await link_resolver.resolve_link("NonExistent")
@@ -227,12 +220,12 @@ async def test_folder_title_pattern_with_md_extension(link_resolver, test_entiti
# Test that it doesn't interfere with exact permalink matches
entity = await link_resolver.resolve_link("components/core-service")
assert entity is not None
assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.permalink == "components/core-service"
# Tests for strict mode parameter combinations
@pytest.mark.asyncio
async def test_strict_mode_parameter_combinations(link_resolver, test_entities, project_prefix):
async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
"""Test all combinations of use_search and strict parameters."""
# Test queries
@@ -243,11 +236,11 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities,
# Case 1: use_search=True, strict=False (default behavior - fuzzy matching allowed)
result = await link_resolver.resolve_link(exact_match, use_search=True, strict=False)
assert result is not None
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=False)
assert result is not None # Should find "Auth Service" via fuzzy matching
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
result = await link_resolver.resolve_link(non_existent, use_search=True, strict=False)
assert result is None
@@ -255,7 +248,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities,
# Case 2: use_search=True, strict=True (exact matches only, even with search enabled)
result = await link_resolver.resolve_link(exact_match, use_search=True, strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=True)
assert result is None # Should NOT find via fuzzy matching in strict mode
@@ -266,7 +259,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities,
# Case 3: use_search=False, strict=False (no search, exact repository matches only)
result = await link_resolver.resolve_link(exact_match, use_search=False, strict=False)
assert result is not None
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=False)
assert result is None # No search means no fuzzy matching
@@ -277,7 +270,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities,
# Case 4: use_search=False, strict=True (redundant but should work same as case 3)
result = await link_resolver.resolve_link(exact_match, use_search=False, strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/auth-service"
assert result.permalink == "components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=True)
assert result is None # No search means no fuzzy matching
@@ -287,28 +280,28 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities,
@pytest.mark.asyncio
async def test_exact_match_types_in_strict_mode(link_resolver, test_entities, project_prefix):
async def test_exact_match_types_in_strict_mode(link_resolver, test_entities):
"""Test that all types of exact matches work in strict mode."""
# 1. Exact permalink match
result = await link_resolver.resolve_link("components/core-service", strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/core-service"
assert result.permalink == "components/core-service"
# 2. Exact title match
result = await link_resolver.resolve_link("Core Service", strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/core-service"
assert result.permalink == "components/core-service"
# 3. Exact file path match
result = await link_resolver.resolve_link("components/Core Service.md", strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/core-service"
assert result.permalink == "components/core-service"
# 4. Folder/title pattern with .md extension added
result = await link_resolver.resolve_link("components/Core Service", strict=True)
assert result is not None
assert result.permalink == f"{project_prefix}/components/core-service"
assert result.permalink == "components/core-service"
# 5. Non-markdown file (Image.png)
result = await link_resolver.resolve_link("Image.png", strict=True)
@@ -336,14 +329,14 @@ async def test_fuzzy_matching_blocked_in_strict_mode(link_resolver, test_entitie
@pytest.mark.asyncio
async def test_link_normalization_with_strict_mode(link_resolver, test_entities, project_prefix):
async def test_link_normalization_with_strict_mode(link_resolver, test_entities):
"""Test that link normalization still works in strict mode."""
# Test bracket removal and alias handling in strict mode
queries_and_expected = [
("[[Core Service]]", f"{project_prefix}/components/core-service"),
("[[Core Service|Main]]", f"{project_prefix}/components/core-service"),
(" [[ Core Service ]] ", f"{project_prefix}/components/core-service"),
("[[Core Service]]", "components/core-service"),
("[[Core Service|Main]]", "components/core-service"), # Alias should be ignored
(" [[ Core Service ]] ", "components/core-service"), # Extra whitespace
]
for query, expected_permalink in queries_and_expected:
@@ -353,7 +346,7 @@ async def test_link_normalization_with_strict_mode(link_resolver, test_entities,
@pytest.mark.asyncio
async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entities, project_prefix):
async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entities):
"""Test how duplicate titles are handled in strict mode."""
# "Core Service" appears twice in test data (components/core-service and components2/core-service)
@@ -363,48 +356,7 @@ async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entit
result = await link_resolver.resolve_link("Core Service", strict=True)
assert result is not None
# Should return the first match (components/core-service based on test fixture order)
assert result.permalink == f"{project_prefix}/components/core-service"
@pytest.mark.asyncio
async def test_cross_project_link_resolution(
session_maker, entity_repository, search_service, tmp_path
):
"""Test resolving explicit cross-project links."""
from basic_memory.repository.project_repository import ProjectRepository
project_repo = ProjectRepository(session_maker)
other_project = await project_repo.create(
{
"name": "other-project",
"description": "Secondary project",
"path": str(tmp_path / "other-project"),
"is_active": True,
"is_default": False,
}
)
now = datetime.now(timezone.utc)
other_entity_repo = EntityRepository(session_maker, project_id=other_project.id)
target = await other_entity_repo.add(
EntityModel(
title="Cross Project Note",
entity_type="note",
content_type="text/markdown",
file_path="docs/Cross Project Note.md",
permalink=f"{other_project.permalink}/docs/cross-project-note",
created_at=now,
updated_at=now,
project_id=other_project.id,
)
)
resolver = LinkResolver(entity_repository, search_service)
resolved = await resolver.resolve_link("other-project::Cross Project Note", strict=True)
assert resolved is not None
assert resolved.id == target.id
assert resolved.project_id == other_project.id
assert result.permalink == "components/core-service"
# ============================================================================
+23 -31
View File
@@ -12,32 +12,28 @@ from basic_memory.schemas.search import SearchQuery, SearchItemType
@pytest.mark.asyncio
async def test_search_permalink(search_service, test_graph):
"""Exact permalink"""
results = await search_service.search(SearchQuery(permalink="test-project/test/root"))
results = await search_service.search(SearchQuery(permalink="test/root"))
assert len(results) == 1
for r in results:
assert "test-project/test/root" in r.permalink
assert "test/root" in r.permalink
@pytest.mark.asyncio
async def test_search_limit_offset(search_service, test_graph):
"""Exact permalink"""
results = await search_service.search(SearchQuery(permalink_match="test-project/test/*"))
results = await search_service.search(SearchQuery(permalink_match="test/*"))
assert len(results) > 1
results = await search_service.search(
SearchQuery(permalink_match="test-project/test/*"), limit=1
)
results = await search_service.search(SearchQuery(permalink_match="test/*"), limit=1)
assert len(results) == 1
results = await search_service.search(
SearchQuery(permalink_match="test-project/test/*"), limit=100
)
results = await search_service.search(SearchQuery(permalink_match="test/*"), limit=100)
num_results = len(results)
# assert offset
offset_results = await search_service.search(
SearchQuery(permalink_match="test-project/test/*"), limit=100, offset=1
SearchQuery(permalink_match="test/*"), limit=100, offset=1
)
assert len(offset_results) == num_results - 1
@@ -45,24 +41,20 @@ async def test_search_limit_offset(search_service, test_graph):
@pytest.mark.asyncio
async def test_search_permalink_observations_wildcard(search_service, test_graph):
"""Pattern matching"""
results = await search_service.search(
SearchQuery(permalink_match="test-project/test/root/observations/*")
)
results = await search_service.search(SearchQuery(permalink_match="test/root/observations/*"))
assert len(results) == 2
permalinks = {r.permalink for r in results}
assert "test-project/test/root/observations/note/root-note-1" in permalinks
assert "test-project/test/root/observations/tech/root-tech-note" in permalinks
assert "test/root/observations/note/root-note-1" in permalinks
assert "test/root/observations/tech/root-tech-note" in permalinks
@pytest.mark.asyncio
async def test_search_permalink_relation_wildcard(search_service, test_graph):
"""Pattern matching"""
results = await search_service.search(
SearchQuery(permalink_match="test-project/test/root/connects-to/*")
)
results = await search_service.search(SearchQuery(permalink_match="test/root/connects-to/*"))
assert len(results) == 1
permalinks = {r.permalink for r in results}
assert "test-project/test/root/connects-to/test-project/test/connected-entity-1" in permalinks
assert "test/root/connects-to/test/connected-entity-1" in permalinks
@pytest.mark.asyncio
@@ -70,13 +62,13 @@ async def test_search_permalink_wildcard2(search_service, test_graph):
"""Pattern matching"""
results = await search_service.search(
SearchQuery(
permalink_match="test-project/test/connected*",
permalink_match="test/connected*",
)
)
assert len(results) >= 2
permalinks = {r.permalink for r in results}
assert "test-project/test/connected-entity-1" in permalinks
assert "test-project/test/connected-entity-2" in permalinks
assert "test/connected-entity-1" in permalinks
assert "test/connected-entity-2" in permalinks
@pytest.mark.asyncio
@@ -86,7 +78,7 @@ async def test_search_text(search_service, test_graph):
SearchQuery(text="Root Entity", entity_types=[SearchItemType.ENTITY])
)
assert len(results) >= 1
assert results[0].permalink == "test-project/test/root"
assert results[0].permalink == "test/root"
@pytest.mark.asyncio
@@ -96,7 +88,7 @@ async def test_search_title(search_service, test_graph):
SearchQuery(title="Root", entity_types=[SearchItemType.ENTITY])
)
assert len(results) >= 1
assert results[0].permalink == "test-project/test/root"
assert results[0].permalink == "test/root"
@pytest.mark.asyncio
@@ -104,7 +96,7 @@ async def test_text_search_case_insensitive(search_service, test_graph):
"""Test text search functionality."""
# Case insensitive
results = await search_service.search(SearchQuery(text="ENTITY"))
assert any("test-project/test/root" in r.permalink for r in results)
assert any("test/root" in r.permalink for r in results)
@pytest.mark.asyncio
@@ -123,16 +115,16 @@ async def test_text_search_multiple_terms(search_service, test_graph):
# Multiple terms
results = await search_service.search(SearchQuery(text="root note"))
assert any("test-project/test/root" in r.permalink for r in results)
assert any("test/root" in r.permalink for r in results)
@pytest.mark.asyncio
async def test_pattern_matching(search_service, test_graph):
"""Test pattern matching with various wildcards."""
# Test wildcards
results = await search_service.search(SearchQuery(permalink_match="test-project/test/*"))
results = await search_service.search(SearchQuery(permalink_match="test/*"))
for r in results:
assert "test-project/test/" in r.permalink
assert "test/" in r.permalink
# Test start wildcards
results = await search_service.search(SearchQuery(permalink_match="*/observations"))
@@ -140,9 +132,9 @@ async def test_pattern_matching(search_service, test_graph):
assert "/observations" in r.permalink
# Test permalink partial match
results = await search_service.search(SearchQuery(permalink_match="test-project/test"))
results = await search_service.search(SearchQuery(permalink_match="test"))
for r in results:
assert "test-project/test/" in r.permalink
assert "test/" in r.permalink
@pytest.mark.asyncio
@@ -336,7 +328,7 @@ async def test_boolean_or_search(search_service, test_graph):
connected_found = False
for result in results:
if result.permalink == "test-project/test/root":
if result.permalink == "test/root":
root_found = True
elif "connected" in result.permalink.lower():
connected_found = True
+20 -30
View File
@@ -14,7 +14,6 @@ from basic_memory.schemas.search import SearchQuery
from basic_memory.services import EntityService, FileService
from basic_memory.services.search_service import SearchService
from basic_memory.sync.sync_service import SyncService
from basic_memory.utils import generate_permalink
async def create_test_file(path: Path, content: str = "test content") -> None:
@@ -75,8 +74,7 @@ type: knowledge
await sync_service.sync(project_config.home)
# Verify forward reference
project_prefix = generate_permalink(project_config.name)
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
source = await entity_service.get_by_permalink("source")
assert len(source.relations) == 1
assert source.relations[0].to_id is None
assert source.relations[0].to_name == "target-doc"
@@ -100,8 +98,8 @@ Target content
await sync_service.sync(project_config.home)
# Verify reference is now resolved
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
target = await entity_service.get_by_permalink(f"{project_prefix}/target-doc")
source = await entity_service.get_by_permalink("source")
target = await entity_service.get_by_permalink("target-doc")
assert len(source.relations) == 1
assert source.relations[0].to_id == target.id
assert source.relations[0].to_name == target.title
@@ -146,9 +144,8 @@ Content
# Sync to create both entities
await sync_service.sync(project_config.home)
project_prefix = generate_permalink(project_config.name)
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
target = await entity_service.get_by_permalink(f"{project_prefix}/target")
source = await entity_service.get_by_permalink("source")
target = await entity_service.get_by_permalink("target")
# Create a resolved relation (already exists) that the unresolved one would become.
resolved_relation = Relation(
@@ -170,7 +167,7 @@ Content
unresolved_id = unresolved_relation.id
# Verify we have the unresolved relation
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
source = await entity_service.get_by_permalink("source")
unresolved_outgoing = [r for r in source.outgoing_relations if r.to_id is None]
assert len(unresolved_outgoing) == 1
assert unresolved_outgoing[0].id == unresolved_id
@@ -189,7 +186,7 @@ Content
assert len(unresolved) == 0
# Verify only the resolved relation remains
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
source = await entity_service.get_by_permalink("source")
assert len(source.outgoing_relations) == 1
assert source.outgoing_relations[0].to_id == target.id
@@ -645,7 +642,6 @@ async def test_permalink_formatting(
sync_service: SyncService, project_config: ProjectConfig, entity_service: EntityService
):
"""Test that permalinks are properly formatted during sync."""
project_prefix = generate_permalink(project_config.name)
# Test cases with different filename formats
test_files = {
@@ -679,9 +675,8 @@ Testing permalink generation.
for filename, expected_permalink in test_files.items():
# Find entity for this file
entity = next(e for e in entities if e.file_path == filename)
expected_full_permalink = f"{project_prefix}/{expected_permalink}"
assert entity.permalink == expected_full_permalink, (
f"File {filename} should have permalink {expected_full_permalink}"
assert entity.permalink == expected_permalink, (
f"File {filename} should have permalink {expected_permalink}"
)
@@ -748,13 +743,12 @@ Testing file timestamps
await sync_service.sync(project_config.home)
# Check explicit frontmatter dates
project_prefix = generate_permalink(project_config.name)
explicit_entity = await entity_service.get_by_permalink(f"{project_prefix}/explicit-dates")
explicit_entity = await entity_service.get_by_permalink("explicit-dates")
assert explicit_entity.created_at is not None
assert explicit_entity.updated_at is not None
# Check file timestamps
file_entity = await entity_service.get_by_permalink(f"{project_prefix}/file-dates3")
file_entity = await entity_service.get_by_permalink("file-dates3")
file_stats = file_path.stat()
# Compare using epoch timestamps to handle timezone differences correctly
@@ -799,8 +793,7 @@ Initial content for timestamp test
await sync_service.sync(project_config.home)
# Get initial entity and timestamps
project_prefix = generate_permalink(project_config.name)
entity_before = await entity_service.get_by_permalink(f"{project_prefix}/timestamp-test")
entity_before = await entity_service.get_by_permalink("timestamp-test")
initial_updated_at = entity_before.updated_at
# Modify the file content and update mtime to be newer than watermark
@@ -831,7 +824,7 @@ Modified content for timestamp test
await sync_service.sync(project_config.home)
# Get entity after re-sync
entity_after = await entity_service.get_by_permalink(f"{project_prefix}/timestamp-test")
entity_after = await entity_service.get_by_permalink("timestamp-test")
# Verify that updated_at changed
assert entity_after.updated_at != initial_updated_at, (
@@ -945,7 +938,6 @@ async def test_sync_permalink_resolved(
):
"""Test that we resolve duplicate permalinks on sync ."""
project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
# Create initial file
content = """
@@ -975,13 +967,13 @@ Content for move test
await sync_service.sync(project_config.home)
file_content, _ = await file_service.read_file(new_path)
assert f"permalink: {project_prefix}/new/moved-file" in file_content
assert "permalink: new/moved-file" in file_content
# Create another that has the same permalink
content = f"""
content = """
---
type: knowledge
permalink: {project_prefix}/new/moved-file
permalink: new/moved-file
---
# Test Move
Content for move test
@@ -999,7 +991,7 @@ Content for move test
# assert permalink is unique
file_content, _ = await file_service.read_file(old_path)
assert f"permalink: {project_prefix}/new/moved-file-1" in file_content
assert "permalink: new/moved-file-1" in file_content
@pytest.mark.asyncio
@@ -1132,7 +1124,6 @@ async def test_sync_permalink_updated_on_move(
):
"""Test that we update a permalink on a file move if set in config ."""
project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
# Create initial file
content = dedent(
@@ -1154,7 +1145,7 @@ async def test_sync_permalink_updated_on_move(
# verify permalink
old_content, _ = await file_service.read_file(old_path)
assert f"permalink: {project_prefix}/old/test-move" in old_content
assert "permalink: old/test-move" in old_content
# Move the file
new_path = project_dir / "new" / "moved_file.md"
@@ -1169,7 +1160,7 @@ async def test_sync_permalink_updated_on_move(
await sync_service.sync(project_config.home)
file_content, _ = await file_service.read_file(new_path)
assert f"permalink: {project_prefix}/new/moved-file" in file_content
assert "permalink: new/moved-file" in file_content
@pytest.mark.asyncio
@@ -1306,7 +1297,6 @@ async def test_sync_relation_to_non_markdown_file(
):
"""Test that sync resolves permalink conflicts on update."""
project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
content = f"""
---
@@ -1331,7 +1321,7 @@ tags: []
title: a note
type: note
tags: []
permalink: {project_prefix}/note
permalink: note
---
- relates_to [[{test_files["pdf"].name}]]
-10
View File
@@ -215,16 +215,6 @@ class TestConfigManager:
config = BasicMemoryConfig(disable_permalinks=True)
assert config.disable_permalinks is True
def test_permalinks_include_project_flag_default(self):
"""Test that permalinks_include_project defaults to True."""
config = BasicMemoryConfig()
assert config.permalinks_include_project is True
def test_permalinks_include_project_flag_can_be_disabled(self):
"""Test that permalinks_include_project can be set to False."""
config = BasicMemoryConfig(permalinks_include_project=False)
assert config.permalinks_include_project is False
def test_config_manager_respects_custom_config_dir(self, monkeypatch):
"""Test that ConfigManager respects BASIC_MEMORY_CONFIG_DIR environment variable."""
with tempfile.TemporaryDirectory() as temp_dir:
+3 -6
View File
@@ -61,14 +61,11 @@ Testing permalink generation.
# Run sync
await sync_service.sync(project_config.home)
# Verify permalinks - with project-prefixed permalinks enabled,
# auto-generated permalinks include the project slug prefix
project_prefix = generate_permalink(project_config.name)
# Verify permalinks
for filename, expected_permalink in test_cases:
entity = await entity_service.repository.get_by_file_path(filename)
expected_full = f"{project_prefix}/{expected_permalink}"
assert entity.permalink == expected_full, (
f"File {filename} should have permalink {expected_full}"
assert entity.permalink == expected_permalink, (
f"File {filename} should have permalink {expected_permalink}"
)