From 8eeec64e283f0cc383ae5f5761f7ad739335aeb0 Mon Sep 17 00:00:00 2001 From: Drew Cain Date: Thu, 14 May 2026 09:46:02 -0500 Subject: [PATCH] fix: `basic-memory project list` does not list projects from all workspaces (#822) Signed-off-by: Drew Cain --- .../cli/commands/cloud/workspace.py | 27 +- src/basic_memory/cli/commands/project.py | 298 ++++++-- src/basic_memory/mcp/project_context.py | 56 +- src/basic_memory/schemas/cloud.py | 45 ++ tests/cli/test_project_list_and_ls.py | 699 ++++++++++++++++++ tests/cli/test_workspace_commands.py | 42 ++ tests/mcp/test_project_context.py | 100 +++ 7 files changed, 1167 insertions(+), 100 deletions(-) diff --git a/src/basic_memory/cli/commands/cloud/workspace.py b/src/basic_memory/cli/commands/cloud/workspace.py index f06b3c9c..8ac1e90f 100644 --- a/src/basic_memory/cli/commands/cloud/workspace.py +++ b/src/basic_memory/cli/commands/cloud/workspace.py @@ -6,10 +6,11 @@ from rich.table import Table from basic_memory.cli.commands.command_utils import run_with_cleanup from basic_memory.config import ConfigManager -from basic_memory.mcp.project_context import ( - _workspace_choices, - _workspace_matches_identifier, - get_available_workspaces, +from basic_memory.mcp.project_context import get_available_workspaces +from basic_memory.schemas.cloud import ( + format_workspace_choices, + format_workspace_selection_choices, + workspace_matches_identifier, ) console = Console() @@ -62,7 +63,10 @@ def list_workspaces() -> None: @workspace_app.command("set-default") def set_default_workspace( - identifier: str = typer.Argument(..., help="Workspace name or tenant_id to set as default"), + identifier: str = typer.Argument( + ..., + help="Workspace name, slug, type, or tenant_id to set as default", + ), ) -> None: """Set the default cloud workspace. @@ -71,6 +75,7 @@ def set_default_workspace( Examples: bm cloud workspace set-default Personal + bm cloud workspace set-default organization bm cloud workspace set-default 11111111-1111-1111-1111-111111111111 """ @@ -87,19 +92,21 @@ def set_default_workspace( console.print("[yellow]No accessible workspaces found.[/yellow]") raise typer.Exit(1) - matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, identifier)] + matches = [ws for ws in workspaces if workspace_matches_identifier(ws, identifier)] if not matches: console.print(f"[red]Error: Workspace '{identifier}' not found[/red]") - console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]") + console.print(f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]") raise typer.Exit(1) if len(matches) > 1: console.print( - f"[red]Error: Workspace name '{identifier}' matches multiple workspaces. " - f"Use tenant_id instead.[/red]" + f"[red]Error: Workspace '{identifier}' matches multiple workspaces.[/red]" + ) + console.print( + "[dim]Choose one of these matching workspaces by slug:\n" + f"{format_workspace_selection_choices(matches)}[/dim]" ) - console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]") raise typer.Exit(1) selected = matches[0] diff --git a/src/basic_memory/cli/commands/project.py b/src/basic_memory/cli/commands/project.py index 2bc4b411..bf0628c3 100644 --- a/src/basic_memory/cli/commands/project.py +++ b/src/basic_memory/cli/commands/project.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import cast import typer +from loguru import logger from rich.console import Console, Group from rich.panel import Panel from rich.table import Table @@ -33,6 +34,10 @@ from basic_memory.schemas.cloud import ( CloudProjectIndexStatus, CloudTenantIndexStatusResponse, ProjectVisibility, + WorkspaceInfo, + format_workspace_choices, + format_workspace_selection_choices, + workspace_matches_identifier, ) from basic_memory.schemas.project_info import ProjectItem, ProjectList from basic_memory.utils import generate_permalink, normalize_project_path @@ -281,27 +286,27 @@ def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility: def _resolve_workspace_id(config, workspace: str | None) -> str | None: - """Resolve a workspace name or tenant_id to a tenant_id.""" + """Resolve a workspace name, slug, type, or tenant_id to a tenant_id.""" from basic_memory.mcp.project_context import ( - _workspace_choices, - _workspace_matches_identifier, get_available_workspaces, ) if workspace is not None: workspaces = run_with_cleanup(get_available_workspaces()) - matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)] + matches = [ws for ws in workspaces if workspace_matches_identifier(ws, workspace)] if not matches: console.print(f"[red]Error: Workspace '{workspace}' not found[/red]") if workspaces: - console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]") + console.print(f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]") raise typer.Exit(1) if len(matches) > 1: console.print( - f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. " - f"Use tenant_id instead.[/red]" + f"[red]Error: Workspace '{workspace}' matches multiple workspaces.[/red]" + ) + console.print( + "[dim]Choose one of these matching workspaces by slug:\n" + f"{format_workspace_selection_choices(matches)}[/dim]" ) - console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]") raise typer.Exit(1) return matches[0].tenant_id @@ -312,9 +317,9 @@ def _resolve_workspace_id(config, workspace: str | None) -> str | None: workspaces = run_with_cleanup(get_available_workspaces()) if len(workspaces) == 1: return workspaces[0].tenant_id - except Exception: + except Exception as exc: # Workspace resolution is optional until a command needs a specific tenant. - pass + logger.debug("Workspace resolution failed: {}", exc) return None @@ -323,7 +328,11 @@ def _resolve_workspace_id(config, workspace: str | None) -> str | None: def list_projects( local: bool = typer.Option(False, "--local", help="Force local routing for this command"), cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"), - workspace: str = typer.Option(None, "--workspace", help="Cloud workspace name or tenant_id"), + workspace: str = typer.Option( + None, + "--workspace", + help="Cloud workspace name, slug, type, or tenant_id", + ), json_output: bool = typer.Option(False, "--json", help="Output in JSON format"), ) -> None: """List Basic Memory projects from local and (when available) cloud.""" @@ -339,17 +348,90 @@ def list_projects( try: config = ConfigManager().config - # Use explicit workspace, fall back to config default - effective_workspace = workspace or config.default_workspace + workspace_filter = workspace + workspace_filter_requested = workspace_filter is not None local_result: ProjectList | None = None - cloud_result: ProjectList | None = None + cloud_results: list[tuple[WorkspaceInfo | None, ProjectList]] = [] + available_cloud_workspaces: list[WorkspaceInfo] = [] cloud_error: Exception | None = None + cloud_workspace_error: Exception | None = None + failed_cloud_workspaces: list[tuple[WorkspaceInfo, Exception]] = [] + + def _fetch_cloud_workspace_results() -> tuple[ + list[tuple[WorkspaceInfo | None, ProjectList]], + list[WorkspaceInfo], + Exception | None, + list[tuple[WorkspaceInfo, Exception]], + ]: + from basic_memory.mcp.project_context import ( + get_available_workspaces, + ) + + try: + workspaces = run_with_cleanup(get_available_workspaces()) + except Exception as exc: + fallback_workspace = workspace_filter or config.default_workspace + return ( + [(None, run_with_cleanup(_list_projects(fallback_workspace)))], + [], + exc, + [], + ) + + selected_workspaces = workspaces + if workspace_filter is not None: + matches = [ + ws for ws in workspaces if workspace_matches_identifier(ws, workspace_filter) + ] + if not matches: + console.print(f"[red]Error: Workspace '{workspace_filter}' not found[/red]") + if workspaces: + console.print( + f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]" + ) + raise typer.Exit(1) + if len(matches) > 1: + console.print( + f"[red]Error: Workspace '{workspace_filter}' matches multiple workspaces.[/red]" + ) + console.print( + "[dim]Choose one of these matching workspaces by slug:\n" + f"{format_workspace_selection_choices(matches)}[/dim]" + ) + raise typer.Exit(1) + selected_workspaces = matches + + if not selected_workspaces: + return [], workspaces, None, [] + + results: list[tuple[WorkspaceInfo | None, ProjectList]] = [] + failed_workspaces: list[tuple[WorkspaceInfo, Exception]] = [] + for cloud_workspace in selected_workspaces: + try: + results.append( + ( + cloud_workspace, + run_with_cleanup(_list_projects(cloud_workspace.tenant_id)), + ) + ) + except Exception as exc: + failed_workspaces.append((cloud_workspace, exc)) + + if not results and failed_workspaces: + raise failed_workspaces[0][1] + + return results, workspaces, None, failed_workspaces if cloud: with console.status("[bold blue]Fetching cloud projects...", spinner="dots"): with force_routing(cloud=True): - cloud_result = run_with_cleanup(_list_projects(effective_workspace)) + ( + cloud_results, + available_cloud_workspaces, + cloud_workspace_error, + failed_cloud_workspaces, + ) = _fetch_cloud_workspace_results() elif local: with force_routing(local=True): local_result = run_with_cleanup(_list_projects()) @@ -362,29 +444,17 @@ def list_projects( try: with console.status("[bold blue]Fetching cloud projects...", spinner="dots"): with force_routing(cloud=True): - cloud_result = run_with_cleanup(_list_projects(effective_workspace)) + ( + cloud_results, + available_cloud_workspaces, + cloud_workspace_error, + failed_cloud_workspaces, + ) = _fetch_cloud_workspace_results() + except typer.Exit: + raise except Exception as exc: # pragma: no cover cloud_error = exc - # Resolve workspace name for cloud projects (best-effort) - cloud_ws_name: str | None = None - cloud_ws_type: str | None = None - if cloud_result and effective_workspace: - try: - from basic_memory.mcp.project_context import get_available_workspaces - - with console.status("[bold blue]Resolving workspace...", spinner="dots"): - workspaces = run_with_cleanup(get_available_workspaces()) - matched = next( - (ws for ws in workspaces if ws.tenant_id == effective_workspace), - None, - ) - if matched: - cloud_ws_name = matched.name - cloud_ws_type = matched.workspace_type - except Exception: - pass - table = Table(title="Basic Memory Projects") table.add_column("Name", style="cyan") table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold") @@ -395,29 +465,128 @@ def list_projects( table.add_column("Sync", style="green") table.add_column("Default", style="magenta") - project_names_by_permalink: dict[str, str] = {} + row_names_by_key: dict[tuple[str | None, str], str] = {} local_projects_by_permalink: dict[str, ProjectItem] = {} - cloud_projects_by_permalink: dict[str, ProjectItem] = {} + cloud_projects_by_key: dict[tuple[str | None, str], ProjectItem] = {} + cloud_workspaces_by_key: dict[tuple[str | None, str], WorkspaceInfo | None] = {} if local_result: for project in local_result.projects: permalink = generate_permalink(project.name) - project_names_by_permalink[permalink] = project.name local_projects_by_permalink[permalink] = project - if cloud_result: + for cloud_workspace, cloud_result in cloud_results: + workspace_key = cloud_workspace.tenant_id if cloud_workspace else None for project in cloud_result.projects: permalink = generate_permalink(project.name) - project_names_by_permalink[permalink] = project.name - cloud_projects_by_permalink[permalink] = project + row_key = (workspace_key, permalink) + row_names_by_key[row_key] = project.name + cloud_projects_by_key[row_key] = project + cloud_workspaces_by_key[row_key] = cloud_workspace + + cloud_permalinks = {permalink for _, permalink in cloud_projects_by_key} + for permalink, project in local_projects_by_permalink.items(): + if permalink not in cloud_permalinks: + row_names_by_key[(None, permalink)] = project.name + + cloud_keys_by_permalink: dict[str, list[tuple[str | None, str]]] = {} + for row_key in cloud_projects_by_key: + cloud_keys_by_permalink.setdefault(row_key[1], []).append(row_key) + + configured_names_by_permalink = { + generate_permalink(project_name): project_name for project_name in config.projects + } + + def _workspace_priority(row_key: tuple[str | None, str]) -> tuple[bool, int, str, str]: + """Prefer the user's default/personal workspace when a project is duplicated.""" + workspace = cloud_workspaces_by_key.get(row_key) + if workspace is None: + return (True, 2, "", row_key[0] or "") + workspace_type_rank = 0 if workspace.workspace_type == "personal" else 1 + return ( + not workspace.is_default, + workspace_type_rank, + workspace.name.casefold(), + row_key[0] or "", + ) + + def _select_attached_row_key( + permalink: str, entry: ProjectEntry | None + ) -> tuple[str | None, str] | None: + """Choose the single row that owns local config/default/sync state.""" + cloud_keys = cloud_keys_by_permalink.get(permalink, []) + if not cloud_keys: + return (None, permalink) + + preferred_workspace_ids: list[str] = [] + if entry and entry.workspace_id: + preferred_workspace_ids.append(entry.workspace_id) + if config.default_workspace and config.default_workspace not in preferred_workspace_ids: + preferred_workspace_ids.append(config.default_workspace) + default_cloud_workspace = next( + (item for item in available_cloud_workspaces if item.is_default), + None, + ) + if ( + default_cloud_workspace + and default_cloud_workspace.tenant_id not in preferred_workspace_ids + ): + preferred_workspace_ids.append(default_cloud_workspace.tenant_id) + + for workspace_id in preferred_workspace_ids: + for row_key in cloud_keys: + if row_key[0] == workspace_id: + return row_key + + if workspace_filter_requested and preferred_workspace_ids: + # A filtered list can exclude the workspace that owns local config state. + # In that case, do not attach local/default/sync state to another workspace row. + return None + + default_workspace_keys = [ + row_key + for row_key in cloud_keys + if (row_workspace := cloud_workspaces_by_key.get(row_key)) is not None + and row_workspace.is_default + ] + if len(default_workspace_keys) == 1: + return default_workspace_keys[0] + + if len(cloud_keys) == 1: + return cloud_keys[0] + + return sorted(cloud_keys, key=_workspace_priority)[0] + + attached_row_by_permalink: dict[str, tuple[str | None, str] | None] = {} + for permalink in set(local_projects_by_permalink) | set(configured_names_by_permalink): + configured_name = configured_names_by_permalink.get(permalink) + local_project = local_projects_by_permalink.get(permalink) + entry_name = configured_name or (local_project.name if local_project else None) + entry = config.projects.get(entry_name) if entry_name else None + attached_row_by_permalink[permalink] = _select_attached_row_key(permalink, entry) # --- Build unified project list --- project_rows: list[dict] = [] - for permalink in sorted(project_names_by_permalink): - project_name = project_names_by_permalink[permalink] - local_project = local_projects_by_permalink.get(permalink) - cloud_project = cloud_projects_by_permalink.get(permalink) - entry = config.projects.get(project_name) + sorted_row_keys = sorted( + row_names_by_key, + key=lambda key: (row_names_by_key[key], key[0] or ""), + ) + for row_key in sorted_row_keys: + _, permalink = row_key + project_name = row_names_by_key[row_key] + is_attached_row = attached_row_by_permalink.get(permalink) == row_key + local_project = ( + local_projects_by_permalink.get(permalink) if is_attached_row else None + ) + cloud_project = cloud_projects_by_key.get(row_key) + cloud_workspace = cloud_workspaces_by_key.get(row_key) + configured_name = configured_names_by_permalink.get(permalink) + configured_entry = ( + config.projects.get(configured_name) + if configured_name + else config.projects.get(project_name) + ) + entry = configured_entry if is_attached_row else None local_path = "" if local_project is not None: @@ -447,9 +616,15 @@ def list_projects( else: cli_route = ProjectMode.LOCAL.value - is_default = config.default_project == project_name + default_permalink = ( + generate_permalink(config.default_project) if config.default_project else None + ) + is_default = bool(is_attached_row and permalink == default_permalink) - has_sync = bool(entry and entry.local_sync_path) + sync_supported = ( + cloud_workspace is None or cloud_workspace.workspace_type == "personal" + ) + has_sync = bool(is_attached_row and entry and entry.local_sync_path and sync_supported) # Determine MCP transport based on project routing mode if entry and entry.mode == ProjectMode.CLOUD: mcp_transport = "https" @@ -459,9 +634,8 @@ def list_projects( mcp_transport = "stdio" # Show workspace name (type) for cloud-sourced projects - ws_label = "" - if cloud_project is not None and cloud_ws_name: - ws_label = f"{cloud_ws_name} ({cloud_ws_type})" if cloud_ws_type else cloud_ws_name + cloud_ws_name = cloud_workspace.name if cloud_workspace else None + cloud_ws_type = cloud_workspace.workspace_type if cloud_workspace else None # display_name is a human label for private UUID-named projects (e.g., "My Project"). # Keep "name" as the canonical identifier for scripting/JSON consumers; @@ -481,8 +655,8 @@ def list_projects( } if display_name: row_data["display_name"] = display_name - if ws_label: - row_data["workspace"] = cloud_ws_name or "" + if cloud_project is not None and cloud_ws_name: + row_data["workspace"] = cloud_ws_name if cloud_ws_type: row_data["workspace_type"] = cloud_ws_type @@ -514,6 +688,20 @@ def list_projects( "[dim]Showing local projects only. " "Run 'bm cloud login' or 'bm cloud api-key save ' if this is a credentials issue.[/dim]" ) + if cloud_workspace_error is not None: + console.print( + f"[yellow]Cloud workspace discovery failed: {cloud_workspace_error}[/yellow]" + ) + console.print( + "[dim]Showing cloud projects from the configured/default workspace only.[/dim]" + ) + for failed_workspace, error in failed_cloud_workspaces: + console.print( + f"[yellow]Cloud project discovery failed for workspace " + f"{failed_workspace.name}: {error}[/yellow]" + ) + except typer.Exit: + raise except Exception as e: console.print(f"[red]Error listing projects: {str(e)}[/red]") raise typer.Exit(1) @@ -531,7 +719,7 @@ def add_project( workspace: str = typer.Option( None, "--workspace", - help="Cloud workspace name or tenant_id (cloud mode only)", + help="Cloud workspace name, slug, type, or tenant_id (cloud mode only)", ), visibility: str = typer.Option( None, @@ -914,7 +1102,7 @@ def set_cloud( workspace: str = typer.Option( None, "--workspace", - help="Cloud workspace name or tenant_id to associate with this project", + help="Cloud workspace name, slug, type, or tenant_id to associate with this project", ), ) -> None: """Set a project to cloud mode (route through cloud API). diff --git a/src/basic_memory/mcp/project_context.py b/src/basic_memory/mcp/project_context.py index 99dcfcf5..b6a4c897 100644 --- a/src/basic_memory/mcp/project_context.py +++ b/src/basic_memory/mcp/project_context.py @@ -11,7 +11,7 @@ compatibility with existing MCP tools. import asyncio from contextlib import asynccontextmanager, nullcontext from dataclasses import dataclass, field -from typing import AsyncIterator, Awaitable, Callable, Optional, List, Tuple, cast +from typing import AsyncIterator, Awaitable, Callable, List, Optional, Sequence, Tuple, cast from uuid import UUID from httpx import AsyncClient @@ -25,7 +25,14 @@ from mcp.server.fastmcp.exceptions import ToolError import logfire from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectMode, has_cloud_credentials from basic_memory.project_resolver import ProjectResolver -from basic_memory.schemas.cloud import WorkspaceInfo, WorkspaceListResponse +from basic_memory.schemas.cloud import ( + WorkspaceInfo, + WorkspaceListResponse, + format_workspace_choices, + format_workspace_selection_choices, + workspace_matches_exact_identifier, + workspace_matches_identifier, +) from basic_memory.schemas.project_info import ProjectItem, ProjectList from basic_memory.schemas.v2 import ProjectResolveResponse from basic_memory.schemas.memory import memory_url_path @@ -299,29 +306,6 @@ async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = N return [project.name for project in project_list.projects] -def _workspace_matches_identifier(workspace: WorkspaceInfo, identifier: str) -> bool: - """Return True when identifier matches workspace tenant_id, slug, or name.""" - if workspace.tenant_id == identifier: - return True - if workspace.slug.casefold() == identifier.casefold(): - return True - return workspace.name.lower() == identifier.lower() - - -def _workspace_choices(workspaces: list[WorkspaceInfo]) -> str: - """Format deterministic workspace choices for prompt-style errors.""" - return "\n".join( - [ - ( - f"- {item.name} " - f"(slug={item.slug}, type={item.workspace_type}, " - f"role={item.role}, tenant_id={item.tenant_id})" - ) - for item in workspaces - ] - ) - - def _workspace_project_index_from_state(raw: object) -> WorkspaceProjectIndex | None: """Deserialize a cached workspace project index from MCP context state.""" if not isinstance(raw, dict): @@ -624,7 +608,7 @@ async def _resolve_workspace_segments( return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path) -def _format_qualified_choices(entries: tuple[WorkspaceProjectEntry, ...]) -> str: +def _format_qualified_choices(entries: Sequence[WorkspaceProjectEntry]) -> str: """Format qualified project choices for collision errors.""" return " or ".join(entry.qualified_name for entry in entries) @@ -853,7 +837,7 @@ async def resolve_workspace_project_identifier( ) return matches[0] - matches = index.entries_by_permalink.get(project_permalink, ()) + matches = list(index.entries_by_permalink.get(project_permalink, ())) if not matches: failed_note = "" if index.failed_workspaces: @@ -987,7 +971,9 @@ async def resolve_workspace_parameter( cached_raw = await context.get_state("active_workspace") if isinstance(cached_raw, dict): cached_workspace = WorkspaceInfo.model_validate(cached_raw) - if workspace is None or _workspace_matches_identifier(cached_workspace, workspace): + if workspace is None or workspace_matches_exact_identifier( + cached_workspace, workspace + ): logger.debug( f"Using cached workspace from context: {cached_workspace.tenant_id}" ) @@ -1004,18 +990,18 @@ async def resolve_workspace_parameter( if workspace: matches = [ - item for item in workspaces if _workspace_matches_identifier(item, workspace) + item for item in workspaces if workspace_matches_identifier(item, workspace) ] if not matches: raise ValueError( f"Workspace '{workspace}' was not found.\n" - f"Available workspaces:\n{_workspace_choices(workspaces)}" + f"Available workspaces:\n{format_workspace_choices(workspaces)}" ) if len(matches) > 1: raise ValueError( - f"Workspace name '{workspace}' matches multiple workspaces. " - "Use tenant_id instead.\n" - f"Available workspaces:\n{_workspace_choices(workspaces)}" + f"Workspace '{workspace}' matches multiple workspaces. " + "Choose one of these matching workspaces by slug or tenant_id:\n" + f"{format_workspace_selection_choices(matches)}" ) selected_workspace = matches[0] elif len(workspaces) == 1: @@ -1023,8 +1009,8 @@ async def resolve_workspace_parameter( else: raise ValueError( "Multiple workspaces are available. Ask the user which workspace to use, then retry " - "with the 'workspace' argument set to the tenant_id or unique name.\n" - f"Available workspaces:\n{_workspace_choices(workspaces)}" + "with the 'workspace' argument set to the tenant_id or unique name/slug/type.\n" + f"Available workspaces:\n{format_workspace_choices(workspaces)}" ) await _set_cached_active_workspace(context, selected_workspace) diff --git a/src/basic_memory/schemas/cloud.py b/src/basic_memory/schemas/cloud.py index efe19c8a..18cdd135 100644 --- a/src/basic_memory/schemas/cloud.py +++ b/src/basic_memory/schemas/cloud.py @@ -88,6 +88,51 @@ class WorkspaceListResponse(BaseModel): ) +def workspace_matches_exact_identifier(workspace: WorkspaceInfo, identifier: str) -> bool: + """Return True when identifier matches workspace tenant_id, slug, or name.""" + if workspace.tenant_id == identifier: + return True + if workspace.slug.casefold() == identifier.casefold(): + return True + return workspace.name.casefold() == identifier.casefold() + + +def workspace_matches_identifier(workspace: WorkspaceInfo, identifier: str) -> bool: + """Return True when identifier matches workspace tenant_id, slug, name, or type.""" + return ( + workspace_matches_exact_identifier(workspace, identifier) + or workspace.workspace_type.casefold() == identifier.casefold() + ) + + +def format_workspace_choices(workspaces: list[WorkspaceInfo]) -> str: + """Format deterministic workspace choices for prompt-style errors.""" + return "\n".join( + [ + ( + f"- {item.name} " + f"(slug={item.slug}, type={item.workspace_type}, " + f"role={item.role}, tenant_id={item.tenant_id})" + ) + for item in workspaces + ] + ) + + +def format_workspace_selection_choices(workspaces: list[WorkspaceInfo]) -> str: + """Format matching workspaces with copyable unique identifiers first.""" + return "\n".join( + [ + ( + f"- {item.name} ({item.workspace_type}, role={item.role})\n" + f" workspace: {item.slug}\n" + f" tenant_id: {item.tenant_id}" + ) + for item in workspaces + ] + ) + + class CloudProjectIndexStatus(BaseModel): """Index freshness summary for one cloud project.""" diff --git a/tests/cli/test_project_list_and_ls.py b/tests/cli/test_project_list_and_ls.py index 6e5cbbd1..2b7efec7 100644 --- a/tests/cli/test_project_list_and_ls.py +++ b/tests/cli/test_project_list_and_ls.py @@ -10,6 +10,7 @@ from typer.testing import CliRunner from basic_memory.cli.app import app from basic_memory.mcp.clients.project import ProjectClient +from basic_memory.schemas.cloud import WorkspaceInfo from basic_memory.schemas.project_info import ProjectList # Importing registers project subcommands on the shared app instance. @@ -53,6 +54,26 @@ def mock_client(monkeypatch): monkeypatch.setattr(project_cmd, "get_client", fake_get_client) +def _workspace( + *, + tenant_id: str, + slug: str, + name: str, + workspace_type: str, + is_default: bool = False, +) -> WorkspaceInfo: + return WorkspaceInfo( + tenant_id=tenant_id, + workspace_type=workspace_type, + slug=slug, + name=name, + role="owner", + is_default=is_default, + organization_id=None, + has_active_subscription=True, + ) + + def test_project_list_shows_local_cloud_presence_and_routes( runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch ): @@ -215,6 +236,684 @@ def test_project_list_shows_display_name_for_private_projects( assert private_project["display_name"] == "My Project" +def test_project_list_cloud_fetches_all_workspaces_and_labels_duplicate_permalinks( + runner: CliRunner, write_config, monkeypatch +): + """Cloud project list should include every workspace without collapsing matching names.""" + write_config( + { + "env": "dev", + "projects": {}, + "default_project": None, + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + team = _workspace( + tenant_id="tenant-team", + slug="team", + name="Team", + workspace_type="organization", + ) + + async def fake_get_available_workspaces(): + return [personal, team] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + payloads_by_workspace = { + None: {"projects": [], "default_project": None}, + "tenant-personal": { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "shared", + "path": "/personal/shared", + "is_default": True, + } + ], + "default_project": "shared", + }, + "tenant-team": { + "projects": [ + { + "id": 2, + "external_id": "22222222-2222-2222-2222-222222222222", + "name": "shared", + "path": "/team/shared", + "is_default": False, + } + ], + "default_project": None, + }, + } + seen_workspaces: list[str | None] = [] + + async def fake_list_projects(self): + workspace = self.http_client.workspace + seen_workspaces.append(workspace) + return ProjectList.model_validate(payloads_by_workspace[workspace]) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"}) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + # The initial None call is the local project fetch before cloud workspaces are listed. + assert seen_workspaces == [None, "tenant-personal", "tenant-team"] + + data = json.loads(result.stdout) + shared_projects = [project for project in data["projects"] if project["name"] == "shared"] + assert len(shared_projects) == 2 + assert {project["cloud_path"] for project in shared_projects} == { + "/personal/shared", + "/team/shared", + } + assert {project["workspace"] for project in shared_projects} == {"Personal", "Team"} + assert {project["workspace_type"] for project in shared_projects} == { + "personal", + "organization", + } + + table_result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"}) + + assert table_result.exit_code == 0 + assert "Personal (personal)" in table_result.stdout + assert "Team (organization)" in table_result.stdout + + +def test_project_list_workspace_discovery_failure_warns_and_uses_fallback( + runner: CliRunner, write_config, monkeypatch +): + """Workspace discovery failures should fall back and explain the degraded result.""" + write_config( + { + "env": "dev", + "projects": {}, + "default_project": None, + "default_workspace": "tenant-default", + "cloud_api_key": "bmc_test_key_123", + } + ) + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + async def fail_get_available_workspaces(): + raise RuntimeError("workspace service unavailable") + + payloads_by_workspace = { + None: {"projects": [], "default_project": None}, + "tenant-default": { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "fallback-project", + "path": "/fallback-project", + "is_default": False, + } + ], + "default_project": None, + }, + } + + async def fake_list_projects(self): + return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace]) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fail_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"}) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + assert "fallback-project" in result.stdout + assert "Cloud workspace discovery failed: workspace service unavailable" in result.stdout + assert "Showing cloud projects from the configured/default workspace only" in result.stdout + + +def test_project_list_partial_workspace_failure_warns_and_keeps_successes( + runner: CliRunner, write_config, monkeypatch +): + """A failed workspace fetch should not hide projects from successful workspaces.""" + write_config( + { + "env": "dev", + "projects": {}, + "default_project": None, + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + team = _workspace( + tenant_id="tenant-team", + slug="team", + name="Team", + workspace_type="organization", + ) + + async def fake_get_available_workspaces(): + return [personal, team] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + payloads_by_workspace = { + None: {"projects": [], "default_project": None}, + "tenant-personal": { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "personal-project", + "path": "/personal-project", + "is_default": False, + } + ], + "default_project": None, + }, + } + + async def fake_list_projects(self): + if self.http_client.workspace == "tenant-team": + raise RuntimeError("team unavailable") + return ProjectList.model_validate(payloads_by_workspace[self.http_client.workspace]) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke(app, ["project", "list"], env={"COLUMNS": "240"}) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + assert "personal-project" in result.stdout + assert "Cloud project discovery failed for workspace Team: team unavailable" in result.stdout + + +def test_project_list_workspace_type_filter_selects_unique_workspace( + runner: CliRunner, write_config, monkeypatch +): + """--workspace can use a workspace type when it resolves to one workspace.""" + write_config( + { + "env": "dev", + "projects": {}, + "default_project": None, + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + team = _workspace( + tenant_id="tenant-team", + slug="team", + name="Team", + workspace_type="organization", + ) + + async def fake_get_available_workspaces(): + return [personal, team] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + payloads_by_workspace = { + None: {"projects": [], "default_project": None}, + "tenant-team": { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "team-project", + "path": "/team-project", + "is_default": True, + } + ], + "default_project": "team-project", + }, + } + seen_workspaces: list[str | None] = [] + + async def fake_list_projects(self): + workspace = self.http_client.workspace + seen_workspaces.append(workspace) + return ProjectList.model_validate(payloads_by_workspace[workspace]) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke( + app, + ["project", "list", "--workspace", "organization", "--json"], + env={"COLUMNS": "240"}, + ) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + assert seen_workspaces == [None, "tenant-team"] + data = json.loads(result.stdout) + assert [project["name"] for project in data["projects"]] == ["team-project"] + assert data["projects"][0]["workspace"] == "Team" + + +def test_project_list_workspace_type_filter_lists_ambiguous_matches( + runner: CliRunner, write_config, monkeypatch +): + """Ambiguous workspace type filters should list copyable matching slugs.""" + write_config( + { + "env": "dev", + "projects": {}, + "default_project": None, + "cloud_api_key": "bmc_test_key_123", + } + ) + + async def fake_get_available_workspaces(): + return [ + _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ), + _workspace( + tenant_id="tenant-team-alpha", + slug="team-alpha", + name="Team Alpha", + workspace_type="organization", + ), + _workspace( + tenant_id="tenant-team-beta", + slug="team-beta", + name="Team Beta", + workspace_type="organization", + ), + ] + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield object() + + async def fake_list_projects(self): + return ProjectList.model_validate({"projects": [], "default_project": None}) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke( + app, + ["project", "list", "--workspace", "organization"], + env={"COLUMNS": "240"}, + ) + + assert result.exit_code == 1 + assert "Workspace 'organization' matches multiple workspaces" in result.stdout + assert "Choose one of these matching workspaces by slug" in result.stdout + assert "workspace: team-alpha" in result.stdout + assert "workspace: team-beta" in result.stdout + assert "tenant_id: tenant-team-alpha" in result.stdout + assert "tenant_id: tenant-team-beta" in result.stdout + assert "workspace: personal" not in result.stdout + + +def test_project_list_invalid_workspace_exits_without_local_fallback( + runner: CliRunner, write_config, tmp_path, monkeypatch +): + """Invalid explicit workspace filters should stop instead of showing local-only rows.""" + local_path = (tmp_path / "main").as_posix() + write_config( + { + "env": "dev", + "projects": {"main": {"path": local_path, "mode": "local"}}, + "default_project": "main", + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + + async def fake_get_available_workspaces(): + return [personal] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + async def fake_list_projects(self): + return ProjectList.model_validate( + { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "main", + "path": local_path, + "is_default": True, + } + ], + "default_project": "main", + } + ) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke( + app, + ["project", "list", "--workspace", "missing"], + env={"COLUMNS": "240"}, + ) + + assert result.exit_code == 1 + assert "Workspace 'missing' not found" in result.stdout + assert "Basic Memory Projects" not in result.stdout + assert "Cloud project discovery failed" not in result.stdout + + +def test_project_list_attaches_local_state_to_one_duplicate_cloud_project( + runner: CliRunner, write_config, tmp_path, monkeypatch +): + """Local path/default/sync state should not appear on every matching workspace.""" + local_path = (tmp_path / "main").as_posix() + write_config( + { + "env": "dev", + "projects": { + "main": { + "path": local_path, + "mode": "local", + "local_sync_path": local_path, + } + }, + "default_project": "main", + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + team = _workspace( + tenant_id="tenant-team", + slug="team", + name="Team", + workspace_type="organization", + ) + + async def fake_get_available_workspaces(): + return [personal, team] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + payloads_by_workspace = { + None: { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "main", + "path": local_path, + "is_default": True, + } + ], + "default_project": "main", + }, + "tenant-personal": { + "projects": [ + { + "id": 2, + "external_id": "22222222-2222-2222-2222-222222222222", + "name": "main", + "path": "/basic-memory", + "is_default": True, + } + ], + "default_project": "main", + }, + "tenant-team": { + "projects": [ + { + "id": 3, + "external_id": "33333333-3333-3333-3333-333333333333", + "name": "main", + "path": "/basic-memory", + "is_default": True, + } + ], + "default_project": "main", + }, + } + + async def fake_list_projects(self): + return ProjectList.model_validate( + payloads_by_workspace[self.http_client.workspace] + ) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"}) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + data = json.loads(result.stdout) + main_rows = [project for project in data["projects"] if project["name"] == "main"] + assert len(main_rows) == 2 + + personal_row = next(project for project in main_rows if project["workspace"] == "Personal") + team_row = next(project for project in main_rows if project["workspace"] == "Team") + + assert personal_row["local_path"] == project_cmd.format_path(local_path) + assert personal_row["cli_route"] == "local" + assert personal_row["mcp_stdio"] == "stdio" + assert personal_row["sync"] is True + assert personal_row["is_default"] is True + + assert team_row["local_path"] == "" + assert team_row["cli_route"] == "cloud" + assert team_row["mcp_stdio"] == "https" + assert team_row["sync"] is False + assert team_row["is_default"] is False + + filtered_result = runner.invoke( + app, + ["project", "list", "--workspace", "organization", "--json"], + env={"COLUMNS": "240"}, + ) + + assert filtered_result.exit_code == 0, ( + f"Exit code: {filtered_result.exit_code}, output: {filtered_result.stdout}" + ) + filtered_data = json.loads(filtered_result.stdout) + assert len(filtered_data["projects"]) == 1 + filtered_team_row = filtered_data["projects"][0] + assert filtered_team_row["workspace"] == "Team" + assert filtered_team_row["local_path"] == "" + assert filtered_team_row["cli_route"] == "cloud" + assert filtered_team_row["mcp_stdio"] == "https" + assert filtered_team_row["sync"] is False + assert filtered_team_row["is_default"] is False + + +def test_project_list_hides_bisync_flag_for_attached_team_workspace( + runner: CliRunner, write_config, tmp_path, monkeypatch +): + """Bisync is only supported for personal workspaces.""" + local_path = (tmp_path / "team-main").as_posix() + write_config( + { + "env": "dev", + "projects": { + "main": { + "path": local_path, + "mode": "cloud", + "workspace_id": "tenant-team", + "local_sync_path": local_path, + } + }, + "default_project": "main", + "cloud_api_key": "bmc_test_key_123", + } + ) + + personal = _workspace( + tenant_id="tenant-personal", + slug="personal", + name="Personal", + workspace_type="personal", + is_default=True, + ) + team = _workspace( + tenant_id="tenant-team", + slug="team", + name="Team", + workspace_type="organization", + ) + + async def fake_get_available_workspaces(): + return [personal, team] + + class FakeClient: + def __init__(self, workspace: str | None): + self.workspace = workspace + + @asynccontextmanager + async def fake_get_client(workspace=None): + yield FakeClient(workspace) + + project_payload = { + "projects": [ + { + "id": 1, + "external_id": "11111111-1111-1111-1111-111111111111", + "name": "main", + "path": "/basic-memory", + "is_default": True, + } + ], + "default_project": "main", + } + payloads_by_workspace = { + None: {"projects": [], "default_project": None}, + "tenant-personal": project_payload, + "tenant-team": project_payload, + } + + async def fake_list_projects(self): + return ProjectList.model_validate( + payloads_by_workspace[self.http_client.workspace] + ) + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + monkeypatch.setattr(project_cmd, "get_client", fake_get_client) + monkeypatch.setattr(ProjectClient, "list_projects", fake_list_projects) + + result = runner.invoke(app, ["project", "list", "--json"], env={"COLUMNS": "240"}) + + assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}" + data = json.loads(result.stdout) + main_rows = [project for project in data["projects"] if project["name"] == "main"] + team_row = next(project for project in main_rows if project["workspace"] == "Team") + + assert team_row["cli_route"] == "cloud" + assert team_row["mcp_stdio"] == "https" + assert team_row["sync"] is False + assert team_row["is_default"] is True + + def test_project_ls_local_mode_defaults_to_local_route( runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch ): diff --git a/tests/cli/test_workspace_commands.py b/tests/cli/test_workspace_commands.py index 809bd27a..531c5b46 100644 --- a/tests/cli/test_workspace_commands.py +++ b/tests/cli/test_workspace_commands.py @@ -182,6 +182,48 @@ class TestWorkspaceSetDefault: assert result.exit_code == 1 assert "not found" in result.stdout + def test_set_default_workspace_ambiguous_type_lists_matching_choices( + self, runner, monkeypatch + ): + async def fake_get_available_workspaces(context=None): + return [ + _workspace( + tenant_id="11111111-1111-1111-1111-111111111111", + workspace_type="personal", + slug="personal", + name="Personal", + role="owner", + is_default=True, + ), + _workspace( + tenant_id="22222222-2222-2222-2222-222222222222", + workspace_type="organization", + slug="team-alpha", + name="Team Alpha", + role="editor", + ), + _workspace( + tenant_id="33333333-3333-3333-3333-333333333333", + workspace_type="organization", + slug="team-beta", + name="Team Beta", + role="owner", + ), + ] + + monkeypatch.setattr( + workspace_cmd, "get_available_workspaces", fake_get_available_workspaces + ) + + result = runner.invoke(app, ["cloud", "workspace", "set-default", "organization"]) + + assert result.exit_code == 1 + assert "Workspace 'organization' matches multiple workspaces" in result.stdout + assert "Choose one of these matching workspaces by slug" in result.stdout + assert "workspace: team-alpha" in result.stdout + assert "workspace: team-beta" in result.stdout + assert "workspace: personal" not in result.stdout + def test_set_default_workspace_no_workspaces(self, runner, monkeypatch): async def fake_get_available_workspaces(context=None): return [] diff --git a/tests/mcp/test_project_context.py b/tests/mcp/test_project_context.py index f99e1eae..ee02f082 100644 --- a/tests/mcp/test_project_context.py +++ b/tests/mcp/test_project_context.py @@ -340,6 +340,106 @@ async def test_workspace_invalid_selection_lists_choices(monkeypatch): await resolve_workspace_parameter(workspace="missing-workspace") +@pytest.mark.asyncio +async def test_workspace_ambiguous_type_selection_lists_matching_choices(monkeypatch): + from basic_memory.mcp.project_context import resolve_workspace_parameter + + workspaces = [ + _workspace( + tenant_id="11111111-1111-1111-1111-111111111111", + workspace_type="personal", + slug="personal", + name="Personal", + role="owner", + is_default=True, + ), + _workspace( + tenant_id="22222222-2222-2222-2222-222222222222", + workspace_type="organization", + slug="team-alpha", + name="Team Alpha", + role="editor", + ), + _workspace( + tenant_id="33333333-3333-3333-3333-333333333333", + workspace_type="organization", + slug="team-beta", + name="Team Beta", + 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) as exc_info: + await resolve_workspace_parameter(workspace="organization") + + message = str(exc_info.value) + assert "Workspace 'organization' matches multiple workspaces" in message + assert "workspace: team-alpha" in message + assert "workspace: team-beta" in message + assert "workspace: personal" not in message + + +@pytest.mark.asyncio +async def test_workspace_type_selection_ignores_cached_workspace_for_ambiguity(monkeypatch): + from basic_memory.mcp.project_context import resolve_workspace_parameter + + cached_workspace = _workspace( + tenant_id="22222222-2222-2222-2222-222222222222", + workspace_type="organization", + slug="team-alpha", + name="Team Alpha", + role="editor", + ) + workspaces = [ + _workspace( + tenant_id="11111111-1111-1111-1111-111111111111", + workspace_type="personal", + slug="personal", + name="Personal", + role="owner", + is_default=True, + ), + cached_workspace, + _workspace( + tenant_id="33333333-3333-3333-3333-333333333333", + workspace_type="organization", + slug="team-beta", + name="Team Beta", + role="owner", + ), + ] + context = _ContextState() + await context.set_state("active_workspace", cached_workspace.model_dump()) + fetches = 0 + + async def fake_get_available_workspaces(context=None): + nonlocal fetches + fetches += 1 + return workspaces + + monkeypatch.setattr( + "basic_memory.mcp.project_context.get_available_workspaces", + fake_get_available_workspaces, + ) + + with pytest.raises(ValueError) as exc_info: + await resolve_workspace_parameter(workspace="organization", context=_ctx(context)) + + message = str(exc_info.value) + assert fetches == 1 + assert "Workspace 'organization' matches multiple workspaces" in message + assert "workspace: team-alpha" in message + assert "workspace: team-beta" in message + + @pytest.mark.asyncio async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch): from basic_memory.mcp.project_context import resolve_workspace_parameter