add background migration task and status tool/prompt

Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/mcp/prompts/__init__.py

@@ -10,10 +10,12 @@ from basic_memory.mcp.prompts import continue_conversation
 from basic_memory.mcp.prompts import recent_activity
 from basic_memory.mcp.prompts import search
 from basic_memory.mcp.prompts import ai_assistant_guide
+from basic_memory.mcp.prompts import migration_status
 
 __all__ = [
     "ai_assistant_guide",
     "continue_conversation",
+    "migration_status",
     "recent_activity",
     "search",
 ]

src/basic_memory/mcp/prompts/migration_status.py

@@ -0,0 +1,114 @@
+"""Migration status prompt for Basic Memory MCP server."""
+
+from typing import Optional
+
+from basic_memory.mcp.server import mcp
+
+
+@mcp.prompt(
+    description="""Get migration status with recommendations for AI assistants.
+    
+    This prompt provides both current migration status and guidance on how
+    AI assistants should respond when migration is in progress or completed.
+    """,
+)
+async def migration_status_prompt(
+) -> str:
+    """Get migration status with AI assistant guidance.
+
+    This prompt provides detailed migration status information along with
+    recommendations for how AI assistants should handle different migration states.
+
+    Returns:
+        Formatted migration status with AI assistant guidance
+    """
+    try:
+        from basic_memory.services.migration_service import migration_manager
+
+        state = migration_manager.state
+
+        # Build status report
+        lines = [
+            "# Basic Memory Migration Status",
+            "",
+            f"**Current Status**: {state.status.value.replace('_', ' ').title()}",
+            f"**System Ready**: {'Yes' if migration_manager.is_ready else 'No'}",
+            "",
+        ]
+
+        if migration_manager.is_ready:
+            lines.extend(
+                [
+                    "✅ **All migrations completed** - System is fully operational",
+                    "",
+                    "All Basic Memory MCP tools are available and functioning normally.",
+                    "You can proceed with any knowledge management tasks.",
+                ]
+            )
+        else:
+            lines.append(f"**Status Message**: {state.message}")
+
+            if state.status.value == "in_progress":
+                if state.projects_total > 0:
+                    progress = f" ({state.projects_migrated}/{state.projects_total})"
+                    lines.append(f"**Progress**: {progress}")
+
+                lines.extend(
+                    [
+                        "",
+                        "🔄 **Migration in progress** - Legacy data is being migrated to new format",
+                        "",
+                        "**Impact**: Some tools may show status messages instead of normal responses",
+                        "until migration completes (usually 1-3 minutes).",
+                    ]
+                )
+
+            elif state.status.value == "failed":
+                lines.extend(
+                    [
+                        "",
+                        f"❌ **Migration failed**: {state.error or 'Unknown error'}",
+                        "",
+                        "**Impact**: System may have limited functionality until issue is resolved.",
+                    ]
+                )
+
+        # Convert string parameter to boolean
+        include_recs = True
+
+        if include_recs and not migration_manager.is_ready:
+            lines.extend(
+                [
+                    "",
+                    "---",
+                    "",
+                    "## AI Assistant Recommendations",
+                    "",
+                    "**When migration is in progress:**",
+                    "- Inform the user about the background migration",
+                    "- Suggest using `migration_status()` tool to check progress",
+                    "- Explain that most tools will work normally once migration completes",
+                    "- Avoid creating complex workflows until migration is done",
+                    "",
+                    "**What to tell users:**",
+                    "- 'Basic Memory is upgrading your knowledge base in the background'",
+                    "- 'This usually takes 1-3 minutes depending on your data size'",
+                    "- 'You can check progress anytime with the migration_status tool'",
+                    "- 'Most functionality will be available once the upgrade completes'",
+                ]
+            )
+
+        return "\n".join(lines)
+
+    except Exception as e:
+        return f"""# Migration Status - Error
+
+❌ **Unable to check migration status**: {str(e)}
+
+## AI Assistant Recommendations
+
+**When status is unavailable:**
+- Assume the system is likely working normally
+- Try proceeding with normal operations
+- If users report issues, suggest checking logs or restarting
+"""

