fix: basic-memory project list does not list projects from all workspaces (#822)

Signed-off-by: Drew Cain <groksrc@gmail.com>
This commit is contained in:
Drew Cain
2026-05-14 09:46:02 -05:00
committed by GitHub
parent c6fa185bf3
commit 8eeec64e28
7 changed files with 1167 additions and 100 deletions
@@ -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]
+243 -55
View File
@@ -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 <key>' 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).
+21 -35
View File
@@ -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)
+45
View File
@@ -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."""
+699
View File
@@ -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
):
+42
View File
@@ -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 []
+100
View File
@@ -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