fix(mcp): route mixed local/cloud projects correctly (#837)

Signed-off-by: Drew Cain <groksrc@gmail.com>

AGENTS.md

@@ -56,6 +56,15 @@ Run `just test-smoke` when you specifically need the MCP smoke flow.
 
 If testmon is “cold,” the first run may be long. Subsequent runs get much faster.
 
+### PR CI Gate
+
+Before opening or updating a PR, run the checks that mirror the common required CI failures:
+
+- Run `just typecheck` in addition to targeted `ruff` and `pytest` commands when tests were added or changed.
+- Sign commits with `git commit -s` so DCO passes. If a PR branch already has unsigned commits, rewrite the branch with signed-off commits before asking for review.
+- Use a semantic PR title accepted by `.github/workflows/pr-title.yml`: `type(scope): summary`.
+- Use one of the allowed scopes: `core`, `cli`, `api`, `mcp`, `sync`, `ui`, `deps`, `installer`.
+
 ### Test Structure
 
 - `tests/` - Unit tests for individual components (mocked, fast)

src/basic_memory/mcp/project_context.py

@@ -142,6 +142,15 @@ async def _set_cached_active_project(
         await context.set_state("default_project_name", active_project.name)
 
 
+async def _clear_cached_active_project(context: Optional[Context]) -> None:
+    """Clear cached project metadata that may no longer match the active route."""
+    if not context:
+        return
+
+    await context.set_state("active_project", None)
+    await context.set_state("default_project_name", None)
+
+
 async def _get_cached_active_workspace(context: Optional[Context]) -> Optional[WorkspaceInfo]:
     """Return the cached active workspace from context when available."""
     if not context:
@@ -167,8 +176,7 @@ async def _set_cached_active_workspace(
         # Why: project names are only unique inside one workspace, so a cached
         #   ProjectItem from the previous tenant can point at the wrong project
         # Outcome: force the next validation call to resolve within the new tenant
-        await context.set_state("active_project", None)
-        await context.set_state("default_project_name", None)
+        await _clear_cached_active_project(context)
 
     await context.set_state("active_workspace", active_workspace.model_dump())
 
@@ -525,6 +533,28 @@ def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool:
     )
 
 
+def _workspace_identifier_discovery_available(
+    identifier: str,
+    config: BasicMemoryConfig,
+) -> bool:
+    """Return True when an identifier is allowed to consult workspace discovery."""
+    if _cloud_workspace_discovery_available(config):
+        return True
+
+    from basic_memory.mcp.async_client import (
+        _explicit_routing,
+        _force_local_mode,
+    )
+
+    if _explicit_routing() and _force_local_mode():
+        return False
+
+    return (
+        has_cloud_credentials(config)
+        and _split_workspace_identifier_segments(identifier) is not None
+    )
+
+
 async def resolve_workspace_qualified_memory_url(
     identifier: str,
     context: Optional[Context] = None,
@@ -1326,11 +1356,23 @@ async def detect_project_from_identifier_prefix(
         # Outcome: keep unqualified search/title input on the active/default project route.
         return None
 
-    if _cloud_workspace_discovery_available(config):
-        workspace_resolution = await resolve_workspace_qualified_identifier(
-            identifier,
-            context=context,
+    if _workspace_identifier_discovery_available(identifier, config):
+        workspace_discovery_fallback_errors = (
+            "not found",
+            "no accessible workspaces",
+            "unable to discover",
         )
+        try:
+            workspace_resolution = await resolve_workspace_qualified_identifier(
+                identifier,
+                context=context,
+            )
+        except ValueError as exc:
+            message = str(exc).lower()
+            if any(error in message for error in workspace_discovery_fallback_errors):
+                return None
+            raise
+
         if workspace_resolution is not None:
             return workspace_resolution.project_identifier
 
@@ -1345,7 +1387,7 @@ async def detect_project_from_identifier_prefix(
             )
         except ValueError as exc:
             message = str(exc).lower()
-            if "not found" in message or "no accessible workspaces" in message:
+            if any(error in message for error in workspace_discovery_fallback_errors):
                 return None
             raise
 
@@ -1462,9 +1504,54 @@ async def get_project_client(
     if project_id and not cloud_available:
         project_mode = ProjectMode.LOCAL
 
+    # Trigger: project_id is a local external_id in a mixed local+cloud setup.
+    # Why: UUIDs are not local config keys, so get_project_mode() treats them as
+    #   cloud projects. A local-first probe avoids making local UUIDs depend on
+    #   healthy cloud workspace discovery.
+    # Outcome: resolve the effective UUID against local ASGI first; if it is not
+    #   local, preserve the existing cloud workspace lookup path.
+    if (
+        project_id
+        and config.projects
+        and not factory_mode
+        and not explicit_cloud_routing
+        and project_mode == ProjectMode.CLOUD
+    ):
+        try:
+            canonical_project_id = str(UUID(resolved_project))
+        except ValueError:
+            pass
+        else:
+            with logfire.span(
+                "routing.local_project_id_probe",
+                project_id=canonical_project_id,
+            ):
+                async with get_client() as client:
+                    try:
+                        active_project = await get_active_project(
+                            client,
+                            canonical_project_id,
+                            context,
+                        )
+                    except ToolError as exc:
+                        if "not found" not in str(exc).lower():
+                            raise
+                    else:
+                        route_mode = "local_asgi"
+                        await _clear_cached_active_workspace_for_local_route(context)
+                        with logfire.span(
+                            "routing.client_session",
+                            project_name=active_project.name,
+                            route_mode=route_mode,
+                        ):
+                            logger.debug("Using local ASGI routing for project_id")
+                            yield client, active_project
+                        return
+
     if factory_mode or project_mode == ProjectMode.CLOUD or explicit_cloud_routing:
         route_mode = "factory" if factory_mode else "cloud_proxy"
         active_ws: WorkspaceInfo | None = None
+        resolved_entry: WorkspaceProjectEntry | None = None
         workspace_id: str
         project_for_api = _unqualified_project_identifier(resolved_project)
 
@@ -1487,6 +1574,13 @@ async def get_project_client(
 
         if active_ws is not None:
             await _set_cached_active_workspace(context, active_ws)
+        if resolved_entry is not None:
+            cached_project = await _get_cached_active_project(context)
+            if (
+                cached_project is not None
+                and cached_project.external_id != resolved_entry.project.external_id
+            ):
+                await _clear_cached_active_project(context)
         with logfire.span(
             "routing.client_session",
             project_name=project_for_api,

src/basic_memory/mcp/tools/edit_note.py

@@ -9,7 +9,7 @@ from pydantic import AliasChoices, Field
 
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import (
-    _cloud_workspace_discovery_available,
+    _workspace_identifier_discovery_available,
     detect_project_from_memory_url_prefix,
     get_project_client,
     add_project_metadata,
@@ -368,8 +368,9 @@ async def edit_note(
                 config,
                 context=context,
             )
-        elif _cloud_workspace_discovery_available(
-            config
+        elif _workspace_identifier_discovery_available(
+            identifier,
+            config,
         ) and is_workspace_qualified_plain_identifier(identifier):
             detected = await detect_project_from_workspace_identifier_prefix(
                 identifier,

src/basic_memory/services/link_resolver.py

@@ -40,17 +40,29 @@ async def detect_project_from_workspace_identifier_prefix(
         return None
 
     from basic_memory.mcp.project_context import (
-        _cloud_workspace_discovery_available,
+        _workspace_identifier_discovery_available,
         resolve_workspace_qualified_identifier,
     )
 
-    if not _cloud_workspace_discovery_available(config):
+    if not _workspace_identifier_discovery_available(identifier, config):
         return None
 
-    workspace_resolution = await resolve_workspace_qualified_identifier(
-        identifier,
-        context=context,
+    workspace_discovery_fallback_errors = (
+        "not found",
+        "no accessible workspaces",
+        "unable to discover",
     )
+    try:
+        workspace_resolution = await resolve_workspace_qualified_identifier(
+            identifier,
+            context=context,
+        )
+    except ValueError as exc:
+        message = str(exc).lower()
+        if any(error in message for error in workspace_discovery_fallback_errors):
+            return None
+        raise
+
     if workspace_resolution is None:
         return None
     return workspace_resolution.project_identifier

tests/mcp/conftest.py

@@ -20,6 +20,31 @@ def mcp() -> FastMCP:
     return cast(Any, mcp_server)
 
 
+class ContextState:
+    """Minimal FastMCP context-state stub for MCP tests."""
+
+    def __init__(self):
+        self._state: dict[str, object] = {}
+
+    async def get_state(self, key: str):
+        return self._state.get(key)
+
+    async def set_state(self, key: str, value: object, **kwargs) -> None:
+        self._state[key] = value
+
+    async def info(self, message: str) -> None:
+        self._state["info_message"] = message
+
+
+def ctx(context: ContextState) -> Any:
+    return cast(Any, context)
+
+
+@pytest.fixture
+def context_state() -> ContextState:
+    return ContextState()
+
+
 @pytest.fixture(scope="function")
 def app(
     app_config, project_config, engine_factory, config_manager

tests/mcp/test_project_context.py

@@ -11,25 +11,7 @@ from typing import Any, AsyncIterator, cast
 
 import pytest
 
-
-class _ContextState:
-    """Minimal FastMCP context-state stub for unit tests."""
-
-    def __init__(self):
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-    async def info(self, message: str) -> None:
-        self._state["info_message"] = message
-
-
-def _ctx(context: _ContextState) -> Any:
-    return cast(Any, context)
+from tests.mcp.conftest import ContextState, ctx
 
 
 def _workspace(
@@ -218,7 +200,7 @@ async def test_env_constraint_overrides_default(config_manager, config_home, mon
 async def test_workspace_auto_selects_single_and_caches(monkeypatch):
     from basic_memory.mcp.project_context import resolve_workspace_parameter
 
-    context = _ContextState()
+    context = ContextState()
     only_workspace = _workspace(
         tenant_id="11111111-1111-1111-1111-111111111111",
         workspace_type="personal",
@@ -236,7 +218,7 @@ async def test_workspace_auto_selects_single_and_caches(monkeypatch):
         fake_get_available_workspaces,
     )
 
-    resolved = await resolve_workspace_parameter(context=_ctx(context))
+    resolved = await resolve_workspace_parameter(context=ctx(context))
     assert resolved.tenant_id == only_workspace.tenant_id
     assert await context.get_state("active_workspace") == only_workspace.model_dump()
 
@@ -272,7 +254,7 @@ async def test_workspace_requires_user_choice_when_multiple(monkeypatch):
     )
 
     with pytest.raises(ValueError, match="Multiple workspaces are available"):
-        await resolve_workspace_parameter(context=_ctx(_ContextState()))
+        await resolve_workspace_parameter(context=ctx(ContextState()))
 
 
 @pytest.mark.asyncio
@@ -416,7 +398,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
             role="owner",
         ),
     ]
-    context = _ContextState()
+    context = ContextState()
     await context.set_state("active_workspace", cached_workspace.model_dump())
     fetches = 0
 
@@ -431,7 +413,7 @@ async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(m
     )
 
     with pytest.raises(ValueError) as exc_info:
-        await resolve_workspace_parameter(workspace="organization", context=_ctx(context))
+        await resolve_workspace_parameter(workspace="organization", context=ctx(context))
 
     message = str(exc_info.value)
     assert fetches == 1
@@ -452,7 +434,7 @@ async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
         role="owner",
         is_default=True,
     )
-    context = _ContextState()
+    context = ContextState()
     await context.set_state("active_workspace", cached_workspace.model_dump())
 
     async def fail_if_called(context=None):  # pragma: no cover
@@ -463,7 +445,7 @@ async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
         fail_if_called,
     )
 
-    resolved = await resolve_workspace_parameter(context=_ctx(context))
+    resolved = await resolve_workspace_parameter(context=ctx(context))
     assert resolved.tenant_id == cached_workspace.tenant_id
 
 
@@ -476,7 +458,7 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
         invalidate_workspace_project_index,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -513,8 +495,8 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
         fake_fetch_workspace_project_entries,
     )
 
-    first = await _ensure_workspace_project_index(context=_ctx(context))
-    second = await _ensure_workspace_project_index(context=_ctx(context))
+    first = await _ensure_workspace_project_index(context=ctx(context))
+    second = await _ensure_workspace_project_index(context=ctx(context))
 
     assert [entry.qualified_name for entry in first.entries] == [
         "personal/personal-notes",
@@ -523,8 +505,8 @@ async def test_workspace_project_index_caches_and_invalidates(monkeypatch):
     assert second.entries == first.entries
     assert calls == ["personal", "acme"]
 
-    await invalidate_workspace_project_index(_ctx(context))
-    await _ensure_workspace_project_index(context=_ctx(context))
+    await invalidate_workspace_project_index(ctx(context))
+    await _ensure_workspace_project_index(context=ctx(context))
     assert calls == ["personal", "acme", "personal", "acme"]
 
 
@@ -539,7 +521,7 @@ async def test_workspace_project_index_keeps_successes_when_workspace_fetch_fail
         resolve_workspace_project_identifier,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -572,21 +554,21 @@ async def test_workspace_project_index_keeps_successes_when_workspace_fetch_fail
         fake_fetch_workspace_project_entries,
     )
 
-    index = await _ensure_workspace_project_index(context=_ctx(context))
+    index = await _ensure_workspace_project_index(context=ctx(context))
 
     assert [entry.qualified_name for entry in index.entries] == ["personal/meeting-notes"]
     assert [workspace.slug for workspace in index.failed_workspaces] == ["acme"]
 
     resolved = await resolve_workspace_project_identifier(
         "personal/meeting-notes",
-        context=_ctx(context),
+        context=ctx(context),
     )
     assert resolved.project.external_id == "personal-meeting-notes"
 
     with pytest.raises(ValueError, match="Use 'personal/meeting-notes'"):
         await resolve_workspace_project_identifier(
             "meeting-notes",
-            context=_ctx(context),
+            context=ctx(context),
         )
 
 
@@ -800,8 +782,11 @@ async def test_detect_project_from_memory_url_prefix_skips_workspace_discovery_f
 
     monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_if_called)
 
+    # Keep this at two path segments. Three-segment memory URLs are valid
+    # workspace-qualified candidates in mixed local+cloud mode, so they should
+    # attempt workspace discovery even when a local project is configured.
     resolved = await detect_project_from_memory_url_prefix(
-        "memory://notes/foo/bar",
+        "memory://notes/foo",
         BasicMemoryConfig(
             projects={"main": ProjectEntry(path="/tmp/main")},
             cloud_api_key="bmc_test123",
@@ -811,6 +796,101 @@ async def test_detect_project_from_memory_url_prefix_skips_workspace_discovery_f
     assert resolved is None
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("identifier", "message"),
+    [
+        ("memory://docs/topic/note", "No accessible workspaces found for this account."),
+        (
+            "docs/topic/note",
+            "Unable to discover projects in any accessible workspace. Failed workspaces: personal",
+        ),
+    ],
+)
+async def test_detect_project_from_identifier_prefix_ignores_workspace_discovery_failures(
+    monkeypatch,
+    identifier,
+    message,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import BasicMemoryConfig, ProjectEntry
+    from basic_memory.mcp.project_context import detect_project_from_identifier_prefix
+
+    async def fail_workspace_index(context=None):
+        raise ValueError(message)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_workspace_index)
+
+    resolved = await detect_project_from_identifier_prefix(
+        identifier,
+        BasicMemoryConfig(
+            projects={"main": ProjectEntry(path="/tmp/main")},
+            cloud_api_key="bmc_test123",
+        ),
+    )
+
+    assert resolved is None
+
+
+@pytest.mark.asyncio
+async def test_detect_project_from_identifier_prefix_resolves_workspace_with_local_config(
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import BasicMemoryConfig, ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+        detect_project_from_identifier_prefix,
+    )
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (
+            WorkspaceProjectEntry(
+                workspace=personal,
+                project=_project("main", id=1, external_id="personal-main-id"),
+            ),
+        ),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+
+    config = BasicMemoryConfig(
+        projects={"hermes-memory": ProjectEntry(path="/tmp/hermes-memory")},
+        cloud_api_key="bmc_test123",
+    )
+
+    assert (
+        await detect_project_from_identifier_prefix(
+            "memory://personal/main/main-to-do-list",
+            config,
+        )
+        == "personal/main"
+    )
+    assert (
+        await detect_project_from_identifier_prefix(
+            "personal/main/main-to-do-list",
+            config,
+        )
+        == "personal/main"
+    )
+
+
 @pytest.mark.asyncio
 async def test_resolve_workspace_qualified_memory_url_ignores_workspace_project_miss(
     monkeypatch,
@@ -1045,7 +1125,7 @@ async def test_resolve_workspace_project_identifier_uses_active_workspace_for_du
         resolve_workspace_project_identifier,
     )
 
-    context = _ContextState()
+    context = ContextState()
     personal = _workspace(
         tenant_id="personal-tenant",
         workspace_type="personal",
@@ -1081,7 +1161,7 @@ async def test_resolve_workspace_project_identifier_uses_active_workspace_for_du
 
     resolved = await resolve_workspace_project_identifier(
         "meeting-notes",
-        context=_ctx(context),
+        context=ctx(context),
     )
     assert resolved.workspace.slug == "acme"
     assert resolved.project.external_id == "acme-project-id"
@@ -1232,6 +1312,318 @@ async def test_get_project_client_with_project_id_routes_locally_without_cloud(
     assert captured["validated_project"] == canonical_uuid
 
 
+@pytest.mark.asyncio
+async def test_get_project_client_with_local_project_id_routes_locally_with_cloud_credentials(
+    config_manager, monkeypatch
+):
+    """A local-only project_id should resolve locally before cloud workspace discovery."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("local project_id should not require cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project("Hermes Memory", id=99, external_id=project)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    local_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(project_id=local_uuid) as (_, active):
+        assert active.external_id == local_uuid
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == local_uuid
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_local_project_id_clears_cached_workspace(
+    config_manager, monkeypatch
+):
+    """Local project_id routing must not inherit a previous cloud workspace context."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("local project_id should not require cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project("Hermes Memory", id=99, external_id=project)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    local_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(
+        project_id=local_uuid,
+        context=ctx(context),
+    ) as (_, active):
+        assert active.external_id == local_uuid
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == local_uuid
+    assert await context.get_state("active_workspace") is None
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_project_id_respects_env_constraint(
+    config_manager, monkeypatch
+):
+    """BASIC_MEMORY_MCP_PROJECT must remain authoritative when project_id is supplied."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+
+    config = config_manager.load_config()
+    config.projects["env-project"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "env-project")
+    )
+    config.projects["other-project"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "other-project")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
+
+    async def fail_index(context=None):  # pragma: no cover
+        raise AssertionError("env-constrained local project should not use cloud discovery")
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_client(**kwargs) -> AsyncIterator[object]:
+        captured["get_client_kwargs"] = kwargs
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        captured["validated_project"] = project
+        return _project(str(project), id=99, external_id="env-project-id")
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fail_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    requested_uuid = "55555555-5555-5555-5555-555555555555"
+    async with project_context.get_project_client(project_id=requested_uuid) as (_, active):
+        assert active.name == "env-project"
+
+    assert captured["get_client_kwargs"] == {}
+    assert captured["validated_project"] == "env-project"
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_with_cloud_project_id_routes_to_workspace_with_local_config(
+    config_manager, monkeypatch
+):
+    """Cloud project_id routing falls through to the workspace index after a local miss."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from mcp.server.fastmcp.exceptions import ToolError
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    cloud_uuid = "22222222-2222-2222-2222-222222222222"
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    cloud_project = _project("main", id=2, external_id=cloud_uuid)
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=cloud_project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    get_client_calls: list[dict[str, str | None]] = []
+    validated_projects: list[str] = []
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        get_client_calls.append({"project_name": project_name, "workspace": workspace})
+        yield object()
+
+    async def fake_get_active_project(client, project, context=None, headers=None):
+        validated_projects.append(project)
+        if project == cloud_uuid:
+            raise ToolError("project not found")
+        return _project(project, id=cloud_project.id, external_id=cloud_project.external_id)
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
+
+    async with project_context.get_project_client(project_id=cloud_uuid) as (_, active):
+        assert active.external_id == cloud_uuid
+
+    assert get_client_calls == [
+        {"project_name": None, "workspace": None},
+        {"project_name": "main", "workspace": "personal-tenant"},
+    ]
+    assert validated_projects == [cloud_uuid, "main"]
+
+
+@pytest.mark.asyncio
+async def test_get_project_client_clears_stale_cached_project_for_workspace_route(
+    config_manager, monkeypatch
+):
+    """Workspace routes must not reuse a same-name project with a different UUID."""
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import WorkspaceProjectEntry
+    from basic_memory.schemas.project_info import ProjectItem
+
+    config = config_manager.load_config()
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = _workspace(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    expected_uuid = "22222222-2222-2222-2222-222222222222"
+    expected_project = _project("main", id=2, external_id=expected_uuid)
+    stale_project = ProjectItem(
+        id=99,
+        external_id="33333333-3333-3333-3333-333333333333",
+        name="main",
+        path="/tmp/stale-main",
+        is_default=False,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+    await context.set_state("active_project", stale_project.model_dump())
+
+    async def fake_resolve_workspace_project_identifier(project_name, context=None):
+        assert project_name == "personal/main"
+        return WorkspaceProjectEntry(workspace=personal, project=expected_project)
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        assert project_name == "main"
+        assert workspace == "personal-tenant"
+        yield object()
+
+    class FakeResponse:
+        def json(self):
+            return {
+                "external_id": expected_uuid,
+                "project_id": expected_project.id,
+                "name": expected_project.name,
+                "permalink": expected_project.permalink,
+                "path": expected_project.path,
+                "is_active": True,
+                "is_default": False,
+                "resolution_method": "permalink",
+            }
+
+    calls = {"count": 0}
+
+    async def fake_call_post(*args, **kwargs):
+        calls["count"] += 1
+        return FakeResponse()
+
+    monkeypatch.setattr(
+        project_context,
+        "resolve_workspace_project_identifier",
+        fake_resolve_workspace_project_identifier,
+    )
+    monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: True)
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+
+    async with project_context.get_project_client(
+        project="personal/main",
+        context=ctx(context),
+    ) as (_, active):
+        assert active.external_id == expected_uuid
+
+    cached_project = await context.get_state("active_project")
+    assert isinstance(cached_project, dict)
+    assert cached_project["external_id"] == expected_uuid
+    assert calls["count"] == 1
+
+
 @pytest.mark.asyncio
 async def test_get_project_client_prefers_project_id_over_project_name(monkeypatch):
     """When both project and project_id are passed, the UUID takes precedence."""
@@ -1270,7 +1662,7 @@ async def test_resolve_project_parameter_uses_cached_active_project_before_api_d
     config.default_project = None
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1288,7 +1680,7 @@ async def test_resolve_project_parameter_uses_cached_active_project_before_api_d
         fail_if_called,
     )
 
-    resolved = await resolve_project_parameter(project=None, context=_ctx(context))
+    resolved = await resolve_project_parameter(project=None, context=ctx(context))
     assert resolved == cached_project.name
 
 
@@ -1302,7 +1694,7 @@ async def test_resolve_project_parameter_caches_api_default_project_name(
     config.default_project = None
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     api_calls = {"count": 0}
 
     async def fake_default_lookup():
@@ -1314,8 +1706,8 @@ async def test_resolve_project_parameter_caches_api_default_project_name(
         fake_default_lookup,
     )
 
-    first = await resolve_project_parameter(project=None, context=_ctx(context))
-    second = await resolve_project_parameter(project=None, context=_ctx(context))
+    first = await resolve_project_parameter(project=None, context=ctx(context))
+    second = await resolve_project_parameter(project=None, context=ctx(context))
 
     assert first == "cloud-default"
     assert second == "cloud-default"
@@ -1327,7 +1719,7 @@ async def test_get_active_project_uses_cached_project_before_resolution(monkeypa
     from basic_memory.mcp.project_context import get_active_project
     from basic_memory.schemas.project_info import ProjectItem
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1345,7 +1737,7 @@ async def test_get_active_project_uses_cached_project_before_resolution(monkeypa
         fail_if_called,
     )
 
-    resolved = await get_active_project(client=cast(Any, None), context=_ctx(context))
+    resolved = await get_active_project(client=cast(Any, None), context=ctx(context))
     assert resolved == cached_project
 
 
@@ -1354,7 +1746,7 @@ async def test_get_active_project_uses_cached_project_for_explicit_permalink(mon
     from basic_memory.mcp.project_context import get_active_project
     from basic_memory.schemas.project_info import ProjectItem
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1375,7 +1767,7 @@ async def test_get_active_project_uses_cached_project_for_explicit_permalink(mon
     )
 
     resolved = await get_active_project(
-        client=cast(Any, None), project="my-research", context=_ctx(context)
+        client=cast(Any, None), project="my-research", context=ctx(context)
     )
     assert resolved == cached_project
 
@@ -1391,7 +1783,7 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
     config.permalinks_include_project = False
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1416,7 +1808,7 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://my-research/notes/roadmap.md",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1438,7 +1830,7 @@ async def test_resolve_project_and_path_keeps_workspace_qualified_canonical_path
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1464,7 +1856,7 @@ async def test_resolve_project_and_path_keeps_workspace_qualified_canonical_path
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://team-paul/main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1486,7 +1878,7 @@ async def test_resolve_project_and_path_preserves_personal_workspace_prefix(
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1513,7 +1905,7 @@ async def test_resolve_project_and_path_preserves_personal_workspace_prefix(
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://personal/main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1532,7 +1924,7 @@ async def test_resolve_project_and_path_preserves_existing_project_prefixed_memo
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1545,7 +1937,7 @@ async def test_resolve_project_and_path_preserves_existing_project_prefixed_memo
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1568,7 +1960,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1599,7 +1991,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
         client=cast(Any, None),
         identifier="memory://notes/foo",
         project="main",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1610,7 +2002,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_active_route(
         client=cast(Any, None),
         identifier="memory://main",
         project="main",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1650,7 +2042,7 @@ async def test_resolve_project_and_path_uses_workspace_context_for_project_root(
             client=cast(Any, None),
             identifier="memory://main",
             project="main",
-            context=_ctx(_ContextState()),
+            context=ctx(ContextState()),
         )
 
     assert active_project == active
@@ -1671,7 +2063,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_cached_project
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     cached_project = ProjectItem(
         id=1,
         external_id="11111111-1111-1111-1111-111111111111",
@@ -1705,7 +2097,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_cached_project
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://main/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project == cached_project
@@ -1725,7 +2117,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_resolved_proje
     config.permalinks_include_project = True
     config_manager.save_config(config)
 
-    context = _ContextState()
+    context = ContextState()
     team_workspace = _workspace(
         tenant_id="team-tenant",
         workspace_type="organization",
@@ -1764,7 +2156,7 @@ async def test_resolve_project_and_path_uses_cached_workspace_for_resolved_proje
     active_project, resolved_path, is_memory_url = await resolve_project_and_path(
         client=cast(Any, None),
         identifier="memory://research/notes/foo",
-        context=_ctx(context),
+        context=ctx(context),
     )
 
     assert active_project.name == "Research"
@@ -1895,7 +2287,7 @@ class TestGetProjectClientRoutingOrder:
         )
         config_manager.save_config(config)
 
-        context = _ContextState()
+        context = ContextState()
         stale_workspace = _workspace(
             tenant_id="team-tenant",
             workspace_type="organization",
@@ -1942,13 +2334,13 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="local-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (client, active_project):
             _, resolved_path, is_memory_url = await resolve_project_and_path(
                 client=cast(Any, client),
                 identifier="memory://notes/foo",
                 project="local-proj",
-                context=_ctx(context),
+                context=ctx(context),
             )
 
         assert active_project == active
@@ -1990,7 +2382,7 @@ class TestGetProjectClientRoutingOrder:
             name="Team Paul",
             role="editor",
         )
-        context = _ContextState()
+        context = ContextState()
         await context.set_state("active_workspace", workspace.model_dump())
 
         async def fail_resolve_workspace_parameter(workspace=None, context=None):
@@ -2034,7 +2426,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 
@@ -2154,7 +2546,7 @@ class TestGetProjectClientRoutingOrder:
             name="Team Paul",
             role="editor",
         )
-        context = _ContextState()
+        context = ContextState()
         await context.set_state("available_workspaces", ["ignored", workspace.model_dump()])
 
         async def fail_workspace_provider():  # pragma: no cover
@@ -2193,7 +2585,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 
@@ -2335,7 +2727,7 @@ class TestGetProjectClientRoutingOrder:
         config.cloud_api_key = "bmc_test123"
         config_manager.save_config(config)
 
-        context = _ContextState()
+        context = ContextState()
         stale_workspace = _workspace(
             tenant_id="other-tenant-id",
             workspace_type="organization",
@@ -2376,7 +2768,7 @@ class TestGetProjectClientRoutingOrder:
 
         async with get_project_client(
             project="cloud-proj",
-            context=_ctx(context),
+            context=ctx(context),
         ) as (_client, active_project):
             assert active_project.external_id == "cloud-project-id"
 

tests/mcp/test_project_context_telemetry.py

@@ -4,30 +4,17 @@ from __future__ import annotations
 
 import importlib
 from contextlib import contextmanager
-from typing import Any, cast
 
 import logfire
 import pytest
 
 from basic_memory.config import ProjectEntry
 from basic_memory.schemas.cloud import WorkspaceInfo
+from tests.mcp.conftest import ContextState, ctx
 
 project_context = importlib.import_module("basic_memory.mcp.project_context")
 
 
-class _ContextState:
-    """Minimal FastMCP context-state stub for unit tests."""
-
-    def __init__(self) -> None:
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-
 def _capture_spans():
     spans: list[tuple[str, dict]] = []
 
@@ -42,7 +29,7 @@ def _capture_spans():
 @pytest.mark.asyncio
 async def test_resolve_workspace_parameter_emits_routing_span(monkeypatch) -> None:
     spans, fake_span = _capture_spans()
-    context = _ContextState()
+    context = ContextState()
     workspace = WorkspaceInfo(
         tenant_id="11111111-1111-1111-1111-111111111111",
         workspace_type="personal",
@@ -58,7 +45,7 @@ async def test_resolve_workspace_parameter_emits_routing_span(monkeypatch) -> No
     monkeypatch.setattr(logfire, "span", fake_span)
     monkeypatch.setattr(project_context, "get_available_workspaces", fake_get_available_workspaces)
 
-    resolved = await project_context.resolve_workspace_parameter(context=cast(Any, context))
+    resolved = await project_context.resolve_workspace_parameter(context=ctx(context))
 
     assert resolved.tenant_id == workspace.tenant_id
     assert spans == [

tests/mcp/test_tool_edit_note.py

@@ -914,8 +914,8 @@ async def test_edit_note_workspace_qualified_plain_permalink_requires_explicit_r
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -958,8 +958,8 @@ async def test_edit_note_workspace_qualified_plain_permalink_json_error(
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -1001,8 +1001,8 @@ async def test_edit_note_ambiguous_namespace_identifier_returns_guidance(
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: True,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         edit_note_module,
@@ -1054,6 +1054,83 @@ async def test_edit_note_workspace_project_args_compose_explicit_route(monkeypat
     assert captured_routes == [("docs/setup", None)]
 
 
+@pytest.mark.asyncio
+async def test_edit_note_plain_workspace_route_returns_guidance_with_local_config(
+    monkeypatch,
+    config_manager,
+    test_project,
+):
+    """Mixed local+cloud configs should still stop ambiguous plain write routes."""
+    from contextlib import asynccontextmanager
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+
+    edit_note_module = importlib.import_module("basic_memory.mcp.tools.edit_note")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (
+            WorkspaceProjectEntry(
+                workspace=personal,
+                project=ProjectItem(
+                    id=1,
+                    external_id="11111111-1111-1111-1111-111111111111",
+                    name="main",
+                    path="/tmp/main",
+                    is_default=False,
+                ),
+            ),
+        ),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    @asynccontextmanager
+    async def fail_if_called(*args, **kwargs):
+        raise AssertionError("ambiguous plain identifiers should not select a project client")
+        yield
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(edit_note_module, "get_project_client", fail_if_called)
+
+    result = await edit_note(
+        identifier="personal/main/team/plain-edit-note",
+        operation="append",
+        content="\nAppended via plain workspace-qualified permalink.",
+        project=None,
+    )
+
+    assert isinstance(result, str)
+    assert "# Edit Failed - Ambiguous Identifier" in result
+    assert 'project="personal/main"' in result
+
+
 @pytest.mark.asyncio
 async def test_edit_note_three_segment_plain_path_stays_local_without_workspace_discovery(
     monkeypatch,
@@ -1077,8 +1154,8 @@ async def test_edit_note_three_segment_plain_path_stays_local_without_workspace_
 
     monkeypatch.setattr(
         edit_note_module,
-        "_cloud_workspace_discovery_available",
-        lambda config: False,
+        "_workspace_identifier_discovery_available",
+        lambda identifier, config: False,
     )
     monkeypatch.setattr(
         edit_note_module,

tests/mcp/test_tool_read_content.py

@@ -6,6 +6,10 @@ We keep these tests focused on path boundary/security checks, and rely on
 
 from __future__ import annotations
 
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+
 import pytest
 from mcp.server.fastmcp.exceptions import ToolError
 
@@ -157,6 +161,109 @@ async def test_read_content_allows_safe_path_integration(client, test_project):
     assert "safe note" in result["text"]
 
 
+@pytest.mark.asyncio
+async def test_read_content_workspace_memory_url_routes_with_local_config(
+    monkeypatch,
+    config_manager,
+):
+    """Workspace-qualified memory URLs should route even when local projects exist."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+
+    read_content_module = importlib.import_module("basic_memory.mcp.tools.read_content")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    @asynccontextmanager
+    async def fake_get_project_client(project=None, context=None, project_id=None):
+        assert project == "personal/main"
+        assert project_id is None
+        yield (
+            object(),
+            SimpleNamespace(
+                name="main",
+                external_id="11111111-1111-1111-1111-111111111111",
+                home=Path("/tmp/main"),
+            ),
+        )
+
+    async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
+        assert identifier == "memory://personal/main/docs/report"
+        assert project == "main"
+        return None, "personal/main/docs/report", True
+
+    async def fake_resolve_entity_id(client, project_id, url):
+        assert project_id == "11111111-1111-1111-1111-111111111111"
+        assert url == "personal/main/docs/report"
+        return "entity-1"
+
+    class FakeResponse:
+        headers = {"content-type": "text/markdown", "content-length": "17"}
+        text = "# Routed Content"
+        content = b"# Routed Content"
+
+    async def fake_call_get(client, path, **kwargs):
+        assert path == "/v2/projects/11111111-1111-1111-1111-111111111111/resource/entity-1"
+        return FakeResponse()
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(read_content_module, "get_project_client", fake_get_project_client)
+    monkeypatch.setattr(
+        read_content_module,
+        "resolve_project_and_path",
+        fake_resolve_project_and_path,
+    )
+    monkeypatch.setattr(read_content_module, "resolve_entity_id", fake_resolve_entity_id)
+    monkeypatch.setattr(read_content_module, "call_get", fake_call_get)
+
+    result = await read_content(path="memory://personal/main/docs/report")
+
+    assert result == {
+        "type": "text",
+        "text": "# Routed Content",
+        "content_type": "text/markdown",
+        "encoding": "utf-8",
+    }
+
+
 @pytest.mark.asyncio
 async def test_read_content_empty_path_does_not_trigger_security_error(client, test_project):
     try:

tests/mcp/test_tool_read_note.py

@@ -1,5 +1,7 @@
 """Tests for note tools that exercise the full stack with SQLite."""
 
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
 from textwrap import dedent
 
 import pytest
@@ -7,6 +9,7 @@ import pytest
 from basic_memory.mcp.tools import write_note, read_note
 from basic_memory.mcp.tools.read_note import _parse_opening_frontmatter
 from basic_memory.utils import normalize_newlines
+from tests.mcp.conftest import ContextState, ctx
 
 
 def test_parse_opening_frontmatter_handles_crlf():
@@ -166,6 +169,130 @@ async def test_read_note_title_fallback_requires_exact_title_match(monkeypatch,
     assert "existing note content" not in result
 
 
+@pytest.mark.asyncio
+async def test_read_note_explicit_workspace_project_ignores_stale_cached_project(
+    monkeypatch,
+    config_manager,
+):
+    """Explicit workspace routing should use the resolved cloud UUID, not stale cache."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import WorkspaceProjectEntry
+    from basic_memory.schemas.project_info import ProjectItem
+
+    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
+    clients_mod = importlib.import_module("basic_memory.mcp.clients")
+
+    config = config_manager.load_config()
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = project_context.WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    expected_uuid = "22222222-2222-2222-2222-222222222222"
+    expected_project = ProjectItem(
+        id=2,
+        external_id=expected_uuid,
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    stale_project = ProjectItem(
+        id=99,
+        external_id="33333333-3333-3333-3333-333333333333",
+        name="main",
+        path="/tmp/stale-main",
+        is_default=False,
+    )
+    context = ContextState()
+    await context.set_state("active_workspace", personal.model_dump())
+    await context.set_state("active_project", stale_project.model_dump())
+
+    async def fake_resolve_workspace_project_identifier(project_name, context=None):
+        assert project_name == "personal/main"
+        return WorkspaceProjectEntry(workspace=personal, project=expected_project)
+
+    @asynccontextmanager
+    async def fake_get_client(project_name=None, workspace=None):
+        assert project_name == "main"
+        assert workspace == "personal-tenant"
+        yield object()
+
+    class FakeProjectResolveResponse:
+        def json(self):
+            return {
+                "external_id": expected_uuid,
+                "project_id": expected_project.id,
+                "name": expected_project.name,
+                "permalink": expected_project.permalink,
+                "path": expected_project.path,
+                "is_active": True,
+                "is_default": False,
+                "resolution_method": "permalink",
+            }
+
+    async def fake_call_post(*args, **kwargs):
+        return FakeProjectResolveResponse()
+
+    class FakeKnowledgeClient:
+        def __init__(self, client, project_id):
+            assert project_id == expected_uuid
+
+        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
+            assert identifier == "personal/main/todo"
+            return "entity-1"
+
+        async def get_entity(self, entity_id: str):
+            assert entity_id == "entity-1"
+            return SimpleNamespace(
+                title="TODO",
+                permalink="personal/main/todo",
+                file_path="TODO.md",
+            )
+
+    class FakeResourceClient:
+        def __init__(self, client, project_id):
+            assert project_id == expected_uuid
+
+        async def read(self, entity_id: str):
+            assert entity_id == "entity-1"
+            return SimpleNamespace(
+                status_code=200,
+                text="---\ntitle: TODO\n---\n\n# TODO - Priorities & Tasks\n",
+            )
+
+    monkeypatch.setattr(
+        project_context,
+        "resolve_workspace_project_identifier",
+        fake_resolve_workspace_project_identifier,
+    )
+    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
+    monkeypatch.setattr(clients_mod, "KnowledgeClient", FakeKnowledgeClient)
+    monkeypatch.setattr(clients_mod, "ResourceClient", FakeResourceClient)
+
+    result = await read_note_module.read_note(
+        "memory://todo",
+        project="personal/main",
+        output_format="json",
+        context=ctx(context),
+    )
+
+    assert isinstance(result, dict)
+    assert result["permalink"] == "personal/main/todo"
+    assert result["content"].strip() == "# TODO - Priorities & Tasks"
+
+
 @pytest.mark.asyncio
 async def test_note_unicode_content(app, test_project):
     """Test handling of unicode content in"""

tests/mcp/test_tool_search.py

@@ -3,6 +3,8 @@
 import pytest
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta
+from types import SimpleNamespace
+from typing import cast
 
 from basic_memory.mcp.tools import write_note
 from basic_memory.mcp.tools.search import (
@@ -171,6 +173,109 @@ async def test_search_memory_url_with_project_prefix(client, test_project):
         pytest.fail(f"Search failed with error: {response}")
 
 
+@pytest.mark.asyncio
+async def test_search_workspace_memory_url_routes_with_local_config(monkeypatch, config_manager):
+    """Workspace-qualified memory URL searches should self-route in mixed mode."""
+    import importlib
+
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+    from basic_memory.schemas.search import SearchItemType, SearchResult
+
+    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
+    clients_mod = importlib.import_module("basic_memory.mcp.clients")
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    personal = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project_item = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (personal,),
+        (WorkspaceProjectEntry(workspace=personal, project=project_item),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def fake_get_project_client(project=None, context=None, project_id=None):
+        captured["project"] = project
+        captured["project_id"] = project_id
+        yield object(), SimpleNamespace(name="main", external_id=project_item.external_id)
+
+    async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
+        assert identifier == "memory://personal/main/tests/search-note"
+        assert project == "main"
+        return None, "personal/main/tests/search-note", True
+
+    class FakeSearchClient:
+        def __init__(self, client, project_id):
+            captured["search_project_id"] = project_id
+
+        async def search(self, payload, *, page, page_size):
+            captured["payload"] = payload
+            return SearchResponse(
+                results=[
+                    SearchResult(
+                        title="Search Note",
+                        type=SearchItemType.ENTITY,
+                        score=1.0,
+                        permalink="personal/main/tests/search-note",
+                        file_path="tests/Search Note.md",
+                    )
+                ],
+                current_page=page,
+                page_size=page_size,
+                total=1,
+            )
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+    monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
+    monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
+    monkeypatch.setattr(clients_mod, "SearchClient", FakeSearchClient)
+
+    response = await search_notes(
+        query="memory://personal/main/tests/search-note",
+        output_format="json",
+    )
+
+    assert captured["project"] == "personal/main"
+    assert captured["project_id"] is None
+    assert captured["search_project_id"] == "11111111-1111-1111-1111-111111111111"
+    payload = cast(dict[str, object], captured["payload"])
+    assert payload["permalink"] == "personal/main/tests/search-note"
+    assert isinstance(response, dict)
+    assert response["results"][0]["permalink"] == "personal/main/tests/search-note"
+
+
 @pytest.mark.asyncio
 async def test_search_pagination(client, test_project):
     """Test basic search functionality."""

tests/mcp/test_tool_workspace_management.py

@@ -1,12 +1,11 @@
 """Tests for workspace MCP tools."""
 
-from typing import Any, cast
-
 import pytest
 
 from basic_memory.mcp.project_context import get_available_workspaces, set_workspace_provider
 from basic_memory.mcp.tools.workspaces import list_workspaces
 from basic_memory.schemas.cloud import WorkspaceInfo
+from tests.mcp.conftest import ContextState, ctx
 
 
 def _workspace(
@@ -28,17 +27,6 @@ def _workspace(
     )
 
 
-class _ContextState:
-    def __init__(self):
-        self._state: dict[str, object] = {}
-
-    async def get_state(self, key: str):
-        return self._state.get(key)
-
-    async def set_state(self, key: str, value: object, **kwargs) -> None:
-        self._state[key] = value
-
-
 @pytest.mark.asyncio
 async def test_list_workspaces_formats_workspace_rows(monkeypatch):
     async def fake_get_available_workspaces(context=None):
@@ -145,7 +133,7 @@ async def test_list_workspaces_oauth_error_bubbles_up(monkeypatch):
 
 @pytest.mark.asyncio
 async def test_list_workspaces_uses_context_cache_path(monkeypatch):
-    context = _ContextState()
+    context = ContextState()
     call_count = {"fetches": 0}
     workspace = _workspace(
         tenant_id="33333333-3333-3333-3333-333333333333",
@@ -169,8 +157,8 @@ async def test_list_workspaces_uses_context_cache_path(monkeypatch):
         fake_get_available_workspaces,
     )
 
-    first = await list_workspaces(context=cast(Any, context))
-    second = await list_workspaces(context=cast(Any, context))
+    first = await list_workspaces(context=ctx(context))
+    second = await list_workspaces(context=ctx(context))
 
     assert "# Available Workspaces (1)" in first
     assert "# Available Workspaces (1)" in second
@@ -254,14 +242,14 @@ async def test_get_available_workspaces_provider_caches_in_context():
         return [workspace]
 
     set_workspace_provider(counting_provider)
-    context = _ContextState()
+    context = ContextState()
 
     # First call: provider is invoked, result cached
-    first = await get_available_workspaces(context=cast(Any, context))
+    first = await get_available_workspaces(context=ctx(context))
     assert len(first) == 1
     assert call_count["provider"] == 1
 
     # Second call: served from context cache, provider not called again
-    second = await get_available_workspaces(context=cast(Any, context))
+    second = await get_available_workspaces(context=ctx(context))
     assert len(second) == 1
     assert call_count["provider"] == 1

tests/services/test_link_resolver.py

@@ -139,11 +139,11 @@ async def test_workspace_identifier_project_detection_requires_workspace_shape(
     """Two-segment project-relative paths should not trigger workspace discovery."""
     from basic_memory.services import link_resolver
 
-    def fail_if_called(config):
+    def fail_if_called(identifier, config):
         raise AssertionError("plain project-relative paths should skip cloud discovery")
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
         fail_if_called,
     )
 
@@ -164,8 +164,8 @@ async def test_workspace_identifier_project_detection_skips_without_discovery(
     from basic_memory.services import link_resolver
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
-        lambda config: False,
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: False,
     )
 
     detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
@@ -189,8 +189,8 @@ async def test_workspace_identifier_project_detection_returns_project(
         return SimpleNamespace(project_identifier="team-acme/research")
 
     monkeypatch.setattr(
-        "basic_memory.mcp.project_context._cloud_workspace_discovery_available",
-        lambda config: True,
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: True,
     )
     monkeypatch.setattr(
         "basic_memory.mcp.project_context.resolve_workspace_qualified_identifier",
@@ -205,6 +205,104 @@ async def test_workspace_identifier_project_detection_returns_project(
     assert detected == "team-acme/research"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Workspace 'team-acme' was not found.",
+        "No accessible workspaces found for this account.",
+        "Unable to discover projects in any accessible workspace. Failed workspaces: team-acme",
+    ],
+)
+async def test_workspace_identifier_project_detection_ignores_discovery_failures(
+    monkeypatch,
+    config_manager,
+    message,
+):
+    """Workspace detection should fall back when cloud discovery is unavailable."""
+    from basic_memory.services import link_resolver
+
+    async def fail_workspace_identifier(identifier, context=None):
+        raise ValueError(message)
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._workspace_identifier_discovery_available",
+        lambda identifier, config: True,
+    )
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context.resolve_workspace_qualified_identifier",
+        fail_workspace_identifier,
+    )
+
+    detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
+        "team-acme/research/note",
+        config_manager.config,
+    )
+
+    assert detected is None
+
+
+@pytest.mark.asyncio
+async def test_workspace_identifier_project_detection_allows_mixed_local_cloud_config(
+    monkeypatch,
+    config_manager,
+):
+    """Plain workspace routes should be discoverable even with local projects configured."""
+    from basic_memory.config import ProjectEntry
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+    )
+    from basic_memory.schemas.cloud import WorkspaceInfo
+    from basic_memory.schemas.project_info import ProjectItem
+    from basic_memory.services import link_resolver
+
+    config = config_manager.load_config()
+    config.projects["hermes-memory"] = ProjectEntry(
+        path=str(config_manager.config_dir.parent / "hermes-memory")
+    )
+    config.cloud_api_key = "bmc_test123"
+    config_manager.save_config(config)
+
+    workspace = WorkspaceInfo(
+        tenant_id="personal-tenant",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    project = ProjectItem(
+        id=1,
+        external_id="11111111-1111-1111-1111-111111111111",
+        name="main",
+        path="/tmp/main",
+        is_default=False,
+    )
+    index = _build_workspace_project_index(
+        (workspace,),
+        (WorkspaceProjectEntry(workspace=workspace, project=project),),
+    )
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._ensure_workspace_project_index",
+        fake_index,
+    )
+    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
+    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
+
+    detected = await link_resolver.detect_project_from_workspace_identifier_prefix(
+        "personal/main/todo",
+        config_manager.config,
+    )
+
+    assert detected == "personal/main"
+
+
 @pytest.mark.asyncio
 async def test_exact_permalink_match(link_resolver, test_entities, project_prefix):
     """Test resolving a link that exactly matches a permalink."""