refactor: add composition roots for API, MCP, and CLI entrypoints (#492)

Introduces composition roots (containers) that centralize reading ConfigManager
and environment variables at entrypoints. This is the foundation for reducing
coupling and scattered cloud_mode/config checks throughout the codebase.

Changes:
- Add `basic_memory/runtime.py` with RuntimeMode enum and resolve_runtime_mode()
- Add `api/container.py` with ApiContainer composition root
- Add `mcp/container.py` with McpContainer composition root
- Add `cli/container.py` with CliContainer composition root
- Update API lifespan to use ApiContainer for config and mode decisions
- Update MCP lifespan to use McpContainer for config and mode decisions
- Update CLI callback to use CliContainer for config access
- Update integration test to patch container methods instead of module imports

The containers provide:
- Single point of config access (only composition roots read ConfigManager)
- Runtime mode resolution (cloud/local/test) in one place
- Centralized sync/watch decision logic (should_sync_files property)

This is step 1 of the refactoring roadmap in issue #490. Future steps will:
- Split deps.py into feature modules (#491)
- Inject config explicitly rather than reading globally (#496)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/api/app.py

@@ -8,7 +8,7 @@
 from loguru import logger
 
 from basic_memory import __version__ as version
-from basic_memory import db
+from basic_memory.api.container import ApiContainer, set_container
 from basic_memory.api.routers import (
     directory_router,
     importer_router,
@@ -30,7 +30,7 @@
     prompt_router as v2_prompt,
     importer_router as v2_importer,
 )
-from basic_memory.config import ConfigManager, init_api_logging
+from basic_memory.config import init_api_logging
 from basic_memory.services.initialization import initialize_file_sync, initialize_app
 
 
@@ -41,35 +41,41 @@ async def lifespan(app: FastAPI):  # pragma: no cover
     # Initialize logging for API (stdout in cloud mode, file otherwise)
     init_api_logging()
 
-    app_config = ConfigManager().config
-    logger.info("Starting Basic Memory API")
+    # --- Composition Root ---
+    # Create container and read config (single point of config access)
+    container = ApiContainer.create()
+    set_container(container)
+    app.state.container = container
 
-    await initialize_app(app_config)
+    logger.info(f"Starting Basic Memory API (mode={container.mode.name})")
+
+    await initialize_app(container.config)
 
     # Cache database connections in app state for performance
     logger.info("Initializing database and caching connections...")
-    engine, session_maker = await db.get_or_create_db(app_config.database_path)
+    engine, session_maker = await container.init_database()
     app.state.engine = engine
     app.state.session_maker = session_maker
     logger.info("Database connections cached in app state")
 
-    # Start file sync if enabled
-    if app_config.sync_changes and not app_config.is_test_env:
-        logger.info(f"Sync changes enabled: {app_config.sync_changes}")
+    # Start file sync if enabled (decision made by container)
+    if container.should_sync_files:
+        logger.info(f"Sync changes enabled: {container.config.sync_changes}")
 
-        # start file sync task in background
+        # Start file sync task in background
         async def _file_sync_runner() -> None:
-            await initialize_file_sync(app_config)
+            await initialize_file_sync(container.config)
 
         app.state.sync_task = asyncio.create_task(_file_sync_runner())
     else:
-        if app_config.is_test_env:
+        # Log why sync was skipped
+        if container.mode.is_test:
             logger.info("Test environment detected. Skipping file sync service.")
         else:
             logger.info("Sync changes disabled. Skipping file sync service.")
         app.state.sync_task = None
 
-    # proceed with startup
+    # Proceed with startup
     yield
 
     logger.info("Shutting down Basic Memory API")
@@ -81,7 +87,7 @@ async def _file_sync_runner() -> None:
         except asyncio.CancelledError:
             logger.info("Sync task cancelled successfully")
 
-    await db.shutdown_db()
+    await container.shutdown_database()
 
 
 # Initialize FastAPI app

src/basic_memory/api/container.py

@@ -0,0 +1,100 @@
+"""API composition root for Basic Memory.
+
+This container owns reading ConfigManager and environment variables for the
+API entrypoint. Downstream modules receive config/dependencies explicitly
+rather than reading globals.
+
+Design principles:
+- Only this module reads ConfigManager directly
+- Runtime mode (cloud/local/test) is resolved here
+- Factories for services are provided, not singletons
+"""
+
+from dataclasses import dataclass
+
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, AsyncSession
+
+from basic_memory import db
+from basic_memory.config import BasicMemoryConfig, ConfigManager
+from basic_memory.runtime import RuntimeMode, resolve_runtime_mode
+
+
+@dataclass
+class ApiContainer:
+    """Composition root for the API entrypoint.
+
+    Holds resolved configuration and runtime context.
+    Created once at app startup, then used to wire dependencies.
+    """
+
+    config: BasicMemoryConfig
+    mode: RuntimeMode
+
+    # --- Database ---
+    # Cached database connections (set during lifespan startup)
+    engine: AsyncEngine | None = None
+    session_maker: async_sessionmaker[AsyncSession] | None = None
+
+    @classmethod
+    def create(cls) -> "ApiContainer":
+        """Create container by reading ConfigManager.
+
+        This is the single point where API reads global config.
+        """
+        config = ConfigManager().config
+        mode = resolve_runtime_mode(
+            cloud_mode_enabled=config.cloud_mode_enabled,
+            is_test_env=config.is_test_env,
+        )
+        return cls(config=config, mode=mode)
+
+    # --- Runtime Mode Properties ---
+
+    @property
+    def should_sync_files(self) -> bool:
+        """Whether file sync should be started.
+
+        Sync is enabled when:
+        - sync_changes is True in config
+        - Not in test mode (tests manage their own sync)
+        """
+        return self.config.sync_changes and not self.mode.is_test
+
+    # --- Database Factory ---
+
+    async def init_database(self) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:
+        """Initialize and cache database connections.
+
+        Returns:
+            Tuple of (engine, session_maker)
+        """
+        engine, session_maker = await db.get_or_create_db(self.config.database_path)
+        self.engine = engine
+        self.session_maker = session_maker
+        return engine, session_maker
+
+    async def shutdown_database(self) -> None:
+        """Clean up database connections."""
+        await db.shutdown_db()
+
+
+# Module-level container instance (set by lifespan)
+# This allows deps.py to access the container without reading ConfigManager
+_container: ApiContainer | None = None
+
+
+def get_container() -> ApiContainer:
+    """Get the current API container.
+
+    Raises:
+        RuntimeError: If container hasn't been initialized
+    """
+    if _container is None:
+        raise RuntimeError("API container not initialized. Call set_container() first.")
+    return _container
+
+
+def set_container(container: ApiContainer) -> None:
+    """Set the API container (called by lifespan)."""
+    global _container
+    _container = container

src/basic_memory/cli/app.py

@@ -14,7 +14,8 @@
 
 import typer  # noqa: E402
 
-from basic_memory.config import ConfigManager, init_cli_logging  # noqa: E402
+from basic_memory.cli.container import CliContainer, set_container  # noqa: E402
+from basic_memory.config import init_cli_logging  # noqa: E402
 from basic_memory.telemetry import show_notice_if_needed, track_app_started  # noqa: E402
 
 
@@ -47,6 +48,11 @@ def app_callback(
     # Initialize logging for CLI (file only, no stdout)
     init_cli_logging()
 
+    # --- Composition Root ---
+    # Create container and read config (single point of config access)
+    container = CliContainer.create()
+    set_container(container)
+
     # Show telemetry notice and track CLI startup
     # Skip for 'mcp' command - it handles its own telemetry in lifespan
     # Skip for 'telemetry' command - avoid issues when user is managing telemetry
@@ -65,8 +71,7 @@ def app_callback(
     ):
         from basic_memory.services.initialization import ensure_initialization
 
-        app_config = ConfigManager().config
-        ensure_initialization(app_config)
+        ensure_initialization(container.config)
 
 
 ## import

src/basic_memory/cli/container.py

@@ -0,0 +1,84 @@
+"""CLI composition root for Basic Memory.
+
+This container owns reading ConfigManager and environment variables for the
+CLI entrypoint. Downstream modules receive config/dependencies explicitly
+rather than reading globals.
+
+Design principles:
+- Only this module reads ConfigManager directly
+- Runtime mode (cloud/local/test) is resolved here
+- Different CLI commands may need different initialization
+"""
+
+from dataclasses import dataclass
+
+from basic_memory.config import BasicMemoryConfig, ConfigManager
+from basic_memory.runtime import RuntimeMode, resolve_runtime_mode
+
+
+@dataclass
+class CliContainer:
+    """Composition root for the CLI entrypoint.
+
+    Holds resolved configuration and runtime context.
+    Created once at CLI startup, then used by subcommands.
+    """
+
+    config: BasicMemoryConfig
+    mode: RuntimeMode
+
+    @classmethod
+    def create(cls) -> "CliContainer":
+        """Create container by reading ConfigManager.
+
+        This is the single point where CLI reads global config.
+        """
+        config = ConfigManager().config
+        mode = resolve_runtime_mode(
+            cloud_mode_enabled=config.cloud_mode_enabled,
+            is_test_env=config.is_test_env,
+        )
+        return cls(config=config, mode=mode)
+
+    # --- Runtime Mode Properties ---
+
+    @property
+    def is_cloud_mode(self) -> bool:
+        """Whether running in cloud mode."""
+        return self.mode.is_cloud
+
+
+# Module-level container instance (set by app callback)
+_container: CliContainer | None = None
+
+
+def get_container() -> CliContainer:
+    """Get the current CLI container.
+
+    Returns:
+        The CLI container
+
+    Raises:
+        RuntimeError: If container hasn't been initialized
+    """
+    if _container is None:
+        raise RuntimeError("CLI container not initialized. Call set_container() first.")
+    return _container
+
+
+def set_container(container: CliContainer) -> None:
+    """Set the CLI container (called by app callback)."""
+    global _container
+    _container = container
+
+
+def get_or_create_container() -> CliContainer:
+    """Get existing container or create new one.
+
+    This is useful for CLI commands that might be called before
+    the main app callback runs (e.g., eager options).
+    """
+    global _container
+    if _container is None:
+        _container = CliContainer.create()
+    return _container

src/basic_memory/mcp/container.py

@@ -0,0 +1,91 @@
+"""MCP composition root for Basic Memory.
+
+This container owns reading ConfigManager and environment variables for the
+MCP server entrypoint. Downstream modules receive config/dependencies explicitly
+rather than reading globals.
+
+Design principles:
+- Only this module reads ConfigManager directly
+- Runtime mode (cloud/local/test) is resolved here
+- File sync decisions are centralized here
+"""
+
+from dataclasses import dataclass
+
+from basic_memory.config import BasicMemoryConfig, ConfigManager
+from basic_memory.runtime import RuntimeMode, resolve_runtime_mode
+
+
+@dataclass
+class McpContainer:
+    """Composition root for the MCP server entrypoint.
+
+    Holds resolved configuration and runtime context.
+    Created once at server startup, then used to wire dependencies.
+    """
+
+    config: BasicMemoryConfig
+    mode: RuntimeMode
+
+    @classmethod
+    def create(cls) -> "McpContainer":
+        """Create container by reading ConfigManager.
+
+        This is the single point where MCP reads global config.
+        """
+        config = ConfigManager().config
+        mode = resolve_runtime_mode(
+            cloud_mode_enabled=config.cloud_mode_enabled,
+            is_test_env=config.is_test_env,
+        )
+        return cls(config=config, mode=mode)
+
+    # --- Runtime Mode Properties ---
+
+    @property
+    def should_sync_files(self) -> bool:
+        """Whether local file sync should be started.
+
+        Sync is enabled when:
+        - sync_changes is True in config
+        - Not in test mode (tests manage their own sync)
+        - Not in cloud mode (cloud handles sync differently)
+        """
+        return (
+            self.config.sync_changes and not self.mode.is_test and not self.mode.is_cloud
+        )
+
+    @property
+    def sync_skip_reason(self) -> str | None:
+        """Reason why sync is skipped, or None if sync should run.
+
+        Useful for logging why sync was disabled.
+        """
+        if self.mode.is_test:
+            return "Test environment detected"
+        if self.mode.is_cloud:
+            return "Cloud mode enabled"
+        if not self.config.sync_changes:
+            return "Sync changes disabled"
+        return None
+
+
+# Module-level container instance (set by lifespan)
+_container: McpContainer | None = None
+
+
+def get_container() -> McpContainer:
+    """Get the current MCP container.
+
+    Raises:
+        RuntimeError: If container hasn't been initialized
+    """
+    if _container is None:
+        raise RuntimeError("MCP container not initialized. Call set_container() first.")
+    return _container
+
+
+def set_container(container: McpContainer) -> None:
+    """Set the MCP container (called by lifespan)."""
+    global _container
+    _container = container