project.py

"""Command module for basic-memory project management."""

import json
import os
from datetime import datetime
from pathlib import Path
from typing import cast

import typer
from rich.console import Console, Group
from rich.panel import Panel
from rich.table import Table
from rich.text import Text

from basic_memory.cli.app import app
from basic_memory.cli.auth import CLIAuth
from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
from basic_memory.cli.commands.cloud.project_sync import (
    _has_cloud_credentials,
    _require_cloud_credentials,
)
from basic_memory.cli.commands.cloud.rclone_commands import (
    SyncProject,
    project_ls,
)
from basic_memory.cli.commands.command_utils import get_project_info, run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.clients import ProjectClient
from basic_memory.schemas.cloud import ProjectVisibility
from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.utils import generate_permalink, normalize_project_path

console = Console()

# Create a project subcommand
project_app = typer.Typer(help="Manage multiple Basic Memory projects")
app.add_typer(project_app, name="project")


def format_path(path: str) -> str:
    """Format a path for display, using ~ for home directory."""
    home = str(Path.home())
    if path.startswith(home):
        return path.replace(home, "~", 1)  # pragma: no cover
    return path


def make_bar(value: int, max_value: int, width: int = 40) -> Text:
    """Create a horizontal bar chart element using Unicode blocks."""
    if max_value == 0:
        return Text("░" * width, style="dim")
    filled = max(1, round(value / max_value * width)) if value > 0 else 0
    bar = Text()
    bar.append("█" * filled, style="cyan")
    bar.append("░" * (width - filled), style="dim")
    return bar


def _normalize_project_visibility(visibility: str | None) -> ProjectVisibility:
    """Normalize CLI visibility input to the cloud API contract."""
    if visibility is None:
        return "workspace"

    normalized = visibility.strip().lower()
    if normalized in {"workspace", "shared", "private"}:
        return cast(ProjectVisibility, normalized)

    raise ValueError("Invalid visibility. Expected one of: workspace, shared, private.")


def _resolve_workspace_id(config, workspace: str | None) -> str | None:
    """Resolve a workspace name or tenant_id to a tenant_id."""
    from basic_memory.mcp.project_context import (
        _workspace_choices,
        _workspace_matches_identifier,
        get_available_workspaces,
    )

    if workspace is not None:
        workspaces = run_with_cleanup(get_available_workspaces())
        matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
        if not matches:
            console.print(f"[red]Error: Workspace '{workspace}' not found[/red]")
            if workspaces:
                console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
            raise typer.Exit(1)
        if len(matches) > 1:
            console.print(
                f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. "
                f"Use tenant_id instead.[/red]"
            )
            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
            raise typer.Exit(1)
        return matches[0].tenant_id

    if config.default_workspace:
        return config.default_workspace

    try:
        workspaces = run_with_cleanup(get_available_workspaces())
        if len(workspaces) == 1:
            return workspaces[0].tenant_id
    except Exception:
        # Workspace resolution is optional until a command needs a specific tenant.
        pass

    return None


