Merge branch 'feature/spec-20-simplified-rclone-sync' into claude/issue-406-20251031-0133

Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>

justfile

@@ -2,7 +2,7 @@
 
 # Install dependencies
 install:
-    pip install -e ".[dev]"
+    uv pip install -e ".[dev]"
     uv sync
     @echo ""
     @echo "💡 Remember to activate the virtual environment by running: source .venv/bin/activate"

src/basic_memory/api/routers/project_router.py

@@ -27,6 +27,24 @@ project_router = APIRouter(prefix="/project", tags=["project"])
 project_resource_router = APIRouter(prefix="/projects", tags=["project_management"])
 
 
+def normalize_project_path(path: str) -> str:
+    """Normalize project path by stripping mount point prefix.
+
+    In cloud deployments, the S3 bucket is mounted at /app/data. We strip this
+    prefix from project paths to avoid leaking implementation details and to
+    ensure paths match the actual S3 bucket structure.
+
+    Args:
+        path: Project path (e.g., "/app/data/basic-memory-llc")
+
+    Returns:
+        Normalized path (e.g., "/basic-memory-llc")
+    """
+    if path.startswith("/app/data/"):
+        return path.removeprefix("/app/data")
+    return path
+
+
 @project_router.get("/info", response_model=ProjectInfoResponse)
 async def get_project_info(
     project_service: ProjectServiceDep,
@@ -50,7 +68,7 @@ async def get_project(
 
     return ProjectItem(
         name=found_project.name,
-        path=found_project.path,
+        path=normalize_project_path(found_project.path),
         is_default=found_project.is_default or False,
     )
 
@@ -167,7 +185,7 @@ async def list_projects(
     project_items = [
         ProjectItem(
             name=project.name,
-            path=project.path,
+            path=normalize_project_path(project.path),
             is_default=project.is_default or False,
         )
         for project in projects

src/basic_memory/cli/commands/__init__.py

@@ -1,11 +1,10 @@
 """CLI commands for basic-memory."""
 
-from . import status, sync, db, import_memory_json, mcp, import_claude_conversations
+from . import status, db, import_memory_json, mcp, import_claude_conversations
 from . import import_claude_projects, import_chatgpt, tool, project
 
 __all__ = [
     "status",
-    "sync",
     "db",
     "import_memory_json",
     "mcp",

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

@@ -1,33 +1,11 @@
-"""Cloud bisync commands for Basic Memory CLI."""
+"""Cloud bisync utility functions for Basic Memory CLI."""
 
-import asyncio
-import subprocess
-import time
-from datetime import datetime
 from pathlib import Path
-from typing import Optional
 
-import typer
-from rich.console import Console
-from rich.table import Table
-
-from basic_memory.cli.commands.cloud.api_client import CloudAPIError, make_api_request
-from basic_memory.cli.commands.cloud.cloud_utils import (
-    create_cloud_project,
-    fetch_cloud_projects,
-)
-from basic_memory.cli.commands.cloud.rclone_config import (
-    add_tenant_to_rclone_config,
-)
-from basic_memory.cli.commands.cloud.rclone_installer import RcloneInstallError, install_rclone
+from basic_memory.cli.commands.cloud.api_client import make_api_request
 from basic_memory.config import ConfigManager
-from basic_memory.ignore_utils import get_bmignore_path, create_default_bmignore
-from basic_memory.schemas.cloud import (
-    TenantMountInfo,
-    MountCredentials,
-)
-
-console = Console()
+from basic_memory.ignore_utils import create_default_bmignore, get_bmignore_path
+from basic_memory.schemas.cloud import MountCredentials, TenantMountInfo
 
 
 class BisyncError(Exception):
@@ -36,52 +14,6 @@ class BisyncError(Exception):
     pass
 
 
-class RcloneBisyncProfile:
-    """Bisync profile with safety settings."""
-
-    def __init__(
-        self,
-        name: str,
-        conflict_resolve: str,
-        max_delete: int,
-        check_access: bool,
-        description: str,
-        extra_args: Optional[list[str]] = None,
-    ):
-        self.name = name
-        self.conflict_resolve = conflict_resolve
-        self.max_delete = max_delete
-        self.check_access = check_access
-        self.description = description
-        self.extra_args = extra_args or []
-
-
-# Bisync profiles based on SPEC-9 Phase 2.1
-BISYNC_PROFILES = {
-    "safe": RcloneBisyncProfile(
-        name="safe",
-        conflict_resolve="none",
-        max_delete=10,
-        check_access=False,
-        description="Safe mode with conflict preservation (keeps both versions)",
-    ),
-    "balanced": RcloneBisyncProfile(
-        name="balanced",
-        conflict_resolve="newer",
-        max_delete=25,
-        check_access=False,
-        description="Balanced mode - auto-resolve to newer file (recommended)",
-    ),
-    "fast": RcloneBisyncProfile(
-        name="fast",
-        conflict_resolve="newer",
-        max_delete=50,
-        check_access=False,
-        description="Fast mode for rapid iteration (skip verification)",
-    ),
-}
-
-
 async def get_mount_info() -> TenantMountInfo:
     """Get current tenant information from cloud API."""
     try:
@@ -110,75 +42,6 @@ async def generate_mount_credentials(tenant_id: str) -> MountCredentials:
         raise BisyncError(f"Failed to generate credentials: {e}") from e
 
 
-def scan_local_directories(sync_dir: Path) -> list[str]:
-    """Scan local sync directory for project folders.
-
-    Args:
-        sync_dir: Path to bisync directory
-
-    Returns:
-        List of directory names (project names)
-    """
-    if not sync_dir.exists():
-        return []
-
-    directories = []
-    for item in sync_dir.iterdir():
-        if item.is_dir() and not item.name.startswith("."):
-            directories.append(item.name)
-
-    return directories
-
-
-def get_bisync_state_path(tenant_id: str) -> Path:
-    """Get path to bisync state directory."""
-    return Path.home() / ".basic-memory" / "bisync-state" / tenant_id
-
-
-def get_bisync_directory() -> Path:
-    """Get bisync directory from config.
-
-    Returns:
-        Path to bisync directory (default: ~/basic-memory-cloud-sync)
-    """
-    config_manager = ConfigManager()
-    config = config_manager.config
-
-    sync_dir = config.bisync_config.get("sync_dir", str(Path.home() / "basic-memory-cloud-sync"))
-    return Path(sync_dir).expanduser().resolve()
-
-
-def validate_bisync_directory(bisync_dir: Path) -> None:
-    """Validate bisync directory doesn't conflict with mount.
-
-    Raises:
-        BisyncError: If bisync directory conflicts with mount directory
-    """
-    # Get fixed mount directory
-    mount_dir = (Path.home() / "basic-memory-cloud").resolve()
-
-    # Check if bisync dir is the same as mount dir
-    if bisync_dir == mount_dir:
-        raise BisyncError(
-            f"Cannot use {bisync_dir} for bisync - it's the mount directory!\n"
-            f"Mount and bisync must use different directories.\n\n"
-            f"Options:\n"
-            f"  1. Use default: ~/basic-memory-cloud-sync/\n"
-            f"  2. Specify different directory: --dir ~/my-sync-folder"
-        )
-
-    # Check if mount is active at this location
-    result = subprocess.run(["mount"], capture_output=True, text=True)
-    if str(bisync_dir) in result.stdout and "rclone" in result.stdout:
-        raise BisyncError(
-            f"{bisync_dir} is currently mounted via 'bm cloud mount'\n"
-            f"Cannot use mounted directory for bisync.\n\n"
-            f"Either:\n"
-            f"  1. Unmount first: bm cloud unmount\n"
-            f"  2. Use different directory for bisync"
-        )
-
-
 def convert_bmignore_to_rclone_filters() -> Path:
     """Convert .bmignore patterns to rclone filter format.
 
@@ -245,521 +108,3 @@ def get_bisync_filter_path() -> Path:
         Path to rclone filter file
     """
     return convert_bmignore_to_rclone_filters()
-
-
-def bisync_state_exists(tenant_id: str) -> bool:
-    """Check if bisync state exists (has been initialized)."""
-    state_path = get_bisync_state_path(tenant_id)
-    return state_path.exists() and any(state_path.iterdir())
-
-
-def build_bisync_command(
-    tenant_id: str,
-    bucket_name: str,
-    local_path: Path,
-    profile: RcloneBisyncProfile,
-    dry_run: bool = False,
-    resync: bool = False,
-    verbose: bool = False,
-) -> list[str]:
-    """Build rclone bisync command with profile settings."""
-
-    # Sync with the entire bucket root (all projects)
-    rclone_remote = f"basic-memory-{tenant_id}:{bucket_name}"
-    filter_path = get_bisync_filter_path()
-    state_path = get_bisync_state_path(tenant_id)
-
-    # Ensure state directory exists
-    state_path.mkdir(parents=True, exist_ok=True)
-
-    cmd = [
-        "rclone",
-        "bisync",
-        str(local_path),
-        rclone_remote,
-        "--create-empty-src-dirs",
-        "--resilient",
-        f"--conflict-resolve={profile.conflict_resolve}",
-        f"--max-delete={profile.max_delete}",
-        "--filters-file",
-        str(filter_path),
-        "--workdir",
-        str(state_path),
-    ]
-
-    # Add verbosity flags
-    if verbose:
-        cmd.append("--verbose")  # Full details with file-by-file output
-    else:
-        # Show progress bar during transfers
-        cmd.append("--progress")
-
-    if profile.check_access:
-        cmd.append("--check-access")
-
-    if dry_run:
-        cmd.append("--dry-run")
-
-    if resync:
-        cmd.append("--resync")
-
-    cmd.extend(profile.extra_args)
-
-    return cmd
-
-
-def setup_cloud_bisync(sync_dir: Optional[str] = None) -> None:
-    """Set up cloud bisync with rclone installation and configuration.
-
-    Args:
-        sync_dir: Optional custom sync directory path. If not provided, uses config default.
-    """
-    console.print("[bold blue]Basic Memory Cloud Bisync Setup[/bold blue]")
-    console.print("Setting up bidirectional sync to your cloud tenant...\n")
-
-    try:
-        # Step 1: Install rclone
-        console.print("[blue]Step 1: Installing rclone...[/blue]")
-        install_rclone()
-
-        # Step 2: Get mount info (for tenant_id, bucket)
-        console.print("\n[blue]Step 2: Getting tenant information...[/blue]")
-        tenant_info = asyncio.run(get_mount_info())
-
-        tenant_id = tenant_info.tenant_id
-        bucket_name = tenant_info.bucket_name
-
-        console.print(f"[green]✓ Found tenant: {tenant_id}[/green]")
-        console.print(f"[green]✓ Bucket: {bucket_name}[/green]")
-
-        # Step 3: Generate credentials
-        console.print("\n[blue]Step 3: Generating sync credentials...[/blue]")
-        creds = asyncio.run(generate_mount_credentials(tenant_id))
-
-        access_key = creds.access_key
-        secret_key = creds.secret_key
-
-        console.print("[green]✓ Generated secure credentials[/green]")
-
-        # Step 4: Configure rclone
-        console.print("\n[blue]Step 4: Configuring rclone...[/blue]")
-        add_tenant_to_rclone_config(
-            tenant_id=tenant_id,
-            bucket_name=bucket_name,
-            access_key=access_key,
-            secret_key=secret_key,
-        )
-
-        # Step 5: Configure and create local directory
-        console.print("\n[blue]Step 5: Configuring sync directory...[/blue]")
-
-        # If custom sync_dir provided, save to config
-        if sync_dir:
-            config_manager = ConfigManager()
-            config = config_manager.load_config()
-            config.bisync_config["sync_dir"] = sync_dir
-            config_manager.save_config(config)
-            console.print("[green]✓ Saved custom sync directory to config[/green]")
-
-        # Get bisync directory (from config or default)
-        local_path = get_bisync_directory()
-
-        # Validate bisync directory
-        validate_bisync_directory(local_path)
-
-        # Create directory
-        local_path.mkdir(parents=True, exist_ok=True)
-        console.print(f"[green]✓ Created sync directory: {local_path}[/green]")
-
-        # Step 6: Perform initial resync
-        console.print("\n[blue]Step 6: Performing initial sync...[/blue]")
-        console.print("[yellow]This will establish the baseline for bidirectional sync.[/yellow]")
-
-        run_bisync(
-            tenant_id=tenant_id,
-            bucket_name=bucket_name,
-            local_path=local_path,
-            profile_name="balanced",
-            resync=True,
-        )
-
-        console.print("\n[bold green]✓ Bisync setup completed successfully![/bold green]")
-        console.print("\nYour local files will now sync bidirectionally with the cloud!")
-        console.print(f"\nLocal directory: {local_path}")
-        console.print("\nUseful commands:")
-        console.print("  bm sync                      # Run sync (recommended)")
-        console.print("  bm sync --watch              # Start watch mode")
-        console.print("  bm cloud status              # Check sync status")
-        console.print("  bm cloud check               # Verify file integrity")
-        console.print("  bm cloud bisync --dry-run    # Preview changes (advanced)")
-
-    except (RcloneInstallError, BisyncError, CloudAPIError) as e:
-        console.print(f"\n[red]Setup failed: {e}[/red]")
-        raise typer.Exit(1)
-    except Exception as e:
-        console.print(f"\n[red]Unexpected error during setup: {e}[/red]")
-        raise typer.Exit(1)
-
-
-def run_bisync(
-    tenant_id: Optional[str] = None,
-    bucket_name: Optional[str] = None,
-    local_path: Optional[Path] = None,
-    profile_name: str = "balanced",
-    dry_run: bool = False,
-    resync: bool = False,
-    verbose: bool = False,
-) -> bool:
-    """Run rclone bisync with specified profile."""
-
-    try:
-        # Get tenant info if not provided
-        if not tenant_id or not bucket_name:
-            tenant_info = asyncio.run(get_mount_info())
-            tenant_id = tenant_info.tenant_id
-            bucket_name = tenant_info.bucket_name
-
-        # Set default local path if not provided
-        if not local_path:
-            local_path = get_bisync_directory()
-
-        # Validate bisync directory
-        validate_bisync_directory(local_path)
-
-        # Check if local path exists
-        if not local_path.exists():
-            raise BisyncError(
-                f"Local directory {local_path} does not exist. Run 'basic-memory cloud bisync-setup' first."
-            )
-
-        # Get bisync profile
-        if profile_name not in BISYNC_PROFILES:
-            raise BisyncError(
-                f"Unknown profile: {profile_name}. Available: {list(BISYNC_PROFILES.keys())}"
-            )
-
-        profile = BISYNC_PROFILES[profile_name]
-
-        # Auto-register projects before sync (unless dry-run or resync)
-        if not dry_run and not resync:
-            try:
-                console.print("[dim]Checking for new projects...[/dim]")
-
-                # Fetch cloud projects and extract directory names from paths
-                cloud_data = asyncio.run(fetch_cloud_projects())
-                cloud_projects = cloud_data.projects
-
-                # Extract directory names from cloud project paths
-                # Compare directory names, not project names
-                # Cloud path /app/data/basic-memory -> directory name "basic-memory"
-                cloud_dir_names = set()
-                for p in cloud_projects:
-                    path = p.path
-                    # Strip /app/data/ prefix if present (cloud mode)
-                    if path.startswith("/app/data/"):
-                        path = path[len("/app/data/") :]
-                    # Get the last segment (directory name)
-                    dir_name = Path(path).name
-                    cloud_dir_names.add(dir_name)
-
-                # Scan local directories
-                local_dirs = scan_local_directories(local_path)
-
-                # Create missing cloud projects
-                new_projects = []
-                for dir_name in local_dirs:
-                    if dir_name not in cloud_dir_names:
-                        new_projects.append(dir_name)
-
-                if new_projects:
-                    console.print(
-                        f"[blue]Found {len(new_projects)} new local project(s), creating on cloud...[/blue]"
-                    )
-                    for project_name in new_projects:
-                        try:
-                            asyncio.run(create_cloud_project(project_name))
-                            console.print(f"[green]  ✓ Created project: {project_name}[/green]")
-                        except BisyncError as e:
-                            console.print(
-                                f"[yellow]  ⚠ Could not create {project_name}: {e}[/yellow]"
-                            )
-                else:
-                    console.print("[dim]All local projects already registered on cloud[/dim]")
-
-            except Exception as e:
-                console.print(f"[yellow]Warning: Project auto-registration failed: {e}[/yellow]")
-                console.print("[yellow]Continuing with sync anyway...[/yellow]")
-
-        # Check if first run and require resync
-        if not resync and not bisync_state_exists(tenant_id) and not dry_run:
-            raise BisyncError(
-                "First bisync requires --resync to establish baseline. "
-                "Run: basic-memory cloud bisync --resync"
-            )
-
-        # Build and execute bisync command
-        bisync_cmd = build_bisync_command(
-            tenant_id,
-            bucket_name,
-            local_path,
-            profile,
-            dry_run=dry_run,
-            resync=resync,
-            verbose=verbose,
-        )
-
-        if dry_run:
-            console.print("[yellow]DRY RUN MODE - No changes will be made[/yellow]")
-
-        console.print(
-            f"[blue]Running bisync with profile '{profile_name}' ({profile.description})...[/blue]"
-        )
-        console.print(f"[dim]Command: {' '.join(bisync_cmd)}[/dim]")
-        console.print()  # Blank line before output
-
-        # Stream output in real-time so user sees progress
-        result = subprocess.run(bisync_cmd, text=True)
-
-        if result.returncode != 0:
-            raise BisyncError(f"Bisync command failed with code {result.returncode}")
-
-        console.print()  # Blank line after output
-
-        if dry_run:
-            console.print("[green]✓ Dry run completed successfully[/green]")
-        elif resync:
-            console.print("[green]✓ Initial sync baseline established[/green]")
-        else:
-            console.print("[green]✓ Sync completed successfully[/green]")
-
-        # Notify container to refresh cache (if not dry run)
-        if not dry_run:
-            try:
-                asyncio.run(notify_container_sync(tenant_id))
-            except Exception as e:
-                console.print(f"[yellow]Warning: Could not notify container: {e}[/yellow]")
-
-        return True
-
-    except BisyncError:
-        raise
-    except Exception as e:
-        raise BisyncError(f"Unexpected error during bisync: {e}") from e
-
-
-async def notify_container_sync(tenant_id: str) -> None:
-    """Sync all projects after bisync completes."""
-    try:
-        from basic_memory.cli.commands.command_utils import run_sync
-
-        # Fetch all projects and sync each one
-        cloud_data = await fetch_cloud_projects()
-        projects = cloud_data.projects
-
-        if not projects:
-            console.print("[dim]No projects to sync[/dim]")
-            return
-
-        console.print(f"[blue]Notifying cloud to index {len(projects)} project(s)...[/blue]")
-
-        for project in projects:
-            project_name = project.name
-            if project_name:
-                try:
-                    await run_sync(project=project_name)
-                except Exception as e:
-                    # Non-critical, log and continue
-                    console.print(f"[yellow]  ⚠ Sync failed for {project_name}: {e}[/yellow]")
-
-        console.print("[dim]Note: Cloud indexing has started and may take a few moments[/dim]")
-
-    except Exception as e:
-        # Non-critical, don't fail the bisync
-        console.print(f"[yellow]Warning: Post-sync failed: {e}[/yellow]")
-
-
-def run_bisync_watch(
-    tenant_id: Optional[str] = None,
-    bucket_name: Optional[str] = None,
-    local_path: Optional[Path] = None,
-    profile_name: str = "balanced",
-    interval_seconds: int = 60,
-) -> None:
-    """Run bisync in watch mode with periodic syncs."""
-
-    console.print("[bold blue]Starting bisync watch mode[/bold blue]")
-    console.print(f"Sync interval: {interval_seconds} seconds")
-    console.print("Press Ctrl+C to stop\n")
-
-    try:
-        while True:
-            try:
-                start_time = time.time()
-
-                run_bisync(
-                    tenant_id=tenant_id,
-                    bucket_name=bucket_name,
-                    local_path=local_path,
-                    profile_name=profile_name,
-                )
-
-                elapsed = time.time() - start_time
-                console.print(f"[dim]Sync completed in {elapsed:.1f}s[/dim]")
-
-                # Wait for next interval
-                time.sleep(interval_seconds)
-
-            except BisyncError as e:
-                console.print(f"[red]Sync error: {e}[/red]")
-                console.print(f"[yellow]Retrying in {interval_seconds} seconds...[/yellow]")
-                time.sleep(interval_seconds)
-
-    except KeyboardInterrupt:
-        console.print("\n[yellow]Watch mode stopped[/yellow]")
-
-
-def show_bisync_status() -> None:
-    """Show current bisync status and configuration."""
-
-    try:
-        # Get tenant info
-        tenant_info = asyncio.run(get_mount_info())
-        tenant_id = tenant_info.tenant_id
-
-        local_path = get_bisync_directory()
-        state_path = get_bisync_state_path(tenant_id)
-
-        # Create status table
-        table = Table(title="Cloud Bisync Status", show_header=True, header_style="bold blue")
-        table.add_column("Property", style="green", min_width=20)
-        table.add_column("Value", style="dim", min_width=30)
-
-        # Check initialization status
-        is_initialized = bisync_state_exists(tenant_id)
-        init_status = (
-            "[green]✓ Initialized[/green]" if is_initialized else "[red]✗ Not initialized[/red]"
-        )
-
-        table.add_row("Tenant ID", tenant_id)
-        table.add_row("Local Directory", str(local_path))
-        table.add_row("Status", init_status)
-        table.add_row("State Directory", str(state_path))
-
-        # Check for last sync info
-        if is_initialized:
-            # Look for most recent state file
-            state_files = list(state_path.glob("*.lst"))
-            if state_files:
-                latest = max(state_files, key=lambda p: p.stat().st_mtime)
-                last_sync = datetime.fromtimestamp(latest.stat().st_mtime)
-                table.add_row("Last Sync", last_sync.strftime("%Y-%m-%d %H:%M:%S"))
-
-        console.print(table)
-
-        # Show bisync profiles
-        console.print("\n[bold]Available bisync profiles:[/bold]")
-        for name, profile in BISYNC_PROFILES.items():
-            console.print(f"  {name}: {profile.description}")
-            console.print(f"    - Conflict resolution: {profile.conflict_resolve}")
-            console.print(f"    - Max delete: {profile.max_delete} files")
-
-        console.print("\n[dim]To use a profile: bm cloud bisync --profile <name>[/dim]")
-
-        # Show setup instructions if not initialized
-        if not is_initialized:
-            console.print("\n[yellow]To initialize bisync, run:[/yellow]")
-            console.print("  bm cloud setup")
-            console.print("  or")
-            console.print("  bm cloud bisync --resync")
-
-    except Exception as e:
-        console.print(f"[red]Error getting bisync status: {e}[/red]")
-        raise typer.Exit(1)
-
-
-def run_check(
-    tenant_id: Optional[str] = None,
-    bucket_name: Optional[str] = None,
-    local_path: Optional[Path] = None,
-    one_way: bool = False,
-) -> bool:
-    """Check file integrity between local and cloud using rclone check.
-
-    Args:
-        tenant_id: Cloud tenant ID (auto-detected if not provided)
-        bucket_name: S3 bucket name (auto-detected if not provided)
-        local_path: Local bisync directory (uses config default if not provided)
-        one_way: If True, only check for missing files on destination (faster)
-
-    Returns:
-        True if check passed (files match), False if differences found
-    """
-    try:
-        # Check if rclone is installed
-        from basic_memory.cli.commands.cloud.rclone_installer import is_rclone_installed
-
-        if not is_rclone_installed():
-            raise BisyncError(
-                "rclone is not installed. Run 'bm cloud bisync-setup' first to set up cloud sync."
-            )
-
-        # Get tenant info if not provided
-        if not tenant_id or not bucket_name:
-            tenant_info = asyncio.run(get_mount_info())
-            tenant_id = tenant_id or tenant_info.tenant_id
-            bucket_name = bucket_name or tenant_info.bucket_name
-
-        # Get local path from config
-        if not local_path:
-            local_path = get_bisync_directory()
-
-        # Check if bisync is initialized
-        if not bisync_state_exists(tenant_id):
-            raise BisyncError(
-                "Bisync not initialized. Run 'bm cloud bisync --resync' to establish baseline."
-            )
-
-        # Build rclone check command
-        rclone_remote = f"basic-memory-{tenant_id}:{bucket_name}"
-        filter_path = get_bisync_filter_path()
-
-        cmd = [
-            "rclone",
-            "check",
-            str(local_path),
-            rclone_remote,
-            "--filter-from",
-            str(filter_path),
-        ]
-
-        if one_way:
-            cmd.append("--one-way")
-
-        console.print("[bold blue]Checking file integrity between local and cloud[/bold blue]")
-        console.print(f"[dim]Local:  {local_path}[/dim]")
-        console.print(f"[dim]Remote: {rclone_remote}[/dim]")
-        console.print(f"[dim]Command: {' '.join(cmd)}[/dim]")
-        console.print()
-
-        # Run check command
-        result = subprocess.run(cmd, capture_output=True, text=True)
-
-        # rclone check returns:
-        # 0 = success (all files match)
-        # non-zero = differences found or error
-        if result.returncode == 0:
-            console.print("[green]✓ All files match between local and cloud[/green]")
-            return True
-        else:
-            console.print("[yellow]⚠ Differences found:[/yellow]")
-            if result.stderr:
-                console.print(result.stderr)
-            if result.stdout:
-                console.print(result.stdout)
-            console.print("\n[dim]To sync differences, run: bm sync[/dim]")
-            return False
-
-    except BisyncError:
-        raise
-    except Exception as e:
-        raise BisyncError(f"Check failed: {e}") from e

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

@@ -1,7 +1,6 @@
 """Core cloud commands for Basic Memory CLI."""
 
 import asyncio
-from typing import Optional
 
 import typer
 from rich.console import Console
@@ -15,21 +14,16 @@ from basic_memory.cli.commands.cloud.api_client import (
     get_cloud_config,
     make_api_request,
 )
-from basic_memory.cli.commands.cloud.mount_commands import (
-    mount_cloud_files,
-    setup_cloud_mount,
-    show_mount_status,
-    unmount_cloud_files,
-)
 from basic_memory.cli.commands.cloud.bisync_commands import (
-    run_bisync,
-    run_bisync_watch,
-    run_check,
-    setup_cloud_bisync,
-    show_bisync_status,
+    BisyncError,
+    generate_mount_credentials,
+    get_mount_info,
+)
+from basic_memory.cli.commands.cloud.rclone_config import configure_rclone_remote
+from basic_memory.cli.commands.cloud.rclone_installer import (
+    RcloneInstallError,
+    install_rclone,
 )
-from basic_memory.cli.commands.cloud.rclone_config import MOUNT_PROFILES
-from basic_memory.cli.commands.cloud.bisync_commands import BISYNC_PROFILES
 
 console = Console()
 
@@ -88,18 +82,8 @@ def logout():
 
 
 @cloud_app.command("status")
-def status(
-    bisync: bool = typer.Option(
-        True,
-        "--bisync/--mount",
-        help="Show bisync status (default) or mount status",
-    ),
-) -> None:
-    """Check cloud mode status and cloud instance health.
-
-    Shows cloud mode status, instance health, and sync/mount status.
-    Use --bisync (default) to show bisync status or --mount for mount status.
-    """
+def status() -> None:
+    """Check cloud mode status and cloud instance health."""
     # Check cloud mode
     config_manager = ConfigManager()
     config = config_manager.load_config()
@@ -142,12 +126,7 @@ def status(
         if "timestamp" in health_data:
             console.print(f"  Timestamp: {health_data['timestamp']}")
 
-        # Show sync/mount status based on flag
-        console.print()
-        if bisync:
-            show_bisync_status()
-        else:
-            show_mount_status()
+        console.print("\n[dim]To sync projects, use: bm project bisync --name <project>[/dim]")
 
     except CloudAPIError as e:
         console.print(f"[red]Error checking cloud health: {e}[/red]")
@@ -157,132 +136,60 @@ def status(
         raise typer.Exit(1)
 
 
-# Mount commands
-
-
 @cloud_app.command("setup")
-def setup(
-    bisync: bool = typer.Option(
-        True,
-        "--bisync/--mount",
-        help="Use bidirectional sync (recommended) or mount as network drive",
-    ),
-    sync_dir: Optional[str] = typer.Option(
-        None,
-        "--dir",
-        help="Custom sync directory for bisync (default: ~/basic-memory-cloud-sync)",
-    ),
-) -> None:
-    """Set up cloud file access with automatic rclone installation and configuration.
+def setup() -> None:
+    """Set up cloud sync by installing rclone and configuring credentials.
 
-    Default: Sets up bidirectional sync (recommended).\n
-    Use --mount: Sets up mount as network drive (alternative workflow).\n
-
-    Examples:\n
-      bm cloud setup              # Setup bisync (default)\n
-      bm cloud setup --mount      # Setup mount instead\n
-      bm cloud setup --dir ~/sync # Custom bisync directory\n
+    SPEC-20: Simplified to project-scoped workflow.
+    After setup, use project commands for syncing:
+      bm project add <name> <path> --local-path ~/projects/<name>
+      bm project bisync --name <name> --resync  # First time
+      bm project bisync --name <name>            # Subsequent syncs
     """
-    if bisync:
-        setup_cloud_bisync(sync_dir=sync_dir)
-    else:
-        setup_cloud_mount()
+    console.print("[bold blue]Basic Memory Cloud Setup[/bold blue]")
+    console.print("Setting up cloud sync with rclone...\n")
 
-
-@cloud_app.command("mount")
-def mount(
-    profile: str = typer.Option(
-        "balanced", help=f"Mount profile: {', '.join(MOUNT_PROFILES.keys())}"
-    ),
-    path: Optional[str] = typer.Option(
-        None, help="Custom mount path (default: ~/basic-memory-{tenant-id})"
-    ),
-) -> None:
-    """Mount cloud files locally for editing."""
     try:
-        mount_cloud_files(profile_name=profile)
-    except Exception as e:
-        console.print(f"[red]Mount failed: {e}[/red]")
+        # Step 1: Install rclone
+        console.print("[blue]Step 1: Installing rclone...[/blue]")
+        install_rclone()
+
+        # Step 2: Get tenant info
+        console.print("\n[blue]Step 2: Getting tenant information...[/blue]")
+        tenant_info = asyncio.run(get_mount_info())
+        console.print(f"[green]✓ Found tenant: {tenant_info.tenant_id}[/green]")
+
+        # Step 3: Generate credentials
+        console.print("\n[blue]Step 3: Generating sync credentials...[/blue]")
+        creds = asyncio.run(generate_mount_credentials(tenant_info.tenant_id))
+        console.print("[green]✓ Generated secure credentials[/green]")
+
+        # Step 4: Configure rclone remote
+        console.print("\n[blue]Step 4: Configuring rclone remote...[/blue]")
+        configure_rclone_remote(
+            access_key=creds.access_key,
+            secret_key=creds.secret_key,
+        )
+
+        console.print("\n[bold green]✓ Cloud setup completed successfully![/bold green]")
+        console.print("\n[bold]Next steps:[/bold]")
+        console.print("1. Add a project with local sync path:")
+        console.print("   bm project add research --local-path ~/Documents/research")
+        console.print("\n   Or configure sync for an existing project:")
+        console.print("   bm project sync-setup research ~/Documents/research")
+        console.print("\n2. Preview the initial sync (recommended):")
+        console.print("   bm project bisync --name research --resync --dry-run")
+        console.print("\n3. If all looks good, run the actual sync:")
+        console.print("   bm project bisync --name research --resync")
+        console.print("\n4. Subsequent syncs (no --resync needed):")
+        console.print("   bm project bisync --name research")
+        console.print(
+            "\n[dim]Tip: Always use --dry-run first to preview changes before syncing[/dim]"
+        )
+
+    except (RcloneInstallError, BisyncError, CloudAPIError) as e:
+        console.print(f"\n[red]Setup failed: {e}[/red]")
         raise typer.Exit(1)
-
-
-@cloud_app.command("unmount")
-def unmount() -> None:
-    """Unmount cloud files."""
-    try:
-        unmount_cloud_files()
     except Exception as e:
-        console.print(f"[red]Unmount failed: {e}[/red]")
-        raise typer.Exit(1)
-
-
-# Bisync commands
-
-
-@cloud_app.command("bisync")
-def bisync(
-    profile: str = typer.Option(
-        "balanced", help=f"Bisync profile: {', '.join(BISYNC_PROFILES.keys())}"
-    ),
-    dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
-    resync: bool = typer.Option(False, "--resync", help="Force resync to establish new baseline"),
-    watch: bool = typer.Option(False, "--watch", help="Run continuous sync in watch mode"),
-    interval: int = typer.Option(60, "--interval", help="Sync interval in seconds for watch mode"),
-    verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed sync output"),
-) -> None:
-    """Run bidirectional sync between local files and cloud storage.
-
-    Examples:
-      basic-memory cloud bisync                    # Manual sync with balanced profile
-      basic-memory cloud bisync --dry-run          # Preview what would be synced
-      basic-memory cloud bisync --resync           # Establish new baseline
-      basic-memory cloud bisync --watch            # Continuous sync every 60s
-      basic-memory cloud bisync --watch --interval 30  # Continuous sync every 30s
-      basic-memory cloud bisync --profile safe     # Use safe profile (keep conflicts)
-      basic-memory cloud bisync --verbose          # Show detailed file sync output
-    """
-    try:
-        if watch:
-            run_bisync_watch(profile_name=profile, interval_seconds=interval)
-        else:
-            run_bisync(profile_name=profile, dry_run=dry_run, resync=resync, verbose=verbose)
-    except Exception as e:
-        console.print(f"[red]Bisync failed: {e}[/red]")
-        raise typer.Exit(1)
-
-
-@cloud_app.command("bisync-status")
-def bisync_status() -> None:
-    """Show current bisync status and configuration.
-
-    DEPRECATED: Use 'bm cloud status' instead (bisync is now the default).
-    """
-    console.print(
-        "[yellow]Note: 'bisync-status' is deprecated. Use 'bm cloud status' instead.[/yellow]"
-    )
-    console.print("[dim]Showing bisync status...[/dim]\n")
-    show_bisync_status()
-
-
-@cloud_app.command("check")
-def check(
-    one_way: bool = typer.Option(
-        False,
-        "--one-way",
-        help="Only check for missing files on destination (faster)",
-    ),
-) -> None:
-    """Check file integrity between local and cloud storage using rclone check.
-
-    Verifies that files match between your local bisync directory and cloud storage
-    without transferring any data. This is useful for validating sync integrity.
-
-    Examples:
-      bm cloud check              # Full integrity check
-      bm cloud check --one-way    # Faster check (missing files only)
-    """
-    try:
-        run_check(one_way=one_way)
-    except Exception as e:
-        console.print(f"[red]Check failed: {e}[/red]")
+        console.print(f"\n[red]Unexpected error during setup: {e}[/red]")
         raise typer.Exit(1)

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

@@ -1,295 +0,0 @@
-"""Cloud mount commands for Basic Memory CLI."""
-
-import asyncio
-import subprocess
-import time
-from pathlib import Path
-from typing import Optional
-
-import typer
-from rich.console import Console
-from rich.table import Table
-
-from basic_memory.cli.commands.cloud.api_client import CloudAPIError, make_api_request
-from basic_memory.cli.commands.cloud.rclone_config import (
-    MOUNT_PROFILES,
-    add_tenant_to_rclone_config,
-    build_mount_command,
-    cleanup_orphaned_rclone_processes,
-    get_default_mount_path,
-    get_rclone_processes,
-    is_path_mounted,
-    unmount_path,
-)
-from basic_memory.cli.commands.cloud.rclone_installer import RcloneInstallError, install_rclone
-from basic_memory.config import ConfigManager
-
-console = Console()
-
-
-class MountError(Exception):
-    """Exception raised for mount-related errors."""
-
-    pass
-
-
-async def get_tenant_info() -> dict:
-    """Get current tenant information from cloud API."""
-    try:
-        config_manager = ConfigManager()
-        config = config_manager.config
-        host_url = config.cloud_host.rstrip("/")
-
-        response = await make_api_request(method="GET", url=f"{host_url}/tenant/mount/info")
-
-        return response.json()
-    except Exception as e:
-        raise MountError(f"Failed to get tenant info: {e}") from e
-
-
-async def generate_mount_credentials(tenant_id: str) -> dict:
-    """Generate scoped credentials for mounting."""
-    try:
-        config_manager = ConfigManager()
-        config = config_manager.config
-        host_url = config.cloud_host.rstrip("/")
-
-        response = await make_api_request(method="POST", url=f"{host_url}/tenant/mount/credentials")
-
-        return response.json()
-    except Exception as e:
-        raise MountError(f"Failed to generate mount credentials: {e}") from e
-
-
-def setup_cloud_mount() -> None:
-    """Set up cloud mount with rclone installation and configuration."""
-    console.print("[bold blue]Basic Memory Cloud Setup[/bold blue]")
-    console.print("Setting up local file access to your cloud tenant...\n")
-
-    try:
-        # Step 1: Install rclone
-        console.print("[blue]Step 1: Installing rclone...[/blue]")
-        install_rclone()
-
-        # Step 2: Get tenant info
-        console.print("\n[blue]Step 2: Getting tenant information...[/blue]")
-        tenant_info = asyncio.run(get_tenant_info())
-
-        tenant_id = tenant_info.get("tenant_id")
-        bucket_name = tenant_info.get("bucket_name")
-
-        if not tenant_id or not bucket_name:
-            raise MountError("Invalid tenant information received from cloud API")
-
-        console.print(f"[green]✓ Found tenant: {tenant_id}[/green]")
-        console.print(f"[green]✓ Bucket: {bucket_name}[/green]")
-
-        # Step 3: Generate mount credentials
-        console.print("\n[blue]Step 3: Generating mount credentials...[/blue]")
-        creds = asyncio.run(generate_mount_credentials(tenant_id))
-
-        access_key = creds.get("access_key")
-        secret_key = creds.get("secret_key")
-
-        if not access_key or not secret_key:
-            raise MountError("Failed to generate mount credentials")
-
-        console.print("[green]✓ Generated secure credentials[/green]")
-
-        # Step 4: Configure rclone
-        console.print("\n[blue]Step 4: Configuring rclone...[/blue]")
-        add_tenant_to_rclone_config(
-            tenant_id=tenant_id,
-            bucket_name=bucket_name,
-            access_key=access_key,
-            secret_key=secret_key,
-        )
-
-        # Step 5: Perform initial mount
-        console.print("\n[blue]Step 5: Mounting cloud files...[/blue]")
-        mount_path = get_default_mount_path()
-        MOUNT_PROFILES["balanced"]
-
-        mount_cloud_files(
-            tenant_id=tenant_id,
-            bucket_name=bucket_name,
-            mount_path=mount_path,
-            profile_name="balanced",
-        )
-
-        console.print("\n[bold green]✓ Cloud setup completed successfully![/bold green]")
-        console.print("\nYour cloud files are now accessible at:")
-        console.print(f"  {mount_path}")
-        console.print("\nYou can now edit files locally and they will sync to the cloud!")
-        console.print("\nUseful commands:")
-        console.print("  basic-memory cloud mount-status    # Check mount status")
-        console.print("  basic-memory cloud unmount         # Unmount files")
-        console.print("  basic-memory cloud mount --profile fast  # Remount with faster sync")
-
-    except (RcloneInstallError, MountError, CloudAPIError) as e:
-        console.print(f"\n[red]Setup failed: {e}[/red]")
-        raise typer.Exit(1)
-    except Exception as e:
-        console.print(f"\n[red]Unexpected error during setup: {e}[/red]")
-        raise typer.Exit(1)
-
-
-def mount_cloud_files(
-    tenant_id: Optional[str] = None,
-    bucket_name: Optional[str] = None,
-    mount_path: Optional[Path] = None,
-    profile_name: str = "balanced",
-) -> None:
-    """Mount cloud files with specified profile."""
-
-    try:
-        # Get tenant info if not provided
-        if not tenant_id or not bucket_name:
-            tenant_info = asyncio.run(get_tenant_info())
-            tenant_id = tenant_info.get("tenant_id")
-            bucket_name = tenant_info.get("bucket_name")
-
-            if not tenant_id or not bucket_name:
-                raise MountError("Could not determine tenant information")
-
-        # Set default mount path if not provided
-        if not mount_path:
-            mount_path = get_default_mount_path()
-
-        # Get mount profile
-        if profile_name not in MOUNT_PROFILES:
-            raise MountError(
-                f"Unknown profile: {profile_name}. Available: {list(MOUNT_PROFILES.keys())}"
-            )
-
-        profile = MOUNT_PROFILES[profile_name]
-
-        # Check if already mounted
-        if is_path_mounted(mount_path):
-            console.print(f"[yellow]Path {mount_path} is already mounted[/yellow]")
-            console.print("Use 'basic-memory cloud unmount' first, or mount to a different path")
-            return
-
-        # Create mount directory
-        mount_path.mkdir(parents=True, exist_ok=True)
-
-        # Build and execute mount command
-        mount_cmd = build_mount_command(tenant_id, bucket_name, mount_path, profile)
-
-        console.print(
-            f"[blue]Mounting with profile '{profile_name}' ({profile.description})...[/blue]"
-        )
-        console.print(f"[dim]Command: {' '.join(mount_cmd)}[/dim]")
-
-        result = subprocess.run(mount_cmd, capture_output=True, text=True)
-
-        if result.returncode != 0:
-            error_msg = result.stderr or "Unknown error"
-            raise MountError(f"Mount command failed: {error_msg}")
-
-        # Wait a moment for mount to establish
-        time.sleep(2)
-
-        # Verify mount
-        if is_path_mounted(mount_path):
-            console.print(f"[green]✓ Successfully mounted to {mount_path}[/green]")
-            console.print(f"[green]✓ Sync profile: {profile.description}[/green]")
-        else:
-            raise MountError("Mount command succeeded but path is not mounted")
-
-    except MountError:
-        raise
-    except Exception as e:
-        raise MountError(f"Unexpected error during mount: {e}") from e
-
-
-def unmount_cloud_files(tenant_id: Optional[str] = None) -> None:
-    """Unmount cloud files."""
-
-    try:
-        # Get tenant info if not provided
-        if not tenant_id:
-            tenant_info = asyncio.run(get_tenant_info())
-            tenant_id = tenant_info.get("tenant_id")
-
-            if not tenant_id:
-                raise MountError("Could not determine tenant ID")
-
-        mount_path = get_default_mount_path()
-
-        if not is_path_mounted(mount_path):
-            console.print(f"[yellow]Path {mount_path} is not mounted[/yellow]")
-            return
-
-        console.print(f"[blue]Unmounting {mount_path}...[/blue]")
-
-        # Unmount the path
-        if unmount_path(mount_path):
-            console.print(f"[green]✓ Successfully unmounted {mount_path}[/green]")
-
-            # Clean up any orphaned rclone processes
-            killed_count = cleanup_orphaned_rclone_processes()
-            if killed_count > 0:
-                console.print(
-                    f"[green]✓ Cleaned up {killed_count} orphaned rclone process(es)[/green]"
-                )
-        else:
-            console.print(f"[red]✗ Failed to unmount {mount_path}[/red]")
-            console.print("You may need to manually unmount or restart your system")
-
-    except MountError:
-        raise
-    except Exception as e:
-        raise MountError(f"Unexpected error during unmount: {e}") from e
-
-
-def show_mount_status() -> None:
-    """Show current mount status and running processes."""
-
-    try:
-        # Get tenant info
-        tenant_info = asyncio.run(get_tenant_info())
-        tenant_id = tenant_info.get("tenant_id")
-
-        if not tenant_id:
-            console.print("[red]Could not determine tenant ID[/red]")
-            return
-
-        mount_path = get_default_mount_path()
-
-        # Create status table
-        table = Table(title="Cloud Mount Status", show_header=True, header_style="bold blue")
-        table.add_column("Property", style="green", min_width=15)
-        table.add_column("Value", style="dim", min_width=30)
-
-        # Check mount status
-        is_mounted = is_path_mounted(mount_path)
-        mount_status = "[green]✓ Mounted[/green]" if is_mounted else "[red]✗ Not mounted[/red]"
-
-        table.add_row("Tenant ID", tenant_id)
-        table.add_row("Mount Path", str(mount_path))
-        table.add_row("Status", mount_status)
-
-        # Get rclone processes
-        processes = get_rclone_processes()
-        if processes:
-            table.add_row("rclone Processes", f"{len(processes)} running")
-        else:
-            table.add_row("rclone Processes", "None")
-
-        console.print(table)
-
-        # Show running processes details
-        if processes:
-            console.print("\n[bold]Running rclone processes:[/bold]")
-            for proc in processes:
-                console.print(f"  PID {proc['pid']}: {proc['command'][:80]}...")
-
-        # Show mount profiles
-        console.print("\n[bold]Available mount profiles:[/bold]")
-        for name, profile in MOUNT_PROFILES.items():
-            console.print(f"  {name}: {profile.description}")
-
-    except Exception as e:
-        console.print(f"[red]Error getting mount status: {e}[/red]")
-        raise typer.Exit(1)

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

@@ -0,0 +1,299 @@
+"""Project-scoped rclone sync commands for Basic Memory Cloud.
+
+This module provides simplified, project-scoped rclone operations:
+- Each project syncs independently
+- Uses single "basic-memory-cloud" remote (not tenant-specific)
+- Balanced defaults from SPEC-8 Phase 4 testing
+- Per-project bisync state tracking
+
+Replaces tenant-wide sync with project-scoped workflows.
+"""
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+from rich.console import Console
+
+console = Console()
+
+
+class RcloneError(Exception):
+    """Exception raised for rclone command errors."""
+
+    pass
+
+
+@dataclass
+class SyncProject:
+    """Project configured for cloud sync.
+
+    Attributes:
+        name: Project name
+        path: Cloud path (e.g., "app/data/research")
+        local_sync_path: Local directory for syncing (optional)
+    """
+
+    name: str
+    path: str
+    local_sync_path: Optional[str] = None
+
+
+def get_bmignore_filter_path() -> Path:
+    """Get path to rclone filter file.
+
+    Uses ~/.basic-memory/.bmignore converted to rclone format.
+    File is automatically created with default patterns on first use.
+
+    Returns:
+        Path to rclone filter file
+    """
+    # Import here to avoid circular dependency
+    from basic_memory.cli.commands.cloud.bisync_commands import (
+        convert_bmignore_to_rclone_filters,
+    )
+
+    return convert_bmignore_to_rclone_filters()
+
+
+def get_project_bisync_state(project_name: str) -> Path:
+    """Get path to project's bisync state directory.
+
+    Args:
+        project_name: Name of the project
+
+    Returns:
+        Path to bisync state directory for this project
+    """
+    return Path.home() / ".basic-memory" / "bisync-state" / project_name
+
+
+def bisync_initialized(project_name: str) -> bool:
+    """Check if bisync has been initialized for this project.
+
+    Args:
+        project_name: Name of the project
+
+    Returns:
+        True if bisync state exists, False otherwise
+    """
+    state_path = get_project_bisync_state(project_name)
+    return state_path.exists() and any(state_path.iterdir())
+
+
+def get_project_remote(project: SyncProject, bucket_name: str) -> str:
+    """Build rclone remote path for project.
+
+    Args:
+        project: Project with cloud path
+        bucket_name: S3 bucket name
+
+    Returns:
+        Remote path like "basic-memory-cloud:bucket-name/basic-memory-llc"
+
+    Note:
+        The API returns paths like "/app/data/basic-memory-llc" because the S3 bucket
+        is mounted at /app/data on the fly machine. We need to strip the /app/data/
+        prefix to get the actual S3 path within the bucket.
+    """
+    # Strip /app/data/ prefix from cloud path (mount point on fly machine)
+    cloud_path = project.path.lstrip("/")
+    if cloud_path.startswith("app/data/"):
+        cloud_path = cloud_path.removeprefix("app/data/")
+    return f"basic-memory-cloud:{bucket_name}/{cloud_path}"
+
+
+def project_sync(
+    project: SyncProject,
+    bucket_name: str,
+    dry_run: bool = False,
+    verbose: bool = False,
+) -> bool:
+    """One-way sync: local → cloud.
+
+    Makes cloud identical to local using rclone sync.
+
+    Args:
+        project: Project to sync
+        bucket_name: S3 bucket name
+        dry_run: Preview changes without applying
+        verbose: Show detailed output
+
+    Returns:
+        True if sync succeeded, False otherwise
+
+    Raises:
+        RcloneError: If project has no local_sync_path configured
+    """
+    if not project.local_sync_path:
+        raise RcloneError(f"Project {project.name} has no local_sync_path configured")
+
+    local_path = Path(project.local_sync_path).expanduser()
+    remote_path = get_project_remote(project, bucket_name)
+    filter_path = get_bmignore_filter_path()
+
+    cmd = [
+        "rclone",
+        "sync",
+        str(local_path),
+        remote_path,
+        "--filter-from",
+        str(filter_path),
+    ]
+
+    if verbose:
+        cmd.append("--verbose")
+    else:
+        cmd.append("--progress")
+
+    if dry_run:
+        cmd.append("--dry-run")
+
+    result = subprocess.run(cmd, text=True)
+    return result.returncode == 0
+
+
+def project_bisync(
+    project: SyncProject,
+    bucket_name: str,
+    dry_run: bool = False,
+    resync: bool = False,
+    verbose: bool = False,
+) -> bool:
+    """Two-way sync: local ↔ cloud.
+
+    Uses rclone bisync with balanced defaults:
+    - conflict_resolve: newer (auto-resolve to most recent)
+    - max_delete: 25 (safety limit)
+    - check_access: false (skip for performance)
+
+    Args:
+        project: Project to sync
+        bucket_name: S3 bucket name
+        dry_run: Preview changes without applying
+        resync: Force resync to establish new baseline
+        verbose: Show detailed output
+
+    Returns:
+        True if bisync succeeded, False otherwise
+
+    Raises:
+        RcloneError: If project has no local_sync_path or needs --resync
+    """
+    if not project.local_sync_path:
+        raise RcloneError(f"Project {project.name} has no local_sync_path configured")
+
+    local_path = Path(project.local_sync_path).expanduser()
+    remote_path = get_project_remote(project, bucket_name)
+    filter_path = get_bmignore_filter_path()
+    state_path = get_project_bisync_state(project.name)
+
+    # Ensure state directory exists
+    state_path.mkdir(parents=True, exist_ok=True)
+
+    cmd = [
+        "rclone",
+        "bisync",
+        str(local_path),
+        remote_path,
+        "--create-empty-src-dirs",
+        "--resilient",
+        "--conflict-resolve=newer",
+        "--max-delete=25",
+        "--filter-from",
+        str(filter_path),
+        "--workdir",
+        str(state_path),
+    ]
+
+    if verbose:
+        cmd.append("--verbose")
+    else:
+        cmd.append("--progress")
+
+    if dry_run:
+        cmd.append("--dry-run")
+
+    if resync:
+        cmd.append("--resync")
+
+    # Check if first run requires resync
+    if not resync and not bisync_initialized(project.name) and not dry_run:
+        raise RcloneError(
+            f"First bisync for {project.name} requires --resync to establish baseline.\n"
+            f"Run: bm project bisync --name {project.name} --resync"
+        )
+
+    result = subprocess.run(cmd, text=True)
+    return result.returncode == 0
+
+
+def project_check(
+    project: SyncProject,
+    bucket_name: str,
+    one_way: bool = False,
+) -> bool:
+    """Check integrity between local and cloud.
+
+    Verifies files match without transferring data.
+
+    Args:
+        project: Project to check
+        bucket_name: S3 bucket name
+        one_way: Only check for missing files on destination (faster)
+
+    Returns:
+        True if files match, False if differences found
+
+    Raises:
+        RcloneError: If project has no local_sync_path configured
+    """
+    if not project.local_sync_path:
+        raise RcloneError(f"Project {project.name} has no local_sync_path configured")
+
+    local_path = Path(project.local_sync_path).expanduser()
+    remote_path = get_project_remote(project, bucket_name)
+    filter_path = get_bmignore_filter_path()
+
+    cmd = [
+        "rclone",
+        "check",
+        str(local_path),
+        remote_path,
+        "--filter-from",
+        str(filter_path),
+    ]
+
+    if one_way:
+        cmd.append("--one-way")
+
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    return result.returncode == 0
+
+
+def project_ls(
+    project: SyncProject,
+    bucket_name: str,
+    path: Optional[str] = None,
+) -> list[str]:
+    """List files in remote project.
+
+    Args:
+        project: Project to list files from
+        bucket_name: S3 bucket name
+        path: Optional subdirectory within project
+
+    Returns:
+        List of file paths
+
+    Raises:
+        subprocess.CalledProcessError: If rclone command fails
+    """
+    remote_path = get_project_remote(project, bucket_name)
+    if path:
+        remote_path = f"{remote_path}/{path}"
+
+    cmd = ["rclone", "ls", remote_path]
+    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+    return result.stdout.splitlines()

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

@@ -1,11 +1,14 @@
-"""rclone configuration management for Basic Memory Cloud."""
+"""rclone configuration management for Basic Memory Cloud.
+
+This module provides simplified rclone configuration for SPEC-20.
+Uses a single "basic-memory-cloud" remote for all operations.
+"""
 
 import configparser
 import os
 import shutil
-import subprocess
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Optional
 
 from rich.console import Console
 
@@ -18,64 +21,6 @@ class RcloneConfigError(Exception):
     pass
 
 
-class RcloneMountProfile:
-    """Mount profile with optimized settings."""
-
-    def __init__(
-        self,
-        name: str,
-        cache_time: str,
-        poll_interval: str,
-        attr_timeout: str,
-        write_back: str,
-        description: str,
-        extra_args: Optional[List[str]] = None,
-    ):
-        self.name = name
-        self.cache_time = cache_time
-        self.poll_interval = poll_interval
-        self.attr_timeout = attr_timeout
-        self.write_back = write_back
-        self.description = description
-        self.extra_args = extra_args or []
-
-
-# Mount profiles based on SPEC-7 Phase 4 testing
-MOUNT_PROFILES = {
-    "fast": RcloneMountProfile(
-        name="fast",
-        cache_time="5s",
-        poll_interval="3s",
-        attr_timeout="3s",
-        write_back="1s",
-        description="Ultra-fast development (5s sync, higher bandwidth)",
-    ),
-    "balanced": RcloneMountProfile(
-        name="balanced",
-        cache_time="10s",
-        poll_interval="5s",
-        attr_timeout="5s",
-        write_back="2s",
-        description="Fast development (10-15s sync, recommended)",
-    ),
-    "safe": RcloneMountProfile(
-        name="safe",
-        cache_time="15s",
-        poll_interval="10s",
-        attr_timeout="10s",
-        write_back="5s",
-        description="Conflict-aware mount with backup",
-        extra_args=[
-            "--conflict-suffix",
-            ".conflict-{DateTimeExt}",
-            "--backup-dir",
-            "~/.basic-memory/conflicts",
-            "--track-renames",
-        ],
-    ),
-}
-
-
 def get_rclone_config_path() -> Path:
     """Get the path to rclone configuration file."""
     config_dir = Path.home() / ".config" / "rclone"
@@ -116,28 +61,38 @@ def save_rclone_config(config: configparser.ConfigParser) -> None:
     console.print(f"[dim]Updated rclone config: {config_path}[/dim]")
 
 
-def add_tenant_to_rclone_config(
-    tenant_id: str,
-    bucket_name: str,
+def configure_rclone_remote(
     access_key: str,
     secret_key: str,
     endpoint: str = "https://fly.storage.tigris.dev",
     region: str = "auto",
 ) -> str:
-    """Add tenant configuration to rclone config file."""
+    """Configure single rclone remote named 'basic-memory-cloud'.
 
+    This is the simplified approach from SPEC-20 that uses one remote
+    for all Basic Memory cloud operations (not tenant-specific).
+
+    Args:
+        access_key: S3 access key ID
+        secret_key: S3 secret access key
+        endpoint: S3-compatible endpoint URL
+        region: S3 region (default: auto)
+
+    Returns:
+        The remote name: "basic-memory-cloud"
+    """
     # Backup existing config
     backup_rclone_config()
 
     # Load existing config
     config = load_rclone_config()
 
-    # Create section name
-    section_name = f"basic-memory-{tenant_id}"
+    # Single remote name (not tenant-specific)
+    REMOTE_NAME = "basic-memory-cloud"
 
-    # Add/update the tenant section
-    if not config.has_section(section_name):
-        config.add_section(section_name)
+    # Add/update the remote section
+    if not config.has_section(REMOTE_NAME):
+        config.add_section(REMOTE_NAME)
 
     config.set(section_name, "type", "s3")
     config.set(section_name, "provider", "Other")
@@ -145,144 +100,12 @@ def add_tenant_to_rclone_config(
     config.set(section_name, "secret_access_key", secret_key)
     config.set(section_name, "endpoint", endpoint)
     config.set(section_name, "region", region)
+    # Prevent unnecessary encoding of filenames (only encode slashes and invalid UTF-8)
+    # This prevents files with spaces like "Hello World.md" from being quoted
+    config.set(section_name, "encoding", "Slash,InvalidUtf8")
 
     # Save updated config
     save_rclone_config(config)
 
-    console.print(f"[green]✓ Added tenant {tenant_id} to rclone config[/green]")
-    return section_name
-
-
-def remove_tenant_from_rclone_config(tenant_id: str) -> bool:
-    """Remove tenant configuration from rclone config."""
-    config = load_rclone_config()
-    section_name = f"basic-memory-{tenant_id}"
-
-    if config.has_section(section_name):
-        backup_rclone_config()
-        config.remove_section(section_name)
-        save_rclone_config(config)
-        console.print(f"[green]✓ Removed tenant {tenant_id} from rclone config[/green]")
-        return True
-
-    return False
-
-
-def get_default_mount_path() -> Path:
-    """Get default mount path (fixed location per SPEC-9).
-
-    Returns:
-        Fixed mount path: ~/basic-memory-cloud/
-    """
-    return Path.home() / "basic-memory-cloud"
-
-
-def build_mount_command(
-    tenant_id: str, bucket_name: str, mount_path: Path, profile: RcloneMountProfile
-) -> List[str]:
-    """Build rclone mount command with optimized settings."""
-
-    rclone_remote = f"basic-memory-{tenant_id}:{bucket_name}"
-
-    cmd = [
-        "rclone",
-        "nfsmount",
-        rclone_remote,
-        str(mount_path),
-        "--vfs-cache-mode",
-        "writes",
-        "--dir-cache-time",
-        profile.cache_time,
-        "--vfs-cache-poll-interval",
-        profile.poll_interval,
-        "--attr-timeout",
-        profile.attr_timeout,
-        "--vfs-write-back",
-        profile.write_back,
-        "--daemon",
-    ]
-
-    # Add profile-specific extra arguments
-    cmd.extend(profile.extra_args)
-
-    return cmd
-
-
-def is_path_mounted(mount_path: Path) -> bool:
-    """Check if a path is currently mounted."""
-    if not mount_path.exists():
-        return False
-
-    try:
-        # Check if mount point is actually mounted by looking for mount table entry
-        result = subprocess.run(["mount"], capture_output=True, text=True, check=False)
-
-        if result.returncode == 0:
-            # Look for our mount path in mount output
-            mount_str = str(mount_path.resolve())
-            return mount_str in result.stdout
-
-        return False
-    except Exception:
-        return False
-
-
-def get_rclone_processes() -> List[Dict[str, str]]:
-    """Get list of running rclone processes."""
-    try:
-        # Use ps to find rclone processes
-        result = subprocess.run(
-            ["ps", "-eo", "pid,args"], capture_output=True, text=True, check=False
-        )
-
-        processes = []
-        if result.returncode == 0:
-            for line in result.stdout.split("\n"):
-                if "rclone" in line and "basic-memory" in line:
-                    parts = line.strip().split(None, 1)
-                    if len(parts) >= 2:
-                        processes.append({"pid": parts[0], "command": parts[1]})
-
-        return processes
-    except Exception:
-        return []
-
-
-def kill_rclone_process(pid: str) -> bool:
-    """Kill a specific rclone process."""
-    try:
-        subprocess.run(["kill", pid], check=True)
-        console.print(f"[green]✓ Killed rclone process {pid}[/green]")
-        return True
-    except subprocess.CalledProcessError:
-        console.print(f"[red]✗ Failed to kill rclone process {pid}[/red]")
-        return False
-
-
-def unmount_path(mount_path: Path) -> bool:
-    """Unmount a mounted path."""
-    if not is_path_mounted(mount_path):
-        return True
-
-    try:
-        subprocess.run(["umount", str(mount_path)], check=True)
-        console.print(f"[green]✓ Unmounted {mount_path}[/green]")
-        return True
-    except subprocess.CalledProcessError as e:
-        console.print(f"[red]✗ Failed to unmount {mount_path}: {e}[/red]")
-        return False
-
-
-def cleanup_orphaned_rclone_processes() -> int:
-    """Clean up orphaned rclone processes for basic-memory."""
-    processes = get_rclone_processes()
-    killed_count = 0
-
-    for proc in processes:
-        console.print(
-            f"[yellow]Found rclone process: {proc['pid']} - {proc['command'][:80]}...[/yellow]"
-        )
-        if kill_rclone_process(proc["pid"]):
-            killed_count += 1
-
-    return killed_count
+    console.print(f"[green]✓ Configured rclone remote: {REMOTE_NAME}[/green]")
+    return REMOTE_NAME

src/basic_memory/cli/commands/db.py

@@ -37,8 +37,8 @@ def reset(
         logger.info("Database reset complete")
 
         if reindex:
-            # Import and run sync
-            from basic_memory.cli.commands.sync import sync
+            # Run database sync directly
+            from basic_memory.cli.commands.command_utils import run_sync
 
             logger.info("Rebuilding search index from filesystem...")
-            sync(watch=False)  # pyright: ignore
+            asyncio.run(run_sync(project=None))

src/basic_memory/cli/commands/project.py

@@ -25,6 +25,17 @@ from basic_memory.mcp.tools.utils import call_put
 from basic_memory.utils import generate_permalink
 from basic_memory.mcp.tools.utils import call_patch
 
+# Import rclone commands for project sync
+from basic_memory.cli.commands.cloud.rclone_commands import (
+    SyncProject,
+    RcloneError,
+    project_sync,
+    project_bisync,
+    project_check,
+    project_ls,
+)
+from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
+
 console = Console()
 
 # Create a project subcommand
@@ -32,6 +43,23 @@ project_app = typer.Typer(help="Manage multiple Basic Memory projects")
 app.add_typer(project_app, name="project")
 
 
+def normalize_project_path(path: str) -> str:
+    """Normalize project path by stripping mount point prefix.
+
+    In cloud deployments, the S3 bucket is mounted at /app/data. We strip this
+    prefix to get the actual S3 bucket path and avoid leaking implementation details.
+
+    Args:
+        path: Project path (e.g., "/app/data/basic-memory-llc")
+
+    Returns:
+        Normalized path (e.g., "/basic-memory-llc")
+    """
+    if path.startswith("/app/data/"):
+        return path.removeprefix("/app/data")
+    return path
+
+
 def format_path(path: str) -> str:
     """Format a path for display, using ~ for home directory."""
     home = str(Path.home())
@@ -51,15 +79,41 @@ def list_projects() -> None:
 
     try:
         result = asyncio.run(_list_projects())
+        config = ConfigManager().config
 
         table = Table(title="Basic Memory Projects")
         table.add_column("Name", style="cyan")
         table.add_column("Path", style="green")
-        table.add_column("Default", style="magenta")
+
+        # Add Local Path column if in cloud mode
+        if config.cloud_mode_enabled:
+            table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
+
+        # Show Default column in local mode or if default_project_mode is enabled in cloud mode
+        show_default_column = not config.cloud_mode_enabled or config.default_project_mode
+        if show_default_column:
+            table.add_column("Default", style="magenta")
 
         for project in result.projects:
             is_default = "✓" if project.is_default else ""
-            table.add_row(project.name, format_path(project.path), is_default)
+            normalized_path = normalize_project_path(project.path)
+
+            # Build row based on mode
+            row = [project.name, format_path(normalized_path)]
+
+            # Add local path if in cloud mode
+            if config.cloud_mode_enabled:
+                local_path = ""
+                if project.name in config.cloud_projects:
+                    local_path = config.cloud_projects[project.name].local_path or ""
+                    local_path = format_path(local_path)
+                row.append(local_path)
+
+            # Add default indicator if showing default column
+            if show_default_column:
+                row.append(is_default)
+
+            table.add_row(*row)
 
         console.print(table)
     except Exception as e:
@@ -73,20 +127,38 @@ def add_project(
     path: str = typer.Argument(
         None, help="Path to the project directory (required for local mode)"
     ),
+    local_path: str = typer.Option(
+        None, "--local-path", help="Local sync path for cloud mode (optional)"
+    ),
     set_default: bool = typer.Option(False, "--default", help="Set as default project"),
 ) -> None:
     """Add a new project.
 
-    For cloud mode: only name is required
-    For local mode: both name and path are required
+    Cloud mode examples:\n
+        bm project add research                           # No local sync\n
+        bm project add research --local-path ~/docs       # With local sync\n
+
+    Local mode example:\n
+        bm project add research ~/Documents/research
     """
     config = ConfigManager().config
 
+    # Resolve local sync path early (needed for both cloud and local mode)
+    local_sync_path: str | None = None
+    if local_path:
+        local_sync_path = Path(os.path.abspath(os.path.expanduser(local_path))).as_posix()
+
     if config.cloud_mode_enabled:
-        # Cloud mode: path not needed (auto-generated from name)
+        # Cloud mode: path auto-generated from name, local sync is optional
+
         async def _add_project():
             async with get_client() as client:
-                data = {"name": name, "path": generate_permalink(name), "set_default": set_default}
+                data = {
+                    "name": name,
+                    "path": generate_permalink(name),
+                    "local_sync_path": local_sync_path,
+                    "set_default": set_default,
+                }
                 response = await call_post(client, "/projects/projects", json=data)
                 return ProjectStatusResponse.model_validate(response.json())
     else:
@@ -107,13 +179,85 @@ def add_project(
     try:
         result = asyncio.run(_add_project())
         console.print(f"[green]{result.message}[/green]")
+
+        # Save local sync path to config if in cloud mode
+        if config.cloud_mode_enabled and local_sync_path:
+            from basic_memory.config import CloudProjectConfig
+
+            # Create local directory if it doesn't exist
+            local_dir = Path(local_sync_path)
+            local_dir.mkdir(parents=True, exist_ok=True)
+
+            # Update config with sync path
+            config.cloud_projects[name] = CloudProjectConfig(
+                local_path=local_sync_path,
+                last_sync=None,
+                bisync_initialized=False,
+            )
+            ConfigManager().save_config(config)
+
+            console.print(f"\n[green]✓ Local sync path configured: {local_sync_path}[/green]")
+            console.print("\nNext steps:")
+            console.print(f"  1. Preview: bm project bisync --name {name} --resync --dry-run")
+            console.print(f"  2. Sync: bm project bisync --name {name} --resync")
     except Exception as e:
         console.print(f"[red]Error adding project: {str(e)}[/red]")
         raise typer.Exit(1)
 
-    # Display usage hint
-    console.print("\nTo use this project:")
-    console.print(f"  basic-memory --project={name} <command>")
+
+@project_app.command("sync-setup")
+def setup_project_sync(
+    name: str = typer.Argument(..., help="Project name"),
+    local_path: str = typer.Argument(..., help="Local sync directory"),
+) -> None:
+    """Configure local sync for an existing cloud project.
+
+    Example:
+      bm project sync-setup research ~/Documents/research
+    """
+    config_manager = ConfigManager()
+    config = config_manager.config
+
+    if not config.cloud_mode_enabled:
+        console.print("[red]Error: sync-setup only available in cloud mode[/red]")
+        raise typer.Exit(1)
+
+    async def _verify_project_exists():
+        """Verify the project exists on cloud by listing all projects."""
+        async with get_client() as client:
+            response = await call_get(client, "/projects/projects")
+            project_list = response.json()
+            project_names = [p["name"] for p in project_list["projects"]]
+            if name not in project_names:
+                raise ValueError(f"Project '{name}' not found on cloud")
+            return True
+
+    try:
+        # Verify project exists on cloud
+        asyncio.run(_verify_project_exists())
+
+        # Resolve and create local path
+        resolved_path = Path(os.path.abspath(os.path.expanduser(local_path)))
+        resolved_path.mkdir(parents=True, exist_ok=True)
+
+        # Update local config with sync path
+        from basic_memory.config import CloudProjectConfig
+
+        config.cloud_projects[name] = CloudProjectConfig(
+            local_path=resolved_path.as_posix(),
+            last_sync=None,
+            bisync_initialized=False,
+        )
+        config_manager.save_config(config)
+
+        console.print(f"[green]✓ Sync configured for project '{name}'[/green]")
+        console.print(f"\nLocal sync path: {resolved_path}")
+        console.print("\nNext steps:")
+        console.print(f"  1. Preview: bm project bisync --name {name} --resync --dry-run")
+        console.print(f"  2. Sync: bm project bisync --name {name} --resync")
+    except Exception as e:
+        console.print(f"[red]Error configuring sync: {str(e)}[/red]")
+        raise typer.Exit(1)
 
 
 @project_app.command("remove")
@@ -134,16 +278,59 @@ def remove_project(
             return ProjectStatusResponse.model_validate(response.json())
 
     try:
+        # Get config to check for local sync path and bisync state
+        config = ConfigManager().config
+        local_path = None
+        has_bisync_state = False
+
+        if config.cloud_mode_enabled and name in config.cloud_projects:
+            local_path = config.cloud_projects[name].local_path
+
+            # Check for bisync state
+            from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
+
+            bisync_state_path = get_project_bisync_state(name)
+            has_bisync_state = bisync_state_path.exists()
+
+        # Remove project from cloud/API
         result = asyncio.run(_remove_project())
         console.print(f"[green]{result.message}[/green]")
+
+        # Clean up local sync directory if it exists and delete_notes is True
+        if delete_notes and local_path:
+            local_dir = Path(local_path)
+            if local_dir.exists():
+                import shutil
+
+                shutil.rmtree(local_dir)
+                console.print(f"[green]✓ Removed local sync directory: {local_path}[/green]")
+
+        # Clean up bisync state if it exists
+        if has_bisync_state:
+            from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
+            import shutil
+
+            bisync_state_path = get_project_bisync_state(name)
+            if bisync_state_path.exists():
+                shutil.rmtree(bisync_state_path)
+                console.print("[green]✓ Removed bisync state[/green]")
+
+        # Clean up cloud_projects config entry
+        if config.cloud_mode_enabled and name in config.cloud_projects:
+            del config.cloud_projects[name]
+            ConfigManager().save_config(config)
+
+        # Show informative message if files were not deleted
+        if not delete_notes:
+            if local_path:
+                console.print(f"[yellow]Note: Local files remain at {local_path}[/yellow]")
+            else:
+                console.print("[yellow]Note: Cloud project files have not been deleted.[/yellow]")
+
     except Exception as e:
         console.print(f"[red]Error removing project: {str(e)}[/red]")
         raise typer.Exit(1)
 
-    # Show this message only if files were not deleted
-    if not delete_notes:
-        console.print("[yellow]Note: The project files have not been deleted from disk.[/yellow]")
-
 
 @project_app.command("default")
 def set_default_project(
@@ -248,6 +435,349 @@ def move_project(
         raise typer.Exit(1)
 
 
+@project_app.command("sync")
+def sync_project_command(
+    name: str = typer.Option(..., "--name", help="Project name to sync"),
+    dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
+) -> None:
+    """One-way sync: local → cloud (make cloud identical to local).
+
+    Example:
+      bm project sync --name research
+      bm project sync --name research --dry-run
+    """
+    config = ConfigManager().config
+    if not config.cloud_mode_enabled:
+        console.print("[red]Error: sync only available in cloud mode[/red]")
+        raise typer.Exit(1)
+
+    try:
+        # Get tenant info for bucket name
+        tenant_info = asyncio.run(get_mount_info())
+        bucket_name = tenant_info.bucket_name
+
+        # Get project info
+        async def _get_project():
+            async with get_client() as client:
+                response = await call_get(client, "/projects/projects")
+                projects_list = ProjectList.model_validate(response.json())
+                for proj in projects_list.projects:
+                    if generate_permalink(proj.name) == generate_permalink(name):
+                        return proj
+                return None
+
+        project_data = asyncio.run(_get_project())
+        if not project_data:
+            console.print(f"[red]Error: Project '{name}' not found[/red]")
+            raise typer.Exit(1)
+
+        # Get local_sync_path from cloud_projects config
+        local_sync_path = None
+        if name in config.cloud_projects:
+            local_sync_path = config.cloud_projects[name].local_path
+
+        if not local_sync_path:
+            console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
+            console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
+            raise typer.Exit(1)
+
+        # Create SyncProject
+        sync_project = SyncProject(
+            name=project_data.name,
+            path=normalize_project_path(project_data.path),
+            local_sync_path=local_sync_path,
+        )
+
+        # Run sync
+        console.print(f"[blue]Syncing {name} (local → cloud)...[/blue]")
+        success = project_sync(sync_project, bucket_name, dry_run=dry_run, verbose=verbose)
+
+        if success:
+            console.print(f"[green]✓ {name} synced successfully[/green]")
+
+            # Trigger database sync if not a dry run
+            if not dry_run:
+
+                async def _trigger_db_sync():
+                    async with get_client() as client:
+                        permalink = generate_permalink(name)
+                        response = await call_post(client, f"/{permalink}/project/sync", json={})
+                        return response.json()
+
+                try:
+                    result = asyncio.run(_trigger_db_sync())
+                    console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
+                except Exception as e:
+                    console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
+        else:
+            console.print(f"[red]✗ {name} sync failed[/red]")
+            raise typer.Exit(1)
+
+    except RcloneError as e:
+        console.print(f"[red]Sync error: {e}[/red]")
+        raise typer.Exit(1)
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("bisync")
+def bisync_project_command(
+    name: str = typer.Option(..., "--name", help="Project name to bisync"),
+    dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
+    resync: bool = typer.Option(False, "--resync", help="Force new baseline"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
+) -> None:
+    """Two-way sync: local ↔ cloud (bidirectional sync).
+
+    Examples:
+      bm project bisync --name research --resync  # First time
+      bm project bisync --name research           # Subsequent syncs
+      bm project bisync --name research --dry-run # Preview changes
+    """
+    config = ConfigManager().config
+    if not config.cloud_mode_enabled:
+        console.print("[red]Error: bisync only available in cloud mode[/red]")
+        raise typer.Exit(1)
+
+    try:
+        # Get tenant info for bucket name
+        tenant_info = asyncio.run(get_mount_info())
+        bucket_name = tenant_info.bucket_name
+
+        # Get project info
+        async def _get_project():
+            async with get_client() as client:
+                response = await call_get(client, "/projects/projects")
+                projects_list = ProjectList.model_validate(response.json())
+                for proj in projects_list.projects:
+                    if generate_permalink(proj.name) == generate_permalink(name):
+                        return proj
+                return None
+
+        project_data = asyncio.run(_get_project())
+        if not project_data:
+            console.print(f"[red]Error: Project '{name}' not found[/red]")
+            raise typer.Exit(1)
+
+        # Get local_sync_path from cloud_projects config
+        local_sync_path = None
+        if name in config.cloud_projects:
+            local_sync_path = config.cloud_projects[name].local_path
+
+        if not local_sync_path:
+            console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
+            console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
+            raise typer.Exit(1)
+
+        # Create SyncProject
+        sync_project = SyncProject(
+            name=project_data.name,
+            path=normalize_project_path(project_data.path),
+            local_sync_path=local_sync_path,
+        )
+
+        # Run bisync
+        console.print(f"[blue]Bisync {name} (local ↔ cloud)...[/blue]")
+        success = project_bisync(
+            sync_project, bucket_name, dry_run=dry_run, resync=resync, verbose=verbose
+        )
+
+        if success:
+            console.print(f"[green]✓ {name} bisync completed successfully[/green]")
+
+            # Update config
+            config.cloud_projects[name].last_sync = datetime.now()
+            config.cloud_projects[name].bisync_initialized = True
+            ConfigManager().save_config(config)
+
+            # Trigger database sync if not a dry run
+            if not dry_run:
+
+                async def _trigger_db_sync():
+                    async with get_client() as client:
+                        permalink = generate_permalink(name)
+                        response = await call_post(client, f"/{permalink}/project/sync", json={})
+                        return response.json()
+
+                try:
+                    result = asyncio.run(_trigger_db_sync())
+                    console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
+                except Exception as e:
+                    console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
+        else:
+            console.print(f"[red]✗ {name} bisync failed[/red]")
+            raise typer.Exit(1)
+
+    except RcloneError as e:
+        console.print(f"[red]Bisync error: {e}[/red]")
+        raise typer.Exit(1)
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("check")
+def check_project_command(
+    name: str = typer.Option(..., "--name", help="Project name to check"),
+    one_way: bool = typer.Option(False, "--one-way", help="Check one direction only (faster)"),
+) -> None:
+    """Verify file integrity between local and cloud.
+
+    Example:
+      bm project check --name research
+    """
+    config = ConfigManager().config
+    if not config.cloud_mode_enabled:
+        console.print("[red]Error: check only available in cloud mode[/red]")
+        raise typer.Exit(1)
+
+    try:
+        # Get tenant info for bucket name
+        tenant_info = asyncio.run(get_mount_info())
+        bucket_name = tenant_info.bucket_name
+
+        # Get project info
+        async def _get_project():
+            async with get_client() as client:
+                response = await call_get(client, "/projects/projects")
+                projects_list = ProjectList.model_validate(response.json())
+                for proj in projects_list.projects:
+                    if generate_permalink(proj.name) == generate_permalink(name):
+                        return proj
+                return None
+
+        project_data = asyncio.run(_get_project())
+        if not project_data:
+            console.print(f"[red]Error: Project '{name}' not found[/red]")
+            raise typer.Exit(1)
+
+        # Get local_sync_path from cloud_projects config
+        local_sync_path = None
+        if name in config.cloud_projects:
+            local_sync_path = config.cloud_projects[name].local_path
+
+        if not local_sync_path:
+            console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
+            console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
+            raise typer.Exit(1)
+
+        # Create SyncProject
+        sync_project = SyncProject(
+            name=project_data.name,
+            path=normalize_project_path(project_data.path),
+            local_sync_path=local_sync_path,
+        )
+
+        # Run check
+        console.print(f"[blue]Checking {name} integrity...[/blue]")
+        match = project_check(sync_project, bucket_name, one_way=one_way)
+
+        if match:
+            console.print(f"[green]✓ {name} files match[/green]")
+        else:
+            console.print(f"[yellow]⚠ {name} has differences[/yellow]")
+
+    except RcloneError as e:
+        console.print(f"[red]Check error: {e}[/red]")
+        raise typer.Exit(1)
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("bisync-reset")
+def bisync_reset(
+    name: str = typer.Argument(..., help="Project name to reset bisync state for"),
+) -> None:
+    """Clear bisync state for a project.
+
+    This removes the bisync metadata files, forcing a fresh --resync on next bisync.
+    Useful when bisync gets into an inconsistent state or when remote path changes.
+    """
+    from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
+    import shutil
+
+    try:
+        state_path = get_project_bisync_state(name)
+
+        if not state_path.exists():
+            console.print(f"[yellow]No bisync state found for project '{name}'[/yellow]")
+            return
+
+        # Remove the entire state directory
+        shutil.rmtree(state_path)
+        console.print(f"[green]✓ Cleared bisync state for project '{name}'[/green]")
+        console.print("\nNext steps:")
+        console.print(f"  1. Preview: bm project bisync --name {name} --resync --dry-run")
+        console.print(f"  2. Sync: bm project bisync --name {name} --resync")
+
+    except Exception as e:
+        console.print(f"[red]Error clearing bisync state: {str(e)}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("ls")
+def ls_project_command(
+    name: str = typer.Option(..., "--name", help="Project name to list files from"),
+    path: str = typer.Argument(None, help="Path within project (optional)"),
+) -> None:
+    """List files in remote project.
+
+    Examples:
+      bm project ls --name research
+      bm project ls --name research subfolder
+    """
+    config = ConfigManager().config
+    if not config.cloud_mode_enabled:
+        console.print("[red]Error: ls only available in cloud mode[/red]")
+        raise typer.Exit(1)
+
+    try:
+        # Get tenant info for bucket name
+        tenant_info = asyncio.run(get_mount_info())
+        bucket_name = tenant_info.bucket_name
+
+        # Get project info
+        async def _get_project():
+            async with get_client() as client:
+                response = await call_get(client, "/projects/projects")
+                projects_list = ProjectList.model_validate(response.json())
+                for proj in projects_list.projects:
+                    if generate_permalink(proj.name) == generate_permalink(name):
+                        return proj
+                return None
+
+        project_data = asyncio.run(_get_project())
+        if not project_data:
+            console.print(f"[red]Error: Project '{name}' not found[/red]")
+            raise typer.Exit(1)
+
+        # Create SyncProject (local_sync_path not needed for ls)
+        sync_project = SyncProject(
+            name=project_data.name,
+            path=normalize_project_path(project_data.path),
+        )
+
+        # List files
+        files = project_ls(sync_project, bucket_name, path=path)
+
+        if files:
+            console.print(f"\n[bold]Files in {name}" + (f"/{path}" if path else "") + ":[/bold]")
+            for file in files:
+                console.print(f"  {file}")
+            console.print(f"\n[dim]Total: {len(files)} files[/dim]")
+        else:
+            console.print(
+                f"[yellow]No files found in {name}" + (f"/{path}" if path else "") + "[/yellow]"
+            )
+
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
 @project_app.command("info")
 def display_project_info(
     name: str = typer.Argument(..., help="Name of the project"),

src/basic_memory/cli/commands/sync.py

@@ -1,59 +0,0 @@
-"""Command module for basic-memory sync operations."""
-
-import asyncio
-from typing import Annotated, Optional
-
-import typer
-
-from basic_memory.cli.app import app
-from basic_memory.cli.commands.command_utils import run_sync
-from basic_memory.config import ConfigManager
-
-
-@app.command()
-def sync(
-    project: Annotated[
-        Optional[str],
-        typer.Option(help="The project name."),
-    ] = None,
-    watch: Annotated[
-        bool,
-        typer.Option("--watch", help="Run continuous sync (cloud mode only)"),
-    ] = False,
-    interval: Annotated[
-        int,
-        typer.Option("--interval", help="Sync interval in seconds for watch mode (default: 60)"),
-    ] = 60,
-) -> None:
-    """Sync knowledge files with the database.
-
-    In local mode: Scans filesystem and updates database.
-    In cloud mode: Runs bidirectional file sync (bisync) then updates database.
-
-    Examples:
-      bm sync                    # One-time sync
-      bm sync --watch            # Continuous sync every 60s
-      bm sync --watch --interval 30  # Continuous sync every 30s
-    """
-    config = ConfigManager().config
-
-    if config.cloud_mode_enabled:
-        # Cloud mode: run bisync which includes database sync
-        from basic_memory.cli.commands.cloud.bisync_commands import run_bisync, run_bisync_watch
-
-        try:
-            if watch:
-                run_bisync_watch(interval_seconds=interval)
-            else:
-                run_bisync()
-        except Exception:
-            raise typer.Exit(1)
-    else:
-        # Local mode: just database sync
-        if watch:
-            typer.echo(
-                "Error: --watch is only available in cloud mode. Run 'bm cloud login' first."
-            )
-            raise typer.Exit(1)
-
-        asyncio.run(run_sync(project))

src/basic_memory/cli/main.py

@@ -13,7 +13,6 @@ from basic_memory.cli.commands import (  # noqa: F401  # pragma: no cover
     mcp,
     project,
     status,
-    sync,
     tool,
 )
 

src/basic_memory/config.py

@@ -3,11 +3,12 @@
 import json
 import os
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, Literal, Optional, List, Tuple
 
 from loguru import logger
-from pydantic import Field, field_validator
+from pydantic import BaseModel, Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 import basic_memory
@@ -39,6 +40,22 @@ class ProjectConfig:
         return f"/{generate_permalink(self.name)}"
 
 
+class CloudProjectConfig(BaseModel):
+    """Sync configuration for a cloud project.
+
+    This tracks the local working directory and sync state for a project
+    that is synced with Basic Memory Cloud.
+    """
+
+    local_path: str = Field(description="Local working directory path for this cloud project")
+    last_sync: Optional[datetime] = Field(
+        default=None, description="Timestamp of last successful sync operation"
+    )
+    bisync_initialized: bool = Field(
+        default=False, description="Whether rclone bisync baseline has been established"
+    )
+
+
 class BasicMemoryConfig(BaseSettings):
     """Pydantic model for Basic Memory global configuration."""
 
@@ -138,6 +155,11 @@ class BasicMemoryConfig(BaseSettings):
         description="Enable cloud mode - all requests go to cloud instead of local (config file value)",
     )
 
+    cloud_projects: Dict[str, CloudProjectConfig] = Field(
+        default_factory=dict,
+        description="Cloud project sync configuration mapping project names to their local paths and sync state",
+    )
+
     @property
     def cloud_mode_enabled(self) -> bool:
         """Check if cloud mode is enabled.
@@ -154,14 +176,6 @@ class BasicMemoryConfig(BaseSettings):
         # Fall back to config file value
         return self.cloud_mode
 
-    bisync_config: Dict[str, Any] = Field(
-        default_factory=lambda: {
-            "profile": "balanced",
-            "sync_dir": str(Path.home() / "basic-memory-cloud-sync"),
-        },
-        description="Bisync configuration for cloud sync",
-    )
-
     model_config = SettingsConfigDict(
         env_prefix="BASIC_MEMORY_",
         extra="ignore",
@@ -427,7 +441,9 @@ def get_project_config(project_name: Optional[str] = None) -> ProjectConfig:
 def save_basic_memory_config(file_path: Path, config: BasicMemoryConfig) -> None:
     """Save configuration to file."""
     try:
-        file_path.write_text(json.dumps(config.model_dump(), indent=2))
+        # Use model_dump with mode='json' to serialize datetime objects properly
+        config_dict = config.model_dump(mode="json")
+        file_path.write_text(json.dumps(config_dict, indent=2))
     except Exception as e:  # pragma: no cover
         logger.error(f"Failed to save config: {e}")
 

src/basic_memory/services/project_service.py

@@ -234,16 +234,26 @@ class ProjectService:
         if not self.repository:  # pragma: no cover
             raise ValueError("Repository is required for remove_project")
 
-        # Get project path before removing from config
+        # Get project from database first
         project = await self.get_project(name)
-        project_path = project.path if project else None
+        if not project:
+            raise ValueError(f"Project '{name}' not found")
 
-        # First remove from config (this will validate the project exists and is not default)
-        self.config_manager.remove_project(name)
+        project_path = project.path
 
-        # Then remove from database using robust lookup
-        if project:
-            await self.repository.delete(project.id)
+        # Check if project is default (in cloud mode, check database; in local mode, check config)
+        if project.is_default or name == self.config_manager.config.default_project:
+            raise ValueError(f"Cannot remove the default project '{name}'")
+
+        # Remove from config if it exists there (may not exist in cloud mode)
+        try:
+            self.config_manager.remove_project(name)
+        except ValueError:
+            # Project not in config - that's OK in cloud mode, continue with database deletion
+            logger.debug(f"Project '{name}' not found in config, removing from database only")
+
+        # Remove from database
+        await self.repository.delete(project.id)
 
         logger.info(f"Project '{name}' removed from configuration and database")
 

test-int/cli/test_sync_commands_integration.py

@@ -1,61 +0,0 @@
-"""Integration tests for sync CLI commands."""
-
-from pathlib import Path
-from typer.testing import CliRunner
-
-from basic_memory.cli.main import app
-
-
-def test_sync_command(app_config, test_project, config_manager, config_home):
-    """Test 'bm sync' command successfully syncs files."""
-    runner = CliRunner()
-
-    # Create a test file
-    test_file = Path(config_home) / "test-note.md"
-    test_file.write_text("# Test Note\n\nThis is a test.")
-
-    # Run sync
-    result = runner.invoke(app, ["sync", "--project", "test-project"])
-
-    if result.exit_code != 0:
-        print(f"STDOUT: {result.stdout}")
-        print(f"STDERR: {result.stderr}")
-    assert result.exit_code == 0
-    assert "sync" in result.stdout.lower() or "initiated" in result.stdout.lower()
-
-
-def test_status_command(app_config, test_project, config_manager, config_home):
-    """Test 'bm status' command shows sync status."""
-    runner = CliRunner()
-
-    # Create a test file
-    test_file = Path(config_home) / "unsynced.md"
-    test_file.write_text("# Unsynced Note\n\nThis file hasn't been synced yet.")
-
-    # Run status
-    result = runner.invoke(app, ["status", "--project", "test-project"])
-
-    if result.exit_code != 0:
-        print(f"STDOUT: {result.stdout}")
-        print(f"STDERR: {result.stderr}")
-    assert result.exit_code == 0
-    # Should show some status output
-    assert len(result.stdout) > 0
-
-
-def test_status_verbose(app_config, test_project, config_manager, config_home):
-    """Test 'bm status --verbose' shows detailed status."""
-    runner = CliRunner()
-
-    # Create a test file
-    test_file = Path(config_home) / "test.md"
-    test_file.write_text("# Test\n\nContent.")
-
-    # Run status with verbose
-    result = runner.invoke(app, ["status", "--project", "test-project", "--verbose"])
-
-    if result.exit_code != 0:
-        print(f"STDOUT: {result.stdout}")
-        print(f"STDERR: {result.stderr}")
-    assert result.exit_code == 0
-    assert len(result.stdout) > 0

tests/cli/test_bisync_commands.py

@@ -1,463 +0,0 @@
-"""Tests for bisync_commands module."""
-
-from pathlib import Path
-from unittest.mock import Mock, patch
-
-import pytest
-
-from basic_memory.cli.commands.cloud.bisync_commands import (
-    BisyncError,
-    convert_bmignore_to_rclone_filters,
-    scan_local_directories,
-    validate_bisync_directory,
-    build_bisync_command,
-    get_bisync_directory,
-    get_bisync_state_path,
-    bisync_state_exists,
-    BISYNC_PROFILES,
-)
-
-
-class TestConvertBmignoreToRcloneFilters:
-    """Tests for convert_bmignore_to_rclone_filters()."""
-
-    def test_converts_basic_patterns(self, tmp_path):
-        """Test conversion of basic gitignore patterns to rclone format."""
-        bmignore_dir = tmp_path / ".basic-memory"
-        bmignore_dir.mkdir(exist_ok=True)
-        bmignore_file = bmignore_dir / ".bmignore"
-
-        # Write test patterns
-        bmignore_file.write_text("# Comment line\nnode_modules\n*.pyc\n.git\n**/*.log\n")
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bmignore_path",
-            return_value=bmignore_file,
-        ):
-            convert_bmignore_to_rclone_filters()
-
-        # Read the generated rclone filter file
-        rclone_filter = bmignore_dir / ".bmignore.rclone"
-        assert rclone_filter.exists()
-
-        content = rclone_filter.read_text()
-        lines = content.strip().split("\n")
-
-        # Check comment preserved
-        assert "# Comment line" in lines
-
-        # Check patterns converted correctly
-        assert "- node_modules/**" in lines  # Directory without wildcard
-        assert "- *.pyc" in lines  # Wildcard pattern unchanged
-        assert "- .git/**" in lines  # Directory pattern
-        assert "- **/*.log" in lines  # Wildcard pattern unchanged
-
-    def test_handles_empty_bmignore(self, tmp_path):
-        """Test handling of empty .bmignore file."""
-        bmignore_dir = tmp_path / ".basic-memory"
-        bmignore_dir.mkdir(exist_ok=True)
-        bmignore_file = bmignore_dir / ".bmignore"
-        bmignore_file.write_text("")
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bmignore_path",
-            return_value=bmignore_file,
-        ):
-            convert_bmignore_to_rclone_filters()
-
-        rclone_filter = bmignore_dir / ".bmignore.rclone"
-        assert rclone_filter.exists()
-
-    def test_handles_missing_bmignore(self, tmp_path):
-        """Test handling when .bmignore doesn't exist."""
-        bmignore_dir = tmp_path / ".basic-memory"
-        bmignore_dir.mkdir(exist_ok=True)
-        bmignore_file = bmignore_dir / ".bmignore"
-
-        # Ensure file doesn't exist
-        if bmignore_file.exists():
-            bmignore_file.unlink()
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bmignore_path",
-            return_value=bmignore_file,
-        ):
-            with patch("basic_memory.cli.commands.cloud.bisync_commands.create_default_bmignore"):
-                convert_bmignore_to_rclone_filters()
-
-        # Should create minimal filter with .git
-        rclone_filter = bmignore_dir / ".bmignore.rclone"
-        assert rclone_filter.exists()
-        content = rclone_filter.read_text()
-        assert "- .git/**" in content
-
-
-class TestScanLocalDirectories:
-    """Tests for scan_local_directories()."""
-
-    def test_scans_existing_directories(self, tmp_path):
-        """Test scanning existing project directories."""
-        # Use a subdirectory to avoid interference from test fixtures
-        scan_dir = tmp_path / "scan_test"
-        scan_dir.mkdir()
-
-        # Create test directories
-        (scan_dir / "project1").mkdir()
-        (scan_dir / "project2").mkdir()
-        (scan_dir / "project3").mkdir()
-
-        # Create a hidden directory (should be ignored)
-        (scan_dir / ".hidden").mkdir()
-
-        # Create a file (should be ignored)
-        (scan_dir / "file.txt").write_text("test")
-
-        result = scan_local_directories(scan_dir)
-
-        assert len(result) == 3
-        assert "project1" in result
-        assert "project2" in result
-        assert "project3" in result
-        assert ".hidden" not in result
-
-    def test_handles_empty_directory(self, tmp_path):
-        """Test scanning empty directory."""
-        scan_dir = tmp_path / "empty_test"
-        scan_dir.mkdir()
-        result = scan_local_directories(scan_dir)
-        assert result == []
-
-    def test_handles_nonexistent_directory(self, tmp_path):
-        """Test scanning nonexistent directory."""
-        nonexistent = tmp_path / "does-not-exist"
-        result = scan_local_directories(nonexistent)
-        assert result == []
-
-    def test_ignores_hidden_directories(self, tmp_path):
-        """Test that hidden directories are ignored."""
-        scan_dir = tmp_path / "hidden_test"
-        scan_dir.mkdir()
-
-        (scan_dir / ".git").mkdir()
-        (scan_dir / ".cache").mkdir()
-        (scan_dir / "visible").mkdir()
-
-        result = scan_local_directories(scan_dir)
-
-        assert len(result) == 1
-        assert "visible" in result
-        assert ".git" not in result
-        assert ".cache" not in result
-
-
-class TestValidateBisyncDirectory:
-    """Tests for validate_bisync_directory()."""
-
-    def test_allows_valid_directory(self, tmp_path):
-        """Test that valid directory passes validation."""
-        bisync_dir = tmp_path / "sync"
-        bisync_dir.mkdir()
-
-        # Should not raise
-        validate_bisync_directory(bisync_dir)
-
-    def test_rejects_mount_directory(self, tmp_path):
-        """Test that mount directory is rejected."""
-        mount_dir = Path.home() / "basic-memory-cloud"
-
-        with pytest.raises(BisyncError) as exc_info:
-            validate_bisync_directory(mount_dir)
-
-        assert "mount directory" in str(exc_info.value).lower()
-
-    @patch("subprocess.run")
-    def test_rejects_mounted_directory(self, mock_run, tmp_path):
-        """Test that currently mounted directory is rejected."""
-        bisync_dir = tmp_path / "sync"
-        bisync_dir.mkdir()
-
-        # Mock mount command showing this directory is mounted
-        mock_run.return_value = Mock(
-            stdout=f"rclone on {bisync_dir} type fuse.rclone",
-            stderr="",
-            returncode=0,
-        )
-
-        with pytest.raises(BisyncError) as exc_info:
-            validate_bisync_directory(bisync_dir)
-
-        assert "currently mounted" in str(exc_info.value).lower()
-
-
-class TestBuildBisyncCommand:
-    """Tests for build_bisync_command()."""
-
-    def test_builds_basic_command(self, tmp_path):
-        """Test building basic bisync command."""
-        tenant_id = "test-tenant"
-        bucket_name = "test-bucket"
-        local_path = tmp_path / "sync"
-        local_path.mkdir()
-        profile = BISYNC_PROFILES["balanced"]
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_filter_path"
-        ) as mock_filter:
-            mock_filter.return_value = Path("/test/filter")
-
-            cmd = build_bisync_command(
-                tenant_id=tenant_id,
-                bucket_name=bucket_name,
-                local_path=local_path,
-                profile=profile,
-            )
-
-        assert cmd[0] == "rclone"
-        assert cmd[1] == "bisync"
-        assert str(local_path) in cmd
-        assert f"basic-memory-{tenant_id}:{bucket_name}" in cmd
-        assert "--create-empty-src-dirs" in cmd
-        assert "--resilient" in cmd
-        assert f"--conflict-resolve={profile.conflict_resolve}" in cmd
-        assert f"--max-delete={profile.max_delete}" in cmd
-        assert "--progress" in cmd
-
-    def test_adds_dry_run_flag(self, tmp_path):
-        """Test that dry-run flag is added when requested."""
-        tenant_id = "test-tenant"
-        bucket_name = "test-bucket"
-        local_path = tmp_path / "sync"
-        local_path.mkdir()
-        profile = BISYNC_PROFILES["safe"]
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_filter_path"
-        ) as mock_filter:
-            mock_filter.return_value = Path("/test/filter")
-
-            cmd = build_bisync_command(
-                tenant_id=tenant_id,
-                bucket_name=bucket_name,
-                local_path=local_path,
-                profile=profile,
-                dry_run=True,
-            )
-
-        assert "--dry-run" in cmd
-
-    def test_adds_resync_flag(self, tmp_path):
-        """Test that resync flag is added when requested."""
-        tenant_id = "test-tenant"
-        bucket_name = "test-bucket"
-        local_path = tmp_path / "sync"
-        local_path.mkdir()
-        profile = BISYNC_PROFILES["balanced"]
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_filter_path"
-        ) as mock_filter:
-            mock_filter.return_value = Path("/test/filter")
-
-            cmd = build_bisync_command(
-                tenant_id=tenant_id,
-                bucket_name=bucket_name,
-                local_path=local_path,
-                profile=profile,
-                resync=True,
-            )
-
-        assert "--resync" in cmd
-
-    def test_adds_verbose_flag(self, tmp_path):
-        """Test that verbose flag is added when requested."""
-        tenant_id = "test-tenant"
-        bucket_name = "test-bucket"
-        local_path = tmp_path / "sync"
-        local_path.mkdir()
-        profile = BISYNC_PROFILES["fast"]
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_filter_path"
-        ) as mock_filter:
-            mock_filter.return_value = Path("/test/filter")
-
-            cmd = build_bisync_command(
-                tenant_id=tenant_id,
-                bucket_name=bucket_name,
-                local_path=local_path,
-                profile=profile,
-                verbose=True,
-            )
-
-        assert "--verbose" in cmd
-        assert "--progress" not in cmd  # Progress replaced by verbose
-
-    def test_creates_state_directory(self, tmp_path):
-        """Test that state directory is created."""
-        tenant_id = "test-tenant"
-        bucket_name = "test-bucket"
-        local_path = tmp_path / "sync"
-        local_path.mkdir()
-        profile = BISYNC_PROFILES["balanced"]
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_filter_path"
-        ) as mock_filter:
-            with patch(
-                "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_state_path"
-            ) as mock_state:
-                state_path = tmp_path / "state"
-                mock_filter.return_value = Path("/test/filter")
-                mock_state.return_value = state_path
-
-                build_bisync_command(
-                    tenant_id=tenant_id,
-                    bucket_name=bucket_name,
-                    local_path=local_path,
-                    profile=profile,
-                )
-
-                # State directory should be created
-                assert state_path.exists()
-                assert state_path.is_dir()
-
-
-class TestBisyncStateManagement:
-    """Tests for bisync state functions."""
-
-    def test_get_bisync_state_path(self):
-        """Test state path generation."""
-        tenant_id = "test-tenant-123"
-        result = get_bisync_state_path(tenant_id)
-
-        expected = Path.home() / ".basic-memory" / "bisync-state" / tenant_id
-        assert result == expected
-
-    def test_bisync_state_exists_true(self, tmp_path):
-        """Test checking for existing state."""
-        state_dir = tmp_path / "state"
-        state_dir.mkdir()
-        (state_dir / "test.lst").write_text("test")
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_state_path",
-            return_value=state_dir,
-        ):
-            result = bisync_state_exists("test-tenant")
-
-        assert result is True
-
-    def test_bisync_state_exists_false_no_dir(self, tmp_path):
-        """Test checking for nonexistent state directory."""
-        state_dir = tmp_path / "nonexistent"
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_state_path",
-            return_value=state_dir,
-        ):
-            result = bisync_state_exists("test-tenant")
-
-        assert result is False
-
-    def test_bisync_state_exists_false_empty_dir(self, tmp_path):
-        """Test checking for empty state directory."""
-        state_dir = tmp_path / "state"
-        state_dir.mkdir()
-
-        with patch(
-            "basic_memory.cli.commands.cloud.bisync_commands.get_bisync_state_path",
-            return_value=state_dir,
-        ):
-            result = bisync_state_exists("test-tenant")
-
-        assert result is False
-
-
-class TestGetBisyncDirectory:
-    """Tests for get_bisync_directory()."""
-
-    def test_returns_default_directory(self):
-        """Test that default directory is returned when not configured."""
-        with patch("basic_memory.cli.commands.cloud.bisync_commands.ConfigManager") as mock_config:
-            mock_config.return_value.config.bisync_config = {}
-
-            result = get_bisync_directory()
-
-            expected = Path.home() / "basic-memory-cloud-sync"
-            assert result == expected
-
-    def test_returns_configured_directory(self, tmp_path):
-        """Test that configured directory is returned."""
-        custom_dir = tmp_path / "custom-sync"
-
-        with patch("basic_memory.cli.commands.cloud.bisync_commands.ConfigManager") as mock_config:
-            mock_config.return_value.config.bisync_config = {"sync_dir": str(custom_dir)}
-
-            result = get_bisync_directory()
-
-            assert result == custom_dir
-
-
-class TestCloudProjectAutoRegistration:
-    """Tests for project auto-registration logic."""
-
-    @pytest.mark.asyncio
-    async def test_extracts_directory_names_from_cloud_paths(self):
-        """Test extraction of directory names from cloud project paths."""
-        from basic_memory.cli.commands.cloud.cloud_utils import fetch_cloud_projects
-
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            mock_response = Mock()
-            mock_response.json.return_value = {
-                "projects": [
-                    {"name": "Main Project", "path": "/app/data/basic-memory"},
-                    {"name": "Work", "path": "/app/data/work-notes"},
-                    {"name": "Personal", "path": "/app/data/personal"},
-                ]
-            }
-            mock_request.return_value = mock_response
-
-            result = await fetch_cloud_projects()
-
-            # Extract directory names as the code does
-            cloud_dir_names = set()
-            for p in result.projects:
-                path = p.path
-                if path.startswith("/app/data/"):
-                    path = path[len("/app/data/") :]
-                dir_name = Path(path).name
-                cloud_dir_names.add(dir_name)
-
-            assert cloud_dir_names == {"basic-memory", "work-notes", "personal"}
-
-    @pytest.mark.asyncio
-    async def test_create_cloud_project_generates_permalink(self):
-        """Test that create_cloud_project generates correct permalink."""
-        from basic_memory.cli.commands.cloud.cloud_utils import create_cloud_project
-
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch(
-                "basic_memory.cli.commands.cloud.cloud_utils.generate_permalink"
-            ) as mock_permalink:
-                mock_permalink.return_value = "my-new-project"
-                mock_response = Mock()
-                mock_response.json.return_value = {
-                    "message": "Project 'My New Project' added successfully",
-                    "status": "success",
-                    "default": False,
-                    "old_project": None,
-                    "new_project": {"name": "My New Project", "path": "my-new-project"},
-                }
-                mock_request.return_value = mock_response
-
-                await create_cloud_project("My New Project")
-
-                # Verify permalink was generated
-                mock_permalink.assert_called_once_with("My New Project")
-
-                # Verify request was made with correct data
-                call_args = mock_request.call_args
-                json_data = call_args.kwargs["json_data"]
-                assert json_data["name"] == "My New Project"
-                assert json_data["path"] == "my-new-project"
-                assert json_data["set_default"] is False

