feat: add force_full parameter to sync API for bisync operations

Add a force_full parameter to the sync API endpoint and service layer
to allow forcing a full scan that bypasses the watermark optimization.
This is critical for ensuring database consistency after rclone bisync
operations, which can modify files without updating mtimes in a way
that incremental scans can detect.

Changes:
- Add force_full query parameter to /project/sync endpoint
- Pass force_full through sync service layers (sync -> scan)
- Add "full_forced" scan type that bypasses watermark optimization
- Update CLI helpers (run_sync, sync_project) to support force_full
- Force full scan after bisync in upload_command (KEY FIX for #407)
- Include "full_forced" in deletion detection logic

This fixes issue #407 where files synced via bisync don't appear in
the web app because the incremental scan doesn't detect changes made
by external tools like rclone.

Performance impact: Full scans are slower (450s vs 2s for 1.4k files),
but this only happens after explicit bisync operations where users
expect some processing time.

Fixes #407

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

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

Author: phernandez

src/basic_memory/api/routers/project_router.py

@@ -127,6 +127,7 @@ async def sync_project(
     background_tasks: BackgroundTasks,
     sync_service: SyncServiceDep,
     project_config: ProjectConfigDep,
+    force_full: bool = Query(False, description="Force full scan, bypassing watermark optimization"),
 ):
     """Force project filesystem sync to database.
 
@@ -136,12 +137,18 @@ async def sync_project(
         background_tasks: FastAPI background tasks
         sync_service: Sync service for this project
         project_config: Project configuration
+        force_full: If True, force a full scan even if watermark exists
 
     Returns:
         Response confirming sync was initiated
     """
-    background_tasks.add_task(sync_service.sync, project_config.home, project_config.name)
-    logger.info(f"Filesystem sync initiated for project: {project_config.name}")
+    background_tasks.add_task(
+        sync_service.sync, project_config.home, project_config.name, force_full=force_full
+    )
+    logger.info(
+        f"Filesystem sync initiated for project: {project_config.name} "
+        f"(force_full={force_full})"
+    )
 
     return {
         "status": "sync_started",

src/basic_memory/cli/commands/cloud/cloud_utils.py

@@ -69,16 +69,17 @@ async def create_cloud_project(project_name: str) -> CloudProjectCreateResponse:
         raise CloudUtilsError(f"Failed to create cloud project '{project_name}': {e}") from e
 
 
-async def sync_project(project_name: str) -> None:
+async def sync_project(project_name: str, force_full: bool = False) -> None:
     """Trigger sync for a specific project on cloud.
 
     Args:
         project_name: Name of project to sync
+        force_full: If True, force a full scan bypassing watermark optimization
     """
     try:
         from basic_memory.cli.commands.command_utils import run_sync
 
-        await run_sync(project=project_name)
+        await run_sync(project=project_name, force_full=force_full)
     except Exception as e:
         raise CloudUtilsError(f"Failed to sync project '{project_name}': {e}") from e
 

src/basic_memory/cli/commands/cloud/upload_command.py

@@ -112,10 +112,11 @@ async def _upload():
             console.print(f"[green]✅ Successfully uploaded to '{project}'[/green]")
 
         # Sync project if requested (skip on dry run)
+        # Force full scan after bisync to ensure database is up-to-date with synced files
         if sync and not dry_run:
             console.print(f"[blue]Syncing project '{project}'...[/blue]")
             try:
-                await sync_project(project)
+                await sync_project(project, force_full=True)
             except Exception as e:
                 console.print(f"[yellow]Warning: Sync failed: {e}[/yellow]")
                 console.print("[dim]Files uploaded but may not be indexed yet[/dim]")

src/basic_memory/cli/commands/command_utils.py

@@ -16,13 +16,21 @@
 console = Console()
 
 
-async def run_sync(project: Optional[str] = None):
-    """Run sync operation via API endpoint."""
+async def run_sync(project: Optional[str] = None, force_full: bool = False):
+    """Run sync operation via API endpoint.
+
+    Args:
+        project: Optional project name
+        force_full: If True, force a full scan bypassing watermark optimization
+    """
 
     try:
         async with get_client() as client:
             project_item = await get_active_project(client, project, None)
-            response = await call_post(client, f"{project_item.project_url}/project/sync")
+            url = f"{project_item.project_url}/project/sync"
+            if force_full:
+                url += "?force_full=true"
+            response = await call_post(client, url)
             data = response.json()
             console.print(f"[green]✓ {data['message']}[/green]")
     except (ToolError, ValueError) as e:

src/basic_memory/sync/sync_service.py

@@ -256,16 +256,24 @@ def _clear_failure(self, path: str) -> None:
             del self._file_failures[path]
 
     @logfire.instrument()
-    async def sync(self, directory: Path, project_name: Optional[str] = None) -> SyncReport:
-        """Sync all files with database and update scan watermark."""
+    async def sync(
+        self, directory: Path, project_name: Optional[str] = None, force_full: bool = False
+    ) -> SyncReport:
+        """Sync all files with database and update scan watermark.
+
+        Args:
+            directory: Directory to sync
+            project_name: Optional project name
+            force_full: If True, force a full scan bypassing watermark optimization
+        """
 
         start_time = time.time()
         sync_start_timestamp = time.time()  # Capture at start for watermark
-        logger.info(f"Sync operation started for directory: {directory}")
+        logger.info(f"Sync operation started for directory: {directory} (force_full={force_full})")
 
         # initial paths from db to sync
         # path -> checksum
-        report = await self.scan(directory)
+        report = await self.scan(directory, force_full=force_full)
 
         # order of sync matters to resolve relations effectively
         logger.info(
@@ -383,7 +391,7 @@ async def sync(self, directory: Path, project_name: Optional[str] = None) -> Syn
         return report
 
     @logfire.instrument()
-    async def scan(self, directory):
+    async def scan(self, directory, force_full: bool = False):
         """Smart scan using watermark and file count for large project optimization.
 
         Uses scan watermark tracking to dramatically reduce scan time for large projects:
@@ -401,6 +409,10 @@ async def scan(self, directory):
         - Compare with last_file_count to detect deletions
         - If no deletions: incremental scan with find -newermt (0.2s)
         - Process changed files with mtime-based comparison
+
+        Args:
+            directory: Directory to scan
+            force_full: If True, bypass watermark optimization and force full scan
         """
         scan_start_time = time.time()
 
@@ -420,7 +432,13 @@ async def scan(self, directory):
         logger.debug(f"Found {current_count} files in directory")
 
         # Step 2: Determine scan strategy based on watermark and file count
-        if project.last_file_count is None:
+        if force_full:
+            # User explicitly requested full scan → bypass watermark optimization
+            scan_type = "full_forced"
+            logger.info("Force full scan requested, bypassing watermark optimization")
+            file_paths_to_scan = await self._scan_directory_full(directory)
+
+        elif project.last_file_count is None:
             # First sync ever → full scan
             scan_type = "full_initial"
             logger.info("First sync for this project, performing full scan")
@@ -550,7 +568,7 @@ async def scan(self, directory):
 
         # Step 5: Detect deletions (only for full scans)
         # Incremental scans can't reliably detect deletions since they only see modified files
-        if scan_type in ("full_initial", "full_deletions", "full_fallback"):
+        if scan_type in ("full_initial", "full_deletions", "full_fallback", "full_forced"):
             # Use optimized query for just file paths (not full entities)
             db_file_paths = await self.entity_repository.get_all_file_paths()
             logger.debug(f"Found {len(db_file_paths)} db paths for deletion detection")