@project_app.command("list")
def list_projects(
    local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
    workspace: str = typer.Option(None, "--workspace", help="Cloud workspace name or tenant_id"),
    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
) -> None:
    """List Basic Memory projects from local and (when available) cloud."""
    try:
        validate_routing_flags(local, cloud)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)

    async def _list_projects(ws: str | None = None):
        async with get_client(workspace=ws) as client:
            return await ProjectClient(client).list_projects()

    try:
        config = ConfigManager().config
        # Use explicit workspace, fall back to config default
        effective_workspace = workspace or config.default_workspace

        local_result: ProjectList | None = None
        cloud_result: ProjectList | None = None
        cloud_error: Exception | None = None

        if cloud:
            with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                with force_routing(cloud=True):
                    cloud_result = run_with_cleanup(_list_projects(effective_workspace))
        elif local:
            with force_routing(local=True):
                local_result = run_with_cleanup(_list_projects())
        else:
            # Default behavior: always show local projects first.
            with force_routing(local=True):
                local_result = run_with_cleanup(_list_projects())

            if _has_cloud_credentials(config):
                try:
                    with console.status("[bold blue]Fetching cloud projects...", spinner="dots"):
                        with force_routing(cloud=True):
                            cloud_result = run_with_cleanup(_list_projects(effective_workspace))
                except Exception as exc:  # pragma: no cover
                    cloud_error = exc

        # Resolve workspace name for cloud projects (best-effort)
        cloud_ws_name: str | None = None
        cloud_ws_type: str | None = None
        if cloud_result and effective_workspace:
            try:
                from basic_memory.mcp.project_context import get_available_workspaces

                with console.status("[bold blue]Resolving workspace...", spinner="dots"):
                    workspaces = run_with_cleanup(get_available_workspaces())
                matched = next(
                    (ws for ws in workspaces if ws.tenant_id == effective_workspace),
                    None,
                )
                if matched:
                    cloud_ws_name = matched.name
                    cloud_ws_type = matched.workspace_type
            except Exception:
                pass

        table = Table(title="Basic Memory Projects")
        table.add_column("Name", style="cyan")
        table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
        table.add_column("Cloud Path", style="green")
        table.add_column("Workspace", style="green")
        table.add_column("CLI Route", style="blue")
        table.add_column("MCP", style="blue")
        table.add_column("Sync", style="green")
        table.add_column("Default", style="magenta")

        project_names_by_permalink: dict[str, str] = {}
        local_projects_by_permalink: dict[str, ProjectItem] = {}
        cloud_projects_by_permalink: dict[str, ProjectItem] = {}

        if local_result:
            for project in local_result.projects:
                permalink = generate_permalink(project.name)
                project_names_by_permalink[permalink] = project.name
                local_projects_by_permalink[permalink] = project

        if cloud_result:
            for project in cloud_result.projects:
                permalink = generate_permalink(project.name)
                project_names_by_permalink[permalink] = project.name
                cloud_projects_by_permalink[permalink] = project

        # --- Build unified project list ---
        project_rows: list[dict] = []
        for permalink in sorted(project_names_by_permalink):
            project_name = project_names_by_permalink[permalink]
            local_project = local_projects_by_permalink.get(permalink)
            cloud_project = cloud_projects_by_permalink.get(permalink)
            entry = config.projects.get(project_name)

            local_path = ""
            if local_project is not None:
                local_path = format_path(normalize_project_path(local_project.path))
            elif entry and entry.local_sync_path:
                local_path = format_path(entry.local_sync_path)
            elif entry and entry.mode == ProjectMode.LOCAL and entry.path:
                local_path = format_path(normalize_project_path(entry.path))

            # Clear local path for cloud-mode projects — only local projects
            # should display a local path
            if entry and entry.mode == ProjectMode.CLOUD:
                local_path = ""

            cloud_path = ""
            if cloud_project is not None:
                cloud_path = normalize_project_path(cloud_project.path)

            if local:
                cli_route = "local (flag)"
            elif cloud:
                cli_route = "cloud (flag)"
            elif entry:
                cli_route = entry.mode.value
            elif cloud_project is not None and local_project is None:
                cli_route = ProjectMode.CLOUD.value
            else:
                cli_route = ProjectMode.LOCAL.value

            is_default = config.default_project == project_name

            has_sync = bool(entry and entry.local_sync_path)
            # Determine MCP transport based on project routing mode
            if entry and entry.mode == ProjectMode.CLOUD:
                mcp_transport = "https"
            elif entry is None and cloud_project is not None:
                mcp_transport = "https"
            else:
                mcp_transport = "stdio"

            # Show workspace name (type) for cloud-sourced projects
            ws_label = ""
            if cloud_project is not None and cloud_ws_name:
                ws_label = f"{cloud_ws_name} ({cloud_ws_type})" if cloud_ws_type else cloud_ws_name

            row_data = {
                "name": project_name,
                "permalink": permalink,
                "local_path": local_path,
                "cloud_path": cloud_path,
                "cli_route": cli_route,
                "mcp_stdio": mcp_transport,
                "sync": has_sync,
                "is_default": is_default,
            }
            if ws_label:
                row_data["workspace"] = cloud_ws_name or ""
                if cloud_ws_type:
                    row_data["workspace_type"] = cloud_ws_type

            project_rows.append(row_data)

        # --- JSON output ---
        if json_output:
            print(json.dumps({"projects": project_rows}, indent=2, default=str))
            return

        # --- Rich table output ---
        for row_data in project_rows:
            table.add_row(
                row_data["name"],
                row_data["local_path"],
                row_data["cloud_path"],
                row_data.get("workspace", "")
                + (f" ({row_data['workspace_type']})" if row_data.get("workspace_type") else ""),
                row_data["cli_route"],
                row_data["mcp_stdio"],
                "[X]" if row_data["sync"] else "",
                "[X]" if row_data["is_default"] else "",
            )

        console.print(table)
        if cloud_error is not None:
            console.print(f"[yellow]Cloud project discovery failed: {cloud_error}[/yellow]")
            console.print(
                "[dim]Showing local projects only. "
                "Run 'bm cloud login' or 'bm cloud api-key save <key>' if this is a credentials issue.[/dim]"
            )
    except Exception as e:
        console.print(f"[red]Error listing projects: {str(e)}[/red]")
        raise typer.Exit(1)


