basicmachines-co-basic-memory/tests/mcp/test_project_context.py

"""Tests for project context utilities (no standard-library mock usage).

These functions are config/env driven, so we use the real ConfigManager-backed
test config file and pytest monkeypatch for environment variables.
"""

from __future__ import annotations

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


@pytest.mark.asyncio
async def test_returns_none_when_no_default_and_no_project(config_manager, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    cfg.default_project = None
    config_manager.save_config(cfg)

    monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)

    # Prevent API fallback from returning a project via stale dependency overrides
    async def _no_api_fallback():
        return None

    monkeypatch.setattr(
        "basic_memory.mcp.project_context._resolve_default_project_from_api",
        _no_api_fallback,
    )
    assert await resolve_project_parameter(project=None, allow_discovery=False) is None


@pytest.mark.asyncio
async def test_allows_discovery_when_enabled(config_manager, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    cfg.default_project = None
    config_manager.save_config(cfg)

    # Prevent API fallback from returning a project via stale dependency overrides
    async def _no_api_fallback():
        return None

    monkeypatch.setattr(
        "basic_memory.mcp.project_context._resolve_default_project_from_api",
        _no_api_fallback,
    )
    assert await resolve_project_parameter(project=None, allow_discovery=True) is None


@pytest.mark.asyncio
async def test_returns_project_when_specified(config_manager):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    config_manager.save_config(cfg)

    assert await resolve_project_parameter(project="my-project") == "my-project"


@pytest.mark.asyncio
async def test_uses_env_var_priority(config_manager, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    config_manager.save_config(cfg)

    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
    assert await resolve_project_parameter(project="explicit-project") == "env-project"


@pytest.mark.asyncio
async def test_uses_explicit_project_when_no_env(config_manager, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    config_manager.save_config(cfg)

    monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
    assert await resolve_project_parameter(project="explicit-project") == "explicit-project"


@pytest.mark.asyncio
async def test_canonicalizes_case_insensitive_project_reference(
    config_manager, config_home, monkeypatch
):
    from basic_memory.config import ProjectEntry
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    project_name = "Personal-Project"
    project_path = config_home / "personal-project"
    project_path.mkdir(parents=True, exist_ok=True)
    cfg.projects[project_name] = ProjectEntry(path=str(project_path))
    config_manager.save_config(cfg)

    monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)

    assert await resolve_project_parameter(project="personal-project") == project_name
    assert await resolve_project_parameter(project="PERSONAL-PROJECT") == project_name


@pytest.mark.asyncio
async def test_uses_default_project(config_manager, config_home, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter
    from basic_memory.config import ProjectEntry

    cfg = config_manager.load_config()
    (config_home / "default-project").mkdir(parents=True, exist_ok=True)
    cfg.projects["default-project"] = ProjectEntry(path=str(config_home / "default-project"))
    cfg.default_project = "default-project"
    config_manager.save_config(cfg)

    monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
    assert await resolve_project_parameter(project=None) == "default-project"


@pytest.mark.asyncio
async def test_returns_none_when_no_default(config_manager, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter

    cfg = config_manager.load_config()
    cfg.default_project = None
    config_manager.save_config(cfg)

    monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)

    # Prevent API fallback from returning a project via stale dependency overrides
    async def _no_api_fallback():
        return None

    monkeypatch.setattr(
        "basic_memory.mcp.project_context._resolve_default_project_from_api",
        _no_api_fallback,
    )
    assert await resolve_project_parameter(project=None) is None


@pytest.mark.asyncio
async def test_env_constraint_overrides_default(config_manager, config_home, monkeypatch):
    from basic_memory.mcp.project_context import resolve_project_parameter
    from basic_memory.config import ProjectEntry

    cfg = config_manager.load_config()
    (config_home / "default-project").mkdir(parents=True, exist_ok=True)
    cfg.projects["default-project"] = ProjectEntry(path=str(config_home / "default-project"))
    cfg.default_project = "default-project"
    config_manager.save_config(cfg)

    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
    assert await resolve_project_parameter(project=None) == "env-project"


@pytest.mark.asyncio
async def test_workspace_auto_selects_single_and_caches(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
    from basic_memory.schemas.cloud import WorkspaceInfo

    context = _ContextState()
    only_workspace = WorkspaceInfo(
        tenant_id="11111111-1111-1111-1111-111111111111",
        workspace_type="personal",
        name="Personal",
        role="owner",
    )

    async def fake_get_available_workspaces(context=None):
        return [only_workspace]

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.get_available_workspaces",
        fake_get_available_workspaces,
    )

    resolved = await resolve_workspace_parameter(context=context)
    assert resolved.tenant_id == only_workspace.tenant_id
    assert await context.get_state("active_workspace") == only_workspace.model_dump()


@pytest.mark.asyncio
async def test_workspace_requires_user_choice_when_multiple(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
    from basic_memory.schemas.cloud import WorkspaceInfo

    workspaces = [
        WorkspaceInfo(
            tenant_id="11111111-1111-1111-1111-111111111111",
            workspace_type="personal",
            name="Personal",
            role="owner",
        ),
        WorkspaceInfo(
            tenant_id="22222222-2222-2222-2222-222222222222",
            workspace_type="organization",
            name="Team",
            role="editor",
        ),
    ]

    async def fake_get_available_workspaces(context=None):
        return workspaces

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.get_available_workspaces",
        fake_get_available_workspaces,
    )

    with pytest.raises(ValueError, match="Multiple workspaces are available"):
        await resolve_workspace_parameter(context=_ContextState())


@pytest.mark.asyncio
async def test_workspace_explicit_selection_by_tenant_id_or_name(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
    from basic_memory.schemas.cloud import WorkspaceInfo

    team_workspace = WorkspaceInfo(
        tenant_id="22222222-2222-2222-2222-222222222222",
        workspace_type="organization",
        name="Team",
        role="editor",
    )
    workspaces = [
        WorkspaceInfo(
            tenant_id="11111111-1111-1111-1111-111111111111",
            workspace_type="personal",
            name="Personal",
            role="owner",
        ),
        team_workspace,
    ]

    async def fake_get_available_workspaces(context=None):
        return workspaces

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.get_available_workspaces",
        fake_get_available_workspaces,
    )

    resolved_by_id = await resolve_workspace_parameter(workspace=team_workspace.tenant_id)
    assert resolved_by_id.tenant_id == team_workspace.tenant_id

    resolved_by_name = await resolve_workspace_parameter(workspace="team")
    assert resolved_by_name.tenant_id == team_workspace.tenant_id


@pytest.mark.asyncio
async def test_workspace_invalid_selection_lists_choices(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
    from basic_memory.schemas.cloud import WorkspaceInfo

    workspaces = [
        WorkspaceInfo(
            tenant_id="11111111-1111-1111-1111-111111111111",
            workspace_type="personal",
            name="Personal",
            role="owner",
        )
    ]

    async def fake_get_available_workspaces(context=None):
        return workspaces

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.get_available_workspaces",
        fake_get_available_workspaces,
    )

    with pytest.raises(ValueError, match="Workspace 'missing-workspace' was not found"):
        await resolve_workspace_parameter(workspace="missing-workspace")


@pytest.mark.asyncio
async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
    from basic_memory.mcp.project_context import resolve_workspace_parameter
    from basic_memory.schemas.cloud import WorkspaceInfo

    cached_workspace = WorkspaceInfo(
        tenant_id="11111111-1111-1111-1111-111111111111",
        workspace_type="personal",
        name="Personal",
        role="owner",
    )
    context = _ContextState()
    await context.set_state("active_workspace", cached_workspace.model_dump())

    async def fail_if_called(context=None):  # pragma: no cover
        raise AssertionError("Workspace fetch should not run when cache is available")

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.get_available_workspaces",
        fail_if_called,
    )

    resolved = await resolve_workspace_parameter(context=context)
    assert resolved.tenant_id == cached_workspace.tenant_id


@pytest.mark.asyncio
async def test_resolve_project_parameter_uses_cached_active_project_before_api_default_lookup(
    config_manager, monkeypatch
):
    from basic_memory.mcp.project_context import resolve_project_parameter
    from basic_memory.schemas.project_info import ProjectItem

    config = config_manager.load_config()
    config.default_project = None
    config_manager.save_config(config)

    context = _ContextState()
    cached_project = ProjectItem(
        id=1,
        external_id="11111111-1111-1111-1111-111111111111",
        name="Cached Project",
        path="/tmp/cached-project",
        is_default=True,
    )
    await context.set_state("active_project", cached_project.model_dump())

    async def fail_if_called():  # pragma: no cover
        raise AssertionError("Default project API lookup should not run when project is cached")

    monkeypatch.setattr(
        "basic_memory.mcp.project_context._resolve_default_project_from_api",
        fail_if_called,
    )

    resolved = await resolve_project_parameter(project=None, context=context)
    assert resolved == cached_project.name


@pytest.mark.asyncio
async def test_resolve_project_parameter_caches_api_default_project_name(
    config_manager, monkeypatch
):
    from basic_memory.mcp.project_context import resolve_project_parameter

    config = config_manager.load_config()
    config.default_project = None
    config_manager.save_config(config)

    context = _ContextState()
    api_calls = {"count": 0}

    async def fake_default_lookup():
        api_calls["count"] += 1
        return "cloud-default"

    monkeypatch.setattr(
        "basic_memory.mcp.project_context._resolve_default_project_from_api",
        fake_default_lookup,
    )

    first = await resolve_project_parameter(project=None, context=context)
    second = await resolve_project_parameter(project=None, context=context)

    assert first == "cloud-default"
    assert second == "cloud-default"
    assert api_calls["count"] == 1


@pytest.mark.asyncio
async def test_get_active_project_uses_cached_project_before_resolution(monkeypatch):
    from basic_memory.mcp.project_context import get_active_project
    from basic_memory.schemas.project_info import ProjectItem

    context = _ContextState()
    cached_project = ProjectItem(
        id=1,
        external_id="11111111-1111-1111-1111-111111111111",
        name="Cached Project",
        path="/tmp/cached-project",
        is_default=True,
    )
    await context.set_state("active_project", cached_project.model_dump())

    async def fail_if_called(*args, **kwargs):  # pragma: no cover
        raise AssertionError("Project resolution should not run when cache matches")

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.resolve_project_parameter",
        fail_if_called,
    )

    resolved = await get_active_project(client=None, context=context)
    assert resolved == cached_project


@pytest.mark.asyncio
async def test_get_active_project_uses_cached_project_for_explicit_permalink(monkeypatch):
    from basic_memory.mcp.project_context import get_active_project
    from basic_memory.schemas.project_info import ProjectItem

    context = _ContextState()
    cached_project = ProjectItem(
        id=1,
        external_id="11111111-1111-1111-1111-111111111111",
        name="My Research",
        path="/tmp/my-research",
        is_default=False,
    )
    await context.set_state("active_project", cached_project.model_dump())

    async def fail_if_called(*args, **kwargs):  # pragma: no cover
        raise AssertionError(
            "Project resolution should not run when explicit project matches cache"
        )

    monkeypatch.setattr(
        "basic_memory.mcp.project_context.resolve_project_parameter",
        fail_if_called,
    )

    resolved = await get_active_project(client=None, project="my-research", context=context)
    assert resolved == cached_project


@pytest.mark.asyncio
async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefix(
    config_manager, monkeypatch
):
    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 = False
    config_manager.save_config(config)

    context = _ContextState()
    cached_project = ProjectItem(
        id=1,
        external_id="11111111-1111-1111-1111-111111111111",
        name="My Research",
        path="/tmp/my-research",
        is_default=False,
    )
    await context.set_state("active_project", cached_project.model_dump())

    async def fail_if_called(*args, **kwargs):  # pragma: no cover
        raise AssertionError("Project resolve API should not run when memory URL matches cache")

    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_if_called)
    monkeypatch.setattr(
        "basic_memory.mcp.project_context.resolve_project_parameter",
        fake_resolve_project_parameter,
    )

    active_project, resolved_path, is_memory_url = await resolve_project_and_path(
        client=None,
        identifier="memory://my-research/notes/roadmap.md",
        context=context,
    )

    assert active_project == cached_project
    assert resolved_path == "notes/roadmap.md"
    assert is_memory_url is True


