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

"""Tests for MCP project management tools."""

from contextlib import asynccontextmanager
from unittest.mock import AsyncMock, patch

import httpx
import pytest
from sqlalchemy import select

from basic_memory import db
from basic_memory.mcp.tools import list_memory_projects, create_memory_project, delete_project
from basic_memory.config import BasicMemoryConfig, ProjectEntry
from basic_memory.mcp.tools.project_management import _merge_projects, _merge_workspace_projects
from basic_memory.models.project import Project
from basic_memory.schemas.project_info import ProjectItem, ProjectList


# --- Helpers ---


def _make_project(
    name: str,
    path: str,
    *,
    id: int = 1,
    external_id: str = "00000000-0000-0000-0000-000000000001",
    is_default: bool = False,
    display_name: str | None = None,
    is_private: bool = False,
) -> ProjectItem:
    return ProjectItem(
        id=id,
        external_id=external_id,
        name=name,
        path=path,
        is_default=is_default,
        display_name=display_name,
        is_private=is_private,
    )


def _make_list(projects: list[ProjectItem], default: str | None = None) -> ProjectList:
    return ProjectList(projects=projects, default_project=default)


# --- Existing tests (updated for source labels) ---


@pytest.mark.asyncio
async def test_list_memory_projects_unconstrained(app, test_project):
    result = await list_memory_projects()
    assert "Available projects:" in result
    assert f"- {test_project.name}" in result


@pytest.mark.asyncio
async def test_list_memory_projects_shows_display_name(app, client, test_project):
    """When a project has display_name set, list_memory_projects shows 'display_name (name)' format."""
    mock_project = _make_project(
        "private-fb83af23",
        "/tmp/private",
        id=1,
        display_name="My Notes",
        is_private=True,
    )
    regular_project = _make_project(
        "main",
        "/tmp/main",
        id=2,
        external_id="00000000-0000-0000-0000-000000000002",
        is_default=True,
    )
    mock_list = _make_list([regular_project, mock_project], default="main")

    with patch(
        "basic_memory.mcp.clients.project.ProjectClient.list_projects",
        new_callable=AsyncMock,
        return_value=mock_list,
    ):
        result = await list_memory_projects()

    # Regular project shows name with source label
    assert "- main (local)" in result
    # Private project shows display_name with slug in parentheses, then source
    assert "- My Notes (private-fb83af23) (local)" in result


@pytest.mark.asyncio
async def test_list_memory_projects_no_display_name_shows_name_only(app, client, test_project):
    """When a project has no display_name, list_memory_projects shows just the name."""
    project = _make_project("my-project", "/tmp/my-project", is_default=True)
    mock_list = _make_list([project], default="my-project")

    with patch(
        "basic_memory.mcp.clients.project.ProjectClient.list_projects",
        new_callable=AsyncMock,
        return_value=mock_list,
    ):
        result = await list_memory_projects()

    assert "- my-project (local)" in result


@pytest.mark.asyncio
async def test_list_memory_projects_constrained_env(monkeypatch, app, test_project):
    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", test_project.name)
    result = await list_memory_projects()
    assert f"Project: {test_project.name}" in result
    assert "constrained to a single project" in result


@pytest.mark.asyncio
async def test_create_and_delete_project_and_name_match_branch(
    app, tmp_path_factory, session_maker
):
    # Create a project through the tool (exercises POST + response formatting).
    project_root = tmp_path_factory.mktemp("extra-project-home")
    result = await create_memory_project(
        project_name="My Project",
        project_path=str(project_root),
        set_default=False,
    )
    assert isinstance(result, str)
    assert result.startswith("✓")
    assert "My Project" in result

    # Make permalink intentionally not derived from name so delete_project hits the name-match branch.
    async with db.scoped_session(session_maker) as session:
        project = (
            await session.execute(select(Project).where(Project.name == "My Project"))
        ).scalar_one()
        project.permalink = "custom-permalink"
        await session.commit()

    delete_result = await delete_project("My Project")
    assert delete_result.startswith("✓")


