Remove sync_status_service and related functionality

- Removed sync_status_service.py tracking service
- Removed sync_status MCP tool
- Removed all sync_status_tracker calls from:
  - sync_service.py (8 calls)
  - initialization.py (4 calls)
  - mcp/tools/utils.py (2 utility functions)
- Removed migration status checks from MCP tools:
  - write_note.py
  - read_note.py
  - build_context.py
- Removed associated test files and test cases
- Total removal: 1,134 lines

Closes #363

Co-authored-by: Paul Hernandez <phernandez@users.noreply.github.com>

src/basic_memory/mcp/tools/build_context.py

@@ -106,28 +106,6 @@ async def build_context(
         # Get the active project using the new stateless approach
         active_project = await get_active_project(client, project, context)
 
-        # 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, project_name=active_project.name
-        )
-        if migration_status:  # pragma: no cover
-            # 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().astimezone(),
-                    primary_count=0,
-                    related_count=0,
-                    uri=migration_status,  # Include status in metadata
-                ),
-            )
         project_url = active_project.project_url
 
         response = await call_get(

src/basic_memory/mcp/tools/read_note.py

@@ -97,14 +97,6 @@ async def read_note(
             )
             return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries"
 
-        # 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, project_name=active_project.name
-        )
-        if migration_status:  # pragma: no cover
-            return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before reading notes."
         project_url = active_project.project_url
 
         # Get the file via REST API - first try direct permalink lookup

src/basic_memory/mcp/tools/sync_status.py

