fix(mcp): preserve workspace paths in build_context (#801)

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

src/basic_memory/mcp/project_context.py

@@ -162,6 +162,17 @@ async def _set_cached_active_workspace(
     await context.set_state("active_workspace", active_workspace.model_dump())
 
 
+async def _clear_cached_active_workspace_for_local_route(context: Optional[Context]) -> None:
+    """Drop tenant workspace metadata before routing through a local project."""
+    if not context:
+        return
+
+    # Trigger: local routing follows a cloud route in the same MCP session
+    # Why: active_workspace is tenant metadata, not part of local project identity
+    # Outcome: memory:// resolution uses project-only local permalinks
+    await context.set_state("active_workspace", None)
+
+
 async def _get_cached_default_project(context: Optional[Context]) -> Optional[str]:
     """Return the cached default project name from context when available."""
     if not context:
@@ -446,6 +457,52 @@ def _canonical_memory_path_for_workspace(
     return f"{prefix}/{normalized_remainder}"
 
 
+def _canonical_memory_path_for_active_route(
+    active_project: ProjectItem,
+    path: str,
+    *,
+    include_project: bool,
+    cached_workspace: WorkspaceInfo | None = None,
+) -> str:
+    """Return the canonical permalink path for the currently routed project/workspace."""
+    project_prefix = active_project.permalink
+    workspace_remainder = path
+    if include_project and (path == project_prefix or path.startswith(f"{project_prefix}/")):
+        # Trigger: the memory URL already names the active project root/prefix
+        # Why: workspace canonicalization adds the project prefix itself, so
+        #   keeping it in the remainder would produce <workspace>/<project>/<project>
+        # Outcome: keep project-root and project-prefixed URLs canonical once
+        workspace_remainder = (
+            "" if path == project_prefix else path.removeprefix(f"{project_prefix}/")
+        )
+
+    workspace_context = current_workspace_permalink_context()
+    if workspace_context is not None:
+        return _canonical_memory_path_for_workspace(
+            workspace_slug=workspace_context.workspace_slug,
+            workspace_type=workspace_context.workspace_type,
+            project_permalink=active_project.permalink,
+            remainder=workspace_remainder,
+            include_project=include_project,
+        )
+
+    if cached_workspace is not None:
+        return _canonical_memory_path_for_workspace(
+            workspace_slug=cached_workspace.slug,
+            workspace_type=cached_workspace.workspace_type,
+            project_permalink=active_project.permalink,
+            remainder=workspace_remainder,
+            include_project=include_project,
+        )
+
+    if not include_project:
+        return path
+
+    if path == project_prefix or path.startswith(f"{project_prefix}/"):
+        return path
+    return f"{project_prefix}/{path}"
+
+
 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 (
@@ -1115,8 +1172,11 @@ async def resolve_project_and_path(
                         f"Project is constrained to '{resolved_project}', cannot use '{project_prefix}'."
                     )
 
-                resolved_path = (
-                    f"{cached_project.permalink}/{remainder}" if include_project else remainder
+                resolved_path = _canonical_memory_path_for_active_route(
+                    cached_project,
+                    remainder,
+                    include_project=include_project,
+                    cached_workspace=cached_workspace,
                 )
                 return cached_project, resolved_path, True
 
@@ -1151,25 +1211,24 @@ async def resolve_project_and_path(
                 )
                 await _set_cached_active_project(context, active_project)
 
-                resolved_path = (
-                    f"{resolved.permalink}/{remainder}" if include_project else remainder
+                resolved_path = _canonical_memory_path_for_active_route(
+                    active_project,
+                    remainder,
+                    include_project=include_project,
+                    cached_workspace=cached_workspace,
                 )
                 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
+        # Trigger: memory URL has no resolvable project route segment
+        # Why: preserve active-project behavior while honoring workspace paths
+        # Outcome: normalize against the already-selected local/cloud route
         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}"
+        resolved_path = _canonical_memory_path_for_active_route(
+            active_project,
+            normalized_path,
+            include_project=include_project,
+            cached_workspace=cached_workspace,
+        )
         return active_project, resolved_path, True
 
 
@@ -1325,6 +1384,7 @@ async def get_project_client(
     # Outcome: route strictly based on explicit flag, no workspace network calls
     if _explicit_routing() and _force_local_mode():
         route_mode = "explicit_local"
+        await _clear_cached_active_workspace_for_local_route(context)
         with logfire.span(
             "routing.client_session",
             project_name=resolved_project,
@@ -1397,6 +1457,7 @@ async def get_project_client(
 
     # Step 4: Local routing (default)
     route_mode = "local_asgi"
+    await _clear_cached_active_workspace_for_local_route(context)
     with logfire.span(
         "routing.client_session",
         project_name=resolved_project,

tests/mcp/test_project_context.py

@@ -1453,6 +1453,225 @@ async def test_resolve_project_and_path_preserves_existing_project_prefixed_memo
     assert is_memory_url is True
 
 
+@pytest.mark.asyncio
+async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
+    config_manager,
+    monkeypatch,
+):
+    from mcp.server.fastmcp.exceptions import ToolError
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import resolve_project_and_path
+    from basic_memory.schemas.project_info import ProjectItem
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    context = _ContextState()
+    cached_project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    team_workspace = _workspace(
+        tenant_id="team-tenant",
+        workspace_type="organization",
+        slug="team-paul",
+        name="Team Paul",
+        role="editor",
+    )
+    await context.set_state("active_project", cached_project.model_dump())
+    await context.set_state("active_workspace", team_workspace.model_dump())
+
+    async def fake_call_post(*args, **kwargs):
+        raise ToolError("project not found")
+
+    async def fake_get_active_project(*args, **kwargs):
+        return cached_project
+
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+        client=cast(Any, None),
+        identifier="memory://notes/foo",
+        project="main",
+        context=_ctx(context),
+    )
+
+    assert active_project == cached_project
+    assert resolved_path == "team-paul/main/notes/foo"
+    assert is_memory_url is True
+
+    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+        client=cast(Any, None),
+        identifier="memory://main",
+        project="main",
+        context=_ctx(context),
+    )
+
+    assert active_project == cached_project
+    assert resolved_path == "team-paul/main"
+    assert is_memory_url is True
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_and_path_uses_workspace_context_for_project_root(
+    config_manager,
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import resolve_project_and_path
+    from basic_memory.schemas.project_info import ProjectItem
+    from basic_memory.workspace_context import workspace_permalink_context
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    active = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+
+    async def fake_get_active_project(*args, **kwargs):
+        return active
+
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
+        active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+            client=cast(Any, None),
+            identifier="memory://main",
+            project="main",
+            context=_ctx(_ContextState()),
+        )
+
+    assert active_project == active
+    assert resolved_path == "team-paul/main"
+    assert is_memory_url is True
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_and_path_uses_cached_workspace_for_cached_project_prefix(
+    config_manager,
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import resolve_project_and_path
+    from basic_memory.schemas.project_info import ProjectItem
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    context = _ContextState()
+    cached_project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    team_workspace = _workspace(
+        tenant_id="team-tenant",
+        workspace_type="organization",
+        slug="team-paul",
+        name="Team Paul",
+        role="editor",
+    )
+    await context.set_state("active_project", cached_project.model_dump())
+    await context.set_state("active_workspace", team_workspace.model_dump())
+
+    async def fail_call_post(*args, **kwargs):  # pragma: no cover
+        raise AssertionError("Cached project prefix should not call project resolve API")
+
+    async def fake_resolve_project_parameter(project=None, **kwargs):
+        return cached_project.name if project else cached_project.name
+
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fail_call_post)
+    monkeypatch.setattr(
+        project_context,
+        "resolve_project_parameter",
+        fake_resolve_project_parameter,
+    )
+
+    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+        client=cast(Any, None),
+        identifier="memory://main/notes/foo",
+        context=_ctx(context),
+    )
+
+    assert active_project == cached_project
+    assert resolved_path == "team-paul/main/notes/foo"
+    assert is_memory_url is True
+
+
+@pytest.mark.asyncio
+async def test_resolve_project_and_path_uses_cached_workspace_for_resolved_project_prefix(
+    config_manager,
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import resolve_project_and_path
+
+    config = config_manager.load_config()
+    config.permalinks_include_project = True
+    config_manager.save_config(config)
+
+    context = _ContextState()
+    team_workspace = _workspace(
+        tenant_id="team-tenant",
+        workspace_type="organization",
+        slug="team-paul",
+        name="Team Paul",
+        role="editor",
+    )
+    await context.set_state("active_workspace", team_workspace.model_dump())
+
+    class FakeResponse:
+        def json(self):
+            return {
+                "external_id": "22222222-2222-2222-2222-222222222222",
+                "project_id": 2,
+                "name": "Research",
+                "permalink": "research",
+                "path": "/tmp/research",
+                "is_active": True,
+                "is_default": False,
+                "resolution_method": "permalink",
+            }
+
+    async def fake_call_post(*args, **kwargs):
+        return FakeResponse()
+
+    async def fake_resolve_project_parameter(project=None, **kwargs):
+        return "Research" if project else "Research"
+
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+    monkeypatch.setattr(
+        project_context,
+        "resolve_project_parameter",
+        fake_resolve_project_parameter,
+    )
+
+    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
+        client=cast(Any, None),
+        identifier="memory://research/notes/foo",
+        context=_ctx(context),
+    )
+
+    assert active_project.name == "Research"
+    assert resolved_path == "team-paul/research/notes/foo"
+    assert is_memory_url is True
+
+
 class TestDetectProjectFromUrlPrefix:
     """Test detect_project_from_url_prefix for URL-based project detection."""
 
@@ -1556,6 +1775,92 @@ class TestGetProjectClientRoutingOrder:
         # The error should NOT be about workspaces
         assert "workspace" not in str(exc_info.value).lower()
 
+    @pytest.mark.asyncio
+    async def test_local_route_clears_stale_cached_workspace(self, config_manager, monkeypatch):
+        """A previous cloud workspace must not decorate later local memory URLs."""
+        from contextlib import asynccontextmanager
+
+        from mcp.server.fastmcp.exceptions import ToolError
+
+        import basic_memory.mcp.project_context as project_context
+        from basic_memory.config import ProjectEntry
+        from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
+        from basic_memory.schemas.project_info import ProjectItem
+        from basic_memory.workspace_context import current_workspace_permalink_context
+
+        config = config_manager.load_config()
+        config.permalinks_include_project = True
+        config.projects["local-proj"] = ProjectEntry(
+            path=str(config_manager.config_dir.parent / "local-proj")
+        )
+        config_manager.save_config(config)
+
+        context = _ContextState()
+        stale_workspace = _workspace(
+            tenant_id="team-tenant",
+            workspace_type="organization",
+            slug="team-paul",
+            name="Team Paul",
+            role="editor",
+        )
+        await context.set_state("active_workspace", stale_workspace.model_dump())
+
+        seen: dict[str, object] = {}
+        active = ProjectItem(
+            id=1,
+            external_id="local-project-id",
+            name="local-proj",
+            path="/local-proj",
+            is_default=False,
+        )
+
+        async def fail_ensure_workspace_project_index(context=None):  # pragma: no cover
+            raise AssertionError("Local routing must not discover cloud workspaces")
+
+        @asynccontextmanager
+        async def fake_get_client(project_name=None, workspace=None):
+            seen["project_name"] = project_name
+            seen["workspace"] = workspace
+            seen["permalink_context"] = current_workspace_permalink_context()
+            yield object()
+
+        async def fake_get_active_project(client, project_name, context=None, headers=None):
+            assert project_name == "local-proj"
+            return active
+
+        async def fake_call_post(*args, **kwargs):
+            raise ToolError("project not found")
+
+        monkeypatch.setattr(
+            project_context,
+            "_ensure_workspace_project_index",
+            fail_ensure_workspace_project_index,
+        )
+        monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+        monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+
+        async with get_project_client(
+            project="local-proj",
+            context=_ctx(context),
+        ) as (client, active_project):
+            _, resolved_path, is_memory_url = await resolve_project_and_path(
+                client=cast(Any, client),
+                identifier="memory://notes/foo",
+                project="local-proj",
+                context=_ctx(context),
+            )
+
+        assert active_project == active
+        assert seen == {
+            "project_name": "local-proj",
+            "workspace": None,
+            "permalink_context": None,
+        }
+        assert resolved_path == "local-proj/notes/foo"
+        assert is_memory_url is True
+        assert await context.get_state("active_workspace") is None
+
     @pytest.mark.asyncio
     async def test_cloud_project_uses_per_project_workspace_id(self, config_manager, monkeypatch):
         """Cloud project with workspace_id uses cached workspace permalink context."""

tests/mcp/test_tool_build_context.py

@@ -4,7 +4,7 @@ import pytest
 
 from mcp.server.fastmcp.exceptions import ToolError
 
-from basic_memory.mcp.tools import build_context
+from basic_memory.mcp.tools import build_context, write_note
 
 
 @pytest.mark.asyncio
@@ -75,6 +75,36 @@ async def test_get_discussion_context_pattern(client, test_graph, test_project):
     assert result["metadata"]["depth"] == 1
 
 
+@pytest.mark.asyncio
+async def test_build_context_project_id_preserves_workspace_contextvar_canonical_path(
+    app, test_project
+):
+    """project_id routing keeps ContextVar workspace prefixes in memory URL lookups."""
+    from basic_memory.workspace_context import workspace_permalink_context
+
+    with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
+        await write_note(
+            project_id=test_project.external_id,
+            title="Workspace Build Context Note",
+            directory="tests",
+            content="Build context should find this workspace note",
+        )
+
+        result = await build_context(
+            project_id=test_project.external_id,
+            url="memory://tests/*",
+            timeframe="30d",
+        )
+
+    assert isinstance(result, dict)
+    assert len(result["results"]) == 1
+    primary = result["results"][0]["primary_result"]
+    assert primary["permalink"] == (
+        f"team-paul/{test_project.name}/tests/workspace-build-context-note"
+    )
+    assert primary["content"] == "Build context should find this workspace note"
+
+
 @pytest.mark.asyncio
 async def test_get_discussion_context_timeframe(client, test_graph, test_project):
     """Test timeframe parameter filtering."""