@pytest.mark.asyncio
async def test_create_memory_project_resolves_workspace_slug(app, tmp_path_factory):
    """A friendly workspace slug resolves to the tenant id used for cloud routing."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    project_root = tmp_path_factory.mktemp("ws-project-home")
    captured: dict[str, str | None] = {}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project created",
        status="success",
        default=False,
        new_project=_make_project("WS Project", str(project_root)),
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
            return_value=_make_workspace(
                "tenant-abc-123",
                "Team Paul",
                workspace_type="organization",
                slug="team-paul",
            ),
        ) as mock_resolve_workspace,
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([], default=None),
        ),
        patch.object(
            ProjectClient,
            "create_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ),
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        await create_memory_project(
            project_name="WS Project",
            project_path=str(project_root),
            workspace="team-paul",
        )

    mock_resolve_workspace.assert_awaited_once_with(workspace="team-paul", context=None)
    assert captured["workspace"] == "tenant-abc-123"


@pytest.mark.asyncio
async def test_create_memory_project_workspace_is_local_noop(app, tmp_path_factory):
    """Local create accepts workspace without requiring cloud workspace discovery."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    project_root = tmp_path_factory.mktemp("local-ws-project-home")
    captured: dict[str, str | None] = {}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project created",
        status="success",
        default=False,
        new_project=_make_project("Local WS Project", str(project_root)),
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=False,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=False,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
        ) as mock_resolve_workspace,
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([], default=None),
        ),
        patch.object(
            ProjectClient,
            "create_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ),
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        await create_memory_project(
            project_name="Local WS Project",
            project_path=str(project_root),
            workspace="team-paul",
        )

    mock_resolve_workspace.assert_not_awaited()
    assert captured["workspace"] == "team-paul"


@pytest.mark.asyncio
async def test_create_memory_project_default_workspace_is_none(app, tmp_path_factory):
    """When workspace is omitted, get_client receives workspace=None (default workspace)."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    project_root = tmp_path_factory.mktemp("default-ws-project-home")
    captured: dict[str, str | None] = {"workspace": "sentinel"}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project created",
        status="success",
        default=False,
        new_project=_make_project("Default WS Project", str(project_root)),
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([], default=None),
        ),
        patch.object(
            ProjectClient,
            "create_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ),
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        await create_memory_project(
            project_name="Default WS Project",
            project_path=str(project_root),
        )

    assert captured["workspace"] is None


@pytest.mark.asyncio
async def test_create_memory_project_constrained_with_workspace_returns_disabled_message(
    monkeypatch, tmp_path_factory
):
    """A constrained MCP session rejects creation before resolving a workspace selector."""
    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "locked-project")
    project_root = tmp_path_factory.mktemp("constrained-create-project-home")

    with (
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
            side_effect=RuntimeError("bad workspace"),
        ) as mock_resolve_workspace,
    ):
        result = await create_memory_project(
            project_name="Any Project",
            project_path=str(project_root),
            workspace="missing-team",
        )

    mock_resolve_workspace.assert_not_awaited()
    assert "Project creation disabled" in result
    assert "locked-project" in result


@pytest.mark.asyncio
async def test_delete_project_resolves_workspace_slug(app):
    """A friendly workspace slug resolves to the tenant id used for delete routing."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    target_project = _make_project(
        "WS Project",
        "/ws-project",
        external_id="project-uuid",
    )
    captured: dict[str, str | None] = {}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project deleted",
        status="success",
        default=False,
        old_project=target_project,
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
            return_value=_make_workspace(
                "tenant-abc-123",
                "Team Paul",
                workspace_type="organization",
                slug="team-paul",
            ),
        ) as mock_resolve_workspace,
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([target_project], default=None),
        ),
        patch.object(
            ProjectClient,
            "delete_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ) as mock_delete_project,
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        result = await delete_project("WS Project", workspace="team-paul")

    mock_resolve_workspace.assert_awaited_once_with(workspace="team-paul", context=None)
    assert captured["workspace"] == "tenant-abc-123"
    mock_delete_project.assert_awaited_once_with("project-uuid")
    assert result.startswith("✓")


@pytest.mark.asyncio
async def test_delete_project_workspace_is_local_noop(app):
    """Local delete accepts workspace without requiring cloud workspace discovery."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    target_project = _make_project(
        "Local WS Project",
        "/local-ws-project",
        external_id="local-project-uuid",
    )
    captured: dict[str, str | None] = {}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project deleted",
        status="success",
        default=False,
        old_project=target_project,
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=False,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=False,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
        ) as mock_resolve_workspace,
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([target_project], default=None),
        ),
        patch.object(
            ProjectClient,
            "delete_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ),
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        await delete_project("Local WS Project", workspace="team-paul")

    mock_resolve_workspace.assert_not_awaited()
    assert captured["workspace"] == "team-paul"


@pytest.mark.asyncio
async def test_delete_project_default_workspace_is_none(app):
    """When workspace is omitted, delete_project routes through the default workspace."""
    from basic_memory.mcp.clients import ProjectClient
    from basic_memory.schemas.project_info import ProjectStatusResponse

    target_project = _make_project(
        "Default WS Project",
        "/default-ws-project",
        external_id="default-project-uuid",
    )
    captured: dict[str, str | None] = {"workspace": "sentinel"}

    @asynccontextmanager
    async def fake_get_client(*, workspace=None, project_name=None):
        captured["workspace"] = workspace
        async with httpx.AsyncClient(base_url="http://testserver") as client:
            yield client

    fake_status = ProjectStatusResponse(
        message="Project deleted",
        status="success",
        default=False,
        old_project=target_project,
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.get_client",
            new=fake_get_client,
        ),
        patch.object(
            ProjectClient,
            "list_projects",
            new_callable=AsyncMock,
            return_value=_make_list([target_project], default=None),
        ),
        patch.object(
            ProjectClient,
            "delete_project",
            new_callable=AsyncMock,
            return_value=fake_status,
        ),
        patch(
            "basic_memory.mcp.project_context.invalidate_workspace_project_index",
            new_callable=AsyncMock,
        ),
    ):
        await delete_project("Default WS Project")

    assert captured["workspace"] is None


@pytest.mark.asyncio
async def test_delete_project_constrained_with_workspace_returns_disabled_message(monkeypatch):
    """A constrained MCP session rejects deletion before resolving a workspace selector."""
    monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "locked-project")

    with (
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
            new_callable=AsyncMock,
            side_effect=RuntimeError("bad workspace"),
        ) as mock_resolve_workspace,
    ):
        result = await delete_project("Any Project", workspace="missing-team")

    mock_resolve_workspace.assert_not_awaited()
    assert "Project deletion disabled" in result
    assert "locked-project" in result


# --- Cloud merge tests ---


@pytest.mark.asyncio
async def test_list_memory_projects_local_and_cloud_merge(app, test_project):
    """When cloud credentials exist, projects from both sources are merged by permalink."""
    local_main = _make_project("main", "/home/user/basic-memory", is_default=True)
    local_specs = _make_project(
        "specs", "/home/user/specs", id=2, external_id="00000000-0000-0000-0000-000000000002"
    )
    local_list = _make_list([local_main, local_specs], default="main")

    cloud_main = _make_project("main", "/main", id=10, external_id="cloud-main-uuid")
    cloud_llc = _make_project(
        "basic-memory-llc", "/basic-memory-llc", id=11, external_id="cloud-llc-uuid"
    )
    workspace = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    workspace_index = _make_workspace_index([(workspace, [cloud_main, cloud_llc])])

    with (
        patch(
            "basic_memory.mcp.clients.project.ProjectClient.list_projects",
            new_callable=AsyncMock,
            return_value=local_list,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            return_value=workspace_index,
        ),
    ):
        result = await list_memory_projects()

    # Both local+cloud project shows merged source
    assert "- main (local+cloud)" in result
    # Local-only project
    assert "- specs (local)" in result
    # Cloud-only project
    assert "- basic-memory-llc (cloud)" in result


@pytest.mark.asyncio
async def test_list_memory_projects_no_cloud_credentials(app, test_project):
    """When no cloud credentials exist, only local projects are shown."""
    with patch(
        "basic_memory.mcp.tools.project_management.has_cloud_credentials",
        return_value=False,
    ):
        result = await list_memory_projects()

    assert "Available projects:" in result
    assert f"- {test_project.name} (local)" in result
    # No cloud source labels
    assert "cloud)" not in result


@pytest.mark.asyncio
async def test_list_memory_projects_cloud_failure_graceful(app, test_project):
    """When cloud fetch fails, local projects are still returned."""
    with (
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            side_effect=RuntimeError("cloud unavailable"),
        ),
    ):
        result = await list_memory_projects()

    assert "Available projects:" in result
    assert f"- {test_project.name} (local)" in result


@pytest.mark.asyncio
async def test_list_memory_projects_factory_mode(app, test_project):
    """Factory mode lists projects from every accessible workspace."""
    personal_project = _make_project(
        "personal-main",
        "/personal-main",
        is_default=True,
        external_id="personal-project-uuid",
    )
    team_project = _make_project(
        "team-specs",
        "/team-specs",
        id=2,
        external_id="team-project-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    team_ws = _make_workspace(
        "team-tenant",
        "Team Paul",
        "organization",
        slug="team-paul",
    )
    workspace_index = _make_workspace_index(
        [
            (personal_ws, [personal_project]),
            (team_ws, [team_project]),
        ]
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            return_value=workspace_index,
        ) as mock_index,
    ):
        result = await list_memory_projects()

    mock_index.assert_awaited_once()
    assert "Workspace: Personal (personal default)" in result
    assert "Workspace: Team Paul (team-paul)" in result
    assert "- personal-main (cloud) [personal-project-uuid]" in result
    assert (
        "- team-specs (cloud) [team-project-uuid] - cloud-only (local sync unsupported)" in result
    )


@pytest.mark.asyncio
async def test_list_memory_projects_factory_mode_json_includes_workspace(app, test_project):
    """In factory mode, JSON output includes workspace metadata for all cloud projects."""
    default_project = _make_project(
        "personal-main",
        "/personal-main",
        is_default=True,
        external_id="personal-project-uuid",
    )
    org_project = _make_project(
        "cloud-proj",
        "/cloud-proj",
        id=2,
        external_id="org-project-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    org_ws = _make_workspace("tenant-abc", "My Org", "organization")
    workspace_index = _make_workspace_index(
        [
            (personal_ws, [default_project]),
            (org_ws, [org_project]),
        ]
    )

    with (
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            return_value=workspace_index,
        ),
    ):
        result = await list_memory_projects(output_format="json")

    assert isinstance(result, dict)
    assert result["default_project"] == "personal-main"
    projects = result["projects"]
    assert len(projects) == 2
    proj = {project["name"]: project for project in projects}["cloud-proj"]
    assert proj["source"] == "cloud"
    assert proj["cloud_path"] == "/cloud-proj"
    assert proj["local_path"] is None
    assert proj["workspace_name"] == "My Org"
    assert proj["workspace_type"] == "organization"
    assert proj["workspace_tenant_id"] == "tenant-abc"
    assert proj["workspace_slug"] == "my-org"
    assert proj["workspace_is_default"] is False
    assert proj["qualified_name"] == "my-org/cloud-proj"
    assert proj["sync_supported"] is False
    assert proj["sync_reason"] == "organization workspace"
    assert proj["local_usage"] == "cloud-only"


@pytest.mark.asyncio
async def test_list_memory_projects_factory_mode_workspace_lookup_failure(app, test_project):
    """In factory mode, workspace discovery failures are surfaced to the caller."""
    with (
        patch(
            "basic_memory.mcp.tools.project_management.is_factory_mode",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            side_effect=RuntimeError("no user context"),
        ),
    ):
        with pytest.raises(RuntimeError, match="no user context"):
            await list_memory_projects()


@pytest.mark.asyncio
async def test_list_memory_projects_json_with_cloud(app, test_project):
    """JSON output includes local_path, cloud_path, and source fields."""
    local_main = _make_project("main", "/home/user/basic-memory", is_default=True)
    local_list = _make_list([local_main], default="main")

    cloud_main = _make_project("main", "/main", id=10, external_id="cloud-main-uuid")
    cloud_only = _make_project("cloud-only", "/cloud-only", id=11, external_id="cloud-only-uuid")
    workspace = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    workspace_index = _make_workspace_index([(workspace, [cloud_main, cloud_only])])

    with (
        patch(
            "basic_memory.mcp.clients.project.ProjectClient.list_projects",
            new_callable=AsyncMock,
            return_value=local_list,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            return_value=workspace_index,
        ),
    ):
        result = await list_memory_projects(output_format="json")

    assert isinstance(result, dict)
    projects = result["projects"]
    assert result["default_project"] == "main"

    # Find projects by name
    by_name = {p["name"]: p for p in projects}

    # main: local+cloud
    main_proj = by_name["main"]
    assert main_proj["source"] == "local+cloud"
    assert main_proj["local_path"] == "/home/user/basic-memory"
    assert main_proj["cloud_path"] == "/main"
    # Backward-compat: path prefers local
    assert main_proj["path"] == "/home/user/basic-memory"
    assert main_proj["is_default"] is True
    assert main_proj["sync_supported"] is True
    assert main_proj["sync_reason"] is None
    assert main_proj["local_usage"] == "sync-supported"

    # cloud-only
    cloud_proj = by_name["cloud-only"]
    assert cloud_proj["source"] == "cloud"
    assert cloud_proj["local_path"] is None
    assert cloud_proj["cloud_path"] == "/cloud-only"
    assert cloud_proj["path"] == "/cloud-only"
    assert cloud_proj["workspace_slug"] == "personal"
    assert cloud_proj["qualified_name"] == "personal/cloud-only"
    assert cloud_proj["sync_supported"] is True
    assert cloud_proj["sync_reason"] is None
    assert cloud_proj["local_usage"] == "sync-supported"


# --- Unit test for _merge_projects ---


def test_merge_projects_empty():
    """Merging two None lists produces an empty result."""
    assert _merge_projects(None, None) == []


def test_merge_projects_local_only():
    """Merging with only local projects sets source to 'local', workspace fields are None."""
    local_list = _make_list(
        [_make_project("alpha", "/alpha"), _make_project("beta", "/beta", id=2)],
        default="alpha",
    )
    merged = _merge_projects(local_list, None)
    assert len(merged) == 2
    assert all(p["source"] == "local" for p in merged)
    # Sorted by permalink
    assert merged[0]["name"] == "alpha"
    assert merged[1]["name"] == "beta"
    # Local-only projects have no workspace info
    assert all(p["workspace_name"] is None for p in merged)
    assert all(p["workspace_type"] is None for p in merged)
    assert all(p["workspace_tenant_id"] is None for p in merged)
    assert all(p["sync_supported"] is True for p in merged)
    assert all(p["sync_reason"] is None for p in merged)
    assert all(p["local_usage"] == "sync-supported" for p in merged)


def test_merge_projects_cloud_only():
    """Merging with only cloud projects sets source to 'cloud' with workspace info."""
    cloud_list = _make_list(
        [_make_project("gamma", "/gamma")],
        default="gamma",
    )
    merged = _merge_projects(
        None,
        cloud_list,
        cloud_workspace_name="Personal",
        cloud_workspace_type="personal",
        cloud_workspace_tenant_id="tenant-123",
    )
    assert len(merged) == 1
    assert merged[0]["source"] == "cloud"
    assert merged[0]["local_path"] is None
    assert merged[0]["cloud_path"] == "/gamma"
    assert merged[0]["workspace_name"] == "Personal"
    assert merged[0]["workspace_type"] == "personal"
    assert merged[0]["workspace_tenant_id"] == "tenant-123"
    assert merged[0]["sync_supported"] is True
    assert merged[0]["sync_reason"] is None
    assert merged[0]["local_usage"] == "sync-supported"


def test_merge_projects_overlap():
    """Overlapping projects carry workspace info from cloud side."""
    local_list = _make_list([_make_project("shared", "/local/shared")])
    cloud_list = _make_list([_make_project("shared", "/cloud/shared")])
    merged = _merge_projects(
        local_list,
        cloud_list,
        cloud_workspace_name="Acme Corp",
        cloud_workspace_type="organization",
        cloud_workspace_tenant_id="org-456",
    )
    assert len(merged) == 1
    assert merged[0]["source"] == "local+cloud"
    assert merged[0]["local_path"] == "/local/shared"
    assert merged[0]["cloud_path"] == "/cloud/shared"
    # Backward compat: path prefers local
    assert merged[0]["path"] == "/local/shared"
    # Cloud workspace info is present because the project has a cloud source
    assert merged[0]["workspace_name"] == "Acme Corp"
    assert merged[0]["workspace_type"] == "organization"
    assert merged[0]["workspace_tenant_id"] == "org-456"
    assert merged[0]["sync_supported"] is False
    assert merged[0]["sync_reason"] == "organization workspace"
    assert merged[0]["local_usage"] == "cloud-only"


def test_merge_workspace_projects_attaches_local_state_to_one_duplicate_workspace(tmp_path):
    """A same-name team workspace project should stay cloud-only (#848)."""
    local_path = str(tmp_path / "main")
    local_main = _make_project("main", local_path, is_default=True)
    local_list = _make_list([local_main], default="main")
    personal_main = _make_project(
        "main",
        "/cloud/personal-main",
        id=10,
        external_id="personal-main-uuid",
    )
    team_main = _make_project(
        "main",
        "/cloud/team-main",
        id=11,
        external_id="team-main-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    team_ws = _make_workspace(
        "team-tenant",
        "Team",
        workspace_type="organization",
        slug="team",
    )
    workspace_index = _make_workspace_index(
        [
            (personal_ws, [personal_main]),
            (team_ws, [team_main]),
        ]
    )
    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})

    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)

    by_qualified_name = {project["qualified_name"]: project for project in merged}
    personal_project = by_qualified_name["personal/main"]
    team_project = by_qualified_name["team/main"]

    assert personal_project["source"] == "local+cloud"
    assert personal_project["local_path"] == local_path
    assert personal_project["path"] == local_path
    assert team_project["source"] == "cloud"
    assert team_project["local_path"] is None
    assert team_project["path"] == "/cloud/team-main"
    assert team_project["sync_supported"] is False
    assert team_project["sync_reason"] == "organization workspace"
    assert team_project["local_usage"] == "cloud-only"


