feat(cli): support cloud project visibility on add

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

@@ -7,6 +7,7 @@
     CloudProjectList,
     CloudProjectCreateRequest,
     CloudProjectCreateResponse,
+    ProjectVisibility,
 )
 from basic_memory.utils import generate_permalink
 
@@ -63,13 +64,15 @@ async def create_cloud_project(
     project_name: str,
     *,
     workspace: str | None = None,
+    visibility: ProjectVisibility = "workspace",
     api_request=make_api_request,
 ) -> CloudProjectCreateResponse:
     """Create a new project on cloud.
 
     Args:
         project_name: Name of project to create
         workspace: Optional workspace override for tenant-scoped project creation
+        visibility: Visibility for the created cloud project
 
     Returns:
         CloudProjectCreateResponse with project details from API
@@ -86,6 +89,7 @@ async def create_cloud_project(
             name=project_name,
             path=project_path,
             set_default=False,
+            visibility=visibility,
         )
 
         response = await api_request(

src/basic_memory/cli/commands/project.py

@@ -4,6 +4,7 @@
 import os
 from datetime import datetime
 from pathlib import Path
+from typing import cast
 
 import typer
 from rich.console import Console, Group
@@ -27,6 +28,7 @@
 from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
 from basic_memory.mcp.async_client import get_client
 from basic_memory.mcp.clients import ProjectClient
+from basic_memory.schemas.cloud import ProjectVisibility
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.utils import generate_permalink, normalize_project_path
 
@@ -56,6 +58,57 @@ def make_bar(value: int, max_value: int, width: int = 40) -> Text:
     return bar
 
 
+def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:
+    """Normalize CLI visibility input to the cloud API contract."""
+    if visibility is None:
+        return "workspace"
+
+    normalized = visibility.strip().lower()
+    if normalized in {"workspace", "shared", "private"}:
+        return cast(ProjectVisibility, normalized)
+
+    raise ValueError("Invalid visibility. Expected one of: workspace, shared, private.")
+
+
+def _resolve_workspace_id(config, workspace: str | None) -> str | None:
+    """Resolve a workspace name 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)]
+        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]")
+            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]"
+            )
+            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+            raise typer.Exit(1)
+        return matches[0].tenant_id
+
+    if config.default_workspace:
+        return config.default_workspace
+
+    try:
+        workspaces = run_with_cleanup(get_available_workspaces())
+        if len(workspaces) == 1:
+            return workspaces[0].tenant_id
+    except Exception:
+        # Workspace resolution is optional until a command needs a specific tenant.
+        pass
+
+    return None
+
+
 @project_app.command("list")
 def list_projects(
     local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
@@ -257,6 +310,16 @@ def add_project(
     local_path: str = typer.Option(
         None, "--local-path", help="Local sync path for cloud mode (optional)"
     ),
+    workspace: str = typer.Option(
+        None,
+        "--workspace",
+        help="Cloud workspace name or tenant_id (cloud mode only)",
+    ),
+    visibility: str = typer.Option(
+        None,
+        "--visibility",
+        help="Cloud project visibility: workspace, shared, or private",
+    ),
     set_default: bool = typer.Option(False, "--default", help="Set as default project"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -271,6 +334,8 @@ def add_project(
     Cloud mode examples:\n
         bm project add research                           # No local sync\n
         bm project add research --local-path ~/docs       # With local sync\n
+        bm project add research --cloud --visibility shared\n
+        bm project add research --cloud --workspace Personal --visibility shared\n
 
     Local mode example:\n
         bm project add research ~/Documents/research
@@ -285,6 +350,7 @@ def add_project(
 
     # Determine effective mode: default local, cloud only when explicitly requested.
     effective_cloud_mode = cloud and not local
+    resolved_workspace_id: str | None = None
 
     # Resolve local sync path early (needed for both cloud and local mode)
     local_sync_path: str | None = None
@@ -293,18 +359,31 @@ def add_project(
 
     if effective_cloud_mode:
         _require_cloud_credentials(config)
+        try:
+            resolved_visibility = _normalize_project_visibility(visibility)
+        except ValueError as e:
+            console.print(f"[red]Error: {e}[/red]")
+            raise typer.Exit(1)
+        resolved_workspace_id = _resolve_workspace_id(config, workspace)
         # Cloud mode: path auto-generated from name, local sync is optional
 
         async def _add_project():
-            async with get_client() as client:
+            async with get_client(workspace=resolved_workspace_id) as client:
                 data = {
                     "name": name,
                     "path": generate_permalink(name),
                     "local_sync_path": local_sync_path,
                     "set_default": set_default,
+                    "visibility": resolved_visibility,
                 }
                 return await ProjectClient(client).create_project(data)
     else:
+        if workspace is not None:
+            console.print("[red]Error: --workspace is only supported in cloud mode[/red]")
+            raise typer.Exit(1)
+        if visibility is not None:
+            console.print("[red]Error: --visibility is only supported in cloud mode[/red]")
+            raise typer.Exit(1)
         # Local mode: path is required
         if path is None:
             console.print("[red]Error: path argument is required in local mode[/red]")
@@ -323,25 +402,34 @@ async def _add_project():
             result = run_with_cleanup(_add_project())
         console.print(f"[green]{result.message}[/green]")
 
-        # Save local sync path to config if in cloud mode
-        if effective_cloud_mode and local_sync_path:
-            # Create local directory if it doesn't exist
-            local_dir = Path(local_sync_path)
-            local_dir.mkdir(parents=True, exist_ok=True)
-
-            # Update project entry — path is always the local directory
+        # Trigger: local config needs enough metadata to route future commands back to cloud.
+        # Why: explicit workspace selection and local sync state should persist across CLI sessions.
+        # Outcome: cloud-backed projects keep cloud mode, workspace_id, and optional local sync path.
+        if effective_cloud_mode and (local_sync_path or resolved_workspace_id):
             entry = config.projects.get(name)
             if entry:
-                entry.path = local_sync_path
-                entry.local_sync_path = local_sync_path
+                entry.mode = ProjectMode.CLOUD
+                if local_sync_path:
+                    entry.path = local_sync_path
+                    entry.local_sync_path = local_sync_path
+                if resolved_workspace_id:
+                    entry.workspace_id = resolved_workspace_id
             else:
                 # Project may not be in local config yet (cloud-only add)
                 config.projects[name] = ProjectEntry(
-                    path=local_sync_path,
+                    path=local_sync_path or "",
+                    mode=ProjectMode.CLOUD,
                     local_sync_path=local_sync_path,
+                    workspace_id=resolved_workspace_id,
                 )
             ConfigManager().save_config(config)
 
+        # Save local sync path to config if in cloud mode
+        if effective_cloud_mode and local_sync_path:
+            # Create local directory if it doesn't exist
+            local_dir = Path(local_sync_path)
+            local_dir.mkdir(parents=True, exist_ok=True)
+
             console.print(f"\n[green]Local sync path configured: {local_sync_path}[/green]")
             console.print("\nNext steps:")
             console.print(f"  1. Preview: bm cloud bisync --name {name} --resync --dry-run")
@@ -575,45 +663,7 @@ def set_cloud(
         console.print("[dim]Run 'bm cloud api-key save <key>' or 'bm cloud login' first[/dim]")
         raise typer.Exit(1)
 
-    # --- Resolve workspace to tenant_id ---
-    resolved_workspace_id: str | None = None
-
-    if workspace is not None:
-        # Explicit --workspace: resolve to tenant_id via cloud lookup
-        from basic_memory.mcp.project_context import (
-            get_available_workspaces,
-            _workspace_matches_identifier,
-            _workspace_choices,
-        )
-
-        workspaces = run_with_cleanup(get_available_workspaces())
-        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]")
-            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]"
-            )
-            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
-            raise typer.Exit(1)
-        resolved_workspace_id = matches[0].tenant_id
-    elif config.default_workspace:
-        # Fall back to global default
-        resolved_workspace_id = config.default_workspace
-    else:
-        # Try auto-select if single workspace
-        try:
-            from basic_memory.mcp.project_context import get_available_workspaces
-
-            workspaces = run_with_cleanup(get_available_workspaces())
-            if len(workspaces) == 1:
-                resolved_workspace_id = workspaces[0].tenant_id
-        except Exception:
-            pass  # Workspace resolution is optional at set-cloud time
+    resolved_workspace_id = _resolve_workspace_id(config, workspace)
 
     config.set_project_mode(name, ProjectMode.CLOUD)
     if resolved_workspace_id:

src/basic_memory/schemas/cloud.py

@@ -1,7 +1,11 @@
 """Schemas for cloud-related API responses."""
 
+from typing import Literal
+
 from pydantic import BaseModel, Field
 
+type ProjectVisibility = Literal["workspace", "shared", "private"]
+
 
 class TenantMountInfo(BaseModel):
     """Response from /tenant/mount/info endpoint."""
@@ -36,6 +40,10 @@ class CloudProjectCreateRequest(BaseModel):
     name: str = Field(..., description="Project name")
     path: str = Field(..., description="Project path (permalink)")
     set_default: bool = Field(default=False, description="Set as default project")
+    visibility: ProjectVisibility = Field(
+        default="workspace",
+        description="Project visibility for team workspaces",
+    )
 
 
 class CloudProjectCreateResponse(BaseModel):

tests/cli/cloud/test_cloud_api_client_and_utils.py

@@ -165,6 +165,45 @@ async def api_request(**kwargs):
     assert created.new_project["name"] == "My Project"
     # Path should be permalink-like (kebab)
     assert seen["create_payload"]["path"] == "my-project"
+    assert seen["create_payload"]["visibility"] == "workspace"
+
+
+@pytest.mark.asyncio
+async def test_create_cloud_project_accepts_visibility_override(config_home, config_manager):
+    """Shared cloud helper should pass explicit visibility through to the API payload."""
+    config = config_manager.load_config()
+    config.cloud_host = "https://cloud.example.test"
+    config_manager.save_config(config)
+
+    seen_payload: dict | None = None
+
+    async def api_request(**kwargs):
+        nonlocal seen_payload
+        seen_payload = kwargs["json_data"]
+        return httpx.Response(
+            200,
+            json={
+                "message": "created",
+                "status": "success",
+                "default": False,
+                "old_project": None,
+                "new_project": {"name": "shared-project", "path": "shared-project"},
+            },
+        )
+
+    created = await create_cloud_project(
+        "Shared Project",
+        visibility="shared",
+        api_request=api_request,
+    )
+
+    assert created.new_project is not None
+    assert seen_payload == {
+        "name": "Shared Project",
+        "path": "shared-project",
+        "set_default": False,
+        "visibility": "shared",
+    }
 
 
 @pytest.mark.asyncio