src/basic_memory/mcp/server.py

@@ -31,23 +31,23 @@ load_dotenv()
 @dataclass
 class AppContext:
     watch_task: Optional[asyncio.Task]
+    migration_manager: Optional[Any] = None
 
 
 @asynccontextmanager
 async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:  # pragma: no cover
     """Manage application lifecycle with type-safe context"""
-    # Initialize on startup
-    watch_task = await initialize_app(app_config)
+    # Initialize on startup (now returns migration_manager)
+    migration_manager = await initialize_app(app_config)
 
     # Initialize project session with default project
     session.initialize(app_config.default_project)
 
     try:
-        yield AppContext(watch_task=watch_task)
+        yield AppContext(watch_task=None, migration_manager=migration_manager)
     finally:
-        # Cleanup on shutdown
-        if watch_task:
-            watch_task.cancel()
+        # Cleanup on shutdown - migration tasks will be cancelled automatically
+        pass
 
 
 # OAuth configuration function

src/basic_memory/mcp/tools/__init__.py

@@ -17,6 +17,7 @@ from basic_memory.mcp.tools.canvas import canvas
 from basic_memory.mcp.tools.list_directory import list_directory
 from basic_memory.mcp.tools.edit_note import edit_note
 from basic_memory.mcp.tools.move_note import move_note
