add sync status tool

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

src/basic_memory/services/initialization.py

@@ -83,7 +83,9 @@ async def migrate_legacy_projects(app_config: BasicMemoryConfig):
             logger.error(f"Project {project_name} not found in database, skipping migration")
             continue
 
+        logger.info(f"Starting migration for project: {project_name} (id: {project.id})")
         await migrate_legacy_project_data(project, legacy_dir)
+        logger.info(f"Completed migration for project: {project_name}")
     logger.info("Legacy projects successfully migrated")
 
 
@@ -104,7 +106,7 @@ async def migrate_legacy_project_data(project: Project, legacy_dir: Path) -> boo
     sync_dir = Path(project.path)
 
     logger.info(f"Sync starting project: {project.name}")
-    await sync_service.sync(sync_dir)
+    await sync_service.sync(sync_dir, project_name=project.name)
     logger.info(f"Sync completed successfully for project: {project.name}")
 
     # After successful sync, remove the legacy directory
@@ -158,12 +160,32 @@ async def initialize_file_sync(
         sync_dir = Path(project.path)
 
         try:
-            await sync_service.sync(sync_dir)
+            await sync_service.sync(sync_dir, project_name=project.name)
             logger.info(f"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 syncing 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))
             # Continue with other projects even if one fails
 
+    # Mark migration complete if it was in progress
+    try:
+        from basic_memory.services.migration_service import migration_manager
+
+        if not migration_manager.is_ready:
+            migration_manager.mark_completed("Migration completed with file sync")
+            logger.info("Marked migration as completed after file sync")
+    except Exception as e:
+        logger.warning(f"Could not update migration status: {e}")
+
     # Then start the watch service in the background
     logger.info("Starting watch service for all projects")
     # run the watch service

src/basic_memory/services/migration_service.py

@@ -9,7 +9,6 @@ from typing import Optional
 from loguru import logger
 
 from basic_memory.config import BasicMemoryConfig
-from basic_memory.models import Project
 
 
 class MigrationStatus(Enum):
@@ -62,11 +61,11 @@ class MigrationManager:
                 if self._state.projects_total > 0
                 else ""
             )
-            return f"🔄 Migration in progress{progress}: {self._state.message}"
+            return f"🔄 File sync in progress{progress}: {self._state.message}. Use sync_status() tool for details."
         elif self._state.status == MigrationStatus.FAILED:
-            return f"❌ Migration failed: {self._state.error or 'Unknown error'}"
+            return f"❌ File sync failed: {self._state.error or 'Unknown error'}. Use sync_status() tool for details."
         elif self._state.status == MigrationStatus.COMPLETED:
-            return "✅ Migration completed successfully"
+            return "✅ File sync completed successfully"
         else:
             return "✅ System ready"
 
@@ -160,6 +159,10 @@ class MigrationManager:
         except asyncio.TimeoutError:
             return False
 
+    def mark_completed(self, message: str = "Migration completed") -> None:
+        """Mark migration as completed externally."""
+        self._state = MigrationState(status=MigrationStatus.COMPLETED, message=message)
+
 
 # Global migration manager instance
 migration_manager = MigrationManager()

src/basic_memory/services/sync_status_service.py

@@ -0,0 +1,181 @@
+"""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(
+        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:
+            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:
+            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:
+        """Check if system is ready (no sync in progress)."""
+        return self._global_status in (SyncStatus.IDLE, SyncStatus.COMPLETED)
+
+    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:
+        """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()