def test_merge_workspace_projects_uses_configured_workspace_for_local_state(tmp_path):
    """Per-project workspace_id should select the attached duplicate row."""
    local_path = str(tmp_path / "main")
    local_main = _make_project("main", local_path, is_default=True)
    local_list = _make_list([local_main], default="main")
    personal_main = _make_project(
        "main",
        "/cloud/personal-main",
        id=10,
        external_id="personal-main-uuid",
    )
    team_main = _make_project(
        "main",
        "/cloud/team-main",
        id=11,
        external_id="team-main-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    team_ws = _make_workspace(
        "team-tenant",
        "Team",
        workspace_type="organization",
        slug="team",
    )
    workspace_index = _make_workspace_index(
        [
            (personal_ws, [personal_main]),
            (team_ws, [team_main]),
        ]
    )
    config = BasicMemoryConfig(
        projects={
            "main": ProjectEntry(
                path=local_path,
                workspace_id="team-tenant",
            )
        }
    )

    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)

    by_qualified_name = {project["qualified_name"]: project for project in merged}
    personal_project = by_qualified_name["personal/main"]
    team_project = by_qualified_name["team/main"]

    assert personal_project["source"] == "cloud"
    assert personal_project["local_path"] is None
    assert personal_project["path"] == "/cloud/personal-main"
    assert team_project["source"] == "local+cloud"
    assert team_project["local_path"] == local_path
    assert team_project["path"] == local_path


