fix(cli): address workspace review feedback

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

Author: groksrc

src/basic_memory/cli/commands/cloud/workspace.py

@@ -6,11 +6,11 @@
 
 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,
-    _workspace_selection_choices,
-    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()
@@ -92,11 +92,11 @@ async def _list():
         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:
@@ -105,7 +105,7 @@ async def _list():
         )
         console.print(
             "[dim]Choose one of these matching workspaces by slug:\n"
-            f"{_workspace_selection_choices(matches)}[/dim]"
+            f"{format_workspace_selection_choices(matches)}[/dim]"
         )
         raise typer.Exit(1)
 

src/basic_memory/cli/commands/project.py

@@ -34,6 +34,9 @@
     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
@@ -284,27 +287,24 @@ def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:
 def _resolve_workspace_id(config, workspace: str | None) -> str | None:
     """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,
-        _workspace_selection_choices,
         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 '{workspace}' matches multiple workspaces.[/red]"
             )
             console.print(
                 "[dim]Choose one of these matching workspaces by slug:\n"
-                f"{_workspace_selection_choices(matches)}[/dim]"
+                f"{format_workspace_selection_choices(matches)}[/dim]"
             )
             raise typer.Exit(1)
         return matches[0].tenant_id
@@ -347,6 +347,8 @@ async def _list_projects(ws: str | None = None):
 
     try:
         config = ConfigManager().config
+        workspace_filter = workspace
+        workspace_filter_requested = workspace_filter is not None
 
         local_result: ProjectList | None = None
         cloud_results: list[tuple[WorkspaceInfo | None, ProjectList]] = []
@@ -355,47 +357,55 @@ async def _list_projects(ws: str | None = None):
         cloud_workspace_error: Exception | None = None
         failed_cloud_workspaces: list[tuple[WorkspaceInfo, Exception]] = []
 
-        def _fetch_cloud_workspace_results() -> list[tuple[WorkspaceInfo | None, ProjectList]]:
-            nonlocal available_cloud_workspaces, cloud_workspace_error
-
+        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 (
-                _workspace_choices,
-                _workspace_matches_identifier,
-                _workspace_selection_choices,
                 get_available_workspaces,
             )
 
             try:
                 workspaces = run_with_cleanup(get_available_workspaces())
-                available_cloud_workspaces = workspaces
             except Exception as exc:
-                cloud_workspace_error = exc
-                fallback_workspace = workspace or config.default_workspace
-                return [(None, run_with_cleanup(_list_projects(fallback_workspace)))]
+                fallback_workspace = workspace_filter or config.default_workspace
+                return (
+                    [(None, run_with_cleanup(_list_projects(fallback_workspace)))],
+                    [],
+                    exc,
+                    [],
+                )
 
             selected_workspaces = workspaces
-            if workspace is not None:
-                matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
+            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}' not found[/red]")
+                    console.print(f"[red]Error: Workspace '{workspace_filter}' 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 '{workspace}' matches multiple workspaces.[/red]"
+                        f"[red]Error: Workspace '{workspace_filter}' matches multiple workspaces.[/red]"
                     )
                     console.print(
                         "[dim]Choose one of these matching workspaces by slug:\n"
-                        f"{_workspace_selection_choices(matches)}[/dim]"
+                        f"{format_workspace_selection_choices(matches)}[/dim]"
                     )
                     raise typer.Exit(1)
                 selected_workspaces = matches
 
             if not selected_workspaces:
-                return []
+                return [], workspaces, None, []
 
             results: list[tuple[WorkspaceInfo | None, ProjectList]] = []
+            failed_workspaces: list[tuple[WorkspaceInfo, Exception]] = []
             for cloud_workspace in selected_workspaces:
                 try:
                     results.append(
@@ -405,17 +415,22 @@ def _fetch_cloud_workspace_results() -> list[tuple[WorkspaceInfo | None, Project
                         )
                     )
                 except Exception as exc:
-                    failed_cloud_workspaces.append((cloud_workspace, exc))
+                    failed_workspaces.append((cloud_workspace, exc))
 
-            if not results and failed_cloud_workspaces:
-                raise failed_cloud_workspaces[0][1]
+            if not results and failed_workspaces:
+                raise failed_workspaces[0][1]
 
-            return results
+            return results, workspaces, None, failed_workspaces
 
         if cloud:
             with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                 with force_routing(cloud=True):
-                    cloud_results = _fetch_cloud_workspace_results()
+                    (
+                        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())
@@ -428,7 +443,12 @@ def _fetch_cloud_workspace_results() -> list[tuple[WorkspaceInfo | None, Project
                 try:
                     with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                         with force_routing(cloud=True):
-                            cloud_results = _fetch_cloud_workspace_results()
+                            (
+                                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
@@ -491,7 +511,7 @@ def _workspace_priority(row_key: tuple[str | None, str]) -> tuple[bool, int, str
 
         def _select_attached_row_key(
             permalink: str, entry: ProjectEntry | None
-        ) -> tuple[str | None, str]:
+        ) -> 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:
@@ -517,8 +537,10 @@ def _select_attached_row_key(
                     if row_key[0] == workspace_id:
                         return row_key
 
-            if workspace is not None and preferred_workspace_ids:
-                return (preferred_workspace_ids[0], permalink)
+            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
@@ -534,7 +556,7 @@ def _select_attached_row_key(
 
             return sorted(cloud_keys, key=_workspace_priority)[0]
 
-        attached_row_by_permalink: dict[str, tuple[str | None, str]] = {}
+        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)
@@ -633,7 +655,7 @@ def _select_attached_row_key(
             if display_name:
                 row_data["display_name"] = display_name
             if cloud_project is not None and cloud_ws_name:
-                row_data["workspace"] = cloud_ws_name or ""
+                row_data["workspace"] = cloud_ws_name
                 if cloud_ws_type:
                     row_data["workspace_type"] = cloud_ws_type
 

src/basic_memory/mcp/project_context.py

@@ -11,7 +11,7 @@
 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,13 @@
 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_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,45 +305,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, name, or type."""
-    if workspace.tenant_id == identifier:
-        return True
-    if workspace.slug.casefold() == identifier.casefold():
-        return True
-    if workspace.workspace_type.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_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
-        ]
-    )
-
-
 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):
@@ -640,7 +607,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)
 
@@ -869,7 +836,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:
@@ -1003,7 +970,7 @@ 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_identifier(cached_workspace, workspace):
                     logger.debug(
                         f"Using cached workspace from context: {cached_workspace.tenant_id}"
                     )
@@ -1020,18 +987,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 '{workspace}' matches multiple workspaces. "
                     "Choose one of these matching workspaces by slug or tenant_id:\n"
-                    f"{_workspace_selection_choices(matches)}"
+                    f"{format_workspace_selection_choices(matches)}"
                 )
             selected_workspace = matches[0]
         elif len(workspaces) == 1:
@@ -1040,7 +1007,7 @@ async def resolve_workspace_parameter(
             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/slug/type.\n"
-                f"Available workspaces:\n{_workspace_choices(workspaces)}"
+                f"Available workspaces:\n{format_workspace_choices(workspaces)}"
             )
 
         await _set_cached_active_workspace(context, selected_workspace)