@pytest.mark.asyncio
async def test_get_project_client_rejects_workspace_for_local_project(config_manager):
    from basic_memory.mcp.project_context import get_project_client
    from basic_memory.config import ProjectEntry

    # Register "main" as a LOCAL project so get_project_mode returns LOCAL
    config = config_manager.load_config()
    (config_manager.config_dir.parent / "main").mkdir(parents=True, exist_ok=True)
    config.projects["main"] = ProjectEntry(path=str(config_manager.config_dir.parent / "main"))
    config_manager.save_config(config)

    with pytest.raises(
        ValueError, match="Workspace 'tenant-123' cannot be used with local project"
    ):
        async with get_project_client(project="main", workspace="tenant-123"):
            pass


class TestDetectProjectFromUrlPrefix:
    """Test detect_project_from_url_prefix for URL-based project detection."""

    def test_detects_project_from_memory_url(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix

        config = config_manager.load_config()
        # The config has "test-project" from the conftest fixture
        result = detect_project_from_url_prefix("memory://test-project/some-note", config)
        assert result == "test-project"

    def test_detects_project_from_plain_path(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix

        config = config_manager.load_config()
        result = detect_project_from_url_prefix("test-project/some-note", config)
        assert result == "test-project"

    def test_returns_none_for_unknown_prefix(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix

        config = config_manager.load_config()
        result = detect_project_from_url_prefix("memory://unknown-project/note", config)
        assert result is None

    def test_returns_none_for_no_slash(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix

        config = config_manager.load_config()
        result = detect_project_from_url_prefix("memory://single-segment", config)
        assert result is None

    def test_returns_none_for_wildcard_prefix(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix

        config = config_manager.load_config()
        result = detect_project_from_url_prefix("memory://*/notes", config)
        assert result is None

    def test_matches_case_insensitive_via_permalink(self, config_manager):
        from basic_memory.mcp.project_context import detect_project_from_url_prefix
        from basic_memory.config import ProjectEntry

        config = config_manager.load_config()
        (config_manager.config_dir.parent / "My Research").mkdir(parents=True, exist_ok=True)
        config.projects["My Research"] = ProjectEntry(
            path=str(config_manager.config_dir.parent / "My Research")
        )
        config_manager.save_config(config)

        result = detect_project_from_url_prefix("memory://my-research/notes", config)
        assert result == "My Research"


class TestGetProjectClientRoutingOrder:
    """Test that get_project_client respects explicit routing before workspace resolution."""

    @pytest.mark.asyncio
    async def test_local_flag_skips_workspace_resolution(self, config_manager, monkeypatch):
        """--local flag should never trigger workspace resolution, even for cloud projects."""
        from basic_memory.mcp.project_context import get_project_client
        from basic_memory.config import ProjectEntry, ProjectMode

        config = config_manager.load_config()
        config.projects["cloud-proj"] = ProjectEntry(
            path=str(config_manager.config_dir.parent / "cloud-proj"),
            mode=ProjectMode.CLOUD,
        )
        config_manager.save_config(config)

        # Set explicit local routing
        monkeypatch.setenv("BASIC_MEMORY_EXPLICIT_ROUTING", "true")
        monkeypatch.setenv("BASIC_MEMORY_FORCE_LOCAL", "true")
        monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)

        # Should not raise "Multiple workspaces" — it should skip workspace entirely
        # It will fail at project validation (no API running), which proves routing worked
        with pytest.raises(Exception) as exc_info:
            async with get_project_client(project="cloud-proj"):
                pass

        # The error should NOT be about workspaces
        assert "workspace" not in str(exc_info.value).lower()

    @pytest.mark.asyncio
    async def test_cloud_project_uses_per_project_workspace_id(self, config_manager, monkeypatch):
        """Cloud project with workspace_id in config should use it without network lookup."""
        from basic_memory.mcp.project_context import get_project_client
        from basic_memory.config import ProjectEntry, ProjectMode

        config = config_manager.load_config()
        config.projects["cloud-proj"] = ProjectEntry(
            path=str(config_manager.config_dir.parent / "cloud-proj"),
            mode=ProjectMode.CLOUD,
            workspace_id="per-project-tenant-id",
        )
        config.cloud_api_key = "bmc_test123"
        config_manager.save_config(config)

        # Patch resolve_workspace_parameter to fail if called — it should be skipped
        async def fail_if_called(**kwargs):  # pragma: no cover
            raise AssertionError(
                "resolve_workspace_parameter should not be called when workspace_id is set"
            )

        monkeypatch.setattr(
            "basic_memory.mcp.project_context.resolve_workspace_parameter",
            fail_if_called,
        )

        # Will fail at cloud client creation (no real cloud), but proves workspace
        # resolution was skipped
        with pytest.raises(Exception) as exc_info:
            async with get_project_client(project="cloud-proj"):
                pass

        # Should not be a workspace resolution error
        error_msg = str(exc_info.value).lower()
        assert "resolve_workspace_parameter should not be called" not in error_msg

    @pytest.mark.asyncio
    async def test_cloud_project_uses_default_workspace(self, config_manager, monkeypatch):
        """Cloud project without workspace_id should fall back to default_workspace."""
        from basic_memory.mcp.project_context import get_project_client
        from basic_memory.config import ProjectEntry, ProjectMode

        config = config_manager.load_config()
        config.projects["cloud-proj"] = ProjectEntry(
            path=str(config_manager.config_dir.parent / "cloud-proj"),
            mode=ProjectMode.CLOUD,
        )
        config.default_workspace = "global-default-tenant-id"
        config.cloud_api_key = "bmc_test123"
        config_manager.save_config(config)

        # Patch resolve_workspace_parameter to fail if called — it should be skipped
        async def fail_if_called(**kwargs):  # pragma: no cover
            raise AssertionError(
                "resolve_workspace_parameter should not be called when default_workspace is set"
            )

        monkeypatch.setattr(
            "basic_memory.mcp.project_context.resolve_workspace_parameter",
            fail_if_called,
        )

        # Will fail at cloud client creation, but proves workspace resolution was skipped
        with pytest.raises(Exception) as exc_info:
            async with get_project_client(project="cloud-proj"):
                pass

        error_msg = str(exc_info.value).lower()
        assert "resolve_workspace_parameter should not be called" not in error_msg

    @pytest.mark.asyncio
    async def test_cloud_only_project_routes_to_cloud(self, config_manager, monkeypatch):
        """Project NOT in local config should route to cloud (not default to LOCAL).

        Cloud-only projects aren't registered in local config. The routing logic
        should detect this and use CLOUD mode, falling back to default_workspace.
        """
        from basic_memory.mcp.project_context import get_project_client

        config = config_manager.load_config()
        # Do NOT add "cloud-only-proj" to config.projects — it's cloud-only
        config.default_workspace = "global-default-tenant-id"
        config.cloud_api_key = "bmc_test123"
        config_manager.save_config(config)

        # Patch resolve_workspace_parameter to fail if called — it should be skipped
        # because default_workspace is set (priority 3)
        async def fail_if_called(**kwargs):  # pragma: no cover
            raise AssertionError(
                "resolve_workspace_parameter should not be called when default_workspace is set"
            )

        monkeypatch.setattr(
            "basic_memory.mcp.project_context.resolve_workspace_parameter",
            fail_if_called,
        )

        # Will fail at cloud client creation (no real cloud), but proves cloud routing
        # was selected instead of local routing
        with pytest.raises(Exception) as exc_info:
            async with get_project_client(project="cloud-only-proj"):
                pass

        # The error should NOT be about workspace resolution or local routing
        error_msg = str(exc_info.value).lower()
        assert "resolve_workspace_parameter should not be called" not in error_msg
        # Should not get a local ASGI routing error
        assert "no project found" not in error_msg

    @pytest.mark.asyncio
    async def test_factory_mode_skips_workspace_resolution(self, config_manager, monkeypatch):
        """When a client factory is set (in-process cloud server), skip workspace resolution.

        The cloud MCP server calls set_client_factory() so that get_client() routes
        requests through TenantASGITransport. In this mode, workspace and tenant context
        are already resolved by the transport layer. Attempting cloud workspace resolution
        would call the production control-plane API and fail with 401.
        """
        from contextlib import asynccontextmanager

        from basic_memory.mcp import async_client
        from basic_memory.mcp.project_context import get_project_client
        from basic_memory.config import ProjectEntry, ProjectMode

        config = config_manager.load_config()
        config.projects["cloud-proj"] = ProjectEntry(
            path=str(config_manager.config_dir.parent / "cloud-proj"),
            mode=ProjectMode.CLOUD,
        )
        config_manager.save_config(config)

        # Set up a factory (simulates what cloud MCP server does)
        @asynccontextmanager
        async def fake_factory():
            from httpx import ASGITransport, AsyncClient
            from basic_memory.api.app import app as fastapi_app

            async with AsyncClient(
                transport=ASGITransport(app=fastapi_app),
                base_url="http://test",
            ) as client:
                yield client

        original_factory = async_client._client_factory
        async_client.set_client_factory(fake_factory)

        # Patch workspace resolution to fail if called — factory mode should skip it
        async def fail_if_called(**kwargs):  # pragma: no cover
            raise AssertionError("resolve_workspace_parameter must not be called in factory mode")

        monkeypatch.setattr(
            "basic_memory.mcp.project_context.resolve_workspace_parameter",
            fail_if_called,
        )

        # Patch get_cloud_control_plane_client to fail if called
        @asynccontextmanager
        async def fail_control_plane():  # pragma: no cover
            raise AssertionError(
                "get_cloud_control_plane_client must not be called in factory mode"
            )

        monkeypatch.setattr(
            "basic_memory.mcp.async_client.get_cloud_control_plane_client",
            fail_control_plane,
        )

        try:
            # Will fail at project validation (no real project in DB), but proves
            # workspace resolution and control-plane calls were skipped
            with pytest.raises(Exception) as exc_info:
                async with get_project_client(project="cloud-proj"):
                    pass

            error_msg = str(exc_info.value).lower()
            assert "resolve_workspace_parameter must not be called" not in error_msg
            assert "get_cloud_control_plane_client must not be called" not in error_msg
        finally:
            # Restore original factory to avoid polluting other tests
            async_client._client_factory = original_factory