@@ -1,261 +0,0 @@
-"""Sync status tool for Basic Memory MCP server."""
-
-from typing import Optional
-
-from loguru import logger
-from fastmcp import Context
-
-from basic_memory.config import ConfigManager
-from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.server import mcp
-from basic_memory.mcp.project_context import get_active_project
-from basic_memory.services.sync_status_service import sync_status_tracker
-
-
-def _get_all_projects_status() -> list[str]:
-    """Get status lines for all configured projects."""
-    status_lines = []
-
-    try:
-        app_config = ConfigManager().config
-
-        if app_config.projects:
-            status_lines.extend(["", "---", "", "**All Projects Status:**"])
-
-            for project_name, project_path in app_config.projects.items():
-                # Check if this project has sync status
-                project_sync_status = sync_status_tracker.get_project_status(project_name)
-
-                if project_sync_status:
-                    # Project has tracked sync activity
-                    if project_sync_status.status.value == "watching":
-                        # Project is actively watching for changes (steady state)
-                        status_icon = "👁️"
-                        status_text = "Watching for changes"
-                    elif project_sync_status.status.value == "completed":
-                        # Sync completed but not yet watching - transitional state
-                        status_icon = "✅"
-                        status_text = "Sync completed"
-                    elif project_sync_status.status.value in ["scanning", "syncing"]:
-                        status_icon = "🔄"
-                        status_text = "Sync in progress"
-                        if project_sync_status.files_total > 0:
-                            progress_pct = (
-                                project_sync_status.files_processed
-                                / project_sync_status.files_total
-                            ) * 100
-                            status_text += f" ({project_sync_status.files_processed}/{project_sync_status.files_total}, {progress_pct:.0f}%)"
-                    elif project_sync_status.status.value == "failed":
-                        status_icon = "❌"
-                        status_text = f"Sync error: {project_sync_status.error or 'Unknown error'}"
-                    else:
-                        status_icon = "⏸️"
-                        status_text = project_sync_status.status.value.title()
-                else:
-                    # Project has no tracked sync activity - will be synced automatically
-                    status_icon = "⏳"
-                    status_text = "Pending sync"
-
-                status_lines.append(f"- {status_icon} **{project_name}**: {status_text}")
-
-    except Exception as e:
-        logger.debug(f"Could not get project config for comprehensive status: {e}")
-
-    return status_lines
-
-
-@mcp.tool(
-    description="""Check the status of file synchronization and background operations.
-    
-    Use this tool to:
-    - Check if file sync is in progress or completed
-    - Get detailed sync progress information  
-    - Understand if your files are fully indexed
-    - Get specific error details if sync operations failed
-    - Monitor initial project setup and legacy migration
-    
-    This covers all sync operations including:
-    - Initial project setup and file indexing
-    - Legacy project migration to unified database
-    - Ongoing file monitoring and updates
-    - Background processing of knowledge graphs
-    """,
-)
-async def sync_status(project: Optional[str] = None, context: Context | None = None) -> str:
-    """Get current sync status and system readiness information.
-
-    This tool provides detailed information about any ongoing or completed
-    sync operations, helping users understand when their files are ready.
-
-    Args:
-        project: Optional project name to get project-specific context
-
-    Returns:
-        Formatted sync status with progress, readiness, and guidance
-    """
-    logger.info("MCP tool call tool=sync_status")
-
-    async with get_client() as client:
-        status_lines = []
-
-        try:
-            from basic_memory.services.sync_status_service import sync_status_tracker
-
-            # Get overall summary
-            summary = sync_status_tracker.get_summary()
-            is_ready = sync_status_tracker.is_ready
-
-            # Header
-            status_lines.extend(
-                [
-                    "# Basic Memory Sync Status",
-                    "",
-                    f"**Current Status**: {summary}",
-                    f"**System Ready**: {'✅ Yes' if is_ready else '🔄 Processing'}",
-                    "",
-                ]
-            )
-
-            if is_ready:
-                status_lines.extend(
-                    [
-                        "✅ **All sync operations completed**",
-                        "",
-                        "- File indexing is complete",
-                        "- Knowledge graphs are up to date",
-                        "- All Basic Memory tools are fully operational",
-                        "",
-                        "Your knowledge base is ready for use!",
-                    ]
-                )
-
-                # Show all projects status even when ready
-                status_lines.extend(_get_all_projects_status())
-            else:
-                # System is still processing - show both active and all projects
-                all_sync_projects = sync_status_tracker.get_all_projects()
-
-                active_projects = [
-                    p
-                    for p in all_sync_projects.values()
-                    if p.status.value in ["scanning", "syncing"]
-                ]
-                failed_projects = [
-                    p for p in all_sync_projects.values() if p.status.value == "failed"
-                ]
-
-                if active_projects:
-                    status_lines.extend(
-                        [
-                            "🔄 **File synchronization in progress**",
-                            "",
-                            "Basic Memory is automatically processing all configured projects and building knowledge graphs.",
-                            "This typically takes 1-3 minutes depending on the amount of content.",
-                            "",
-                            "**Currently Processing:**",
-                        ]
-                    )
-
-                    for project_status in active_projects:
-                        progress = ""
-                        if project_status.files_total > 0:
-                            progress_pct = (
-                                project_status.files_processed / project_status.files_total
-                            ) * 100
-                            progress = f" ({project_status.files_processed}/{project_status.files_total}, {progress_pct:.0f}%)"
-
-                        status_lines.append(
-                            f"- **{project_status.project_name}**: {project_status.message}{progress}"
-                        )
-
-                    status_lines.extend(
-                        [
-                            "",
-                            "**What's happening:**",
-                            "- Scanning and indexing markdown files",
-                            "- Building entity and relationship graphs",
-                            "- Settings up full-text search indexes",
-                            "- Processing file changes and updates",
-                            "",
-                            "**What you can do:**",
-                            "- Wait for automatic processing to complete - no action needed",
-                            "- Use this tool again to check progress",
-                            "- Simple operations may work already",
-                            "- All projects will be available once sync finishes",
-                        ]
-                    )
-
-                # Handle failed projects (independent of active projects)
-                if failed_projects:
-                    status_lines.extend(["", "❌ **Some projects failed to sync:**", ""])
-
-                    for project_status in failed_projects:
-                        status_lines.append(
-                            f"- **{project_status.project_name}**: {project_status.error or 'Unknown error'}"
-                        )
-
-                    status_lines.extend(
-                        [
-                            "",
-                            "**Next steps:**",
-                            "1. Check the logs for detailed error information",
-                            "2. Ensure file permissions allow read/write access",
-                            "3. Try restarting the MCP server",
-                            "4. If issues persist, consider filing a support issue",
-                        ]
-                    )
-                elif not active_projects:
-                    # No active or failed projects - must be pending
-                    status_lines.extend(
-                        [
-                            "⏳ **Sync operations pending**",
-                            "",
-                            "File synchronization has been queued but hasn't started yet.",
-                            "This usually resolves automatically within a few seconds.",
-                        ]
-                    )
-
-            # Add comprehensive project status for all configured projects
-            all_projects_status = _get_all_projects_status()
-            if all_projects_status:
-                status_lines.extend(all_projects_status)
-
-                # Add explanation about automatic syncing if there are unsynced projects
-                unsynced_count = sum(1 for line in all_projects_status if "⏳" in line)
-                if unsynced_count > 0 and not is_ready:
-                    status_lines.extend(
-                        [
-                            "",
-                            "**Note**: All configured projects will be automatically synced during startup.",
-                        ]
-                    )
-
-            # Add project context if provided
-            if project:
-                try:
-                    active_project = await get_active_project(client, project, context)
-                    status_lines.extend(
-                        [
-                            "",
-                            "---",
-                            "",
-                            f"**Active Project**: {active_project.name}",
-                            f"**Project Path**: {active_project.home}",
-                        ]
-                    )
-                except Exception as e:
-                    logger.debug(f"Could not get project info: {e}")
-
-            return "\n".join(status_lines)
-
-        except Exception as e:
-            return f"""# Sync Status - Error
-
-❌ **Unable to check sync status**: {str(e)}
-
-**Troubleshooting:**
-- The system may still be starting up
-- Try waiting a few seconds and checking again
-- Check logs for detailed error information
-- Consider restarting if the issue persists
-"""

src/basic_memory/mcp/tools/utils.py