@project_app.command("add")
def add_project(
    name: str = typer.Argument(..., help="Name of the 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)"
    ),
    workspace: str = typer.Option(
        None,
        "--workspace",
        help="Cloud workspace name or tenant_id (cloud mode only)",
    ),
    visibility: str = typer.Option(
        None,
        "--visibility",
        help="Cloud project visibility: workspace, shared, or private",
    ),
    set_default: bool = typer.Option(False, "--default", help="Set as default project"),
    local: bool = typer.Option(
        False, "--local", help="Force local API routing (ignore cloud mode)"
    ),
    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
) -> None:
    """Add a new project.

    Use --local to force local routing when cloud mode is enabled.
    Use --cloud to force cloud routing when cloud mode is disabled.

    Cloud mode examples:\n
        bm project add research                           # No local sync\n
        bm project add research --local-path ~/docs       # With local sync\n
        bm project add research --cloud --visibility shared\n
        bm project add research --cloud --workspace Personal --visibility shared\n

    Local mode example:\n
        bm project add research ~/Documents/research
    """
    try:
        validate_routing_flags(local, cloud)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)

    config = ConfigManager().config

    # Determine effective mode: default local, cloud only when explicitly requested.
    effective_cloud_mode = cloud and not local
    resolved_workspace_id: str | None = None

    # Resolve local sync path early (needed for both cloud and local mode)
    local_sync_path: str | None = None
    if local_path:
        local_sync_path = Path(os.path.abspath(os.path.expanduser(local_path))).as_posix()

    if effective_cloud_mode:
        _require_cloud_credentials(config)
        try:
            resolved_visibility = _normalize_project_visibility(visibility)
        except ValueError as e:
            console.print(f"[red]Error: {e}[/red]")
            raise typer.Exit(1)
        resolved_workspace_id = _resolve_workspace_id(config, workspace)
        # Cloud mode: path auto-generated from name, local sync is optional

        async def _add_project():
            async with get_client(workspace=resolved_workspace_id) as client:
                data = {
                    "name": name,
                    "path": generate_permalink(name),
                    "local_sync_path": local_sync_path,
                    "set_default": set_default,
                    "visibility": resolved_visibility,
                }
                return await ProjectClient(client).create_project(data)
    else:
        if workspace is not None:
            console.print("[red]Error: --workspace is only supported in cloud mode[/red]")
            raise typer.Exit(1)
        if visibility is not None:
            console.print("[red]Error: --visibility is only supported in cloud mode[/red]")
            raise typer.Exit(1)
        # Local mode: path is required
        if path is None:
            console.print("[red]Error: path argument is required in local mode[/red]")
            raise typer.Exit(1)

        # Resolve to absolute path
        resolved_path = Path(os.path.abspath(os.path.expanduser(path))).as_posix()

        async def _add_project():
            async with get_client() as client:
                data = {"name": name, "path": resolved_path, "set_default": set_default}
                return await ProjectClient(client).create_project(data)

    try:
        with force_routing(local=local, cloud=cloud):
            result = run_with_cleanup(_add_project())
        console.print(f"[green]{result.message}[/green]")

        # Trigger: local config needs enough metadata to route future commands back to cloud.
        # Why: explicit workspace selection and local sync state should persist across CLI sessions.
        # Outcome: cloud-backed projects keep cloud mode, workspace_id, and optional local sync path.
        if effective_cloud_mode and (local_sync_path or resolved_workspace_id):
            entry = config.projects.get(name)
            if entry:
                entry.mode = ProjectMode.CLOUD
                if local_sync_path:
                    entry.path = local_sync_path
                    entry.local_sync_path = local_sync_path
                if resolved_workspace_id:
                    entry.workspace_id = resolved_workspace_id
            else:
                # Project may not be in local config yet (cloud-only add)
                config.projects[name] = ProjectEntry(
                    path=local_sync_path or "",
                    mode=ProjectMode.CLOUD,
                    local_sync_path=local_sync_path,
                    workspace_id=resolved_workspace_id,
                )
            ConfigManager().save_config(config)

        # Save local sync path to config if in cloud mode
        if effective_cloud_mode and local_sync_path:
            # Create local directory if it doesn't exist
            local_dir = Path(local_sync_path)
            local_dir.mkdir(parents=True, exist_ok=True)

            console.print(f"\n[green]Local sync path configured: {local_sync_path}[/green]")
            console.print("\nNext steps:")
            console.print(f"  1. Preview: bm cloud bisync --name {name} --resync --dry-run")
            console.print(f"  2. Sync: bm cloud bisync --name {name} --resync")
    except Exception as e:
        console.print(f"[red]Error adding project: {str(e)}[/red]")
        raise typer.Exit(1)


