add background migration task and status tool/prompt

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

Author: phernandez

src/basic_memory/mcp/prompts/__init__.py

@@ -10,10 +10,12 @@
 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 @@
 @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.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 @@
     "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