+from basic_memory.mcp.tools.migration_status import migration_status
 from basic_memory.mcp.tools.project_management import (
     list_projects,
     switch_project,
@@ -36,6 +37,7 @@ __all__ = [
     "get_current_project",
     "list_directory",
     "list_projects",
+    "migration_status",
     "move_note",
     "read_content",
     "read_note",

src/basic_memory/mcp/tools/build_context.py

@@ -82,6 +82,27 @@ async def build_context(
     logger.info(f"Building context from {url}")
     # URL is already validated and normalized by MemoryUrl type annotation
 
+    # Check migration status and wait briefly if needed
+    from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
+
+    migration_status = await wait_for_migration_or_return_status(timeout=5.0)
+    if migration_status:
+        # Return a proper GraphContext with status message
+        from basic_memory.schemas.memory import MemoryMetadata
+        from datetime import datetime
+
+        return GraphContext(
+            results=[],
+            metadata=MemoryMetadata(
+                depth=depth or 1,
+                timeframe=timeframe,
+                generated_at=datetime.now(),
+                primary_count=0,
+                related_count=0,
+                uri=migration_status,  # Include status in metadata
+            ),
+        )
+
     active_project = get_active_project(project)
     project_url = active_project.project_url
 

src/basic_memory/mcp/tools/migration_status.py

@@ -0,0 +1,166 @@
+"""Migration status tool for Basic Memory MCP server."""
+
+from typing import Optional
+
+from loguru import logger
+
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.project_session import get_active_project
+
+
+@mcp.tool(
+    description="""Check the status of system migration and background operations.
+    
+    Use this tool to:
+    - Check if migration is in progress or completed
+    - Get detailed migration progress information
+    - Understand if the system is ready for normal operations
+    - Get specific error details if migration failed
+    """,
+)
+async def migration_status(project: Optional[str] = None) -> str:
+    """Get current migration status and system readiness information.
+
+    This tool provides detailed information about any ongoing or completed
+    migration operations, helping users understand system availability.
+
+    Args:
+        project: Optional project name (included for consistency with other tools)
+
+    Returns:
+        Detailed migration status including:
+        - Current migration state (ready, in progress, failed, etc.)
+        - Progress information if migration is running
+        - Error details if migration failed
+        - Estimated completion information
+        - Guidance on next steps
+
+    Examples:
+        # Check current migration status
+        migration_status()
+
+        # Get migration status for specific project context
+        migration_status(project="work-project")
+    """
+    logger.info("MCP tool call tool=migration_status")
+
+    try:
+        from basic_memory.services.migration_service import migration_manager
+
+        # Get current migration state
+        state = migration_manager.state
+
+        # Build detailed status response
+        status_lines = [
+            "# Migration Status",
+            "",
+            f"**Current Status**: {state.status.value.replace('_', ' ').title()}",
+            "",
+        ]
+
+        if migration_manager.is_ready:
+            status_lines.extend(
+                [
+                    "✅ **System Ready**: All migrations completed successfully",
+                    "",
+                    "The system is fully operational and ready for normal use. All MCP tools",
+                    "are available and functioning normally.",
+                ]
+            )
+        else:
+            # Migration in progress or failed
+            status_lines.append(f"**Message**: {state.message}")
+
+            if state.status.value == "in_progress":
+                status_lines.extend(
+                    [
+                        "",
+                        "🔄 **Migration in Progress**",
+                        "",
+                    ]
+                )
+
+                if state.projects_total > 0:
+                    progress_pct = (state.projects_migrated / state.projects_total) * 100
+                    status_lines.extend(
+                        [
+                            f"- **Progress**: {state.projects_migrated}/{state.projects_total} projects ({progress_pct:.0f}%)",
+                            f"- **Remaining**: {state.projects_total - state.projects_migrated} projects",
+                        ]
+                    )
+
+                status_lines.extend(
+                    [
+                        "",
+                        "**What's happening**: Basic Memory is migrating legacy project data",
+                        "to the new unified database format. This process runs in the background",
+                        "and most tools will show status messages until completion.",
+                        "",
+                        "**Estimated time**: Usually 1-3 minutes depending on knowledge base size",
+                        "",
+                        "**What you can do**: Wait for migration to complete, or check status",
+                        "again in a few moments. The system will be fully operational once finished.",
+                    ]
+                )
+
+            elif state.status.value == "failed":
+                status_lines.extend(
+                    [
+                        "",
+                        "❌ **Migration Failed**",
+                        "",
+                        f"**Error**: {state.error or 'Unknown error occurred'}",
+                        "",
+                        "**What this means**: The automatic migration encountered an issue.",
+                        "Basic Memory may still work, but some legacy data might not be available.",
+                        "",
+                        "**Recommended actions**:",
+                        "1. Try running `basic-memory sync` manually from the command line",
+                        "2. Check the logs for more detailed error information",
+                        "3. If issues persist, consider filing a support issue",
+                    ]
+                )
+
+            elif state.status.value == "pending":
+                status_lines.extend(
+                    [
+                        "",
+                        "⏳ **Migration Pending**",
+                        "",
+                        "Migration has been detected as needed but hasn't started yet.",
+                        "This usually resolves automatically within a few seconds.",
+                    ]
+                )
+
+        # Add project context if provided
+        if project:
+            try:
+                active_project = get_active_project(project)
+                status_lines.extend(
+                    [
+                        "",
+                        "---",
+                        "",
+                        f"**Active Project**: {active_project.name}",
+                        f"**Project Path**: {active_project.path}",
+                    ]
+                )
+            except Exception as e:
+                logger.debug(f"Could not get project info: {e}")
+                # Don't fail the tool for project info issues
+
+        return "\n".join(status_lines)
+
+    except Exception as e:
+        logger.error(f"Error checking migration status: {e}")
+        return f"""# Migration Status - Error
+
+❌ **Unable to check migration status**
+
+**Error**: {str(e)}
+
+**What this means**: There was a technical issue checking the migration status.
+The system is likely functioning normally, but status information is unavailable.
+
+**Recommended action**: Try again in a moment, or proceed with normal operations.
+"""

src/basic_memory/mcp/tools/read_note.py

@@ -52,6 +52,13 @@ async def read_note(
         read_note("Meeting Notes", project="work-project")
     """
 
+    # Check migration status and wait briefly if needed
+    from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
+
+    migration_status = await wait_for_migration_or_return_status(timeout=5.0)
+    if migration_status:
+        return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before reading notes."
+
     active_project = get_active_project(project)
     project_url = active_project.project_url
 

src/basic_memory/mcp/tools/utils.py

@@ -506,3 +506,47 @@ async def call_delete(
 
     except HTTPStatusError as e:
         raise ToolError(error_message) from e
+
+
+def check_migration_status() -> Optional[str]:
+    """Check if migration is in progress and return status message if so.
+
+    Returns:
+        Status message if migration is in progress, None if system is ready
+    """
+    try:
+        from basic_memory.services.migration_service import migration_manager
+
+        if not migration_manager.is_ready:
+            return migration_manager.status_message
+        return None
+    except Exception:
+        # If there's any error checking migration status, assume ready
+        return None
+
+
+async def wait_for_migration_or_return_status(timeout: float = 5.0) -> Optional[str]:
+    """Wait briefly for migration to complete, or return status message.
+
+    Args:
+        timeout: Maximum time to wait for migration completion
+
+    Returns:
+        Status message if migration is still in progress, None if ready
+    """
+    try:
+        from basic_memory.services.migration_service import migration_manager
+
+        if migration_manager.is_ready:
+            return None
+
+        # Wait briefly for migration to complete
+        completed = await migration_manager.wait_for_completion(timeout=timeout)
+
+        if completed:
+            return None
+        else:
+            return migration_manager.status_message
+    except Exception:
+        # If there's any error, assume ready
+        return None

src/basic_memory/mcp/tools/write_note.py

@@ -70,6 +70,13 @@ async def write_note(
     """
     logger.info(f"MCP tool call tool=write_note folder={folder}, title={title}, tags={tags}")
 
+    # Check migration status and wait briefly if needed
+    from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
+
+    migration_status = await wait_for_migration_or_return_status(timeout=5.0)
+    if migration_status:
+        return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before creating notes."
+
     # Process tags using the helper function
     tag_list = parse_tags(tags)
     # Create the entity request

src/basic_memory/services/initialization.py

@@ -185,7 +185,7 @@ async def initialize_app(
     - Running database migrations
     - Reconciling projects from config.json with projects table
     - Setting up file synchronization
-    - Migrating legacy project data
+    - Starting background migration for legacy project data
 
     Args:
         app_config: The Basic Memory project configuration
@@ -197,8 +197,13 @@ async def initialize_app(
     # Reconcile projects from config.json with projects table
     await reconcile_projects_with_config(app_config)
 
-    # migrate legacy project data
-    await migrate_legacy_projects(app_config)
+    # Start background migration for legacy project data (non-blocking)
+    from basic_memory.services.migration_service import migration_manager
+
+    await migration_manager.start_background_migration(app_config)
+
+    logger.info("App initialization completed (migration running in background if needed)")
+    return migration_manager
 
 
 def ensure_initialization(app_config: BasicMemoryConfig) -> None:

src/basic_memory/services/migration_service.py

@@ -0,0 +1,165 @@
+"""Migration service for handling background migrations and status tracking."""
+
+import asyncio
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+from typing import Optional
+
+from loguru import logger
+
+from basic_memory.config import BasicMemoryConfig
+from basic_memory.models import Project
+
+
+class MigrationStatus(Enum):
+    """Status of migration operations."""
+
+    NOT_NEEDED = "not_needed"
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+@dataclass
+class MigrationState:
+    """Current state of migration operations."""
+
+    status: MigrationStatus
+    message: str
+    progress: Optional[str] = None
+    error: Optional[str] = None
+    projects_migrated: int = 0
+    projects_total: int = 0
+
+
+class MigrationManager:
+    """Manages background migration operations and status tracking."""
+
+    def __init__(self):
+        self._state = MigrationState(
+            status=MigrationStatus.NOT_NEEDED, message="No migration required"
+        )
+        self._migration_task: Optional[asyncio.Task] = None
+
+    @property
+    def state(self) -> MigrationState:
+        """Get current migration state."""
+        return self._state
+
+    @property
+    def is_ready(self) -> bool:
+        """Check if the system is ready for normal operations."""
+        return self._state.status in (MigrationStatus.NOT_NEEDED, MigrationStatus.COMPLETED)
+
+    @property
+    def status_message(self) -> str:
+        """Get a user-friendly status message."""
+        if self._state.status == MigrationStatus.IN_PROGRESS:
+            progress = (
+                f" ({self._state.projects_migrated}/{self._state.projects_total})"
+                if self._state.projects_total > 0
+                else ""
+            )
+            return f"🔄 Migration in progress{progress}: {self._state.message}"
+        elif self._state.status == MigrationStatus.FAILED:
+            return f"❌ Migration failed: {self._state.error or 'Unknown error'}"
+        elif self._state.status == MigrationStatus.COMPLETED:
+            return "✅ Migration completed successfully"
+        else:
+            return "✅ System ready"
+
+    async def check_migration_needed(self, app_config: BasicMemoryConfig) -> bool:
+        """Check if migration is needed without performing it."""
+        from basic_memory import db
+        from basic_memory.repository import ProjectRepository
+
+        try:
+            # Get database session
+            _, session_maker = await db.get_or_create_db(
+                db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM
+            )
+            project_repository = ProjectRepository(session_maker)
+
+            # Check for legacy projects
+            legacy_projects = []
+            for project_name, project_path in app_config.projects.items():
+                legacy_dir = Path(project_path) / ".basic-memory"
+                if legacy_dir.exists():
+                    project = await project_repository.get_by_name(project_name)
+                    if project:
+                        legacy_projects.append(project)
+
+            if legacy_projects:
+                self._state = MigrationState(
+                    status=MigrationStatus.PENDING,
+                    message="Legacy projects detected",
+                    projects_total=len(legacy_projects),
+                )
+                return True
+            else:
+                self._state = MigrationState(
+                    status=MigrationStatus.NOT_NEEDED, message="No migration required"
+                )
+                return False
+
+        except Exception as e:
+            logger.error(f"Error checking migration status: {e}")
+            self._state = MigrationState(
+                status=MigrationStatus.FAILED, message="Migration check failed", error=str(e)
+            )
+            return False
+
+    async def start_background_migration(self, app_config: BasicMemoryConfig) -> None:
+        """Start migration in background if needed."""
+        if not await self.check_migration_needed(app_config):
+            return
+
+        if self._migration_task and not self._migration_task.done():
+            logger.info("Migration already in progress")
+            return
+
+        logger.info("Starting background migration")
+        self._migration_task = asyncio.create_task(self._run_migration(app_config))
+
+    async def _run_migration(self, app_config: BasicMemoryConfig) -> None:
+        """Run the actual migration process."""
+        try:
+            self._state.status = MigrationStatus.IN_PROGRESS
+            self._state.message = "Migrating legacy projects"
+
+            # Import here to avoid circular imports
+            from basic_memory.services.initialization import migrate_legacy_projects
+
+            # Run the migration
+            await migrate_legacy_projects(app_config)
+
+            self._state = MigrationState(
+                status=MigrationStatus.COMPLETED, message="Migration completed successfully"
+            )
+            logger.info("Background migration completed successfully")
+
+        except Exception as e:
+            logger.error(f"Background migration failed: {e}")
+            self._state = MigrationState(
+                status=MigrationStatus.FAILED, message="Migration failed", error=str(e)
+            )
+
+    async def wait_for_completion(self, timeout: Optional[float] = None) -> bool:
+        """Wait for migration to complete."""
+        if self.is_ready:
+            return True
+
+        if not self._migration_task:
+            return False
+
+        try:
+            await asyncio.wait_for(self._migration_task, timeout=timeout)
+            return self.is_ready
+        except asyncio.TimeoutError:
+            return False
+
+
+# Global migration manager instance
+migration_manager = MigrationManager()