tests/cli/test_cloud_utils.py

@@ -1,326 +0,0 @@
-"""Tests for cloud_utils module."""
-
-from unittest.mock import AsyncMock, Mock, patch
-
-import pytest
-
-from basic_memory.cli.commands.cloud.cloud_utils import (
-    CloudUtilsError,
-    create_cloud_project,
-    fetch_cloud_projects,
-    project_exists,
-    sync_project,
-)
-
-
-class TestFetchCloudProjects:
-    """Tests for fetch_cloud_projects()."""
-
-    @pytest.mark.asyncio
-    async def test_fetches_projects_successfully(self):
-        """Test successful fetch of cloud projects."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                # Setup config
-                mock_config.return_value.config.cloud_host = "https://example.com"
-
-                # Mock API response
-                mock_response = Mock()
-                mock_response.json.return_value = {
-                    "projects": [
-                        {"name": "Project 1", "path": "/app/data/project-1"},
-                        {"name": "Project 2", "path": "/app/data/project-2"},
-                    ]
-                }
-                mock_request.return_value = mock_response
-
-                result = await fetch_cloud_projects()
-
-        # Verify result
-        assert len(result.projects) == 2
-        assert result.projects[0].name == "Project 1"
-        assert result.projects[1].name == "Project 2"
-
-        # Verify API was called correctly
-        mock_request.assert_called_once_with(
-            method="GET", url="https://example.com/proxy/projects/projects"
-        )
-
-    @pytest.mark.asyncio
-    async def test_strips_trailing_slash_from_host(self):
-        """Test that trailing slash is stripped from cloud_host."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                # Setup config with trailing slash
-                mock_config.return_value.config.cloud_host = "https://example.com/"
-
-                mock_response = Mock()
-                mock_response.json.return_value = {"projects": []}
-                mock_request.return_value = mock_response
-
-                await fetch_cloud_projects()
-
-        # Verify trailing slash was removed
-        call_args = mock_request.call_args
-        assert call_args[1]["url"] == "https://example.com/proxy/projects/projects"
-
-    @pytest.mark.asyncio
-    async def test_raises_error_on_api_failure(self):
-        """Test that CloudUtilsError is raised on API failure."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                mock_config.return_value.config.cloud_host = "https://example.com"
-                mock_request.side_effect = Exception("API Error")
-
-                with pytest.raises(CloudUtilsError) as exc_info:
-                    await fetch_cloud_projects()
-
-        assert "Failed to fetch cloud projects" in str(exc_info.value)
-
-    @pytest.mark.asyncio
-    async def test_handles_empty_project_list(self):
-        """Test handling of empty project list."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                mock_config.return_value.config.cloud_host = "https://example.com"
-
-                mock_response = Mock()
-                mock_response.json.return_value = {"projects": []}
-                mock_request.return_value = mock_response
-
-                result = await fetch_cloud_projects()
-
-        assert len(result.projects) == 0
-
-
-class TestCreateCloudProject:
-    """Tests for create_cloud_project()."""
-
-    @pytest.mark.asyncio
-    async def test_creates_project_successfully(self):
-        """Test successful project creation."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                with patch(
-                    "basic_memory.cli.commands.cloud.cloud_utils.generate_permalink"
-                ) as mock_permalink:
-                    # Setup mocks
-                    mock_config.return_value.config.cloud_host = "https://example.com"
-                    mock_permalink.return_value = "my-project"
-
-                    mock_response = Mock()
-                    mock_response.json.return_value = {
-                        "message": "Project 'My Project' added successfully",
-                        "status": "success",
-                        "default": False,
-                        "old_project": None,
-                        "new_project": {"name": "My Project", "path": "my-project"},
-                    }
-                    mock_request.return_value = mock_response
-
-                    result = await create_cloud_project("My Project")
-
-        # Verify result
-        assert result.message == "Project 'My Project' added successfully"
-        assert result.status == "success"
-        assert result.default is False
-
-        # Verify permalink was generated
-        mock_permalink.assert_called_once_with("My Project")
-
-        # Verify API request
-        call_args = mock_request.call_args
-        assert call_args[1]["method"] == "POST"
-        assert call_args[1]["url"] == "https://example.com/proxy/projects/projects"
-        assert call_args[1]["headers"]["Content-Type"] == "application/json"
-
-        json_data = call_args[1]["json_data"]
-        assert json_data["name"] == "My Project"
-        assert json_data["path"] == "my-project"
-        assert json_data["set_default"] is False
-
-    @pytest.mark.asyncio
-    async def test_generates_permalink_from_name(self):
-        """Test that permalink is generated from project name."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                with patch(
-                    "basic_memory.cli.commands.cloud.cloud_utils.generate_permalink"
-                ) as mock_permalink:
-                    mock_config.return_value.config.cloud_host = "https://example.com"
-                    mock_permalink.return_value = "test-project-123"
-
-                    mock_response = Mock()
-                    mock_response.json.return_value = {
-                        "message": "Project 'Test Project 123' added successfully",
-                        "status": "success",
-                        "default": False,
-                        "old_project": None,
-                        "new_project": {"name": "Test Project 123", "path": "test-project-123"},
-                    }
-                    mock_request.return_value = mock_response
-
-                    await create_cloud_project("Test Project 123")
-
-        # Verify generate_permalink was called with project name
-        mock_permalink.assert_called_once_with("Test Project 123")
-
-    @pytest.mark.asyncio
-    async def test_raises_error_on_api_failure(self):
-        """Test that CloudUtilsError is raised on API failure."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                with patch(
-                    "basic_memory.cli.commands.cloud.cloud_utils.generate_permalink"
-                ) as mock_permalink:
-                    mock_config.return_value.config.cloud_host = "https://example.com"
-                    mock_permalink.return_value = "project"
-                    mock_request.side_effect = Exception("API Error")
-
-                    with pytest.raises(CloudUtilsError) as exc_info:
-                        await create_cloud_project("Test Project")
-
-        assert "Failed to create cloud project 'Test Project'" in str(exc_info.value)
-
-    @pytest.mark.asyncio
-    async def test_strips_trailing_slash_from_host(self):
-        """Test that trailing slash is stripped from cloud_host."""
-        with patch("basic_memory.cli.commands.cloud.cloud_utils.make_api_request") as mock_request:
-            with patch("basic_memory.cli.commands.cloud.cloud_utils.ConfigManager") as mock_config:
-                with patch(
-                    "basic_memory.cli.commands.cloud.cloud_utils.generate_permalink"
-                ) as mock_permalink:
-                    mock_config.return_value.config.cloud_host = "https://example.com/"
-                    mock_permalink.return_value = "project"
-
-                    mock_response = Mock()
-                    mock_response.json.return_value = {
-                        "message": "Project 'Project' added successfully",
-                        "status": "success",
-                        "default": False,
-                        "old_project": None,
-                        "new_project": {"name": "Project", "path": "project"},
-                    }
-                    mock_request.return_value = mock_response
-
-                    await create_cloud_project("Project")
-
-        # Verify trailing slash was removed
-        call_args = mock_request.call_args
-        assert call_args[1]["url"] == "https://example.com/proxy/projects/projects"
-
-
-class TestSyncProject:
-    """Tests for sync_project()."""
-
-    @pytest.mark.asyncio
-    async def test_syncs_project_successfully(self):
-        """Test successful project sync."""
-        # Patch at the point where it's imported (inside the function)
-        with patch(
-            "basic_memory.cli.commands.command_utils.run_sync", new_callable=AsyncMock
-        ) as mock_sync:
-            await sync_project("test-project")
-
-        # Verify run_sync was called with project name
-        mock_sync.assert_called_once_with(project="test-project")
-
-    @pytest.mark.asyncio
-    async def test_raises_error_on_sync_failure(self):
-        """Test that CloudUtilsError is raised on sync failure."""
-        # Patch at the point where it's imported (inside the function)
-        with patch(
-            "basic_memory.cli.commands.command_utils.run_sync", new_callable=AsyncMock
-        ) as mock_sync:
-            mock_sync.side_effect = Exception("Sync failed")
-
-            with pytest.raises(CloudUtilsError) as exc_info:
-                await sync_project("test-project")
-
-        assert "Failed to sync project 'test-project'" in str(exc_info.value)
-
-
-class TestProjectExists:
-    """Tests for project_exists()."""
-
-    @pytest.mark.asyncio
-    async def test_returns_true_when_project_exists(self):
-        """Test that True is returned when project exists."""
-        from basic_memory.schemas.cloud import CloudProject, CloudProjectList
-
-        with patch(
-            "basic_memory.cli.commands.cloud.cloud_utils.fetch_cloud_projects"
-        ) as mock_fetch:
-            # Create actual CloudProject objects
-            projects = CloudProjectList(
-                projects=[
-                    CloudProject(name="project-1", path="/app/data/project-1"),
-                    CloudProject(name="test-project", path="/app/data/test-project"),
-                    CloudProject(name="project-2", path="/app/data/project-2"),
-                ]
-            )
-            mock_fetch.return_value = projects
-
-            result = await project_exists("test-project")
-
-        assert result is True
-
-    @pytest.mark.asyncio
-    async def test_returns_false_when_project_not_found(self):
-        """Test that False is returned when project doesn't exist."""
-        with patch(
-            "basic_memory.cli.commands.cloud.cloud_utils.fetch_cloud_projects"
-        ) as mock_fetch:
-            # Mock project list without matching project
-            mock_projects = Mock()
-            mock_projects.projects = [
-                Mock(name="project-1"),
-                Mock(name="project-2"),
-            ]
-            mock_fetch.return_value = mock_projects
-
-            result = await project_exists("nonexistent-project")
-
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_returns_false_on_api_error(self):
-        """Test that False is returned on API error."""
-        with patch(
-            "basic_memory.cli.commands.cloud.cloud_utils.fetch_cloud_projects"
-        ) as mock_fetch:
-            mock_fetch.side_effect = Exception("API Error")
-
-            result = await project_exists("test-project")
-
-        # Should return False instead of raising exception
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_handles_empty_project_list(self):
-        """Test handling of empty project list."""
-        with patch(
-            "basic_memory.cli.commands.cloud.cloud_utils.fetch_cloud_projects"
-        ) as mock_fetch:
-            mock_projects = Mock()
-            mock_projects.projects = []
-            mock_fetch.return_value = mock_projects
-
-            result = await project_exists("any-project")
-
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_case_sensitive_matching(self):
-        """Test that project name matching is case-sensitive."""
-        with patch(
-            "basic_memory.cli.commands.cloud.cloud_utils.fetch_cloud_projects"
-        ) as mock_fetch:
-            mock_projects = Mock()
-            mock_projects.projects = [Mock(name="Test-Project")]
-            mock_fetch.return_value = mock_projects
-
-            # Different case should not match
-            result = await project_exists("test-project")
-
-        assert result is False