def test_merge_workspace_projects_uses_default_workspace_for_local_state(tmp_path):
    """Global default_workspace should attach local state before cloud default fallback."""
    local_path = str(tmp_path / "main")
    local_main = _make_project("main", local_path, is_default=True)
    local_list = _make_list([local_main], default="main")
    personal_main = _make_project(
        "main",
        "/cloud/personal-main",
        id=10,
        external_id="personal-main-uuid",
    )
    team_main = _make_project(
        "main",
        "/cloud/team-main",
        id=11,
        external_id="team-main-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=True,
    )
    team_ws = _make_workspace(
        "team-tenant",
        "Team",
        workspace_type="organization",
        slug="team",
    )
    workspace_index = _make_workspace_index(
        [
            (personal_ws, [personal_main]),
            (team_ws, [team_main]),
        ]
    )
    config = BasicMemoryConfig(
        projects={"main": ProjectEntry(path=local_path)},
        default_workspace="team-tenant",
    )

    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)

    by_qualified_name = {project["qualified_name"]: project for project in merged}
    personal_project = by_qualified_name["personal/main"]
    team_project = by_qualified_name["team/main"]

    assert personal_project["source"] == "cloud"
    assert personal_project["local_path"] is None
    assert personal_project["path"] == "/cloud/personal-main"
    assert team_project["source"] == "local+cloud"
    assert team_project["local_path"] == local_path
    assert team_project["path"] == local_path


