fix(mcp): normalize personal workspace memory urls

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/mcp/project_context.py

@@ -422,6 +422,30 @@ def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str
     return workspace_slug, project_identifier, remainder
 
 
+def _canonical_memory_path_for_workspace(
+    *,
+    workspace_slug: str,
+    workspace_type: str,
+    project_permalink: str,
+    remainder: str,
+    include_project: bool,
+) -> str:
+    """Return the stored canonical path for a workspace-qualified memory URL."""
+    normalized_remainder = remainder.strip("/")
+    if workspace_type == "organization":
+        prefix = f"{generate_permalink(workspace_slug)}/{project_permalink}"
+    elif workspace_type == "personal":
+        prefix = project_permalink if include_project else ""
+    else:
+        raise ValueError(f"Unsupported workspace_type for memory URL routing: {workspace_type}")
+
+    if not prefix:
+        return normalized_remainder
+    if not normalized_remainder:
+        return prefix
+    return f"{prefix}/{normalized_remainder}"
+
+
 def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool:
     """Return True when workspace discovery can be used without forcing local routing."""
     from basic_memory.mcp.async_client import (
@@ -496,7 +520,13 @@ async def resolve_workspace_qualified_memory_url(
         )
 
     entry = matches[0]
-    canonical_path = f"{entry.workspace.slug}/{entry.project.permalink}/{remainder}"
+    canonical_path = _canonical_memory_path_for_workspace(
+        workspace_slug=entry.workspace.slug,
+        workspace_type=entry.workspace.workspace_type,
+        project_permalink=entry.project.permalink,
+        remainder=remainder,
+        include_project=ConfigManager().config.permalinks_include_project,
+    )
     return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path)
 
 
@@ -1005,7 +1035,19 @@ async def resolve_project_and_path(
             if normalized_path == qualified_prefix or normalized_path.startswith(
                 f"{qualified_prefix}/"
             ):
-                return cached_project, normalized_path, True
+                remainder = (
+                    ""
+                    if normalized_path == qualified_prefix
+                    else normalized_path.removeprefix(f"{qualified_prefix}/")
+                )
+                resolved_path = _canonical_memory_path_for_workspace(
+                    workspace_slug=cached_workspace.slug,
+                    workspace_type=cached_workspace.workspace_type,
+                    project_permalink=cached_project.permalink,
+                    remainder=remainder,
+                    include_project=bool(include_project),
+                )
+                return cached_project, resolved_path, True
 
         workspace_context = current_workspace_permalink_context()
         if workspace_context and project:
@@ -1016,7 +1058,19 @@ async def resolve_project_and_path(
                 f"{qualified_prefix}/"
             ):
                 active_project = await get_active_project(client, project, context, headers)
-                return active_project, normalized_path, True
+                remainder = (
+                    ""
+                    if normalized_path == qualified_prefix
+                    else normalized_path.removeprefix(f"{qualified_prefix}/")
+                )
+                resolved_path = _canonical_memory_path_for_workspace(
+                    workspace_slug=workspace_context.workspace_slug,
+                    workspace_type=workspace_context.workspace_type,
+                    project_permalink=project_permalink,
+                    remainder=remainder,
+                    include_project=bool(include_project),
+                )
+                return active_project, resolved_path, True
 
         project_prefix, remainder = _split_project_prefix(normalized_path)
         include_project = config.permalinks_include_project

tests/mcp/test_project_context.py

@@ -786,6 +786,49 @@ async def fake_index(context=None):
         await resolve_workspace_qualified_memory_url("memory://team-paul/main/notes/foo")
 
 
+@pytest.mark.asyncio
+async def test_resolve_workspace_qualified_memory_url_uses_personal_canonical_path(
+    config_manager,
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+        resolve_workspace_qualified_memory_url,
+    )
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    entries = (
+        WorkspaceProjectEntry(
+            workspace=personal,
+            project=_project("main", id=1, external_id="personal-main-id"),
+        ),
+    )
+    index = _build_workspace_project_index((personal,), entries)
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+
+    resolved = await resolve_workspace_qualified_memory_url("memory://personal/main/notes/foo")
+
+    assert resolved is not None
+    assert resolved.canonical_path == "main/notes/foo"
+
+
 @pytest.mark.asyncio
 async def test_get_project_client_routes_duplicate_project_through_workspace_slug(
     config_manager,
@@ -1287,6 +1330,55 @@ async def fake_call_post(*args, **kwargs):
     assert is_memory_url is True
 
 
+@pytest.mark.asyncio
+async def test_resolve_project_and_path_strips_personal_workspace_prefix(
+    config_manager,
+    monkeypatch,
+):
+    from mcp.server.fastmcp.exceptions import ToolError
+
+    from basic_memory.mcp.project_context import resolve_project_and_path
+    from basic_memory.schemas.project_info import ProjectItem
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    context = _ContextState()
+    cached_project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    personal_workspace = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    await context.set_state("active_project", cached_project.model_dump())
+    await context.set_state("active_workspace", personal_workspace.model_dump())
+
+    async def fake_call_post(*args, **kwargs):
+        raise ToolError("project not found")
+
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+
+    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+        client=cast(Any, None),
+        identifier="memory://personal/main/notes/foo",
+        context=_ctx(context),
+    )
+
+    assert active_project == cached_project
+    assert resolved_path == "main/notes/foo"
+    assert is_memory_url is True
+
+
 @pytest.mark.asyncio
 async def test_resolve_project_and_path_preserves_existing_project_prefixed_memory_url(
     config_manager,