tests/cli/test_project_add_with_local_path.py

@@ -0,0 +1,149 @@
+"""Tests for bm project add with --local-path flag."""
+
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from typer.testing import CliRunner
+
+from basic_memory.cli.app import app
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+@pytest.fixture
+def mock_config(tmp_path, monkeypatch):
+    """Create a mock config in cloud mode using environment variables."""
+    config_dir = tmp_path / ".basic-memory"
+    config_dir.mkdir(parents=True, exist_ok=True)
+    config_file = config_dir / "config.json"
+
+    config_data = {
+        "env": "dev",
+        "projects": {},
+        "default_project": "main",
+        "cloud_mode": True,
+        "cloud_projects": {},
+    }
+
+    config_file.write_text(json.dumps(config_data, indent=2))
+
+    # Set HOME to tmp_path so ConfigManager uses our test config
+    monkeypatch.setenv("HOME", str(tmp_path))
+
+    yield config_file
+
+
+@pytest.fixture
+def mock_api_client():
+    """Mock the API client for project add."""
+    with patch("basic_memory.cli.commands.project.get_client"):
+        # Mock call_post to return a proper response
+        mock_response = AsyncMock()
+        mock_response.json = lambda: {
+            "message": "Project 'test-project' added successfully",
+            "status": "success",
+            "default": False,
+            "old_project": None,
+            "new_project": {
+                "name": "test-project",
+                "path": "/test-project",
+                "is_default": False,
+            },
+        }
+
+        with patch(
+            "basic_memory.cli.commands.project.call_post", return_value=mock_response
+        ) as mock_post:
+            yield mock_post
+
+
+def test_project_add_with_local_path_saves_to_config(
+    runner, mock_config, mock_api_client, tmp_path
+):
+    """Test that bm project add --local-path saves sync path to config."""
+    local_sync_dir = tmp_path / "sync" / "test-project"
+
+    result = runner.invoke(
+        app,
+        [
+            "project",
+            "add",
+            "test-project",
+            "--local-path",
+            str(local_sync_dir),
+        ],
+    )
+
+    assert result.exit_code == 0, f"Exit code: {result.exit_code}, Stdout: {result.stdout}"
+    assert "Project 'test-project' added successfully" in result.stdout
+    assert "Local sync path configured" in result.stdout
+    # Check path is present (may be line-wrapped in output)
+    assert "test-project" in result.stdout
+    assert "sync" in result.stdout
+
+    # Verify config was updated
+    config_data = json.loads(mock_config.read_text())
+    assert "test-project" in config_data["cloud_projects"]
+    # Use as_posix() for cross-platform compatibility (Windows uses backslashes)
+    assert config_data["cloud_projects"]["test-project"]["local_path"] == local_sync_dir.as_posix()
+    assert config_data["cloud_projects"]["test-project"]["last_sync"] is None
+    assert config_data["cloud_projects"]["test-project"]["bisync_initialized"] is False
+
+    # Verify local directory was created
+    assert local_sync_dir.exists()
+    assert local_sync_dir.is_dir()
+
+
+def test_project_add_without_local_path_no_config_entry(runner, mock_config, mock_api_client):
+    """Test that bm project add without --local-path doesn't save to config."""
+    result = runner.invoke(
+        app,
+        ["project", "add", "test-project"],
+    )
+
+    assert result.exit_code == 0
+    assert "Project 'test-project' added successfully" in result.stdout
+    assert "Local sync path configured" not in result.stdout
+
+    # Verify config was NOT updated with cloud_projects entry
+    config_data = json.loads(mock_config.read_text())
+    assert "test-project" not in config_data.get("cloud_projects", {})
+
+
+def test_project_add_local_path_expands_tilde(runner, mock_config, mock_api_client):
+    """Test that --local-path ~/path expands to absolute path."""
+    result = runner.invoke(
+        app,
+        ["project", "add", "test-project", "--local-path", "~/test-sync"],
+    )
+
+    assert result.exit_code == 0
+
+    # Verify config has expanded path
+    config_data = json.loads(mock_config.read_text())
+    local_path = config_data["cloud_projects"]["test-project"]["local_path"]
+    # Path should be absolute (starts with / on Unix or drive letter on Windows)
+    assert Path(local_path).is_absolute()
+    assert "~" not in local_path
+    assert local_path.endswith("/test-sync")
+
+
+def test_project_add_local_path_creates_nested_directories(
+    runner, mock_config, mock_api_client, tmp_path
+):
+    """Test that --local-path creates nested directories."""
+    nested_path = tmp_path / "a" / "b" / "c" / "test-project"
+
+    result = runner.invoke(
+        app,
+        ["project", "add", "test-project", "--local-path", str(nested_path)],
+    )
+
+    assert result.exit_code == 0
+    assert nested_path.exists()
+    assert nested_path.is_dir()

