diff --git a/CLAUDE.md b/CLAUDE.md index 7ea958c2..cc571adf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -310,6 +310,26 @@ Basic Memory now supports cloud synchronization and storage (requires active sub - Background relation resolution (non-blocking startup) - API performance optimizations (SPEC-11) +**CLI Routing Flags:** + +When cloud mode is enabled, CLI commands route to the cloud API by default. Use `--local` and `--cloud` flags to override: + +```bash +# Force local routing (ignore cloud mode) +basic-memory status --local +basic-memory project list --local + +# Force cloud routing (when cloud mode is disabled) +basic-memory status --cloud +basic-memory project info my-project --cloud +``` + +Key behaviors: +- The local MCP server (`basic-memory mcp`) automatically uses local routing +- This allows simultaneous use of local Claude Desktop and cloud-based clients +- Some commands (like `project default`, `project sync-config`, `project move`) require `--local` in cloud mode since they modify local configuration +- Environment variable `BASIC_MEMORY_FORCE_LOCAL=true` forces local routing globally + ## AI-Human Collaborative Development Basic Memory emerged from and enables a new kind of development process that combines human and AI capabilities. Instead diff --git a/README.md b/README.md index 825df68e..3986f785 100644 --- a/README.md +++ b/README.md @@ -357,6 +357,22 @@ basic-memory cloud check basic-memory cloud mount ``` +**Routing Flags** (for users with cloud subscriptions): + +When cloud mode is enabled, CLI commands communicate with the cloud API by default. Use routing flags to override this: + +```bash +# Force local routing (useful for local MCP server while cloud mode is enabled) +basic-memory status --local +basic-memory project list --local + +# Force cloud routing (when cloud mode is disabled but you want cloud access) +basic-memory status --cloud +basic-memory project info my-project --cloud +``` + +The local MCP server (`basic-memory mcp`) automatically uses local routing, so you can use both local Claude Desktop and cloud-based clients simultaneously. + 4. In Claude Desktop, the LLM can now use these tools: **Content Management:** @@ -433,6 +449,7 @@ Basic Memory uses [Loguru](https://github.com/Delgan/loguru) for logging. The lo |----------|---------|-------------| | `BASIC_MEMORY_LOG_LEVEL` | `INFO` | Log level: DEBUG, INFO, WARNING, ERROR | | `BASIC_MEMORY_CLOUD_MODE` | `false` | When `true`, API logs to stdout with structured context | +| `BASIC_MEMORY_FORCE_LOCAL` | `false` | When `true`, forces local API routing (ignores cloud mode) | | `BASIC_MEMORY_ENV` | `dev` | Set to `test` for test mode (stderr only) | ### Examples diff --git a/src/basic_memory/cli/commands/mcp.py b/src/basic_memory/cli/commands/mcp.py index 4ca86d42..764cd468 100644 --- a/src/basic_memory/cli/commands/mcp.py +++ b/src/basic_memory/cli/commands/mcp.py @@ -17,60 +17,64 @@ import basic_memory.mcp.prompts # noqa: F401 # pragma: no cover from loguru import logger -config = ConfigManager().config - -if not config.cloud_mode_enabled: - - @app.command() - def mcp( - transport: str = typer.Option( - "stdio", help="Transport type: stdio, streamable-http, or sse" - ), - host: str = typer.Option( - "0.0.0.0", help="Host for HTTP transports (use 0.0.0.0 to allow external connections)" - ), - port: int = typer.Option(8000, help="Port for HTTP transports"), - path: str = typer.Option("/mcp", help="Path prefix for streamable-http transport"), - project: Optional[str] = typer.Option(None, help="Restrict MCP server to single project"), - ): # pragma: no cover - """Run the MCP server with configurable transport options. - - This command starts an MCP server using one of three transport options: - - - stdio: Standard I/O (good for local usage) - - streamable-http: Recommended for web deployments (default) - - sse: Server-Sent Events (for compatibility with existing clients) - - Initialization, file sync, and cleanup are handled by the MCP server's lifespan. - """ - # Initialize logging for MCP (file only, stdout breaks protocol) - init_mcp_logging() - - # Validate and set project constraint if specified - if project: - config_manager = ConfigManager() - project_name, _ = config_manager.get_project(project) - if not project_name: - typer.echo(f"No project found named: {project}", err=True) - raise typer.Exit(1) - - # Set env var with validated project name - os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name - logger.info(f"MCP server constrained to project: {project_name}") - - # Run the MCP server (blocks) - # Lifespan handles: initialization, migrations, file sync, cleanup - logger.info(f"Starting MCP server with {transport.upper()} transport") - - if transport == "stdio": - mcp_server.run( - transport=transport, - ) - elif transport == "streamable-http" or transport == "sse": - mcp_server.run( - transport=transport, - host=host, - port=port, - path=path, - log_level="INFO", - ) + +@app.command() +def mcp( + transport: str = typer.Option("stdio", help="Transport type: stdio, streamable-http, or sse"), + host: str = typer.Option( + "0.0.0.0", help="Host for HTTP transports (use 0.0.0.0 to allow external connections)" + ), + port: int = typer.Option(8000, help="Port for HTTP transports"), + path: str = typer.Option("/mcp", help="Path prefix for streamable-http transport"), + project: Optional[str] = typer.Option(None, help="Restrict MCP server to single project"), +): # pragma: no cover + """Run the MCP server with configurable transport options. + + This command starts an MCP server using one of three transport options: + + - stdio: Standard I/O (good for local usage) + - streamable-http: Recommended for web deployments (default) + - sse: Server-Sent Events (for compatibility with existing clients) + + Initialization, file sync, and cleanup are handled by the MCP server's lifespan. + + Note: This command is available regardless of cloud mode setting. + Users who have cloud mode enabled can still use local MCP for Claude Code + and Claude Desktop while using cloud MCP for web and mobile access. + """ + # Force local routing for local MCP server + # Why: The local MCP server should always talk to the local API, not the cloud proxy. + # Even when cloud_mode_enabled is True, stdio MCP runs locally and needs local API access. + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true" + + # Initialize logging for MCP (file only, stdout breaks protocol) + init_mcp_logging() + + # Validate and set project constraint if specified + if project: + config_manager = ConfigManager() + project_name, _ = config_manager.get_project(project) + if not project_name: + typer.echo(f"No project found named: {project}", err=True) + raise typer.Exit(1) + + # Set env var with validated project name + os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name + logger.info(f"MCP server constrained to project: {project_name}") + + # Run the MCP server (blocks) + # Lifespan handles: initialization, migrations, file sync, cleanup + logger.info(f"Starting MCP server with {transport.upper()} transport") + + if transport == "stdio": + mcp_server.run( + transport=transport, + ) + elif transport == "streamable-http" or transport == "sse": + mcp_server.run( + transport=transport, + host=host, + port=port, + path=path, + log_level="INFO", + ) diff --git a/src/basic_memory/cli/commands/project.py b/src/basic_memory/cli/commands/project.py index 88c344c1..8744c57a 100644 --- a/src/basic_memory/cli/commands/project.py +++ b/src/basic_memory/cli/commands/project.py @@ -1,32 +1,32 @@ """Command module for basic-memory project management.""" +import json import os +from datetime import datetime from pathlib import Path import typer from rich.console import Console +from rich.panel import Panel from rich.table import Table from basic_memory.cli.app import app 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 -import json -from datetime import datetime - -from rich.panel import Panel from basic_memory.mcp.async_client import get_client -from basic_memory.mcp.tools.utils import call_get, call_post, call_delete, call_put, call_patch +from basic_memory.mcp.tools.utils import call_delete, call_get, call_patch, call_post, call_put from basic_memory.schemas.project_info import ProjectList, ProjectStatusResponse from basic_memory.utils import generate_permalink, normalize_project_path # Import rclone commands for project sync from basic_memory.cli.commands.cloud.rclone_commands import ( - SyncProject, RcloneError, - project_sync, + SyncProject, project_bisync, project_check, project_ls, + project_sync, ) from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info @@ -46,8 +46,22 @@ def format_path(path: str) -> str: @project_app.command("list") -def list_projects() -> None: - """List all Basic Memory projects.""" +def list_projects( + 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: + """List all Basic Memory projects. + + 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 _list_projects(): async with get_client() as client: @@ -55,19 +69,20 @@ async def _list_projects(): return ProjectList.model_validate(response.json()) try: - result = run_with_cleanup(_list_projects()) + with force_routing(local=local, cloud=cloud): + result = run_with_cleanup(_list_projects()) config = ConfigManager().config table = Table(title="Basic Memory Projects") table.add_column("Name", style="cyan") table.add_column("Path", style="green") - # Add Local Path column if in cloud mode - if config.cloud_mode_enabled: + # Add Local Path column if in cloud mode and not forcing local + if config.cloud_mode_enabled and not local: 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 + show_default_column = local or not config.cloud_mode_enabled or config.default_project_mode if show_default_column: table.add_column("Default", style="magenta") @@ -78,8 +93,8 @@ async def _list_projects(): # Build row based on mode row = [project.name, format_path(normalized_path)] - # Add local path if in cloud mode - if config.cloud_mode_enabled: + # Add local path if in cloud mode and not forcing local + if config.cloud_mode_enabled and not local: local_path = "" if project.name in config.cloud_projects: local_path = config.cloud_projects[project.name].local_path or "" @@ -108,9 +123,16 @@ def add_project( None, "--local-path", help="Local sync path for cloud mode (optional)" ), 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 @@ -118,14 +140,23 @@ def add_project( 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: local flag forces local mode behavior + effective_cloud_mode = config.cloud_mode_enabled and not local + # 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: + if effective_cloud_mode: # Cloud mode: path auto-generated from name, local sync is optional async def _add_project(): @@ -154,11 +185,12 @@ async def _add_project(): return ProjectStatusResponse.model_validate(response.json()) try: - result = run_with_cleanup(_add_project()) + with force_routing(local=local, cloud=cloud): + result = run_with_cleanup(_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: + if effective_cloud_mode and local_sync_path: from basic_memory.config import CloudProjectConfig # Create local directory if it doesn't exist @@ -243,8 +275,21 @@ def remove_project( 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.""" + """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(): async with get_client() as client: @@ -265,11 +310,11 @@ async def _remove_project(): try: # Get config to check for local sync path and bisync state config = ConfigManager().config - local_path = None + local_path_config = None has_bisync_state = False - if config.cloud_mode_enabled and name in config.cloud_projects: - local_path = config.cloud_projects[name].local_path + if config.cloud_mode_enabled and not local and name in config.cloud_projects: + local_path_config = config.cloud_projects[name].local_path # Check for bisync state from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state @@ -278,17 +323,18 @@ async def _remove_project(): has_bisync_state = bisync_state_path.exists() # Remove project from cloud/API - result = run_with_cleanup(_remove_project()) + 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: - local_dir = Path(local_path) + 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}[/green]") + console.print(f"[green]Removed local sync directory: {local_path_config}[/green]") # Clean up bisync state if it exists if has_bisync_state: @@ -301,14 +347,14 @@ async def _remove_project(): console.print("[green]Removed bisync state[/green]") # Clean up cloud_projects config entry - if config.cloud_mode_enabled and name in config.cloud_projects: + if config.cloud_mode_enabled and not local 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]") + 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]") @@ -318,15 +364,24 @@ async def _remove_project(): @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 when 'config.default_project_mode' is set. - Note: This command is only available in local mode. + In cloud mode, use --local to modify the local configuration. """ config = ConfigManager().config - if config.cloud_mode_enabled: - console.print("[red]Error: 'default' command is not available in cloud mode[/red]") + # Trigger: cloud mode enabled without --local flag + # Why: default project is a local configuration concept + # Outcome: require explicit --local flag to modify local config in cloud mode + if config.cloud_mode_enabled and not local: + console.print( + "[red]Error: 'default' command requires --local flag in cloud mode[/red]\n" + "[yellow]Hint: Use 'bm project default --local' to set local default[/yellow]" + ) raise typer.Exit(1) async def _set_default(): @@ -346,7 +401,8 @@ async def _set_default(): return ProjectStatusResponse.model_validate(response.json()) try: - result = run_with_cleanup(_set_default()) + 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]") @@ -354,15 +410,25 @@ async def _set_default(): @project_app.command("sync-config") -def synchronize_projects() -> None: +def synchronize_projects( + local: bool = typer.Option( + False, "--local", help="Force local API routing (required in cloud mode)" + ), +) -> None: """Synchronize project config between configuration file and database. - Note: This command is only available in local mode. + In cloud mode, use --local to sync local configuration. """ config = ConfigManager().config - if config.cloud_mode_enabled: - console.print("[red]Error: 'sync-config' command is not available in cloud mode[/red]") + # Trigger: cloud mode enabled without --local flag + # Why: sync-config syncs local config file with local database + # Outcome: require explicit --local flag to clarify intent in cloud mode + if config.cloud_mode_enabled and not local: + console.print( + "[red]Error: 'sync-config' command requires --local flag in cloud mode[/red]\n" + "[yellow]Hint: Use 'bm project sync-config --local' to sync local config[/yellow]" + ) raise typer.Exit(1) async def _sync_config(): @@ -371,7 +437,8 @@ async def _sync_config(): return ProjectStatusResponse.model_validate(response.json()) try: - result = run_with_cleanup(_sync_config()) + with force_routing(local=local): + result = run_with_cleanup(_sync_config()) console.print(f"[green]{result.message}[/green]") except Exception as e: # pragma: no cover console.print(f"[red]Error synchronizing projects: {str(e)}[/red]") @@ -382,15 +449,24 @@ async def _sync_config(): 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"), + local: bool = typer.Option( + False, "--local", help="Force local API routing (required in cloud mode)" + ), ) -> None: """Move a project to a new location. - Note: This command is only available in local mode. + In cloud mode, use --local to modify local project paths. """ config = ConfigManager().config - if config.cloud_mode_enabled: - console.print("[red]Error: 'move' command is not available in cloud mode[/red]") + # Trigger: cloud mode enabled without --local flag + # Why: moving a project is a local file system operation + # Outcome: require explicit --local flag to clarify intent in cloud mode + if config.cloud_mode_enabled and not local: + console.print( + "[red]Error: 'move' command requires --local flag in cloud mode[/red]\n" + "[yellow]Hint: Use 'bm project move --local' to move local project[/yellow]" + ) raise typer.Exit(1) # Resolve to absolute path @@ -406,7 +482,8 @@ async def _move_project(): return ProjectStatusResponse.model_validate(response.json()) try: - result = run_with_cleanup(_move_project()) + with force_routing(local=local): + result = run_with_cleanup(_move_project()) console.print(f"[green]{result.message}[/green]") # Show important file movement reminder @@ -779,11 +856,26 @@ async def _get_project(): 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.""" + """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 - info = run_with_cleanup(get_project_info(name)) + with force_routing(local=local, cloud=cloud): + info = run_with_cleanup(get_project_info(name)) if json_output: # Convert to JSON and print diff --git a/src/basic_memory/cli/commands/routing.py b/src/basic_memory/cli/commands/routing.py new file mode 100644 index 00000000..2ff09913 --- /dev/null +++ b/src/basic_memory/cli/commands/routing.py @@ -0,0 +1,73 @@ +"""CLI routing utilities for --local/--cloud flag handling. + +This module provides utilities for CLI commands to override the default routing +behavior (determined by cloud_mode_enabled in config). This allows users to: + +1. Use local MCP server even when cloud mode is enabled +2. Force local routing for specific CLI commands with --local flag +3. Force cloud routing with --cloud flag (requires authentication) + +The routing is controlled via environment variables: +- BASIC_MEMORY_FORCE_LOCAL: When "true", forces local ASGI transport +- These are checked in basic_memory.mcp.async_client.get_client() +""" + +import os +from contextlib import contextmanager +from typing import Generator + + +@contextmanager +def force_routing(local: bool = False, cloud: bool = False) -> Generator[None, None, None]: + """Context manager to temporarily override routing mode. + + Sets environment variables that are checked by get_client() to determine + whether to use local ASGI transport or cloud proxy transport. + + Args: + local: If True, force local ASGI transport (ignores cloud_mode_enabled) + cloud: If True, clear force_local to allow cloud routing + + Usage: + with force_routing(local=True): + # All API calls will use local ASGI transport + await some_api_call() + + Raises: + ValueError: If both local and cloud are True + """ + if local and cloud: + raise ValueError("Cannot specify both --local and --cloud") + + # Save original values + original_force_local = os.environ.get("BASIC_MEMORY_FORCE_LOCAL") + + try: + if local: + # Force local routing by setting the env var + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true" + elif cloud: + # Ensure force_local is NOT set, let cloud_mode_enabled take effect + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + # If neither is set, don't change anything (use default behavior) + yield + finally: + # Restore original value + if original_force_local is None: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + else: + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = original_force_local + + +def validate_routing_flags(local: bool, cloud: bool) -> None: + """Validate that --local and --cloud flags are not both specified. + + Args: + local: Value of --local flag + cloud: Value of --cloud flag + + Raises: + ValueError: If both flags are True + """ + if local and cloud: + raise ValueError("Cannot specify both --local and --cloud flags") diff --git a/src/basic_memory/cli/commands/status.py b/src/basic_memory/cli/commands/status.py index cb8cfe51..d8ef3620 100644 --- a/src/basic_memory/cli/commands/status.py +++ b/src/basic_memory/cli/commands/status.py @@ -11,6 +11,7 @@ from rich.tree import Tree from basic_memory.cli.app import app +from basic_memory.cli.commands.routing import force_routing, validate_routing_flags from basic_memory.mcp.async_client import get_client from basic_memory.mcp.tools.utils import call_post from basic_memory.schemas import SyncReportResponse @@ -162,12 +163,25 @@ def status( typer.Option(help="The project name."), ] = None, verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed file information"), + 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"), ): - """Show sync status between files and database.""" + """Show sync status between files and database. + + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + """ from basic_memory.cli.commands.command_utils import run_with_cleanup try: - run_with_cleanup(run_status(project, verbose)) # pragma: no cover + validate_routing_flags(local, cloud) + with force_routing(local=local, cloud=cloud): + run_with_cleanup(run_status(project, verbose)) # pragma: no cover + except ValueError as e: + console.print(f"[red]Error: {e}[/red]") + raise typer.Exit(code=1) except Exception as e: logger.error(f"Error checking status: {e}") typer.echo(f"Error checking status: {e}", err=True) diff --git a/src/basic_memory/cli/commands/tool.py b/src/basic_memory/cli/commands/tool.py index f7996d57..b1c387d6 100644 --- a/src/basic_memory/cli/commands/tool.py +++ b/src/basic_memory/cli/commands/tool.py @@ -9,6 +9,7 @@ from basic_memory.cli.app import app from basic_memory.cli.commands.command_utils import run_with_cleanup +from basic_memory.cli.commands.routing import force_routing, validate_routing_flags from basic_memory.config import ConfigManager # Import prompts @@ -50,6 +51,10 @@ def write_note( tags: Annotated[ Optional[List[str]], typer.Option(help="A list of tags to apply to the note") ] = None, + 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"), ): """Create or update a markdown note. Content can be provided as an argument or read from stdin. @@ -57,6 +62,9 @@ def write_note( 1. Using the --content parameter 2. Piping content through stdin (if --content is not provided) + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + Examples: # Using content parameter @@ -77,8 +85,13 @@ def write_note( # Reading from a file cat document.md | basic-memory tools write-note --title "Document" --folder "docs" + + # Force local routing in cloud mode + basic-memory tools write-note --title "My Note" --folder "notes" --content "..." --local """ try: + validate_routing_flags(local, cloud) + # If content is not provided, read from stdin if content is None: # Check if we're getting data from a pipe or redirect @@ -109,8 +122,12 @@ def write_note( # use the project name, or the default from the config project_name = project_name or config_manager.default_project - note = run_with_cleanup(mcp_write_note.fn(title, content, folder, project_name, tags)) + with force_routing(local=local, cloud=cloud): + note = run_with_cleanup(mcp_write_note.fn(title, content, folder, project_name, tags)) rprint(note) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): typer.echo(f"Error during write_note: {e}", err=True) @@ -129,24 +146,37 @@ def read_note( ] = None, page: int = 1, page_size: int = 10, + 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"), ): - """Read a markdown note from the knowledge base.""" - - # look for the project in the config - config_manager = ConfigManager() - project_name = None - if project is not None: - project_name, _ = config_manager.get_project(project) - if not project_name: - typer.echo(f"No project found named: {project}", err=True) - raise typer.Exit(1) - - # use the project name, or the default from the config - project_name = project_name or config_manager.default_project + """Read a markdown note from the knowledge base. + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + """ try: - note = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size)) + validate_routing_flags(local, cloud) + + # look for the project in the config + config_manager = ConfigManager() + project_name = None + if project is not None: + project_name, _ = config_manager.get_project(project) + if not project_name: + typer.echo(f"No project found named: {project}", err=True) + raise typer.Exit(1) + + # use the project name, or the default from the config + project_name = project_name or config_manager.default_project + + with force_routing(local=local, cloud=cloud): + note = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size)) rprint(note) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): typer.echo(f"Error during read_note: {e}", err=True) @@ -166,38 +196,51 @@ def build_context( page: int = 1, page_size: int = 10, max_related: int = 10, + 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"), ): - """Get context needed to continue a discussion.""" - - # look for the project in the config - config_manager = ConfigManager() - project_name = None - if project is not None: - project_name, _ = config_manager.get_project(project) - if not project_name: - typer.echo(f"No project found named: {project}", err=True) - raise typer.Exit(1) - - # use the project name, or the default from the config - project_name = project_name or config_manager.default_project + """Get context needed to continue a discussion. + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + """ try: - context = run_with_cleanup( - mcp_build_context.fn( - project=project_name, - url=url, - depth=depth, - timeframe=timeframe, - page=page, - page_size=page_size, - max_related=max_related, + validate_routing_flags(local, cloud) + + # look for the project in the config + config_manager = ConfigManager() + project_name = None + if project is not None: + project_name, _ = config_manager.get_project(project) + if not project_name: + typer.echo(f"No project found named: {project}", err=True) + raise typer.Exit(1) + + # use the project name, or the default from the config + project_name = project_name or config_manager.default_project + + with force_routing(local=local, cloud=cloud): + context = run_with_cleanup( + mcp_build_context.fn( + project=project_name, + url=url, + depth=depth, + timeframe=timeframe, + page=page, + page_size=page_size, + max_related=max_related, + ) ) - ) # Use json module for more controlled serialization import json context_dict = context.model_dump(exclude_none=True) print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str)) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): typer.echo(f"Error during build_context: {e}", err=True) @@ -210,18 +253,32 @@ def recent_activity( type: Annotated[Optional[List[SearchItemType]], typer.Option()] = None, depth: Optional[int] = 1, timeframe: Optional[TimeFrame] = "7d", + 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"), ): - """Get recent activity across the knowledge base.""" + """Get recent activity across the knowledge base. + + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + """ try: - result = run_with_cleanup( - mcp_recent_activity.fn( - type=type, # pyright: ignore [reportArgumentType] - depth=depth, - timeframe=timeframe, + validate_routing_flags(local, cloud) + + with force_routing(local=local, cloud=cloud): + result = run_with_cleanup( + mcp_recent_activity.fn( + type=type, # pyright: ignore [reportArgumentType] + depth=depth, + timeframe=timeframe, + ) ) - ) # The tool now returns a formatted string directly print(result) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): typer.echo(f"Error during recent_activity: {e}", err=True) @@ -246,26 +303,31 @@ def search_notes( ] = None, page: int = 1, page_size: int = 10, + 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"), ): - """Search across all content in the knowledge base.""" - - # look for the project in the config - config_manager = ConfigManager() - project_name = None - if project is not None: - project_name, _ = config_manager.get_project(project) - if not project_name: - typer.echo(f"No project found named: {project}", err=True) - raise typer.Exit(1) + """Search across all content in the knowledge base. - # use the project name, or the default from the config - project_name = project_name or config_manager.default_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) - if permalink and title: # pragma: no cover - print("Cannot search both permalink and title") - raise typer.Abort() + # look for the project in the config + config_manager = ConfigManager() + project_name = None + if project is not None: + project_name, _ = config_manager.get_project(project) + if not project_name: + typer.echo(f"No project found named: {project}", err=True) + raise typer.Exit(1) + + # use the project name, or the default from the config + project_name = project_name or config_manager.default_project - try: if permalink and title: # pragma: no cover typer.echo( "Use either --permalink or --title, not both. Exiting.", @@ -279,21 +341,25 @@ def search_notes( search_type = ("title" if title else None,) search_type = "text" if search_type is None else search_type - results = run_with_cleanup( - mcp_search.fn( - query, - project_name, - search_type=search_type, - page=page, - after_date=after_date, - page_size=page_size, + with force_routing(local=local, cloud=cloud): + results = run_with_cleanup( + mcp_search.fn( + query, + project_name, + search_type=search_type, + page=page, + after_date=after_date, + page_size=page_size, + ) ) - ) # Use json module for more controlled serialization import json results_dict = results.model_dump(exclude_none=True) print(json.dumps(results_dict, indent=2, ensure_ascii=True, default=str)) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): logger.exception("Error during search", e) @@ -308,12 +374,28 @@ def continue_conversation( timeframe: Annotated[ Optional[str], typer.Option(help="How far back to look for activity") ] = None, + 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"), ): - """Prompt to continue a previous conversation or work session.""" + """Prompt to continue a previous conversation or work session. + + Use --local to force local routing when cloud mode is enabled. + Use --cloud to force cloud routing when cloud mode is disabled. + """ try: - # Prompt functions return formatted strings directly - session = run_with_cleanup(mcp_continue_conversation.fn(topic=topic, timeframe=timeframe)) # type: ignore + validate_routing_flags(local, cloud) + + with force_routing(local=local, cloud=cloud): + # Prompt functions return formatted strings directly + session = run_with_cleanup( + mcp_continue_conversation.fn(topic=topic, timeframe=timeframe) # type: ignore[arg-type] + ) rprint(session) + except ValueError as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) except Exception as e: # pragma: no cover if not isinstance(e, typer.Exit): logger.exception("Error continuing conversation", e) diff --git a/src/basic_memory/mcp/async_client.py b/src/basic_memory/mcp/async_client.py index da2a12e3..44d2ee3f 100644 --- a/src/basic_memory/mcp/async_client.py +++ b/src/basic_memory/mcp/async_client.py @@ -1,3 +1,4 @@ +import os from contextlib import asynccontextmanager, AbstractAsyncContextManager from typing import AsyncIterator, Callable, Optional @@ -8,6 +9,19 @@ from basic_memory.config import ConfigManager +def _force_local_mode() -> bool: + """Check if local mode is forced via environment variable. + + This allows commands like `bm mcp` to force local routing even when + cloud_mode_enabled is True in config. The local MCP server should + always talk to the local API, not the cloud proxy. + + Returns: + True if BASIC_MEMORY_FORCE_LOCAL is set to a truthy value + """ + return os.environ.get("BASIC_MEMORY_FORCE_LOCAL", "").lower() in ("true", "1", "yes") + + # Optional factory override for dependency injection _client_factory: Optional[Callable[[], AbstractAsyncContextManager[AsyncClient]]] = None @@ -71,7 +85,17 @@ async def get_client() -> AsyncIterator[AsyncClient]: pool=30.0, # 30 seconds for connection pool ) - if config.cloud_mode_enabled: + # Trigger: BASIC_MEMORY_FORCE_LOCAL env var is set + # Why: allows local MCP server and CLI commands to route locally + # even when cloud_mode_enabled is True + # Outcome: uses ASGI transport for in-process local API calls + if _force_local_mode(): + logger.info("Force local mode enabled - using ASGI client for local Basic Memory API") + async with AsyncClient( + transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout + ) as client: + yield client + elif config.cloud_mode_enabled: # CLI cloud mode: inject auth when creating client from basic_memory.cli.auth import CLIAuth @@ -126,7 +150,13 @@ def create_client() -> AsyncClient: pool=30.0, # 30 seconds for connection pool ) - if config.cloud_mode_enabled: + # Check force local first (for local MCP server and CLI --local flag) + if _force_local_mode(): + logger.info("Force local mode enabled - using ASGI client for local Basic Memory API") + return AsyncClient( + transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout + ) + elif config.cloud_mode_enabled: # Use HTTP transport to proxy endpoint proxy_base_url = f"{config.cloud_host}/proxy" logger.info(f"Creating HTTP client for proxy at: {proxy_base_url}") diff --git a/test-int/cli/test_routing_integration.py b/test-int/cli/test_routing_integration.py new file mode 100644 index 00000000..3a0970f2 --- /dev/null +++ b/test-int/cli/test_routing_integration.py @@ -0,0 +1,181 @@ +"""Integration tests for CLI routing flags (--local/--cloud). + +These tests verify that the --local and --cloud flags work correctly +across CLI commands, and that the MCP command forces local routing. + +Note: Environment variable behavior during command execution is tested +in unit tests (tests/cli/test_routing.py) which can properly monkeypatch +the modules before they are imported. These integration tests focus on +CLI behavior: flag acceptance and error handling. +""" + +import os + +import pytest +from typer.testing import CliRunner + +from basic_memory.cli.main import app as cli_app + + +runner = CliRunner() + + +class TestRoutingFlagsValidation: + """Tests for --local/--cloud flag validation. + + These tests verify that using both --local and --cloud together + produces an appropriate error message. + """ + + def test_status_both_flags_error(self): + """Using both --local and --cloud should produce an error.""" + result = runner.invoke(cli_app, ["status", "--local", "--cloud"]) + # Exit code can be 1 or 2 depending on how typer handles the exception + assert result.exit_code != 0 + assert "Cannot specify both --local and --cloud" in result.output + + def test_project_list_both_flags_error(self): + """Using both --local and --cloud should produce an error.""" + result = runner.invoke(cli_app, ["project", "list", "--local", "--cloud"]) + assert result.exit_code != 0 + assert "Cannot specify both --local and --cloud" in result.output + + def test_tool_search_both_flags_error(self): + """Using both --local and --cloud should produce an error.""" + result = runner.invoke(cli_app, ["tool", "search-notes", "test", "--local", "--cloud"]) + assert result.exit_code != 0 + assert "Cannot specify both --local and --cloud" in result.output + + def test_tool_read_note_both_flags_error(self): + """Using both --local and --cloud should produce an error.""" + result = runner.invoke(cli_app, ["tool", "read-note", "test", "--local", "--cloud"]) + assert result.exit_code != 0 + assert "Cannot specify both --local and --cloud" in result.output + + def test_tool_build_context_both_flags_error(self): + """Using both --local and --cloud should produce an error.""" + result = runner.invoke( + cli_app, ["tool", "build-context", "memory://test", "--local", "--cloud"] + ) + assert result.exit_code != 0 + assert "Cannot specify both --local and --cloud" in result.output + + +class TestMcpCommandForcesLocal: + """Tests that the MCP command forces local routing.""" + + def test_mcp_sets_force_local_env(self, monkeypatch): + """MCP command should set BASIC_MEMORY_FORCE_LOCAL before server starts.""" + # Track what environment variable was set + env_set_value = [] + + # Mock the MCP server run to capture env state without actually starting server + import basic_memory.cli.commands.mcp as mcp_mod + + def mock_run(*args, **kwargs): + env_set_value.append(os.environ.get("BASIC_MEMORY_FORCE_LOCAL")) + # Don't actually start the server + raise SystemExit(0) + + # Get the actual mcp_server from the module + monkeypatch.setattr(mcp_mod.mcp_server, "run", mock_run) + + # Also mock init_mcp_logging to avoid file operations + monkeypatch.setattr(mcp_mod, "init_mcp_logging", lambda: None) + + runner.invoke(cli_app, ["mcp"]) + + # Environment variable should have been set to "true" + assert len(env_set_value) == 1 + assert env_set_value[0] == "true" + + +class TestToolCommandsAcceptFlags: + """Tests that tool commands accept routing flags without parsing errors.""" + + @pytest.mark.parametrize( + "command,args", + [ + ("search-notes", ["test query"]), + ("recent-activity", []), + ("read-note", ["test"]), + ("build-context", ["memory://test"]), + ("continue-conversation", []), + ], + ) + def test_tool_commands_accept_local_flag(self, command, args, app_config): + """Tool commands should accept --local flag without parsing error.""" + full_args = ["tool", command] + args + ["--local"] + result = runner.invoke(cli_app, full_args) + # Should not fail due to flag parsing (No such option error) + assert "No such option: --local" not in result.output + + @pytest.mark.parametrize( + "command,args", + [ + ("search-notes", ["test query"]), + ("recent-activity", []), + ("read-note", ["test"]), + ("build-context", ["memory://test"]), + ("continue-conversation", []), + ], + ) + def test_tool_commands_accept_cloud_flag(self, command, args, app_config): + """Tool commands should accept --cloud flag without parsing error.""" + full_args = ["tool", command] + args + ["--cloud"] + result = runner.invoke(cli_app, full_args) + # Should not fail due to flag parsing (No such option error) + assert "No such option: --cloud" not in result.output + + +class TestProjectCommandsAcceptFlags: + """Tests that project commands accept routing flags without parsing errors.""" + + def test_project_list_accepts_local_flag(self, app_config): + """project list should accept --local flag.""" + result = runner.invoke(cli_app, ["project", "list", "--local"]) + assert "No such option: --local" not in result.output + + def test_project_list_accepts_cloud_flag(self, app_config): + """project list should accept --cloud flag.""" + result = runner.invoke(cli_app, ["project", "list", "--cloud"]) + assert "No such option: --cloud" not in result.output + + def test_project_info_accepts_local_flag(self, app_config): + """project info should accept --local flag.""" + result = runner.invoke(cli_app, ["project", "info", "--local"]) + assert "No such option: --local" not in result.output + + def test_project_info_accepts_cloud_flag(self, app_config): + """project info should accept --cloud flag.""" + result = runner.invoke(cli_app, ["project", "info", "--cloud"]) + assert "No such option: --cloud" not in result.output + + def test_project_default_accepts_local_flag(self, app_config): + """project default should accept --local flag.""" + result = runner.invoke(cli_app, ["project", "default", "test", "--local"]) + assert "No such option: --local" not in result.output + + def test_project_sync_config_accepts_local_flag(self, app_config): + """project sync-config should accept --local flag.""" + result = runner.invoke(cli_app, ["project", "sync-config", "test", "--local"]) + assert "No such option: --local" not in result.output + + def test_project_move_accepts_local_flag(self, app_config): + """project move should accept --local flag.""" + result = runner.invoke(cli_app, ["project", "move", "test", "/tmp/dest", "--local"]) + assert "No such option: --local" not in result.output + + +class TestStatusCommandAcceptsFlags: + """Tests that status command accepts routing flags.""" + + def test_status_accepts_local_flag(self, app_config): + """status should accept --local flag.""" + result = runner.invoke(cli_app, ["status", "--local"]) + assert "No such option: --local" not in result.output + + def test_status_accepts_cloud_flag(self, app_config): + """status should accept --cloud flag.""" + result = runner.invoke(cli_app, ["status", "--cloud"]) + assert "No such option: --cloud" not in result.output diff --git a/tests/cli/test_routing.py b/tests/cli/test_routing.py new file mode 100644 index 00000000..8aea9b79 --- /dev/null +++ b/tests/cli/test_routing.py @@ -0,0 +1,102 @@ +"""Tests for CLI routing utilities.""" + +import os + +import pytest + +from basic_memory.cli.commands.routing import force_routing, validate_routing_flags + + +class TestValidateRoutingFlags: + """Tests for validate_routing_flags function.""" + + def test_neither_flag(self): + """Should not raise when neither flag is set.""" + validate_routing_flags(local=False, cloud=False) + + def test_local_only(self): + """Should not raise when only local is set.""" + validate_routing_flags(local=True, cloud=False) + + def test_cloud_only(self): + """Should not raise when only cloud is set.""" + validate_routing_flags(local=False, cloud=True) + + def test_both_flags_raises(self): + """Should raise ValueError when both flags are set.""" + with pytest.raises(ValueError, match="Cannot specify both --local and --cloud"): + validate_routing_flags(local=True, cloud=True) + + +class TestForceRouting: + """Tests for force_routing context manager.""" + + def test_local_sets_env_var(self): + """Local flag should set BASIC_MEMORY_FORCE_LOCAL.""" + # Ensure env var is not set + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + with force_routing(local=True): + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") == "true" + + # Should be cleaned up after context exits + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") is None + + def test_cloud_clears_env_var(self): + """Cloud flag should clear BASIC_MEMORY_FORCE_LOCAL if set.""" + # Set env var + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true" + + with force_routing(cloud=True): + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") is None + + # Should restore original value after context exits + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") == "true" + + # Cleanup + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_neither_flag_no_change(self): + """Neither flag should not change env vars.""" + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + with force_routing(): + # Should not be set + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") is None + + # Should still not be set + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") is None + + def test_preserves_original_env_var(self): + """Should restore original env var value after context exits.""" + original_value = "original" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = original_value + + with force_routing(local=True): + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") == "true" + + # Should restore original value + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") == original_value + + # Cleanup + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_both_flags_raises(self): + """Should raise ValueError when both flags are set.""" + with pytest.raises(ValueError, match="Cannot specify both --local and --cloud"): + with force_routing(local=True, cloud=True): + pass + + def test_restores_on_exception(self): + """Should restore env vars even when exception is raised.""" + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + try: + with force_routing(local=True): + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") == "true" + raise RuntimeError("Test exception") + except RuntimeError: + pass + + # Should be cleaned up even after exception + assert os.environ.get("BASIC_MEMORY_FORCE_LOCAL") is None diff --git a/tests/mcp/test_async_client_force_local.py b/tests/mcp/test_async_client_force_local.py new file mode 100644 index 00000000..a05c740c --- /dev/null +++ b/tests/mcp/test_async_client_force_local.py @@ -0,0 +1,79 @@ +"""Tests for async_client force_local_mode functionality.""" + +import os + + +from basic_memory.mcp.async_client import _force_local_mode + + +class TestForceLocalMode: + """Tests for _force_local_mode function.""" + + def test_returns_false_when_not_set(self): + """Should return False when env var is not set.""" + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + assert _force_local_mode() is False + + def test_returns_true_for_true(self): + """Should return True when env var is 'true'.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true" + try: + assert _force_local_mode() is True + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_true_for_1(self): + """Should return True when env var is '1'.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "1" + try: + assert _force_local_mode() is True + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_true_for_yes(self): + """Should return True when env var is 'yes'.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "yes" + try: + assert _force_local_mode() is True + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_true_for_TRUE_uppercase(self): + """Should return True when env var is 'TRUE' (case insensitive).""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "TRUE" + try: + assert _force_local_mode() is True + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_false_for_false(self): + """Should return False when env var is 'false'.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "false" + try: + assert _force_local_mode() is False + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_false_for_0(self): + """Should return False when env var is '0'.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "0" + try: + assert _force_local_mode() is False + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_false_for_empty(self): + """Should return False when env var is empty string.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "" + try: + assert _force_local_mode() is False + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) + + def test_returns_false_for_random_string(self): + """Should return False when env var is random string.""" + os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "random" + try: + assert _force_local_mode() is False + finally: + os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None) diff --git a/tests/sync/test_sync_service.py b/tests/sync/test_sync_service.py index 91ec14b6..6453e50a 100644 --- a/tests/sync/test_sync_service.py +++ b/tests/sync/test_sync_service.py @@ -757,10 +757,11 @@ async def test_sync_preserves_timestamps( entity_created_epoch = file_entity.created_at.timestamp() entity_updated_epoch = file_entity.updated_at.timestamp() - # Allow 2s difference on Windows due to filesystem timing precision - tolerance = 2 if os.name == "nt" else 1 + # Allow 2s difference due to filesystem timing precision and database processing delays + # Windows has coarser filesystem timestamps, but Postgres can also have slight timing differences + tolerance = 2 assert abs(entity_created_epoch - file_stats.st_ctime) < tolerance - assert abs(entity_updated_epoch - file_stats.st_mtime) < tolerance # Allow tolerance difference + assert abs(entity_updated_epoch - file_stats.st_mtime) < tolerance @pytest.mark.asyncio