@@ -512,71 +512,3 @@ async def call_delete(
         raise ToolError(error_message) from e
 
 
-def check_migration_status() -> Optional[str]:
-    """Check if sync/migration is in progress and return status message if so.
-
-    Returns:
-        Status message if sync is in progress, None if system is ready
-    """
-    try:
-        from basic_memory.services.sync_status_service import sync_status_tracker
-
-        if not sync_status_tracker.is_ready:
-            return sync_status_tracker.get_summary()
-        return None
-    except Exception:
-        # If there's any error checking sync status, assume ready
-        return None
-
-
-async def wait_for_migration_or_return_status(
-    timeout: float = 5.0, project_name: Optional[str] = None
-) -> Optional[str]:
-    """Wait briefly for sync/migration to complete, or return status message.
-
-    Args:
-        timeout: Maximum time to wait for sync completion
-        project_name: Optional project name to check specific project status.
-                     If provided, only checks that project's readiness.
-                     If None, uses global status check (legacy behavior).
-
-    Returns:
-        Status message if sync is still in progress, None if ready
-    """
-    try:
-        from basic_memory.services.sync_status_service import sync_status_tracker
-        import asyncio
-
-        # Check if we should use project-specific or global status
-        def is_ready() -> bool:
-            if project_name:
-                return sync_status_tracker.is_project_ready(project_name)
-            return sync_status_tracker.is_ready
-
-        if is_ready():
-            return None
-
-        # Wait briefly for sync to complete
-        start_time = asyncio.get_event_loop().time()
-        while (asyncio.get_event_loop().time() - start_time) < timeout:
-            if is_ready():
-                return None
-            await asyncio.sleep(0.1)  # Check every 100ms
-
-        # Still not ready after timeout
-        if project_name:
-            # For project-specific checks, get project status details
-            project_status = sync_status_tracker.get_project_status(project_name)
-            if project_status and project_status.status.value == "failed":
-                error_msg = project_status.error or "Unknown sync error"
-                return f"❌ Sync failed for project '{project_name}': {error_msg}"
-            elif project_status:
-                return f"🔄 Project '{project_name}' is still syncing: {project_status.message}"
-            else:
-                return f"⚠️ Project '{project_name}' status unknown"
-        else:
-            # Fall back to global summary for legacy calls
-            return sync_status_tracker.get_summary()
-    except Exception:  # pragma: no cover
-        # If there's any error, assume ready
-        return None

src/basic_memory/mcp/tools/write_note.py

@@ -140,15 +140,6 @@ async def write_note(
             )
             return f"# Error\n\nFolder path '{folder}' is not allowed - paths must stay within project boundaries"
 
-        # 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, project_name=active_project.name
-        )
-        if migration_status:  # pragma: no cover
-            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