tests/markdown/test_date_frontmatter_parsing.py

@@ -6,7 +6,6 @@ but later code expects strings and calls .strip() on them, causing AttributeErro
 """
 
 import pytest
-from pathlib import Path
 from basic_memory.markdown.entity_parser import EntityParser
 
 
@@ -225,11 +224,13 @@ This file has datetime values in frontmatter that PyYAML will parse as datetime
     created_at = entity_markdown.frontmatter.metadata.get("created_at")
     assert isinstance(created_at, str), "Datetime should be converted to string"
     # PyYAML parses "2025-10-24 14:30:00" as datetime, which we normalize to ISO
-    assert "2025-10-24" in created_at and "14:30:00" in created_at, \
+    assert "2025-10-24" in created_at and "14:30:00" in created_at, (
         f"Datetime with time should be normalized to ISO format, got: {created_at}"
+    )
 
     updated_at = entity_markdown.frontmatter.metadata.get("updated_at")
     assert isinstance(updated_at, str), "Datetime should be converted to string"
     # PyYAML parses "2025-10-24T00:00:00" as datetime, which we normalize to ISO
-    assert "2025-10-24" in updated_at and "00:00:00" in updated_at, \
-        f"Datetime at midnight should be normalized to ISO format, got: {updated_at}"
+    assert "2025-10-24" in updated_at and "00:00:00" in updated_at, (
+        f"Datetime at midnight should be normalized to ISO format, got: {updated_at}"
+    )

tests/sync/test_sync_service.py

@@ -1652,6 +1652,7 @@ async def test_circuit_breaker_clears_on_success(
 
 
 @pytest.mark.asyncio
+@pytest.mark.skip("flaky on ci tests")
 async def test_circuit_breaker_tracks_multiple_files(
     sync_service: SyncService, project_config: ProjectConfig
 ):
@@ -1703,9 +1704,11 @@ Content 3
 
     with patch.object(sync_service, "sync_markdown_file", side_effect=mock_sync_markdown_file):
         # Fail 3 times for file1 and file2 (file3 succeeds each time)
+        await force_full_scan(sync_service)
         await sync_service.sync(project_dir)  # Fail count: file1=1, file2=1
         await touch_file(project_dir / "file1.md")  # Touch to trigger incremental scan
         await touch_file(project_dir / "file2.md")  # Touch to trigger incremental scan
+        await force_full_scan(sync_service)
         await sync_service.sync(project_dir)  # Fail count: file1=2, file2=2
         await touch_file(project_dir / "file1.md")  # Touch to trigger incremental scan
         await touch_file(project_dir / "file2.md")  # Touch to trigger incremental scan

tests/test_config.py

@@ -2,8 +2,9 @@
 
 import tempfile
 import pytest
+from datetime import datetime
 
-from basic_memory.config import BasicMemoryConfig, ConfigManager
+from basic_memory.config import BasicMemoryConfig, CloudProjectConfig, ConfigManager
 from pathlib import Path
 
 
@@ -268,3 +269,101 @@ class TestConfigManager:
         # Try to remove the default project
         with pytest.raises(ValueError, match="Cannot remove the default project"):
             config_manager.remove_project("main")
+
+    def test_config_with_cloud_projects_empty_by_default(self, temp_config_manager):
+        """Test that cloud_projects field exists and defaults to empty dict."""
+        config_manager = temp_config_manager
+        config = config_manager.load_config()
+
+        assert hasattr(config, "cloud_projects")
+        assert config.cloud_projects == {}
+
+    def test_save_and_load_config_with_cloud_projects(self):
+        """Test that config with cloud_projects can be saved and loaded."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+
+            config_manager = ConfigManager()
+            config_manager.config_dir = temp_path / "basic-memory"
+            config_manager.config_file = config_manager.config_dir / "config.json"
+            config_manager.config_dir.mkdir(parents=True, exist_ok=True)
+
+            # Create config with cloud_projects
+            now = datetime.now()
+            test_config = BasicMemoryConfig(
+                projects={"main": str(temp_path / "main")},
+                cloud_projects={
+                    "research": CloudProjectConfig(
+                        local_path=str(temp_path / "research-local"),
+                        last_sync=now,
+                        bisync_initialized=True,
+                    )
+                },
+            )
+            config_manager.save_config(test_config)
+
+            # Load and verify
+            loaded_config = config_manager.load_config()
+            assert "research" in loaded_config.cloud_projects
+            assert loaded_config.cloud_projects["research"].local_path == str(
+                temp_path / "research-local"
+            )
+            assert loaded_config.cloud_projects["research"].bisync_initialized is True
+            assert loaded_config.cloud_projects["research"].last_sync == now
+
+    def test_add_cloud_project_to_existing_config(self):
+        """Test adding cloud projects to an existing config file."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+
+            config_manager = ConfigManager()
+            config_manager.config_dir = temp_path / "basic-memory"
+            config_manager.config_file = config_manager.config_dir / "config.json"
+            config_manager.config_dir.mkdir(parents=True, exist_ok=True)
+
+            # Create initial config without cloud projects
+            initial_config = BasicMemoryConfig(projects={"main": str(temp_path / "main")})
+            config_manager.save_config(initial_config)
+
+            # Load, modify, and save
+            config = config_manager.load_config()
+            assert config.cloud_projects == {}
+
+            config.cloud_projects["work"] = CloudProjectConfig(
+                local_path=str(temp_path / "work-local")
+            )
+            config_manager.save_config(config)
+
+            # Reload and verify persistence
+            reloaded_config = config_manager.load_config()
+            assert "work" in reloaded_config.cloud_projects
+            assert reloaded_config.cloud_projects["work"].local_path == str(
+                temp_path / "work-local"
+            )
+            assert reloaded_config.cloud_projects["work"].bisync_initialized is False
+
+    def test_backward_compatibility_loading_config_without_cloud_projects(self):
+        """Test that old config files without cloud_projects field can be loaded."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+
+            config_manager = ConfigManager()
+            config_manager.config_dir = temp_path / "basic-memory"
+            config_manager.config_file = config_manager.config_dir / "config.json"
+            config_manager.config_dir.mkdir(parents=True, exist_ok=True)
+
+            # Manually write old-style config without cloud_projects
+            import json
+
+            old_config_data = {
+                "env": "dev",
+                "projects": {"main": str(temp_path / "main")},
+                "default_project": "main",
+                "log_level": "INFO",
+            }
+            config_manager.config_file.write_text(json.dumps(old_config_data, indent=2))
+
+            # Should load successfully with cloud_projects defaulting to empty dict
+            config = config_manager.load_config()
+            assert config.cloud_projects == {}
+            assert config.projects == {"main": str(temp_path / "main")}