def test_merge_workspace_projects_sorted_fallback_attaches_personal_workspace(tmp_path):
    """When config has no preference and no cloud default exists, use stable priority."""
    local_path = str(tmp_path / "main")
    local_main = _make_project("main", local_path, is_default=True)
    local_list = _make_list([local_main], default="main")
    personal_main = _make_project(
        "main",
        "/cloud/personal-main",
        id=10,
        external_id="personal-main-uuid",
    )
    team_main = _make_project(
        "main",
        "/cloud/team-main",
        id=11,
        external_id="team-main-uuid",
    )
    personal_ws = _make_workspace(
        "personal-tenant",
        "Personal",
        slug="personal",
        is_default=False,
    )
    team_ws = _make_workspace(
        "team-tenant",
        "Team",
        workspace_type="organization",
        slug="team",
    )
    workspace_index = _make_workspace_index(
        [
            (team_ws, [team_main]),
            (personal_ws, [personal_main]),
        ]
    )
    config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})

    merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)

    by_qualified_name = {project["qualified_name"]: project for project in merged}
    personal_project = by_qualified_name["personal/main"]
    team_project = by_qualified_name["team/main"]

    assert personal_project["source"] == "local+cloud"
    assert personal_project["local_path"] == local_path
    assert personal_project["path"] == local_path
    assert team_project["source"] == "cloud"
    assert team_project["local_path"] is None
    assert team_project["path"] == "/cloud/team-main"


