simpler and more efficient solution:

1- just deletes the directory with shutil.rmtree
2- Uses onerror callback to properly handle Windows read-only files (git creates these in .git/objects)
3- Single exception handler
4- No git commands needed

Author: mohamedashrraf222

codeflash/code_utils/git_worktree_utils.py

@@ -1,15 +1,16 @@
 from __future__ import annotations
 
 import configparser
+import os
 import shutil
+import stat
 import subprocess
 import tempfile
 import time
 from pathlib import Path
 from typing import Optional
 
 import git
-from git.exc import GitCommandError
 
 from codeflash.cli_cmds.console import logger
 from codeflash.code_utils.compat import codeflash_cache_dir
@@ -97,62 +98,24 @@ def create_detached_worktree(module_root: Path) -> Optional[Path]:
         return worktree_dir
 
 
-def _fallback_remove_worktree(worktree_dir: Path) -> None:
-    """Fallback worktree removal using shutil.rmtree when git commands fail."""
-    if worktree_dir.exists():
-        shutil.rmtree(worktree_dir, ignore_errors=True)
-        logger.debug(f"Removed worktree directory using fallback method: {worktree_dir}")
+def _handle_remove_readonly(func, path, exc_info):
+    """Error handler for shutil.rmtree to handle read-only files on Windows."""
+    if isinstance(exc_info[1], PermissionError):
+        os.chmod(path, stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR)
+        func(path)
+    else:
+        raise exc_info[1]
 
 
 def remove_worktree(worktree_dir: Path) -> None:
-    """Remove a git worktree, with retry logic for Windows permission errors."""
-    # Try to get repository reference
-    try:
-        repository = git.Repo(worktree_dir, search_parent_directories=True)
-    except git.InvalidGitRepositoryError:
-        # Worktree is not a valid git repository (corrupted or partially created)
-        logger.debug(f"Worktree is not a valid git repository, using fallback deletion: {worktree_dir}")
-        _fallback_remove_worktree(worktree_dir)
+    """Remove a git worktree directory."""
+    if not worktree_dir.exists():
         return
+    try:
+        shutil.rmtree(worktree_dir, onerror=_handle_remove_readonly)
+        logger.debug(f"Removed worktree: {worktree_dir}")
     except Exception:
-        logger.exception(f"Failed to open worktree repository: {worktree_dir}")
-        _fallback_remove_worktree(worktree_dir)
-        return
-
-    # Try git worktree remove first
-    for attempt in range(2):
-        try:
-            repository.git.worktree("remove", "--force", worktree_dir)
-            logger.debug(f"Successfully removed worktree: {worktree_dir}")
-            return
-        except GitCommandError as e:
-            error_msg = str(e).lower()
-            # Check if it's a permission error or not a git repository error
-            if "permission denied" in error_msg or "failed to delete" in error_msg:
-                if attempt == 0:
-                    # Retry once with a small delay to allow file handles to close
-                    logger.debug(f"Permission error removing worktree (attempt {attempt + 1}), retrying after delay: {worktree_dir}")
-                    time.sleep(0.5)
-                    continue
-            elif "not a git repository" in error_msg:
-                # Worktree reference is broken, just delete the directory
-                logger.debug(f"Worktree git reference is broken, using fallback deletion: {worktree_dir}")
-                _fallback_remove_worktree(worktree_dir)
-                return
-            
-            # Fallback to shutil.rmtree for any persistent error
-            logger.warning(f"Git worktree remove failed, using fallback deletion: {worktree_dir}")
-            _fallback_remove_worktree(worktree_dir)
-            # Try to prune stale worktree references
-            try:
-                repository.git.worktree("prune")
-            except Exception:
-                pass
-            return
-        except Exception:
-            logger.exception(f"Failed to remove worktree: {worktree_dir}")
-            _fallback_remove_worktree(worktree_dir)
-            return
+        logger.exception(f"Failed to remove worktree: {worktree_dir}")
 
 
 def create_diff_patch_from_worktree(