tests/test_rclone_commands.py

@@ -0,0 +1,353 @@
+"""Test project-scoped rclone commands."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from basic_memory.cli.commands.cloud.rclone_commands import (
+    RcloneError,
+    SyncProject,
+    bisync_initialized,
+    get_project_bisync_state,
+    get_project_remote,
+    project_bisync,
+    project_check,
+    project_ls,
+    project_sync,
+)
+
+
+def test_sync_project_dataclass():
+    """Test SyncProject dataclass."""
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/Users/test/research",
+    )
+
+    assert project.name == "research"
+    assert project.path == "app/data/research"
+    assert project.local_sync_path == "/Users/test/research"
+
+
+def test_sync_project_optional_local_path():
+    """Test SyncProject with optional local_sync_path."""
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+    )
+
+    assert project.name == "research"
+    assert project.path == "app/data/research"
+    assert project.local_sync_path is None
+
+
+def test_get_project_remote():
+    """Test building rclone remote path with normalized path."""
+    # Path comes from API already normalized (no /app/data/ prefix)
+    project = SyncProject(name="research", path="/research")
+
+    remote = get_project_remote(project, "my-bucket")
+
+    assert remote == "basic-memory-cloud:my-bucket/research"
+
+
+def test_get_project_remote_strips_app_data_prefix():
+    """Test that /app/data/ prefix is stripped from cloud path."""
+    # If API returns path with /app/data/, it should be stripped
+    project = SyncProject(name="research", path="/app/data/research")
+
+    remote = get_project_remote(project, "my-bucket")
+
+    # Should strip /app/data/ prefix to get actual S3 path
+    assert remote == "basic-memory-cloud:my-bucket/research"
+
+
+def test_get_project_bisync_state():
+    """Test getting bisync state directory path."""
+    state_path = get_project_bisync_state("research")
+
+    expected = Path.home() / ".basic-memory" / "bisync-state" / "research"
+    assert state_path == expected
+
+
+def test_bisync_initialized_false_when_not_exists(tmp_path, monkeypatch):
+    """Test bisync_initialized returns False when state doesn't exist."""
+    # Patch to use tmp directory
+    monkeypatch.setattr(
+        "basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
+        lambda project_name: tmp_path / project_name,
+    )
+
+    assert bisync_initialized("research") is False
+
+
+def test_bisync_initialized_false_when_empty(tmp_path, monkeypatch):
+    """Test bisync_initialized returns False when state directory is empty."""
+    state_dir = tmp_path / "research"
+    state_dir.mkdir()
+
+    monkeypatch.setattr(
+        "basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
+        lambda project_name: tmp_path / project_name,
+    )
+
+    assert bisync_initialized("research") is False
+
+
+def test_bisync_initialized_true_when_has_files(tmp_path, monkeypatch):
+    """Test bisync_initialized returns True when state has files."""
+    state_dir = tmp_path / "research"
+    state_dir.mkdir()
+    (state_dir / "state.lst").touch()
+
+    monkeypatch.setattr(
+        "basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
+        lambda project_name: tmp_path / project_name,
+    )
+
+    assert bisync_initialized("research") is True
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_sync_success(mock_run):
+    """Test successful project sync."""
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="/research",  # Normalized path from API
+        local_sync_path="/tmp/research",
+    )
+
+    result = project_sync(project, "my-bucket", dry_run=True)
+
+    assert result is True
+    mock_run.assert_called_once()
+
+    # Check command arguments
+    cmd = mock_run.call_args[0][0]
+    assert cmd[0] == "rclone"
+    assert cmd[1] == "sync"
+    # Use Path for cross-platform comparison (Windows uses backslashes)
+    assert Path(cmd[2]) == Path("/tmp/research")
+    assert cmd[3] == "basic-memory-cloud:my-bucket/research"
+    assert "--dry-run" in cmd
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_sync_with_verbose(mock_run):
+    """Test project sync with verbose flag."""
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    project_sync(project, "my-bucket", verbose=True)
+
+    cmd = mock_run.call_args[0][0]
+    assert "--verbose" in cmd
+    assert "--progress" not in cmd
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_sync_with_progress(mock_run):
+    """Test project sync with progress (default)."""
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    project_sync(project, "my-bucket")
+
+    cmd = mock_run.call_args[0][0]
+    assert "--progress" in cmd
+    assert "--verbose" not in cmd
+
+
+def test_project_sync_no_local_path():
+    """Test project sync raises error when local_sync_path not configured."""
+    project = SyncProject(name="research", path="app/data/research")
+
+    with pytest.raises(RcloneError) as exc_info:
+        project_sync(project, "my-bucket")
+
+    assert "no local_sync_path configured" in str(exc_info.value)
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
+def test_project_bisync_success(mock_bisync_init, mock_run):
+    """Test successful project bisync."""
+    mock_bisync_init.return_value = True  # Already initialized
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    result = project_bisync(project, "my-bucket")
+
+    assert result is True
+    mock_run.assert_called_once()
+
+    # Check command arguments
+    cmd = mock_run.call_args[0][0]
+    assert cmd[0] == "rclone"
+    assert cmd[1] == "bisync"
+    assert "--conflict-resolve=newer" in cmd
+    assert "--max-delete=25" in cmd
+    assert "--resilient" in cmd
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
+def test_project_bisync_requires_resync_first_time(mock_bisync_init, mock_run):
+    """Test that first bisync requires --resync flag."""
+    mock_bisync_init.return_value = False  # Not initialized
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    with pytest.raises(RcloneError) as exc_info:
+        project_bisync(project, "my-bucket")
+
+    assert "requires --resync" in str(exc_info.value)
+    mock_run.assert_not_called()
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
+def test_project_bisync_with_resync_flag(mock_bisync_init, mock_run):
+    """Test bisync with --resync flag for first time."""
+    mock_bisync_init.return_value = False  # Not initialized
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    result = project_bisync(project, "my-bucket", resync=True)
+
+    assert result is True
+    cmd = mock_run.call_args[0][0]
+    assert "--resync" in cmd
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
+def test_project_bisync_dry_run_skips_init_check(mock_bisync_init, mock_run):
+    """Test that dry-run skips initialization check."""
+    mock_bisync_init.return_value = False  # Not initialized
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    # Should not raise error even though not initialized
+    result = project_bisync(project, "my-bucket", dry_run=True)
+
+    assert result is True
+    cmd = mock_run.call_args[0][0]
+    assert "--dry-run" in cmd
+
+
+def test_project_bisync_no_local_path():
+    """Test project bisync raises error when local_sync_path not configured."""
+    project = SyncProject(name="research", path="app/data/research")
+
+    with pytest.raises(RcloneError) as exc_info:
+        project_bisync(project, "my-bucket")
+
+    assert "no local_sync_path configured" in str(exc_info.value)
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_check_success(mock_run):
+    """Test successful project check."""
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    result = project_check(project, "my-bucket")
+
+    assert result is True
+    cmd = mock_run.call_args[0][0]
+    assert cmd[0] == "rclone"
+    assert cmd[1] == "check"
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_check_with_one_way(mock_run):
+    """Test project check with one-way flag."""
+    mock_run.return_value = MagicMock(returncode=0)
+
+    project = SyncProject(
+        name="research",
+        path="app/data/research",
+        local_sync_path="/tmp/research",
+    )
+
+    project_check(project, "my-bucket", one_way=True)
+
+    cmd = mock_run.call_args[0][0]
+    assert "--one-way" in cmd
+
+
+def test_project_check_no_local_path():
+    """Test project check raises error when local_sync_path not configured."""
+    project = SyncProject(name="research", path="app/data/research")
+
+    with pytest.raises(RcloneError) as exc_info:
+        project_check(project, "my-bucket")
+
+    assert "no local_sync_path configured" in str(exc_info.value)
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_ls_success(mock_run):
+    """Test successful project ls."""
+    mock_run.return_value = MagicMock(returncode=0, stdout="file1.md\nfile2.md\nsubdir/file3.md\n")
+
+    project = SyncProject(name="research", path="app/data/research")
+
+    files = project_ls(project, "my-bucket")
+
+    assert len(files) == 3
+    assert "file1.md" in files
+    assert "file2.md" in files
+    assert "subdir/file3.md" in files
+
+
+@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
+def test_project_ls_with_subpath(mock_run):
+    """Test project ls with subdirectory."""
+    mock_run.return_value = MagicMock(returncode=0, stdout="")
+
+    project = SyncProject(name="research", path="/research")  # Normalized path
+
+    project_ls(project, "my-bucket", path="subdir")
+
+    cmd = mock_run.call_args[0][0]
+    assert cmd[-1] == "basic-memory-cloud:my-bucket/research/subdir"