# --- Workspace passthrough tests ---


def _make_workspace(
    tenant_id: str,
    name: str,
    workspace_type: str = "personal",
    role: str = "owner",
    organization_id: str | None = None,
    slug: str | None = None,
    is_default: bool = False,
):
    """Create a WorkspaceInfo for testing."""
    from basic_memory.schemas.cloud import WorkspaceInfo

    return WorkspaceInfo(
        tenant_id=tenant_id,
        name=name,
        workspace_type=workspace_type,
        slug=slug or name.casefold().replace(" ", "-"),
        role=role,
        organization_id=organization_id,
        is_default=is_default,
        has_active_subscription=True,
    )


def _make_workspace_index(workspace_projects):
    """Create a WorkspaceProjectIndex from (workspace, projects) tuples."""
    from basic_memory.mcp.project_context import (
        WorkspaceProjectEntry,
        _build_workspace_project_index,
    )

    workspaces = tuple(workspace for workspace, _projects in workspace_projects)
    entries = tuple(
        WorkspaceProjectEntry(workspace=workspace, project=project)
        for workspace, projects in workspace_projects
        for project in projects
    )
    return _build_workspace_project_index(workspaces, entries)


@pytest.mark.asyncio
async def test_list_memory_projects_aggregates_without_config_workspace(app, test_project):
    """When no explicit workspace is given, cloud discovery fans out across workspaces."""
    cloud_project = _make_project("cloud-proj", "/cloud-proj")
    workspace = _make_workspace(
        "config-default-ws",
        "Default WS",
        slug="default",
        is_default=True,
    )
    workspace_index = _make_workspace_index([(workspace, [cloud_project])])

    with (
        patch("basic_memory.mcp.tools.project_management.ConfigManager") as mock_cm_cls,
        patch(
            "basic_memory.mcp.tools.project_management.has_cloud_credentials",
            return_value=True,
        ),
        patch(
            "basic_memory.mcp.tools.project_management.ensure_workspace_project_index",
            new_callable=AsyncMock,
            return_value=workspace_index,
        ) as mock_index,
    ):
        mock_config = mock_cm_cls.return_value.config
        mock_config.default_workspace = "config-default-ws"
        result = await list_memory_projects()

    mock_index.assert_awaited_once()
    assert "- cloud-proj (cloud) [00000000-0000-0000-0000-000000000001]" in result