fix(backup): Google Drive folder duplication + dedup-wipe — restore GOLD's lock scope (whole-method _folder_cache_lock on project/nested folders, lock-free backup-folder short-circuit, guarded tracker reset). Fix drive_* underscore command routing + declare 3 google libs. Verified by artifact (seedgo 100%, 197 tests incl. 5-thread concurrency) + live (real Drive backup, no duplicate folders)

CHANGELOG.md

@@ -96,6 +96,21 @@ PyPI version — not the changelog header.
   embeds (384-dim) → `drone @memory search` returns it at 91% similarity; audit 100%,
   876 tests (+4).
 
+- **Backup Google Drive folder duplication + dedup-wipe fixed (GOLD-faithful lock
+  restoration).** The Phase-4 port had narrowed `GoogleDriveSync`'s folder lock: a
+  single `drive_sync` run's 3 upload workers raced the folder search+create →
+  multiple "AIPass Backups" root folders, and `get_or_create_backup_folder` reset
+  the dedup tracker on every call (re-uploading everything = the slowness). Restored
+  GOLD's structure exactly: `get_or_create_project_folder` / `get_or_create_nested_folder`
+  hold `_folder_cache_lock` across the **entire** method (cache + root-ensure + search
+  + create); `get_or_create_backup_folder` is lock-free (called inside the project
+  lock — no re-entrant deadlock), short-circuits cached ids via `_verify_folder_id`,
+  and clears the tracker only on a genuine brand-new root folder. Also: all four
+  `drive_*` commands route by their underscore names (were hyphenated → "Unknown
+  command"); `requirements.project.txt` now declares the three google libs. Verified
+  by artifact (seedgo 100%, 197 tests incl. a 5-thread concurrency test → exactly one
+  create) + live (real Drive backup: no duplicate folders).
+
 - **Backup rich CLI output restored end-to-end (FPLAN-0263 + drone passthrough).**
   `drone @backup snapshot|versioned|all` rendered a flat text block instead of the
   original rich output. Two independent causes, both closed: (1) the rich rendering

src/aipass/backup/apps/handlers/drive/client.py

@@ -1,7 +1,7 @@
 # =================== AIPass ====================
 # Name: client.py
 # Description: Google Drive client — auth, folders, file lookup via @api gateway
-# Version: 1.0.0
+# Version: 2.0.0
 # Created: 2026-04-16
 # Modified: 2026-06-12
 # =============================================
@@ -12,6 +12,14 @@ Core Drive v3 client routed through the @api gateway.  Handles
 authentication, folder creation/lookup, and file discovery.
 Never uses console-OAuth -- all auth flows through
 ``aipass.api.apps.modules.google_client``.
+
+Lock pattern ported from GOLD (drive_sync_client.py):
+- get_or_create_backup_folder has NO lock (always called inside
+  project_folder's lock).
+- get_or_create_project_folder wraps its ENTIRE body in
+  _folder_cache_lock (cache check + backup-folder-ensure + search +
+  create).
+- get_or_create_nested_folder wraps the entire path walk in the lock.
 """
 
 from __future__ import annotations
@@ -63,11 +71,7 @@ class DriveClient:
     # -- auth ----------------------------------------------------------------
 
     def authenticate(self) -> bool:
-        """Authenticate through the @api gateway.
-
-        Returns:
-            True if a Drive service was obtained, False otherwise.
-        """
+        """Authenticate through the @api gateway."""
         if not GOOGLE_API_AVAILABLE:
             self.last_error = "Google API libraries not installed"
             json_handler.log_operation(
@@ -99,14 +103,10 @@ class DriveClient:
     # -- low-level API -------------------------------------------------------
 
     def _api_call(self, request: Any, max_retries: int = 3) -> Any:
-        """Execute a Google API request with retry.
-
-        On failure, rebuilds the thread-local service and retries once.
-        """
+        """Execute a Google API request with retry."""
         try:
             return api_call_with_retry(request, max_retries=max_retries)  # type: ignore[misc]
         except Exception as first_exc:
-            # Rebuild thread service and retry once
             logger.info(f"API call failed, rebuilding thread service: {first_exc}")
             try:
                 self._thread_local.service = self._build_thread_service()
@@ -139,16 +139,26 @@ class DriveClient:
     def get_or_create_backup_folder(self) -> str | None:
         """Get or create the root 'AIPass Backups' folder.
 
-        Returns:
-            Folder ID or None on failure.
+        NO lock — always called inside get_or_create_project_folder's lock
+        (or single-threaded during pre-resolve).  Matches GOLD's pattern.
         """
+        # Short-circuit: verify cached ID
+        if self.backup_folder_id:
+            if self._verify_folder_id(self.backup_folder_id):
+                return self.backup_folder_id
+            self.backup_folder_id = None
+
         if not self.drive_service:
             return None
 
         # Search for existing
         query = f"name='{BACKUP_FOLDER_NAME}' and mimeType='{FOLDER_MIME}' and trashed=false"
         try:
-            request = self.drive_service.files().list(q=query, spaces="drive", fields="files(id,name)")
+            request = self.drive_service.files().list(
+                q=query,
+                spaces="drive",
+                fields="files(id,name)",
+            )
             result = self._api_call(request)
             if result and result.get("files"):
                 self.backup_folder_id = result["files"][0]["id"]
@@ -167,14 +177,38 @@ class DriveClient:
             metadata = {"name": BACKUP_FOLDER_NAME, "mimeType": FOLDER_MIME}
             request = self.drive_service.files().create(body=metadata, fields="id")
             result = self._api_call(request)
-            if result:
-                self.backup_folder_id = result["id"]
-                self.file_tracker = {}
+            if not result:
+                return None
+
+            new_id: str = result["id"]
+            self.backup_folder_id = new_id
+
+            # Conditional tracker reset (GOLD pattern):
+            # old drive_ids point to dead files under the old root folder
+            old_count = len(self.file_tracker)
+            if old_count > 0:
+                self.file_tracker.clear()
+                self.project_folder_cache.clear()
                 json_handler.log_operation(
-                    "get_backup_folder",
-                    {"action": "created_new", "folder_id": self.backup_folder_id},
+                    "tracker_reset",
+                    {
+                        "message": f"New backup folder - reset {old_count} tracker entries",
+                        "old_tracker_count": old_count,
+                        "new_folder_id": new_id,
+                    },
                 )
-                return self.backup_folder_id
+
+            # Verify accessible
+            if not self._verify_folder_id(new_id):
+                self.last_error = f"Backup folder {new_id} created but not accessible"
+                self.backup_folder_id = None
+                return None
+
+            json_handler.log_operation(
+                "get_backup_folder",
+                {"action": "created_new", "folder_id": new_id},
+            )
+            return self.backup_folder_id
         except Exception as exc:
             self.last_error = str(exc)
             logger.warning(f"Failed to create backup folder: {exc}")
@@ -184,58 +218,80 @@ class DriveClient:
     def get_or_create_project_folder(self, project_name: str) -> str | None:
         """Get or create a project subfolder under AIPass Backups.
 
-        Thread-safe via lock.
-
-        Returns:
-            Folder ID or None on failure.
+        Lock covers cache check + backup-folder-ensure + search + create
+        to prevent duplicate folders (GOLD's pattern).
         """
         with self._folder_cache_lock:
+            # Cache check with verify
             if project_name in self.project_folder_cache:
-                return self.project_folder_cache[project_name]
+                folder_id = self.project_folder_cache[project_name]
+                if self._verify_folder_id(folder_id):
+                    return folder_id
+                del self.project_folder_cache[project_name]
 
-        if not self.backup_folder_id:
-            self.backup_folder_id = self.get_or_create_backup_folder()
-            if not self.backup_folder_id:
+            # Ensure backup folder (no deadlock: backup_folder has no lock)
+            backup_folder_id = self.get_or_create_backup_folder()
+            if not backup_folder_id:
                 return None
 
-        query = (
-            f"name='{project_name}' "
-            f"and mimeType='{FOLDER_MIME}' "
-            f"and '{self.backup_folder_id}' in parents "
-            f"and trashed=false"
-        )
-        try:
-            request = self.drive_service.files().list(q=query, spaces="drive", fields="files(id,name)")
-            result = self._api_call(request)
-            if result and result.get("files"):
-                folder_id = result["files"][0]["id"]
-                with self._folder_cache_lock:
+            # Search
+            query = (
+                f"name='{project_name}' "
+                f"and mimeType='{FOLDER_MIME}' "
+                f"and '{backup_folder_id}' in parents "
+                f"and trashed=false"
+            )
+            try:
+                request = self.drive_service.files().list(
+                    q=query,
+                    spaces="drive",
+                    fields="files(id,name)",
+                )
+                result = self._api_call(request)
+                if result and result.get("files"):
+                    folder_id = result["files"][0]["id"]
                     self.project_folder_cache[project_name] = folder_id
-                return folder_id
-        except Exception as exc:
-            self.last_error = str(exc)
-            logger.warning(f"Failed to search for project folder '{project_name}': {exc}")
+                    return folder_id
+            except Exception as exc:
+                self.last_error = str(exc)
+                logger.warning(f"Failed to search for project folder '{project_name}': {exc}")
+                return None
+
+            # Create
+            try:
+                metadata = {
+                    "name": project_name,
+                    "mimeType": FOLDER_MIME,
+                    "parents": [backup_folder_id],
+                }
+                request = self.drive_service.files().create(body=metadata, fields="id")
+                result = self._api_call(request)
+                if result:
+                    folder_id = result["id"]
+                    self.project_folder_cache[project_name] = folder_id
+                    return folder_id
+            except Exception as exc:
+                self.last_error = str(exc)
+                logger.warning(f"Failed to create project folder '{project_name}': {exc}")
+
             return None
 
-        # Create
-        try:
-            metadata = {
-                "name": project_name,
-                "mimeType": FOLDER_MIME,
-                "parents": [self.backup_folder_id],
-            }
-            request = self.drive_service.files().create(body=metadata, fields="id")
-            result = self._api_call(request)
-            if result:
-                folder_id = result["id"]
-                with self._folder_cache_lock:
-                    self.project_folder_cache[project_name] = folder_id
-                return folder_id
-        except Exception as exc:
-            self.last_error = str(exc)
-            logger.warning(f"Failed to create project folder '{project_name}': {exc}")
+    def _find_or_create_segment(self, parent_id: str, name: str) -> str | None:
+        """Search for or create a single folder segment under parent_id."""
+        query = f"name='{name}' and mimeType='{FOLDER_MIME}' and '{parent_id}' in parents and trashed=false"
+        request = self.drive_service.files().list(
+            q=query,
+            spaces="drive",
+            fields="files(id,name)",
+        )
+        result = self._api_call(request)
+        if result and result.get("files"):
+            return result["files"][0]["id"]
 
-        return None
+        metadata = {"name": name, "mimeType": FOLDER_MIME, "parents": [parent_id]}
+        request = self.drive_service.files().create(body=metadata, fields="id")
+        result = self._api_call(request)
+        return result["id"] if result else None
 
     def get_or_create_nested_folder(
         self,
@@ -244,63 +300,46 @@ class DriveClient:
     ) -> str | None:
         """Create a nested folder hierarchy segment by segment.
 
-        Thread-safe via lock.
-
-        Args:
-            parent_id: ID of the parent folder.
-            folder_path: Slash-separated path of nested folders.
-
-        Returns:
-            ID of the deepest folder, or None on failure.
+        Lock covers entire walk — full-path + per-segment caching with
+        verify (GOLD's pattern).
         """
-        current_parent = parent_id
-        segments = [s for s in folder_path.split("/") if s]
+        if not folder_path or folder_path == ".":
+            return parent_id
 
-        for segment in segments:
-            cache_key = f"{current_parent}/{segment}"
-            with self._folder_cache_lock:
-                if cache_key in self.project_folder_cache:
-                    current_parent = self.project_folder_cache[cache_key]
-                    continue
+        with self._folder_cache_lock:
+            cache_key = f"{parent_id}:{folder_path}"
+            if cache_key in self.project_folder_cache:
+                folder_id = self.project_folder_cache[cache_key]
+                if self._verify_folder_id(folder_id):
+                    return folder_id
+                del self.project_folder_cache[cache_key]
 
-            # Search for existing
-            query = f"name='{segment}' and mimeType='{FOLDER_MIME}' and '{current_parent}' in parents and trashed=false"
-            try:
-                request = self.drive_service.files().list(q=query, spaces="drive", fields="files(id,name)")
-                result = self._api_call(request)
-                if result and result.get("files"):
-                    folder_id = result["files"][0]["id"]
-                    with self._folder_cache_lock:
-                        self.project_folder_cache[cache_key] = folder_id
-                    current_parent = folder_id
-                    continue
-            except Exception as exc:
-                self.last_error = str(exc)
-                logger.info(f"Failed to search for nested folder '{segment}': {exc}")
-                return None
+            current_parent = parent_id
+            segments = [s for s in folder_path.split("/") if s]
 
-            # Create
-            try:
-                metadata = {
-                    "name": segment,
-                    "mimeType": FOLDER_MIME,
-                    "parents": [current_parent],
-                }
-                request = self.drive_service.files().create(body=metadata, fields="id")
-                result = self._api_call(request)
-                if result:
-                    folder_id = result["id"]
-                    with self._folder_cache_lock:
-                        self.project_folder_cache[cache_key] = folder_id
-                    current_parent = folder_id
-                else:
+            for segment in segments:
+                segment_key = f"{current_parent}:{segment}"
+
+                if segment_key in self.project_folder_cache:
+                    cached_id = self.project_folder_cache[segment_key]
+                    if self._verify_folder_id(cached_id):
+                        current_parent = cached_id
+                        continue
+                    del self.project_folder_cache[segment_key]
+
+                try:
+                    folder_id = self._find_or_create_segment(current_parent, segment)
+                except Exception as exc:
+                    self.last_error = str(exc)
+                    logger.info(f"Failed to handle nested folder '{segment}': {exc}")
                     return None
-            except Exception as exc:
-                self.last_error = str(exc)
-                logger.info(f"Failed to create nested folder '{segment}': {exc}")
-                return None
+                if not folder_id:
+                    return None
+                current_parent = folder_id
+                self.project_folder_cache[segment_key] = current_parent
 
-        return current_parent
+            self.project_folder_cache[cache_key] = current_parent
+            return current_parent
 
     # -- file ops ------------------------------------------------------------
 
@@ -309,14 +348,14 @@ class DriveClient:
         filename: str,
         parent_folder_id: str,
     ) -> dict | None:
-        """Find a file by name in a folder (excludes trashed).
-
-        Returns:
-            File metadata dict with id/name, or None if not found.
-        """
+        """Find a file by name in a folder (excludes trashed)."""
         query = f"name='{filename}' and '{parent_folder_id}' in parents and trashed=false"
         try:
-            request = self.drive_service.files().list(q=query, spaces="drive", fields="files(id,name)")
+            request = self.drive_service.files().list(
+                q=query,
+                spaces="drive",
+                fields="files(id,name)",
+            )
             result = self._api_call(request)
             if result and result.get("files"):
                 return result["files"][0]

src/aipass/backup/apps/modules/drive_clear.py

@@ -17,7 +17,7 @@ from aipass.backup.apps.handlers.json import json_handler
 
 
 MODULE_NAME = "drive_clear"
-PRIMARY_COMMAND = "drive-clear-tracker"
+PRIMARY_COMMAND = "drive_clear"
 
 
 def print_introspection():
@@ -32,7 +32,7 @@ def print_help():
     """Display help for this module."""
     print_introspection()
     console.print()
-    console.print("Usage: drive-clear-tracker <project_root> --force")
+    console.print("Usage: drive_clear <project_root> --force")
     console.print("  --force   Required to confirm tracker deletion")
 
 

src/aipass/backup/apps/modules/drive_stats.py

@@ -17,7 +17,7 @@ from aipass.backup.apps.handlers.json import json_handler
 
 
 MODULE_NAME = "drive_stats"
-PRIMARY_COMMAND = "drive-stats"
+PRIMARY_COMMAND = "drive_stats"
 
 
 def print_introspection():
@@ -32,7 +32,7 @@ def print_help():
     """Display help for this module."""
     print_introspection()
     console.print()
-    console.print("Usage: drive-stats <project_root>")
+    console.print("Usage: drive_stats <project_root>")
 
 
 def run_drive_stats(project_root: str) -> bool:

src/aipass/backup/apps/modules/drive_sync.py

@@ -1,7 +1,7 @@
 # =================== AIPass ====================
 # Name: drive_sync.py
 # Description: Drive sync module — uploads versioned store to Google Drive
-# Version: 1.0.0
+# Version: 1.1.0
 # Created: 2026-04-17
 # Modified: 2026-06-12
 # =============================================
@@ -10,6 +10,9 @@
 
 Scans the versioned store, checks the tracker for changes, and uploads
 new or modified files via the Drive upload engine.
+
+Flow: auth → store path → scan → tracker filter → upload_batch → save tracker.
+No pre-resolve — workers create folders on demand via the client's lock pattern.
 """
 
 import sys
@@ -23,7 +26,7 @@ from aipass.backup.apps.handlers.path.builder import build_versioned_store
 
 
 MODULE_NAME = "drive_sync"
-PRIMARY_COMMAND = "drive-sync"
+PRIMARY_COMMAND = "drive_sync"
 
 
 def print_introspection():
@@ -38,7 +41,7 @@ def print_help():
     """Display help for this module."""
     print_introspection()
     console.print()
-    console.print("Usage: drive-sync <project_root> [options]")
+    console.print("Usage: drive_sync <project_root> [options]")
     console.print("  --force       Force re-upload of all files")
     console.print("  --project     Override project name")
     console.print("  --note        Add a note to uploaded files")
@@ -53,16 +56,6 @@ def run_drive_sync(
 ) -> dict:
     """Run Drive sync -- upload versioned store to Google Drive.
 
-    Flow:
-        1. Create DriveClient, authenticate
-        2. Build versioned store path from project_root
-        3. Scan versioned store for all files (skip dotfiles)
-        4. Load tracker, check each file with check_needs_upload()
-        5. Show Rich progress bar if show_panels
-        6. Call upload_batch() with files that need upload
-        7. Save tracker
-        8. Return result dict
-
     Args:
         project_root: Absolute path to the project.
         project_name: Override project name (defaults to dir name).
@@ -117,7 +110,11 @@ def run_drive_sync(
             console.print("[dim]No files found in versioned store.[/dim]")
         return result
 
-    # 4. Load tracker + filter
+    # 4. Resolve project name
+    if not project_name:
+        project_name = Path(project_root).name
+
+    # 5. Load tracker + filter
     tracker = load_tracker(project_root)
     if force:
         files_to_upload = all_files
@@ -133,10 +130,6 @@ def run_drive_sync(
             console.print(f"[green]All {len(all_files)} files up to date.[/green]")
         return result
 
-    # 5. Resolve project name
-    if not project_name:
-        project_name = Path(project_root).name
-
     # 6. Upload with progress
     progress_fn = None
     progress = None
@@ -203,7 +196,7 @@ def run_drive_sync(
 
 
 def handle_command(command: str, args: list) -> bool:
-    """Handle the drive-sync command. Returns True if handled."""
+    """Handle the drive_sync command. Returns True if handled."""
     if command != PRIMARY_COMMAND:
         return False
 

src/aipass/backup/apps/modules/drive_test.py

@@ -17,7 +17,7 @@ from aipass.backup.apps.handlers.json import json_handler
 
 
 MODULE_NAME = "drive_test"
-PRIMARY_COMMAND = "drive-test"
+PRIMARY_COMMAND = "drive_test"
 
 
 def print_introspection():

src/aipass/backup/requirements.project.txt

@@ -1 +1,4 @@
 rich>=13.0.0
+google-api-python-client>=2.0.0
+google-auth>=2.0.0
+google-auth-oauthlib>=1.0.0

src/aipass/backup/tests/test_drive_pipeline.py

@@ -1,7 +1,7 @@
 # =================== AIPass ====================
 # Name: test_drive_pipeline.py
 # Description: Tests for Drive sync pipeline -- fully mocked, zero real Google calls
-# Version: 1.0.0
+# Version: 2.0.0
 # Created: 2026-06-12
 # Modified: 2026-06-12
 # =============================================
@@ -84,13 +84,14 @@ def _fresh_import(module_path: str, extra_mocks: dict | None = None):
     if extra_mocks:
         mocks.update(extra_mocks)
 
+    # Clear stale drive entries BEFORE patch.dict so they won't be restored
+    for key in list(sys.modules.keys()):
+        if key.startswith("aipass.backup.apps.handlers.drive"):
+            del sys.modules[key]
+    if module_path in sys.modules:
+        del sys.modules[module_path]
+
     with patch.dict(sys.modules, mocks):
-        # Clear cached module if present
-        for key in list(sys.modules.keys()):
-            if key.startswith("aipass.backup.apps.handlers.drive"):
-                del sys.modules[key]
-        if module_path in sys.modules:
-            del sys.modules[module_path]
         mod = importlib.import_module(module_path)
         return mod
 
@@ -172,10 +173,9 @@ class TestDriveClient:
         mock_service = MagicMock()
         client._drive_service = mock_service
 
-        # files().list().execute() returns folder
-        mock_list = MagicMock()
-        mock_service.files.return_value.list.return_value = mock_list
-        mod.api_call_with_retry = MagicMock(return_value={"files": [{"id": "folder_123", "name": "AIPass Backups"}]})
+        mod.api_call_with_retry = MagicMock(
+            return_value={"files": [{"id": "folder_123", "name": "AIPass Backups"}]}
+        )
 
         result = client.get_or_create_backup_folder()
         assert result == "folder_123"
@@ -188,22 +188,21 @@ class TestDriveClient:
         mock_service = MagicMock()
         client._drive_service = mock_service
 
-        # First call (list) returns empty, second call (create) returns new folder
         call_count = {"n": 0}
 
         def _side_effect(request, **kwargs):
             call_count["n"] += 1
             if call_count["n"] == 1:
                 return {"files": []}
-            return {"id": "new_folder_456"}
+            if call_count["n"] == 2:
+                return {"id": "new_folder_456"}
+            return {"id": "new_folder_456", "trashed": False}
 
         mod.api_call_with_retry = MagicMock(side_effect=_side_effect)
 
         result = client.get_or_create_backup_folder()
         assert result == "new_folder_456"
         assert client.backup_folder_id == "new_folder_456"
-        # Tracker should be reset on new folder creation
-        assert client.file_tracker == {}
 
     def test_get_or_create_backup_folder_no_service(self) -> None:
         """No drive service -- returns None."""
@@ -221,7 +220,9 @@ class TestDriveClient:
         client._drive_service = mock_service
         client.backup_folder_id = "root_folder"
 
-        mod.api_call_with_retry = MagicMock(return_value={"files": [{"id": "proj_folder_789", "name": "myproject"}]})
+        mod.api_call_with_retry = MagicMock(
+            return_value={"files": [{"id": "proj_folder_789", "name": "myproject"}]}
+        )
 
         result = client.get_or_create_project_folder("myproject")
         assert result == "proj_folder_789"
@@ -231,8 +232,11 @@ class TestDriveClient:
         """Cached project folder returned without API call."""
         mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
         client = mod.DriveClient()
+        client._drive_service = MagicMock()
         client.project_folder_cache["cached_proj"] = "cached_id"
 
+        mod.api_call_with_retry = MagicMock(return_value={"id": "cached_id", "trashed": False})
+
         result = client.get_or_create_project_folder("cached_proj")
         assert result == "cached_id"
 
@@ -505,8 +509,7 @@ class TestDriveUpload:
         client = client_mod.DriveClient()
         mock_service = MagicMock()
         client._drive_service = mock_service
-        client.backup_folder_id = "root_folder"
-        client.project_folder_cache["testproj"] = "proj_folder"
+        client.get_or_create_project_folder = MagicMock(return_value="proj_folder")
 
         # Create test file
         test_file = tmp_path / "hello.py"
@@ -526,8 +529,7 @@ class TestDriveUpload:
         client = client_mod.DriveClient()
         mock_service = MagicMock()
         client._drive_service = mock_service
-        client.backup_folder_id = "root_folder"
-        client.project_folder_cache["testproj"] = "proj_folder"
+        client.get_or_create_project_folder = MagicMock(return_value="proj_folder")
 
         # Pre-populate tracker with existing drive_id
         client.file_tracker = {"existing.py": {"drive_id": "existing_drive_id"}}
@@ -570,8 +572,7 @@ class TestDriveUpload:
         client = client_mod.DriveClient()
         mock_service = MagicMock()
         client._drive_service = mock_service
-        client.backup_folder_id = "root"
-        client.project_folder_cache["proj"] = "proj_folder"
+        client.get_or_create_project_folder = MagicMock(return_value="proj_folder")
 
         # Create test files
         files = []
@@ -580,7 +581,7 @@ class TestDriveUpload:
             f.write_text(f"content {i}", encoding="utf-8")
             files.append(f)
 
-        client_mod.api_call_with_retry = MagicMock(return_value={"id": f"id_{id}"})
+        client_mod.api_call_with_retry = MagicMock(return_value={"id": "file_id"})
         client_mod.get_drive_service = MagicMock(return_value=mock_service)
 
         progress_calls = []
@@ -610,8 +611,7 @@ class TestDriveUpload:
 
         client = client_mod.DriveClient()
         client._drive_service = MagicMock()
-        client.backup_folder_id = "root"
-        client.project_folder_cache["proj"] = "proj_folder"
+        client.get_or_create_project_folder = MagicMock(return_value="proj_folder")
 
         test_file = tmp_path / "test.txt"
         test_file.write_text("data", encoding="utf-8")
@@ -820,12 +820,12 @@ class TestDriveSync:
     def test_handle_command_help(self) -> None:
         """--help returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_sync")
-        assert mod.handle_command("drive-sync", ["--help"]) is True
+        assert mod.handle_command("drive_sync", ["--help"]) is True
 
     def test_handle_command_no_args(self) -> None:
         """No args prints introspection."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_sync")
-        assert mod.handle_command("drive-sync", []) is True
+        assert mod.handle_command("drive_sync", []) is True
 
     def test_handle_command_wrong_command(self) -> None:
         """Wrong command returns False."""
@@ -842,18 +842,14 @@ class TestDriveTestModule:
     """Tests for drive_test module."""
 
     def test_handle_command_primary(self) -> None:
-        """drive-test returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_test")
-        # No args triggers introspection
-        assert mod.handle_command("drive-test", []) is True
+        assert mod.handle_command("drive_test", []) is True
 
     def test_handle_command_help(self) -> None:
-        """--help returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_test")
-        assert mod.handle_command("drive-test", ["--help"]) is True
+        assert mod.handle_command("drive_test", ["--help"]) is True
 
     def test_handle_command_wrong(self) -> None:
-        """Wrong command returns False."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_test")
         assert mod.handle_command("wrong", []) is False
 
@@ -861,7 +857,6 @@ class TestDriveTestModule:
         """Run drive test with mocked success."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_test")
 
-        # Mock the late-imported modules
         mock_client_module = MagicMock()
         mock_client_instance = MagicMock()
         mock_client_module.DriveClient.return_value = mock_client_instance
@@ -888,17 +883,14 @@ class TestDriveStatsModule:
     """Tests for drive_stats module."""
 
     def test_handle_command_primary(self) -> None:
-        """drive-stats with no args returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_stats")
-        assert mod.handle_command("drive-stats", []) is True
+        assert mod.handle_command("drive_stats", []) is True
 
     def test_handle_command_help(self) -> None:
-        """--help returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_stats")
-        assert mod.handle_command("drive-stats", ["--help"]) is True
+        assert mod.handle_command("drive_stats", ["--help"]) is True
 
     def test_handle_command_wrong(self) -> None:
-        """Wrong command returns False."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_stats")
         assert mod.handle_command("wrong", []) is False
 
@@ -925,24 +917,20 @@ class TestDriveClearModule:
     """Tests for drive_clear module."""
 
     def test_handle_command_primary(self) -> None:
-        """drive-clear-tracker with no args returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_clear")
-        assert mod.handle_command("drive-clear-tracker", []) is True
+        assert mod.handle_command("drive_clear", []) is True
 
     def test_handle_command_help(self) -> None:
-        """--help returns True."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_clear")
-        assert mod.handle_command("drive-clear-tracker", ["--help"]) is True
+        assert mod.handle_command("drive_clear", ["--help"]) is True
 
     def test_handle_command_wrong(self) -> None:
-        """Wrong command returns False."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_clear")
         assert mod.handle_command("wrong", []) is False
 
     def test_run_drive_clear_no_force(self) -> None:
         """Without --force, returns False."""
         mod = _fresh_import("aipass.backup.apps.modules.drive_clear")
-        # force=False means early return, no late import needed
         result = mod.run_drive_clear("/tmp/project", force=False)
         assert result is False
 
@@ -961,4 +949,186 @@ class TestDriveClearModule:
         assert result is True
 
 
+# ---------------------------------------------------------------------------
+# TestThreadSafety — concurrent folder operations
+# ---------------------------------------------------------------------------
+
+
+class TestThreadSafety:
+    """Verify folder get-or-create is thread-safe (GOLD lock pattern)."""
+
+    def test_concurrent_project_folder_single_create(self) -> None:
+        """N threads calling get_or_create_project_folder -> exactly 1 create."""
+        import threading
+
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
+        client = mod.DriveClient()
+        mock_service = MagicMock()
+        client._drive_service = mock_service
+        client.backup_folder_id = "root_folder"
+
+        create_calls = {"n": 0}
+        lock = threading.Lock()
+
+        def _side_effect(request, **kwargs):
+            with lock:
+                create_calls["n"] += 1
+                n = create_calls["n"]
+            if n == 1:
+                return {"id": "root_folder", "trashed": False}
+            if n == 2:
+                return {"files": []}
+            if n == 3:
+                return {"id": "proj_folder_unique"}
+            return {"trashed": False}
+
+        mod.api_call_with_retry = MagicMock(side_effect=_side_effect)
+
+        results = []
+
+        def _worker():
+            r = client.get_or_create_project_folder("myproj")
+            results.append(r)
+
+        threads = [threading.Thread(target=_worker) for _ in range(5)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert all(r == "proj_folder_unique" for r in results), f"Got different IDs: {results}"
+
+    def test_backup_folder_short_circuits(self) -> None:
+        """Once backup_folder_id is set+valid, returns without re-searching."""
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
+        client = mod.DriveClient()
+        client._drive_service = MagicMock()
+        client.backup_folder_id = "already_set"
+
+        mod.api_call_with_retry = MagicMock(return_value={"id": "already_set", "trashed": False})
+
+        result = client.get_or_create_backup_folder()
+        assert result == "already_set"
+        assert mod.api_call_with_retry.call_count == 1
+
+    def test_tracker_preserved_on_existing_folder(self) -> None:
+        """Tracker NOT cleared when backup folder already exists in Drive."""
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
+        client = mod.DriveClient()
+        client._drive_service = MagicMock()
+        client.file_tracker = {"existing.txt": {"drive_id": "abc"}}
+
+        mod.api_call_with_retry = MagicMock(
+            return_value={"files": [{"id": "found_folder", "name": "AIPass Backups"}]}
+        )
+
+        result = client.get_or_create_backup_folder()
+        assert result == "found_folder"
+        assert client.file_tracker == {"existing.txt": {"drive_id": "abc"}}
+
+    def test_tracker_reset_on_new_folder_with_old_entries(self) -> None:
+        """Tracker cleared ONLY when creating a NEW root folder AND old_count>0."""
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
+        client = mod.DriveClient()
+        client._drive_service = MagicMock()
+        client.file_tracker = {"old.txt": {"drive_id": "dead_id"}}
+
+        call_count = {"n": 0}
+
+        def _side_effect(request, **kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                return {"files": []}
+            if call_count["n"] == 2:
+                return {"id": "brand_new_folder"}
+            return {"id": "brand_new_folder", "trashed": False}
+
+        mod.api_call_with_retry = MagicMock(side_effect=_side_effect)
+
+        result = client.get_or_create_backup_folder()
+        assert result == "brand_new_folder"
+        assert client.file_tracker == {}
+
+    def test_tracker_not_reset_on_new_folder_empty_tracker(self) -> None:
+        """Tracker NOT cleared when creating new folder with empty tracker."""
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.client")
+        client = mod.DriveClient()
+        client._drive_service = MagicMock()
+        client.file_tracker = {}
+
+        call_count = {"n": 0}
+
+        def _side_effect(request, **kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                return {"files": []}
+            if call_count["n"] == 2:
+                return {"id": "new_folder"}
+            return {"id": "new_folder", "trashed": False}
+
+        mod.api_call_with_retry = MagicMock(side_effect=_side_effect)
+
+        result = client.get_or_create_backup_folder()
+        assert result == "new_folder"
+        assert client.file_tracker == {}
+
+
+class TestDedup:
+    """Verify tracker-based dedup skips unchanged files."""
+
+    def test_rerun_unchanged_zero_uploads(self, tmp_path: Path) -> None:
+        """All files tracked with matching mtime+size -> 0 uploads."""
+        mod = _fresh_import("aipass.backup.apps.handlers.drive.tracker")
+
+        files = []
+        tracker = {}
+        for i in range(5):
+            f = tmp_path / f"file_{i}.txt"
+            f.write_text(f"content {i}", encoding="utf-8")
+            files.append(f)
+            stat = f.stat()
+            tracker[f"file_{i}.txt"] = {
+                "local_size": stat.st_size,
+                "local_mtime": stat.st_mtime,
+                "drive_id": f"drive_{i}",
+                "last_sync": "2026-06-12T00:00:00",
+            }
+
+        needs_upload = [f for f in files if mod.check_needs_upload(tracker, f, tmp_path)]
+        assert len(needs_upload) == 0
+
+
+# ---------------------------------------------------------------------------
+# TestCommandRouting — all 4 drive commands route by underscore name
+# ---------------------------------------------------------------------------
+
+
+class TestCommandRouting:
+    """Verify drive commands route by underscore names."""
+
+    def test_drive_sync_routes_underscore(self) -> None:
+        mod = _fresh_import("aipass.backup.apps.modules.drive_sync")
+        assert mod.PRIMARY_COMMAND == "drive_sync"
+        assert mod.handle_command("drive_sync", []) is True
+        assert mod.handle_command("drive-sync", []) is False
+
+    def test_drive_test_routes_underscore(self) -> None:
+        mod = _fresh_import("aipass.backup.apps.modules.drive_test")
+        assert mod.PRIMARY_COMMAND == "drive_test"
+        assert mod.handle_command("drive_test", []) is True
+        assert mod.handle_command("drive-test", []) is False
+
+    def test_drive_stats_routes_underscore(self) -> None:
+        mod = _fresh_import("aipass.backup.apps.modules.drive_stats")
+        assert mod.PRIMARY_COMMAND == "drive_stats"
+        assert mod.handle_command("drive_stats", []) is True
+        assert mod.handle_command("drive-stats", []) is False
+
+    def test_drive_clear_routes_underscore(self) -> None:
+        mod = _fresh_import("aipass.backup.apps.modules.drive_clear")
+        assert mod.PRIMARY_COMMAND == "drive_clear"
+        assert mod.handle_command("drive_clear", []) is True
+        assert mod.handle_command("drive-clear-tracker", []) is False
+
+
 # =============================================