fix(cli): fetch projects from all workspaces and populate Workspace column

Fixes two related bugs in `bm project list` (issue #820):

1. Workspace column was always empty when no `--workspace` flag or
   `default_workspace` config was set. The workspace name was only
   resolved when `effective_workspace` was truthy, so the column
   stayed blank for every row in the common case.

2. Only the single default/specified workspace was queried. Projects
   in any other workspace the user had access to were silently omitted.

The fix introduces `_fetch_cloud_workspace_results()`, an inner helper
that calls `get_available_workspaces()` and iterates every reachable
workspace (or just the specified one if `--workspace` / `default_workspace`
is set). Each fetch returns a `(WorkspaceInfo, ProjectList)` pair so the
workspace name and type are available directly when building table rows,
without a separate resolution pass. Graceful fallback is preserved: if
`get_available_workspaces()` fails, an un-scoped fetch is used and the
Workspace column is empty rather than crashing.

Adds two regression tests:
- workspace column populated without `default_workspace` set
- projects from all workspaces appear in the listing

Co-authored-by: Drew Cain <groksrc@users.noreply.github.com>
Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>

src/basic_memory/cli/commands/project.py

@@ -29,6 +29,7 @@ from basic_memory.cli.commands.routing import force_routing, validate_routing_fl
 from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectEntry, ProjectMode
 from basic_memory.mcp.async_client import get_client, resolve_configured_workspace
 from basic_memory.mcp.clients import ProjectClient
+from basic_memory.mcp.project_context import get_available_workspaces
 from basic_memory.schemas.cloud import (
     CloudProjectIndexStatus,
     CloudTenantIndexStatusResponse,
@@ -343,13 +344,66 @@ def list_projects(
         effective_workspace = workspace or config.default_workspace
 
         local_result: ProjectList | None = None
-        cloud_result: ProjectList | None = None
+        # cloud_workspace_results: list of (WorkspaceInfo | None, ProjectList) pairs.
+        # Each pair carries the workspace object so rows can be labelled correctly.
+        cloud_workspace_results: list[tuple] = []
         cloud_error: Exception | None = None
 
+        def _fetch_cloud_workspace_results() -> list[tuple]:
+            """Return (WorkspaceInfo | None, ProjectList) pairs for all reachable workspaces.
+
+            Trigger: cloud credentials are present and projects need cloud augmentation.
+            Why: fetching from every available workspace prevents silently omitting projects
+                 in workspaces other than the default, and ensures the Workspace column is
+                 populated for every cloud-sourced row.
+            Outcome: each (ws, result) pair carries the workspace object for display labeling.
+            """
+            try:
+                all_workspaces = run_with_cleanup(get_available_workspaces())
+            except Exception:
+                all_workspaces = []
+
+            if effective_workspace:
+                # Trigger: user requested a specific workspace (--workspace / default_workspace).
+                # Why: honour the explicit filter; don't silently expand to other workspaces.
+                # Outcome: only the selected workspace's projects appear.
+                target = [ws for ws in all_workspaces if ws.tenant_id == effective_workspace]
+                if not target:
+                    # Not in available list; try the raw ID anyway (best-effort).
+                    result = run_with_cleanup(_list_projects(effective_workspace))
+                    return [(None, result)] if result else []
+                pairs: list[tuple] = []
+                for ws in target:
+                    try:
+                        result = run_with_cleanup(_list_projects(ws.tenant_id))
+                        if result:
+                            pairs.append((ws, result))
+                    except Exception:
+                        pass
+                return pairs
+
+            if all_workspaces:
+                # Trigger: no workspace filter — fetch from every available workspace.
+                # Why: ensures cross-workspace projects are not silently omitted.
+                # Outcome: projects from all workspaces appear in the listing.
+                pairs = []
+                for ws in all_workspaces:
+                    try:
+                        result = run_with_cleanup(_list_projects(ws.tenant_id))
+                        if result:
+                            pairs.append((ws, result))
+                    except Exception:
+                        pass
+                return pairs
+
+            # No workspaces found; fall back to un-scoped fetch.
+            result = run_with_cleanup(_list_projects(None))
+            return [(None, result)] if result else []
+
         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_workspace_results = _fetch_cloud_workspace_results()
         elif local:
             with force_routing(local=True):
                 local_result = run_with_cleanup(_list_projects())
@@ -362,29 +416,10 @@ 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_workspace_results = _fetch_cloud_workspace_results()
                 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,97 +430,129 @@ def list_projects(
         table.add_column("Sync", style="green")
         table.add_column("Default", style="magenta")
 
-        project_names_by_permalink: dict[str, str] = {}
+        # --- Build lookup structures ---
         local_projects_by_permalink: dict[str, ProjectItem] = {}
-        cloud_projects_by_permalink: dict[str, ProjectItem] = {}
-
         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 project in cloud_result.projects:
+        # cloud_entries_by_permalink: permalink → list of (WorkspaceInfo | None, ProjectItem).
+        # Multiple workspaces can contribute the same permalink — each produces its own row.
+        cloud_entries_by_permalink: dict[str, list[tuple]] = {}
+        for ws_obj, ws_result in cloud_workspace_results:
+            for project in ws_result.projects:
                 permalink = generate_permalink(project.name)
-                project_names_by_permalink[permalink] = project.name
-                cloud_projects_by_permalink[permalink] = project
+                cloud_entries_by_permalink.setdefault(permalink, []).append((ws_obj, project))
 
         # --- Build unified project list ---
+        # One row per (workspace, permalink) pair for cloud entries.
+        # Local-only projects get a single row after all cloud rows.
         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)
+        covered_local_permalinks: set[str] = set()
 
-            local_path = ""
-            if local_project is not None:
-                local_path = format_path(normalize_project_path(local_project.path))
-            elif entry and entry.local_sync_path:
-                local_path = format_path(entry.local_sync_path)
-            elif entry and entry.mode == ProjectMode.LOCAL and entry.path:
-                local_path = format_path(normalize_project_path(entry.path))
+        for permalink, entries in sorted(cloud_entries_by_permalink.items()):
+            for ws_obj, cloud_project in entries:
+                local_project = local_projects_by_permalink.get(permalink)
+                if local_project is not None:
+                    covered_local_permalinks.add(permalink)
 
-            # Clear local path for cloud-mode projects — only local projects
-            # should display a local path
+                entry = config.projects.get(cloud_project.name)
+
+                local_path = ""
+                if local_project is not None:
+                    local_path = format_path(normalize_project_path(local_project.path))
+                elif entry and entry.local_sync_path:
+                    local_path = format_path(entry.local_sync_path)
+                elif entry and entry.mode == ProjectMode.LOCAL and entry.path:
+                    local_path = format_path(normalize_project_path(entry.path))
+
+                # Clear local path for cloud-mode projects — only local projects
+                # should display a local path.
+                if entry and entry.mode == ProjectMode.CLOUD:
+                    local_path = ""
+
+                cloud_path = normalize_project_path(cloud_project.path)
+
+                if local:
+                    cli_route = "local (flag)"
+                elif cloud:
+                    cli_route = "cloud (flag)"
+                elif entry:
+                    cli_route = entry.mode.value
+                else:
+                    cli_route = ProjectMode.CLOUD.value
+
+                is_default = config.default_project == cloud_project.name
+                has_sync = bool(entry and entry.local_sync_path)
+
+                # Determine MCP transport based on project routing mode.
+                if entry and entry.mode == ProjectMode.CLOUD:
+                    mcp_transport = "https"
+                elif entry is None:
+                    mcp_transport = "https"
+                else:
+                    mcp_transport = "stdio"
+
+                # 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;
+                # the Rich table uses display_name when available.
+                display_name = cloud_project.display_name or None
+
+                row_data: dict = {
+                    "name": cloud_project.name,
+                    "permalink": permalink,
+                    "local_path": local_path,
+                    "cloud_path": cloud_path,
+                    "cli_route": cli_route,
+                    "mcp_stdio": mcp_transport,
+                    "sync": has_sync,
+                    "is_default": is_default,
+                }
+                if display_name:
+                    row_data["display_name"] = display_name
+                if ws_obj is not None:
+                    row_data["workspace"] = ws_obj.name
+                    if ws_obj.workspace_type:
+                        row_data["workspace_type"] = ws_obj.workspace_type
+
+                project_rows.append(row_data)
+
+        # Add local-only projects (not present in any cloud workspace result).
+        for permalink in sorted(local_projects_by_permalink):
+            if permalink in covered_local_permalinks:
+                continue
+            local_project = local_projects_by_permalink[permalink]
+            entry = config.projects.get(local_project.name)
+
+            local_path = format_path(normalize_project_path(local_project.path))
+            # Clear local path for cloud-mode projects.
             if entry and entry.mode == ProjectMode.CLOUD:
                 local_path = ""
 
-            cloud_path = ""
-            if cloud_project is not None:
-                cloud_path = normalize_project_path(cloud_project.path)
-
             if local:
                 cli_route = "local (flag)"
             elif cloud:
                 cli_route = "cloud (flag)"
             elif entry:
                 cli_route = entry.mode.value
-            elif cloud_project is not None and local_project is None:
-                cli_route = ProjectMode.CLOUD.value
             else:
                 cli_route = ProjectMode.LOCAL.value
 
-            is_default = config.default_project == project_name
-
+            is_default = config.default_project == local_project.name
             has_sync = bool(entry and entry.local_sync_path)
-            # Determine MCP transport based on project routing mode
-            if entry and entry.mode == ProjectMode.CLOUD:
-                mcp_transport = "https"
-            elif entry is None and cloud_project is not None:
-                mcp_transport = "https"
-            else:
-                mcp_transport = "stdio"
+            mcp_transport = "https" if (entry and entry.mode == ProjectMode.CLOUD) else "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
-
-            # 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;
-            # the Rich table uses display_name when available.
-            display_name = (
-                cloud_project.display_name if cloud_project and cloud_project.display_name else None
-            )
             row_data = {
-                "name": project_name,
+                "name": local_project.name,
                 "permalink": permalink,
                 "local_path": local_path,
-                "cloud_path": cloud_path,
+                "cloud_path": "",
                 "cli_route": cli_route,
                 "mcp_stdio": mcp_transport,
                 "sync": has_sync,
                 "is_default": is_default,
             }
-            if display_name:
-                row_data["display_name"] = display_name
-            if ws_label:
-                row_data["workspace"] = cloud_ws_name or ""
-                if cloud_ws_type:
-                    row_data["workspace_type"] = cloud_ws_type
-
             project_rows.append(row_data)
 
         # --- JSON output ---

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.
@@ -44,13 +45,17 @@ def write_config(tmp_path, monkeypatch):
 
 @pytest.fixture
 def mock_client(monkeypatch):
-    """Mock get_client with a no-op async context manager."""
+    """Mock get_client and get_available_workspaces with no-ops for project list tests."""
 
     @asynccontextmanager
     async def fake_get_client(workspace=None):
         yield object()
 
+    async def fake_get_available_workspaces():
+        return []
+
     monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(project_cmd, "get_available_workspaces", fake_get_available_workspaces)
 
 
 def test_project_list_shows_local_cloud_presence_and_routes(
@@ -357,3 +362,155 @@ def test_project_ls_cloud_route_uses_cloud_listing(
     assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
     assert "Files in alpha (CLOUD)" in result.stdout
     assert "cloud.md" in result.stdout
+
+
+# --- Bug #820 regression tests ---
+
+
+def test_project_list_workspace_column_populated_without_explicit_workspace(
+    runner: CliRunner, write_config, tmp_path, monkeypatch
+):
+    """Workspace column should be populated even when no default_workspace is set.
+
+    Bug: the column was always empty because workspace resolution was gated on
+    ``effective_workspace`` being truthy. With no ``--workspace`` flag and no
+    ``default_workspace`` in config, the column showed empty for every row.
+    """
+    ws = WorkspaceInfo(
+        tenant_id="ws-personal-111",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+
+    async def fake_get_available_workspaces():
+        return [ws]
+
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": "main",
+            "cloud_api_key": "bmc_test_key_123",
+            # Intentionally no "default_workspace" key — this triggers Bug 1.
+        }
+    )
+
+    local_payload = {
+        "projects": [
+            {"id": 1, "external_id": "11111111-1111-1111-1111-111111111111", "name": "main", "path": "/main", "is_default": True}
+        ],
+        "default_project": "main",
+    }
+    cloud_payload = {
+        "projects": [
+            {"id": 1, "external_id": "11111111-1111-1111-1111-111111111111", "name": "main", "path": "/main", "is_default": True}
+        ],
+        "default_project": "main",
+    }
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        yield object()
+
+    async def fake_list_projects(self):
+        if os.getenv("BASIC_MEMORY_FORCE_CLOUD", "").lower() in ("true", "1", "yes"):
+            return ProjectList.model_validate(cloud_payload)
+        return ProjectList.model_validate(local_payload)
+
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(project_cmd, "get_available_workspaces", fake_get_available_workspaces)
+    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}"
+    # The Workspace column must show the workspace name for the cloud-sourced project.
+    assert "Personal" in result.stdout
+
+
+def test_project_list_fetches_from_all_workspaces(
+    runner: CliRunner, write_config, tmp_path, monkeypatch
+):
+    """Projects from every cloud workspace should appear in the listing.
+
+    Bug: only the single default/specified workspace was queried; projects in
+    any other workspace were silently omitted.
+    """
+    ws_a = WorkspaceInfo(
+        tenant_id="ws-aaa",
+        workspace_type="personal",
+        slug="personal",
+        name="Personal",
+        role="owner",
+        is_default=True,
+    )
+    ws_b = WorkspaceInfo(
+        tenant_id="ws-bbb",
+        workspace_type="organization",
+        slug="my-org",
+        name="My Org",
+        role="admin",
+        is_default=False,
+    )
+
+    async def fake_get_available_workspaces():
+        return [ws_a, ws_b]
+
+    write_config(
+        {
+            "env": "dev",
+            "projects": {},
+            "default_project": None,
+            "cloud_api_key": "bmc_test_key_123",
+        }
+    )
+
+    local_payload = {"projects": [], "default_project": None}
+
+    ws_a_payload = {
+        "projects": [
+            {"id": 1, "external_id": "aaaa-1111-1111-1111-111111111111", "name": "personal-notes", "path": "/personal-notes", "is_default": False}
+        ],
+        "default_project": None,
+    }
+    ws_b_payload = {
+        "projects": [
+            {"id": 2, "external_id": "bbbb-2222-2222-2222-222222222222", "name": "org-docs", "path": "/org-docs", "is_default": False}
+        ],
+        "default_project": None,
+    }
+
+    # Track which workspace get_client was last called with so list_projects can
+    # return the correct per-workspace payload.
+    current_workspace: list[str | None] = [None]
+
+    @asynccontextmanager
+    async def fake_get_client(workspace=None):
+        current_workspace[0] = workspace
+        yield object()
+
+    async def fake_list_projects(self):
+        if os.getenv("BASIC_MEMORY_FORCE_CLOUD", "").lower() not in ("true", "1", "yes"):
+            return ProjectList.model_validate(local_payload)
+        if current_workspace[0] == "ws-aaa":
+            return ProjectList.model_validate(ws_a_payload)
+        if current_workspace[0] == "ws-bbb":
+            return ProjectList.model_validate(ws_b_payload)
+        return ProjectList.model_validate({"projects": [], "default_project": None})
+
+    monkeypatch.setattr(project_cmd, "get_client", fake_get_client)
+    monkeypatch.setattr(project_cmd, "get_available_workspaces", fake_get_available_workspaces)
+    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}"
+    # Projects from both workspaces must appear.
+    assert "personal-notes" in result.stdout
+    assert "org-docs" in result.stdout
+    # Each row must show its workspace name.
+    assert "Personal" in result.stdout
+    assert "My Org" in result.stdout