@project_app.command("remove")
def remove_project(
    name: str = typer.Argument(..., help="Name of the project to remove"),
    delete_notes: bool = typer.Option(
        False, "--delete-notes", help="Delete project files from disk"
    ),
    local: bool = typer.Option(
        False, "--local", help="Force local API routing (ignore cloud mode)"
    ),
    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
) -> None:
    """Remove a project.

    Use --local to force local routing when cloud mode is enabled.
    Use --cloud to force cloud routing when cloud mode is disabled.
    """
    try:
        validate_routing_flags(local, cloud)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)

    async def _remove_project():
        # Resolve workspace so cloud-only projects auto-route without --cloud
        config = ConfigManager().config
        entry = config.projects.get(name)
        ws = None
        if entry and entry.workspace_id:
            ws = entry.workspace_id
        elif config.default_workspace:
            ws = config.default_workspace

        async with get_client(project_name=name, workspace=ws) as client:
            project_client = ProjectClient(client)
            # Convert name to permalink for efficient resolution
            project_permalink = generate_permalink(name)
            target_project = await project_client.resolve_project(project_permalink)
            return await project_client.delete_project(
                target_project.external_id, delete_notes=delete_notes
            )

    try:
        # Get config to check for local sync path and bisync state
        config = ConfigManager().config
        local_path_config = None
        has_bisync_state = False

        entry = config.projects.get(name)
        if cloud and entry and entry.local_sync_path:
            local_path_config = entry.local_sync_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
        with force_routing(local=local, cloud=cloud):
            result = run_with_cleanup(_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_config:
            local_dir = Path(local_path_config)
            if local_dir.exists():
                import shutil

                shutil.rmtree(local_dir)
                console.print(f"[green]Removed local sync directory: {local_path_config}[/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 sync fields on the project entry
        if cloud and entry and entry.local_sync_path:
            entry.local_sync_path = None
            entry.bisync_initialized = False
            entry.last_sync = None
            ConfigManager().save_config(config)

        # Show informative message if files were not deleted
        if not delete_notes:
            if local_path_config:
                console.print(f"[yellow]Note: Local files remain at {local_path_config}[/yellow]")

    except Exception as e:
        console.print(f"[red]Error removing project: {str(e)}[/red]")
        raise typer.Exit(1)


@project_app.command("default")
def set_default_project(
    name: str = typer.Argument(..., help="Name of the project to set as CLI default"),
    local: bool = typer.Option(
        False, "--local", help="Force local API routing (required in cloud mode)"
    ),
) -> None:
    """Set the default project used as fallback when no project is specified.

    In cloud mode, use --local to modify the local configuration.
    """

    async def _set_default():
        # Resolve workspace so cloud-only projects auto-route without flags
        config = ConfigManager().config
        entry = config.projects.get(name)
        ws = None
        if entry and entry.workspace_id:
            ws = entry.workspace_id
        elif config.default_workspace:
            ws = config.default_workspace

        async with get_client(project_name=name, workspace=ws) as client:
            project_client = ProjectClient(client)
            # Convert name to permalink for efficient resolution
            project_permalink = generate_permalink(name)
            target_project = await project_client.resolve_project(project_permalink)
            return await project_client.set_default(target_project.external_id)

    try:
        with force_routing(local=local):
            result = run_with_cleanup(_set_default())
        console.print(f"[green]{result.message}[/green]")
    except Exception as e:
        console.print(f"[red]Error setting default project: {str(e)}[/red]")
        raise typer.Exit(1)


@project_app.command("move")
def move_project(
    name: str = typer.Argument(..., help="Name of the project to move"),
    new_path: str = typer.Argument(..., help="New absolute path for the project"),
) -> None:
    """Move a local project to a new filesystem location.

    This command only applies to local projects — it updates the project's
    configured path in the local database.
    """
    # Resolve to absolute path
    resolved_path = Path(os.path.abspath(os.path.expanduser(new_path))).as_posix()

    async def _move_project():
        async with get_client() as client:
            project_client = ProjectClient(client)
            project_info = await project_client.resolve_project(name)
            return await project_client.update_project(
                project_info.external_id, {"path": resolved_path}
            )

    try:
        with force_routing(local=True):
            result = run_with_cleanup(_move_project())
        console.print(f"[green]{result.message}[/green]")

        # Show important file movement reminder
        console.print()  # Empty line for spacing
        console.print(
            Panel(
                "[bold red]IMPORTANT:[/bold red] Project configuration updated successfully.\n\n"
                "[yellow]You must manually move your project files from the old location to:[/yellow]\n"
                f"[cyan]{resolved_path}[/cyan]\n\n"
                "[dim]Basic Memory has only updated the configuration - your files remain in their original location.[/dim]",
                title="Manual File Movement Required",
                border_style="yellow",
                expand=False,
            )
        )

    except Exception as e:
        console.print(f"[red]Error moving project: {str(e)}[/red]")
        raise typer.Exit(1)


@project_app.command("set-cloud")
def set_cloud(
    name: str = typer.Argument(..., help="Name of the project to route through cloud"),
    workspace: str = typer.Option(
        None,
        "--workspace",
        help="Cloud workspace name or tenant_id to associate with this project",
    ),
) -> None:
    """Set a project to cloud mode (route through cloud API).

    Requires either an API key or an active OAuth session.

    Use --workspace to associate a specific workspace with this project.
    If omitted, uses the default workspace (if set) or auto-selects when
    only one workspace is available.

    Examples:
      bm project set-cloud research --workspace Personal
      bm project set-cloud research --workspace 11111111-...
      bm project set-cloud research   # uses default workspace
    """

    config_manager = ConfigManager()
    config = config_manager.config

    # Validate project exists in config
    if name not in config.projects:
        console.print(f"[red]Error: Project '{name}' not found in config[/red]")
        raise typer.Exit(1)

    # Validate credentials: API key or OAuth session
    has_api_key = bool(config.cloud_api_key)
    has_oauth = False
    if not has_api_key:
        auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
        has_oauth = auth.load_tokens() is not None

    if not has_api_key and not has_oauth:
        console.print("[red]Error: No cloud credentials found[/red]")
        console.print("[dim]Run 'bm cloud api-key save <key>' or 'bm cloud login' first[/dim]")
        raise typer.Exit(1)

    resolved_workspace_id = _resolve_workspace_id(config, workspace)

    config.set_project_mode(name, ProjectMode.CLOUD)
    if resolved_workspace_id:
        config.projects[name].workspace_id = resolved_workspace_id
    config_manager.save_config(config)

    console.print(f"[green]Project '{name}' set to cloud mode[/green]")
    if resolved_workspace_id:
        console.print(f"[dim]Workspace: {resolved_workspace_id}[/dim]")
    console.print("[dim]MCP tools and CLI commands for this project will route through cloud[/dim]")


@project_app.command("set-local")
def set_local(
    name: str = typer.Argument(..., help="Name of the project to revert to local mode"),
) -> None:
    """Revert a project to local mode (use in-process ASGI transport).

    Clears any associated cloud workspace.

    Example:
      bm project set-local research
    """
    config_manager = ConfigManager()
    config = config_manager.config

    # Validate project exists in config
    if name not in config.projects:
        console.print(f"[red]Error: Project '{name}' not found in config[/red]")
        raise typer.Exit(1)

    config.set_project_mode(name, ProjectMode.LOCAL)
    config.projects[name].workspace_id = None
    config_manager.save_config(config)

    console.print(f"[green]Project '{name}' set to local mode[/green]")
    console.print("[dim]MCP tools and CLI commands for this project will use local transport[/dim]")


@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)"),
    local: bool = typer.Option(False, "--local", help="List files from local project instance"),
    cloud: bool = typer.Option(False, "--cloud", help="List files from cloud project instance"),
) -> None:
    """List files in a project.

    Examples:
      bm project ls --name research
      bm project ls --name research --local
      bm project ls --name research --cloud
      bm project ls --name research subfolder
    """
    try:
        validate_routing_flags(local, cloud)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)

    # Determine routing: explicit flags take precedence, otherwise check project mode
    if cloud or local:
        use_cloud_route = cloud and not local
    else:
        config = ConfigManager().config
        project_mode = config.get_project_mode(name)
        use_cloud_route = project_mode == ProjectMode.CLOUD

    def _list_local_files(project_path: str, subpath: str | None = None) -> list[str]:
        project_root = Path(normalize_project_path(project_path)).expanduser().resolve()
        target_dir = project_root

        if subpath:
            requested = Path(subpath)
            if requested.is_absolute():
                raise ValueError("Path must be relative to the project root")
            target_dir = (project_root / requested).resolve()
            if not target_dir.is_relative_to(project_root):
                raise ValueError("Path must stay within the project root")

        if not target_dir.exists():
            raise ValueError(f"Path not found: {target_dir}")
        if not target_dir.is_dir():
            raise ValueError(f"Path is not a directory: {target_dir}")

        files: list[str] = []
        for file_path in sorted(target_dir.rglob("*")):
            if file_path.is_file():
                size = file_path.stat().st_size
                relative = file_path.relative_to(project_root).as_posix()
                files.append(f"{size:10d} {relative}")

        return files

    try:
        # Get project info
        async def _get_project():
            async with get_client() as client:
                projects_list = await ProjectClient(client).list_projects()
                for proj in projects_list.projects:
                    if generate_permalink(proj.name) == generate_permalink(name):
                        return proj
                return None

        if use_cloud_route:
            config = ConfigManager().config
            _require_cloud_credentials(config)

            tenant_info = run_with_cleanup(get_mount_info())
            bucket_name = tenant_info.bucket_name

            with force_routing(cloud=True):
                project_data = run_with_cleanup(_get_project())
            if not project_data:
                console.print(f"[red]Error: Project '{name}' not found[/red]")
                raise typer.Exit(1)

            sync_project = SyncProject(
                name=project_data.name,
                path=normalize_project_path(project_data.path),
            )
            files = project_ls(sync_project, bucket_name, path=path)
            target_label = "CLOUD"
        else:
            with force_routing(local=True):
                project_data = run_with_cleanup(_get_project())
            if not project_data:
                console.print(f"[red]Error: Project '{name}' not found[/red]")
                raise typer.Exit(1)

            # For cloud-mode projects accessed with --local, use local_sync_path
            # (the actual local directory) instead of project_data.path from the API
            local_dir = project_data.path
            if local:
                entry = ConfigManager().config.projects.get(name)
                if entry and entry.local_sync_path:
                    local_dir = entry.local_sync_path

            files = _list_local_files(local_dir, path)
            target_label = "LOCAL"

        if files:
            heading = f"\n[bold]Files in {name} ({target_label})"
            if path:
                heading += f"/{path}"
            heading += ":[/bold]"
            console.print(heading)
            for file in files:
                console.print(f"  {file}")
            console.print(f"\n[dim]Total: {len(files)} files[/dim]")
        else:
            prefix = f"[yellow]No files found in {name} ({target_label})"
            console.print(prefix + (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"),
    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
    local: bool = typer.Option(
        False, "--local", help="Force local API routing (ignore cloud mode)"
    ),
    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
):
    """Display detailed information and statistics about the current project.

    Use --local to force local routing when cloud mode is enabled.
    Use --cloud to force cloud routing when cloud mode is disabled.
    """
    try:
        validate_routing_flags(local, cloud)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)

    try:
        # Get project info
        with force_routing(local=local, cloud=cloud):
            info = run_with_cleanup(get_project_info(name))

        if json_output:
            # Convert to JSON and print
            print(json.dumps(info.model_dump(), indent=2, default=str))
        else:
            # --- Left column: Knowledge Graph stats ---
            left = Table.grid(padding=(0, 2))
            left.add_column("metric", style="cyan")
            left.add_column("value", style="green", justify="right")

            left.add_row("[bold]Knowledge Graph[/bold]", "")
            left.add_row("Entities", str(info.statistics.total_entities))
            left.add_row("Observations", str(info.statistics.total_observations))
            left.add_row("Relations", str(info.statistics.total_relations))
            left.add_row("Unresolved", str(info.statistics.total_unresolved_relations))
            left.add_row("Isolated", str(info.statistics.isolated_entities))

            # --- Right column: Embeddings ---
            right = Table.grid(padding=(0, 2))
            right.add_column("property", style="cyan")
            right.add_column("value", style="green")

            right.add_row("[bold]Embeddings[/bold]", "")
            if info.embedding_status:
                es = info.embedding_status
                if not es.semantic_search_enabled:
                    right.add_row("[green]●[/green] Semantic Search", "Disabled")
                else:
                    right.add_row("[green]●[/green] Semantic Search", "Enabled")
                    if es.embedding_provider:
                        right.add_row("  Provider", es.embedding_provider)
                    if es.embedding_model:
                        right.add_row("  Model", es.embedding_model)
                    # Embedding coverage bar
                    if es.total_indexed_entities > 0:
                        coverage_bar = make_bar(
                            es.total_entities_with_chunks,
                            es.total_indexed_entities,
                            width=20,
                        )
                        count_text = Text(
                            f" {es.total_entities_with_chunks}/{es.total_indexed_entities}",
                            style="green",
                        )
                        bar_with_count = Text.assemble("  Indexed  ", coverage_bar, count_text)
                        right.add_row(bar_with_count, "")
                    right.add_row("  Chunks", str(es.total_chunks))
                    if es.reindex_recommended:
                        right.add_row(
                            "[yellow]●[/yellow] Status",
                            "[yellow]Reindex recommended[/yellow]",
                        )
                        if es.reindex_reason:
                            right.add_row("  Reason", f"[yellow]{es.reindex_reason}[/yellow]")
                    else:
                        right.add_row("[green]●[/green] Status", "[green]Up to date[/green]")

            # --- Compose two-column layout (content-sized, NOT Layout) ---
            columns = Table.grid(padding=(0, 4), expand=False)
            columns.add_row(left, right)

            # --- Note Types bar chart (top 5 by count) ---
            bars_section = None
            if info.statistics.note_types:
                sorted_types = sorted(
                    info.statistics.note_types.items(), key=lambda x: x[1], reverse=True
                )
                top_types = sorted_types[:5]
                max_count = top_types[0][1] if top_types else 1

                bars = Table.grid(padding=(0, 2), expand=False)
                bars.add_column("type", style="cyan", width=16, justify="right")
                bars.add_column("bar")
                bars.add_column("count", style="green", justify="right")

                for note_type, count in top_types:
                    bars.add_row(note_type, make_bar(count, max_count), str(count))

                remaining = len(sorted_types) - len(top_types)
                bars_section = Group(
                    "[bold]Note Types[/bold]",
                    bars,
                    f"[dim]+{remaining} more types[/dim]" if remaining > 0 else "",
                )

            # --- Footer ---
            current_time = (
                datetime.fromisoformat(str(info.system.timestamp))
                if isinstance(info.system.timestamp, str)
                else info.system.timestamp
            )
            footer = (
                f"[dim]{format_path(info.project_path)}  "
                f"default: {info.default_project}  "
                f"{current_time.strftime('%Y-%m-%d %H:%M')}[/dim]"
            )

            # --- Assemble dashboard ---
            parts: list = [columns, ""]
            if bars_section:
                parts.extend([bars_section, ""])
            parts.append(footer)
            body = Group(*parts)

            console.print(
                Panel(
                    body,
                    title=f"[bold]{info.project_name}[/bold]",
                    subtitle=f"Basic Memory {info.system.version}",
                    expand=False,
                )
            )

    except typer.Exit:
        raise
    except Exception as e:  # pragma: no cover
        typer.echo(f"Error getting project info: {e}", err=True)
        raise typer.Exit(1)