@@ -118,18 +118,8 @@ async def initialize_file_sync(
             sync_dir = Path(project.path)
             await sync_service.sync(sync_dir, project_name=project.name)
             logger.info(f"Background sync completed successfully for project: {project.name}")
-
-            # Mark project as watching for changes after successful sync
-            from basic_memory.services.sync_status_service import sync_status_tracker
-
-            sync_status_tracker.start_project_watch(project.name)
-            logger.info(f"Project {project.name} is now watching for changes")
         except Exception as e:  # pragma: no cover
             logger.error(f"Error in background sync for project {project.name}: {e}")
-            # Mark sync as failed for this project
-            from basic_memory.services.sync_status_service import sync_status_tracker
-
-            sync_status_tracker.fail_project_sync(project.name, str(e))
 
     # Create background tasks for all project syncs (non-blocking)
     sync_tasks = [

src/basic_memory/services/sync_status_service.py

@@ -1,198 +0,0 @@
-"""Simple sync status tracking service."""
-
-from dataclasses import dataclass
-from enum import Enum
-from typing import Dict, Optional
-
-
-class SyncStatus(Enum):
-    """Status of sync operations."""
-
-    IDLE = "idle"
-    SCANNING = "scanning"
-    SYNCING = "syncing"
-    COMPLETED = "completed"
-    FAILED = "failed"
-    WATCHING = "watching"
-
-
-@dataclass
-class ProjectSyncStatus:
-    """Sync status for a single project."""
-
-    project_name: str
-    status: SyncStatus
-    message: str = ""
-    files_total: int = 0
-    files_processed: int = 0
-    error: Optional[str] = None
-
-
-class SyncStatusTracker:
-    """Global tracker for all sync operations."""
-
-    def __init__(self):
-        self._project_statuses: Dict[str, ProjectSyncStatus] = {}
-        self._global_status: SyncStatus = SyncStatus.IDLE
-
-    def start_project_sync(self, project_name: str, files_total: int = 0) -> None:
-        """Start tracking sync for a project."""
-        self._project_statuses[project_name] = ProjectSyncStatus(
-            project_name=project_name,
-            status=SyncStatus.SCANNING,
-            message="Scanning files",
-            files_total=files_total,
-            files_processed=0,
-        )
-        self._update_global_status()
-
-    def update_project_progress(  # pragma: no cover
-        self,
-        project_name: str,
-        status: SyncStatus,
-        message: str = "",
-        files_processed: int = 0,
-        files_total: Optional[int] = None,
-    ) -> None:
-        """Update progress for a project."""
-        if project_name not in self._project_statuses:  # pragma: no cover
-            return
-
-        project_status = self._project_statuses[project_name]
-        project_status.status = status
-        project_status.message = message
-        project_status.files_processed = files_processed
-
-        if files_total is not None:
-            project_status.files_total = files_total
-
-        self._update_global_status()
-
-    def complete_project_sync(self, project_name: str) -> None:
-        """Mark project sync as completed."""
-        if project_name in self._project_statuses:
-            self._project_statuses[project_name].status = SyncStatus.COMPLETED
-            self._project_statuses[project_name].message = "Sync completed"
-            self._update_global_status()
-
-    def fail_project_sync(self, project_name: str, error: str) -> None:
-        """Mark project sync as failed."""
-        if project_name in self._project_statuses:
-            self._project_statuses[project_name].status = SyncStatus.FAILED
-            self._project_statuses[project_name].error = error
-            self._update_global_status()
-
-    def start_project_watch(self, project_name: str) -> None:
-        """Mark project as watching for changes (steady state after sync)."""
-        if project_name in self._project_statuses:
-            self._project_statuses[project_name].status = SyncStatus.WATCHING
-            self._project_statuses[project_name].message = "Watching for changes"
-            self._update_global_status()
-        else:
-            # Create new status if project isn't tracked yet
-            self._project_statuses[project_name] = ProjectSyncStatus(
-                project_name=project_name,
-                status=SyncStatus.WATCHING,
-                message="Watching for changes",
-                files_total=0,
-                files_processed=0,
-            )
-            self._update_global_status()
-
-    def _update_global_status(self) -> None:
-        """Update global status based on project statuses."""
-        if not self._project_statuses:  # pragma: no cover
-            self._global_status = SyncStatus.IDLE
-            return
-
-        statuses = [p.status for p in self._project_statuses.values()]
-
-        if any(s == SyncStatus.FAILED for s in statuses):
-            self._global_status = SyncStatus.FAILED
-        elif any(s in (SyncStatus.SCANNING, SyncStatus.SYNCING) for s in statuses):
-            self._global_status = SyncStatus.SYNCING
-        elif all(s in (SyncStatus.COMPLETED, SyncStatus.WATCHING) for s in statuses):
-            self._global_status = SyncStatus.COMPLETED
-        else:
-            self._global_status = SyncStatus.SYNCING
-
-    @property
-    def global_status(self) -> SyncStatus:
-        """Get overall sync status."""
-        return self._global_status
-
-    @property
-    def is_syncing(self) -> bool:
-        """Check if any sync operation is in progress."""
-        return self._global_status in (SyncStatus.SCANNING, SyncStatus.SYNCING)
-
-    @property
-    def is_ready(self) -> bool:  # pragma: no cover
-        """Check if system is ready (no sync in progress)."""
-        return self._global_status in (SyncStatus.IDLE, SyncStatus.COMPLETED)
-
-    def is_project_ready(self, project_name: str) -> bool:
-        """Check if a specific project is ready for operations.
-
-        Args:
-            project_name: Name of the project to check
-
-        Returns:
-            True if the project is ready (completed, watching, or not tracked),
-            False if the project is syncing, scanning, or failed
-        """
-        project_status = self._project_statuses.get(project_name)
-        if not project_status:
-            # Project not tracked = ready (likely hasn't been synced yet)
-            return True
-
-        return project_status.status in (SyncStatus.COMPLETED, SyncStatus.WATCHING, SyncStatus.IDLE)
-
-    def get_project_status(self, project_name: str) -> Optional[ProjectSyncStatus]:
-        """Get status for a specific project."""
-        return self._project_statuses.get(project_name)
-
-    def get_all_projects(self) -> Dict[str, ProjectSyncStatus]:
-        """Get all project statuses."""
-        return self._project_statuses.copy()
-
-    def get_summary(self) -> str:  # pragma: no cover
-        """Get a user-friendly summary of sync status."""
-        if self._global_status == SyncStatus.IDLE:
-            return "✅ System ready"
-        elif self._global_status == SyncStatus.COMPLETED:
-            return "✅ All projects synced successfully"
-        elif self._global_status == SyncStatus.FAILED:
-            failed_projects = [
-                p.project_name
-                for p in self._project_statuses.values()
-                if p.status == SyncStatus.FAILED
-            ]
-            return f"❌ Sync failed for: {', '.join(failed_projects)}"
-        else:
-            active_projects = [
-                p.project_name
-                for p in self._project_statuses.values()
-                if p.status in (SyncStatus.SCANNING, SyncStatus.SYNCING)
-            ]
-            total_files = sum(p.files_total for p in self._project_statuses.values())
-            processed_files = sum(p.files_processed for p in self._project_statuses.values())
-
-            if total_files > 0:
-                progress_pct = (processed_files / total_files) * 100
-                return f"🔄 Syncing {len(active_projects)} projects ({processed_files}/{total_files} files, {progress_pct:.0f}%)"
-            else:
-                return f"🔄 Syncing {len(active_projects)} projects"
-
-    def clear_completed(self) -> None:
-        """Remove completed project statuses to clean up memory."""
-        self._project_statuses = {
-            name: status
-            for name, status in self._project_statuses.items()
-            if status.status != SyncStatus.COMPLETED
-        }
-        self._update_global_status()
-
-
-# Global sync status tracker instance
-sync_status_tracker = SyncStatusTracker()

src/basic_memory/sync/sync_service.py

@@ -24,7 +24,6 @@ from basic_memory.repository.search_repository import SearchRepository
 from basic_memory.services import EntityService, FileService
 from basic_memory.services.link_resolver import LinkResolver
 from basic_memory.services.search_service import SearchService
-from basic_memory.services.sync_status_service import sync_status_tracker, SyncStatus
 
 
 @dataclass
@@ -131,23 +130,11 @@ class SyncService:
         start_time = time.time()
         logger.info(f"Sync operation started for directory: {directory}")
 
-        # Start tracking sync for this project if project name provided
-        if project_name:
-            sync_status_tracker.start_project_sync(project_name)
 
         # initial paths from db to sync
         # path -> checksum
         report = await self.scan(directory)
 
-        # Update progress with file counts
-        if project_name:
-            sync_status_tracker.update_project_progress(
-                project_name=project_name,
-                status=SyncStatus.SYNCING,
-                message="Processing file changes",
-                files_total=report.total,
-                files_processed=0,
-            )
 
         # order of sync matters to resolve relations effectively
         logger.info(
@@ -170,55 +157,23 @@ class SyncService:
                 await self.handle_move(old_path, new_path)
 
             files_processed += 1
-            if project_name:
-                sync_status_tracker.update_project_progress(  # pragma: no cover
-                    project_name=project_name,
-                    status=SyncStatus.SYNCING,
-                    message="Processing moves",
-                    files_processed=files_processed,
-                )
 
         # deleted next
         for path in report.deleted:
             await self.handle_delete(path)
             files_processed += 1
-            if project_name:
-                sync_status_tracker.update_project_progress(  # pragma: no cover
-                    project_name=project_name,
-                    status=SyncStatus.SYNCING,
-                    message="Processing deletions",
-                    files_processed=files_processed,
-                )
 
         # then new and modified
         for path in report.new:
             await self.sync_file(path, new=True)
             files_processed += 1
-            if project_name:
-                sync_status_tracker.update_project_progress(
-                    project_name=project_name,
-                    status=SyncStatus.SYNCING,
-                    message="Processing new files",
-                    files_processed=files_processed,
-                )
 
         for path in report.modified:
             await self.sync_file(path, new=False)
             files_processed += 1
-            if project_name:
-                sync_status_tracker.update_project_progress(  # pragma: no cover
-                    project_name=project_name,
-                    status=SyncStatus.SYNCING,
-                    message="Processing modified files",
-                    files_processed=files_processed,
-                )
 
         await self.resolve_relations()
 
-        # Mark sync as completed
-        if project_name:
-            sync_status_tracker.complete_project_sync(project_name)
-
         duration_ms = int((time.time() - start_time) * 1000)
         logger.info(
             f"Sync operation completed: directory={directory}, total_changes={report.total}, duration_ms={duration_ms}"

tests/mcp/test_tool_sync_status.py

@@ -1,170 +0,0 @@
-"""Tests for sync_status MCP tool."""
-
-import pytest
-from unittest.mock import MagicMock, patch
-
-from basic_memory.mcp.tools.sync_status import sync_status
-from basic_memory.services.sync_status_service import (
-    SyncStatus,
-    ProjectSyncStatus,
-    SyncStatusTracker,
-)
-
-
-@pytest.mark.asyncio
-async def test_sync_status_completed():
-    """Test sync_status when all operations are completed."""
-    # Mock sync status tracker with ready status
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = True
-    mock_tracker.get_summary.return_value = "✅ All projects synced successfully"
-    mock_tracker.get_all_projects.return_value = {}
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn()
-
-    assert "Basic Memory Sync Status" in result
-    assert "System Ready**: ✅ Yes" in result
-    assert "All sync operations completed" in result
-    assert "File indexing is complete" in result
-    assert "knowledge base is ready for use" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_in_progress():
-    """Test sync_status when sync is in progress."""
-    # Mock sync status tracker with in progress status
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = False
-    mock_tracker.get_summary.return_value = "🔄 Syncing 2 projects (5/10 files, 50%)"
-
-    # Mock active projects
-    project1 = ProjectSyncStatus(
-        project_name="project1",
-        status=SyncStatus.SYNCING,
-        message="Processing new files",
-        files_total=5,
-        files_processed=3,
-    )
-    project2 = ProjectSyncStatus(
-        project_name="project2",
-        status=SyncStatus.SCANNING,
-        message="Scanning files",
-        files_total=5,
-        files_processed=2,
-    )
-
-    mock_tracker.get_all_projects.return_value = {"project1": project1, "project2": project2}
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn()
-
-    assert "Basic Memory Sync Status" in result
-    assert "System Ready**: 🔄 Processing" in result
-    assert "File synchronization in progress" in result
-    assert "project1**: Processing new files (3/5, 60%)" in result
-    assert "project2**: Scanning files (2/5, 40%)" in result
-    assert "Scanning and indexing markdown files" in result
-    assert "Use this tool again to check progress" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_failed():
-    """Test sync_status when sync has failed."""
-    # Mock sync status tracker with failed project
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = False
-    mock_tracker.get_summary.return_value = "❌ Sync failed for: project1"
-
-    # Mock failed project
-    failed_project = ProjectSyncStatus(
-        project_name="project1",
-        status=SyncStatus.FAILED,
-        message="Sync failed",
-        error="Permission denied",
-    )
-
-    mock_tracker.get_all_projects.return_value = {"project1": failed_project}
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn()
-
-    assert "Basic Memory Sync Status" in result
-    assert "System Ready**: 🔄 Processing" in result
-    assert "Some projects failed to sync" in result
-    assert "project1**: Permission denied" in result
-    assert "Check the logs for detailed error information" in result
-    assert "Try restarting the MCP server" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_idle():
-    """Test sync_status when system is idle."""
-    # Mock sync status tracker with idle status
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = True
-    mock_tracker.get_summary.return_value = "✅ System ready"
-    mock_tracker.get_all_projects.return_value = {}
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn()
-
-    assert "Basic Memory Sync Status" in result
-    assert "System Ready**: ✅ Yes" in result
-    assert "All sync operations completed" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_with_project():
-    """Test sync_status with specific project context."""
-    # Mock sync status tracker
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = True
-    mock_tracker.get_summary.return_value = "✅ All projects synced successfully"
-
-    # Mock specific project status
-    project_status = ProjectSyncStatus(
-        project_name="test-project",
-        status=SyncStatus.COMPLETED,
-        message="Sync completed",
-        files_total=10,
-        files_processed=10,
-    )
-    mock_tracker.get_project_status.return_value = project_status
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn(project="test-project")
-
-    # The function should use the original logic for project-specific queries
-    # But since we changed the implementation, let's just verify it doesn't crash
-    assert "Basic Memory Sync Status" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_pending():
-    """Test sync_status when no projects are active."""
-    # Mock sync status tracker with no active projects
-    mock_tracker = MagicMock(spec=SyncStatusTracker)
-    mock_tracker.is_ready = False
-    mock_tracker.get_summary.return_value = "✅ System ready"
-    mock_tracker.get_all_projects.return_value = {}
-
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-        result = await sync_status.fn()
-
-    assert "Basic Memory Sync Status" in result
-    assert "Sync operations pending" in result
-    assert "usually resolves automatically" in result
-
-
-@pytest.mark.asyncio
-async def test_sync_status_error_handling():
-    """Test sync_status handles errors gracefully."""
-    # Mock sync status tracker that raises an exception
-    with patch("basic_memory.services.sync_status_service.sync_status_tracker") as mock_tracker:
-        mock_tracker.is_ready = True
-        mock_tracker.get_summary.side_effect = Exception("Test error")
-
-        result = await sync_status.fn()
-
-    assert "Unable to check sync status**: Test error" in result

tests/mcp/test_tool_utils.py

@@ -12,8 +12,6 @@ from basic_memory.mcp.tools.utils import (
     call_put,
     call_delete,
     get_error_message,
-    check_migration_status,
-    wait_for_migration_or_return_status,
 )
 
 
@@ -184,82 +182,3 @@ async def test_call_post_with_json(mock_response):
     mock_post.assert_called_once()
     call_kwargs = mock_post.call_args[1]
     assert call_kwargs["json"] == json_data
-
-
-class TestMigrationStatus:
-    """Test migration status checking functions."""
-
-    def test_check_migration_status_ready(self):
-        """Test check_migration_status when system is ready."""
-        mock_tracker = MagicMock()
-        mock_tracker.is_ready = True
-
-        with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-            result = check_migration_status()
-            assert result is None
-
-    def test_check_migration_status_not_ready(self):
-        """Test check_migration_status when sync is in progress."""
-        mock_tracker = MagicMock()
-        mock_tracker.is_ready = False
-        mock_tracker.get_summary.return_value = "Sync in progress..."
-
-        with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-            result = check_migration_status()
-            assert result == "Sync in progress..."
-            mock_tracker.get_summary.assert_called_once()
-
-    def test_check_migration_status_exception(self):
-        """Test check_migration_status with import/other exception."""
-        # Mock the import itself to raise an exception
-        with patch("builtins.__import__", side_effect=ImportError("Module not found")):
-            result = check_migration_status()
-            assert result is None
-
-    @pytest.mark.asyncio
-    async def test_wait_for_migration_ready(self):
-        """Test wait_for_migration when system is already ready."""
-        mock_tracker = MagicMock()
-        mock_tracker.is_ready = True
-
-        with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-            result = await wait_for_migration_or_return_status()
-            assert result is None
-
-    @pytest.mark.asyncio
-    async def test_wait_for_migration_becomes_ready(self):
-        """Test wait_for_migration when system becomes ready during wait."""
-        mock_tracker = MagicMock()
-        mock_tracker.is_ready = False
-
-        with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-            # Mock asyncio.sleep to make tracker ready after first check
-            async def mock_sleep(delay):
-                mock_tracker.is_ready = True
-
-            with patch("asyncio.sleep", side_effect=mock_sleep):
-                result = await wait_for_migration_or_return_status(timeout=1.0)
-                assert result is None
-
-    @pytest.mark.asyncio
-    async def test_wait_for_migration_timeout(self):
-        """Test wait_for_migration when timeout occurs."""
-        mock_tracker = MagicMock()
-        mock_tracker.is_ready = False
-        mock_tracker.get_summary.return_value = "Still syncing..."
-
-        with patch("basic_memory.services.sync_status_service.sync_status_tracker", mock_tracker):
-            with patch("asyncio.sleep", new_callable=AsyncMock):
-                result = await wait_for_migration_or_return_status(timeout=0.1)
-                assert result == "Still syncing..."
-                mock_tracker.get_summary.assert_called_once()
-
-    @pytest.mark.asyncio
-    async def test_wait_for_migration_exception(self):
-        """Test wait_for_migration with exception during checking."""
-        with patch(
-            "basic_memory.services.sync_status_service.sync_status_tracker",
-            side_effect=Exception("Test error"),
-        ):
-            result = await wait_for_migration_or_return_status()
-            assert result is None

tests/services/test_sync_status_service.py

@@ -1,262 +0,0 @@
-"""Test sync status service functionality."""
-
-import pytest
-from basic_memory.services.sync_status_service import SyncStatusTracker, SyncStatus
-
-
-@pytest.fixture
-def sync_tracker():
-    """Create a fresh sync status tracker for testing."""
-    return SyncStatusTracker()
-
-
-def test_sync_tracker_initial_state(sync_tracker):
-    """Test initial state of sync tracker."""
-    assert sync_tracker.is_ready
-    assert not sync_tracker.is_syncing
-    assert sync_tracker.global_status == SyncStatus.IDLE
-    assert sync_tracker.get_summary() == "✅ System ready"
-
-    # Test project-specific ready check for unknown project
-    assert sync_tracker.is_project_ready("unknown-project")
-
-
-def test_start_project_sync(sync_tracker):
-    """Test starting project sync."""
-    sync_tracker.start_project_sync("test-project", files_total=10)
-
-    assert not sync_tracker.is_ready
-    assert sync_tracker.is_syncing
-    assert sync_tracker.global_status == SyncStatus.SYNCING
-
-    project_status = sync_tracker.get_project_status("test-project")
-    assert project_status is not None
-    assert project_status.status == SyncStatus.SCANNING
-    assert project_status.message == "Scanning files"
-    assert project_status.files_total == 10
-
-
-def test_update_project_progress(sync_tracker):
-    """Test updating project progress."""
-    sync_tracker.start_project_sync("test-project")  # Use default files_total=0
-    sync_tracker.update_project_progress(
-        "test-project", SyncStatus.SYNCING, "Processing files", files_processed=5, files_total=10
-    )
-
-    project_status = sync_tracker.get_project_status("test-project")
-    assert project_status.status == SyncStatus.SYNCING
-    assert project_status.message == "Processing files"
-    assert project_status.files_processed == 5
-    assert project_status.files_total == 10
-    assert sync_tracker.global_status == SyncStatus.SYNCING
-
-
-def test_complete_project_sync(sync_tracker):
-    """Test completing project sync."""
-    sync_tracker.start_project_sync("test-project")
-    sync_tracker.complete_project_sync("test-project")
-
-    assert sync_tracker.is_ready
-    assert not sync_tracker.is_syncing
-    assert sync_tracker.global_status == SyncStatus.COMPLETED
-
-    project_status = sync_tracker.get_project_status("test-project")
-    assert project_status.status == SyncStatus.COMPLETED
-    assert project_status.message == "Sync completed"
-
-
-def test_fail_project_sync(sync_tracker):
-    """Test failing project sync."""
-    sync_tracker.start_project_sync("test-project")
-    sync_tracker.fail_project_sync("test-project", "Connection error")
-
-    assert not sync_tracker.is_ready
-    assert not sync_tracker.is_syncing
-    assert sync_tracker.global_status == SyncStatus.FAILED
-
-    project_status = sync_tracker.get_project_status("test-project")
-    assert project_status.status == SyncStatus.FAILED
-    assert project_status.error == "Connection error"
-
-
-def test_start_project_watch(sync_tracker):
-    """Test starting project watch mode."""
-    sync_tracker.start_project_watch("test-project")
-
-    assert sync_tracker.is_ready
-    assert not sync_tracker.is_syncing
-    assert sync_tracker.global_status == SyncStatus.COMPLETED
-
-    project_status = sync_tracker.get_project_status("test-project")
-    assert project_status.status == SyncStatus.WATCHING
-    assert project_status.message == "Watching for changes"
-
-
-def test_multiple_projects_status(sync_tracker):
-    """Test status with multiple projects."""
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.start_project_sync("project2")
-
-    # Both scanning - should be syncing
-    assert sync_tracker.global_status == SyncStatus.SYNCING
-    assert sync_tracker.is_syncing
-
-    # Complete one project
-    sync_tracker.complete_project_sync("project1")
-    assert sync_tracker.global_status == SyncStatus.SYNCING  # Still syncing
-
-    # Complete second project
-    sync_tracker.complete_project_sync("project2")
-    assert sync_tracker.global_status == SyncStatus.COMPLETED
-    assert sync_tracker.is_ready
-
-
-def test_mixed_project_statuses(sync_tracker):
-    """Test mixed project statuses."""
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.start_project_sync("project2")
-
-    # Fail one project
-    sync_tracker.fail_project_sync("project1", "Error")
-    # Complete other project
-    sync_tracker.complete_project_sync("project2")
-
-    # Should show failed status
-    assert sync_tracker.global_status == SyncStatus.FAILED
-    assert not sync_tracker.is_ready
-
-
-def test_get_summary_with_progress(sync_tracker):
-    """Test summary with progress information."""
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.update_project_progress(
-        "project1", SyncStatus.SYNCING, "Processing", files_processed=25, files_total=100
-    )
-
-    summary = sync_tracker.get_summary()
-    assert "🔄 Syncing 1 projects" in summary
-    assert "(25/100 files, 25%)" in summary
-
-
-def test_get_all_projects(sync_tracker):
-    """Test getting all project statuses."""
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.start_project_sync("project2")
-
-    all_projects = sync_tracker.get_all_projects()
-    assert len(all_projects) == 2
-    assert "project1" in all_projects
-    assert "project2" in all_projects
-    assert all_projects["project1"].status == SyncStatus.SCANNING
-    assert all_projects["project2"].status == SyncStatus.SCANNING
-
-
-def test_clear_completed(sync_tracker):
-    """Test clearing completed project statuses."""
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.start_project_sync("project2")
-
-    sync_tracker.complete_project_sync("project1")
-    sync_tracker.fail_project_sync("project2", "Error")
-
-    # Should have 2 projects before clearing
-    assert len(sync_tracker.get_all_projects()) == 2
-
-    sync_tracker.clear_completed()
-
-    # Should only have the failed project after clearing
-    remaining = sync_tracker.get_all_projects()
-    assert len(remaining) == 1
-    assert "project2" in remaining
-    assert remaining["project2"].status == SyncStatus.FAILED
-
-
-def test_summary_messages(sync_tracker):
-    """Test various summary messages."""
-    # Initial state
-    assert sync_tracker.get_summary() == "✅ System ready"
-
-    # All completed
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.complete_project_sync("project1")
-    assert sync_tracker.get_summary() == "✅ All projects synced successfully"
-
-    # Failed projects
-    sync_tracker.fail_project_sync("project1", "Test error")
-    assert "❌ Sync failed for: project1" in sync_tracker.get_summary()
-
-
-def test_global_status_edge_cases(sync_tracker):
-    """Test edge cases for global status calculation."""
-    # Test mixed statuses (some completed, some watching) - should be completed
-    sync_tracker.start_project_sync("project1")
-    sync_tracker.start_project_sync("project2")
-
-    sync_tracker.complete_project_sync("project1")
-    sync_tracker.start_project_watch("project2")
-
-    assert sync_tracker.global_status == SyncStatus.COMPLETED
-
-    # Test fallback case - create a scenario that doesn't match specific conditions
-    sync_tracker.start_project_sync("project3")
-    sync_tracker.update_project_progress("project3", SyncStatus.IDLE, "Idle")
-
-    # This should trigger the "else" clause in _update_global_status
-    assert sync_tracker.global_status == SyncStatus.SYNCING
-
-
-def test_summary_without_file_counts(sync_tracker):
-    """Test summary when projects don't have file counts."""
-    sync_tracker.start_project_sync("project1")  # files_total defaults to 0
-    sync_tracker.start_project_sync("project2")  # files_total defaults to 0
-
-    # Don't set file counts - should use the fallback message
-    summary = sync_tracker.get_summary()
-    assert "🔄 Syncing 2 projects" in summary
-    assert "files" not in summary  # Should not show file progress
-
-
-def test_is_project_ready_functionality(sync_tracker):
-    """Test project-specific ready checks."""
-    # Unknown project should be ready
-    assert sync_tracker.is_project_ready("unknown-project")
-
-    # Project in different states
-    sync_tracker.start_project_sync("scanning-project")
-    assert not sync_tracker.is_project_ready("scanning-project")  # SCANNING = not ready
-
-    sync_tracker.update_project_progress("scanning-project", SyncStatus.SYNCING, "Processing")
-    assert not sync_tracker.is_project_ready("scanning-project")  # SYNCING = not ready
-
-    sync_tracker.fail_project_sync("scanning-project", "Test error")
-    assert not sync_tracker.is_project_ready("scanning-project")  # FAILED = not ready
-
-    sync_tracker.complete_project_sync("scanning-project")
-    assert sync_tracker.is_project_ready("scanning-project")  # COMPLETED = ready
-
-    # Test watching project
-    sync_tracker.start_project_watch("watching-project")
-    assert sync_tracker.is_project_ready("watching-project")  # WATCHING = ready
-
-
-def test_project_isolation_scenario(sync_tracker):
-    """Test the specific bug scenario: project isolation with mixed sync states."""
-    # Set up the bug scenario: one failed project, one healthy project
-    sync_tracker.start_project_sync("main")
-    sync_tracker.fail_project_sync(
-        "main", "UNIQUE constraint failed: entity.file_path, entity.project_id"
-    )
-
-    sync_tracker.start_project_sync("basic-memory-testing-20250626-1009")
-    sync_tracker.complete_project_sync("basic-memory-testing-20250626-1009")
-    sync_tracker.start_project_watch("basic-memory-testing-20250626-1009")
-
-    # Global status should be failed due to "main" project
-    assert sync_tracker.global_status == SyncStatus.FAILED
-    assert not sync_tracker.is_ready
-
-    # But the healthy project should be ready for operations
-    assert sync_tracker.is_project_ready("basic-memory-testing-20250626-1009")
-    assert not sync_tracker.is_project_ready("main")
-
-    # This demonstrates the fix: project-specific checks allow isolation