Fix for VSC windows

codeflash/benchmarking/trace_benchmarks.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import contextlib
 import os
 import re
 import subprocess
+import sys
 from pathlib import Path
 
 from codeflash.cli_cmds.console import logger
@@ -17,21 +19,42 @@ def trace_benchmarks_pytest(
         benchmark_env["PYTHONPATH"] = str(project_root)
     else:
         benchmark_env["PYTHONPATH"] += os.pathsep + str(project_root)
-    result = subprocess.run(
-        [
-            SAFE_SYS_EXECUTABLE,
-            Path(__file__).parent / "pytest_new_process_trace_benchmarks.py",
-            benchmarks_root,
-            tests_root,
-            trace_file,
-        ],
-        cwd=project_root,
-        check=False,
-        capture_output=True,
-        text=True,
-        env=benchmark_env,
-        timeout=timeout,
-    )
+
+    is_windows = sys.platform == "win32"
+    cmd_list = [
+        SAFE_SYS_EXECUTABLE,
+        Path(__file__).parent / "pytest_new_process_trace_benchmarks.py",
+        benchmarks_root,
+        tests_root,
+        trace_file,
+    ]
+
+    if is_windows:
+        # Use Windows-safe subprocess handling to avoid file locking issues
+        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
+        process = subprocess.Popen(
+            cmd_list,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            stdin=subprocess.DEVNULL,
+            cwd=project_root,
+            env=benchmark_env,
+            text=True,
+            creationflags=creationflags,
+        )
+        try:
+            stdout_content, stderr_content = process.communicate(timeout=timeout)
+            returncode = process.returncode
+        except subprocess.TimeoutExpired:
+            with contextlib.suppress(OSError):
+                process.kill()
+            stdout_content, stderr_content = process.communicate(timeout=5)
+            raise subprocess.TimeoutExpired(cmd_list, timeout, output=stdout_content, stderr=stderr_content) from None
+        result = subprocess.CompletedProcess(cmd_list, returncode, stdout_content, stderr_content)
+    else:
+        result = subprocess.run(
+            cmd_list, cwd=project_root, check=False, capture_output=True, text=True, env=benchmark_env, timeout=timeout
+        )
     if result.returncode != 0:
         if "ERROR collecting" in result.stdout:
             # Pattern matches "===== ERRORS =====" (any number of =) and captures everything after

codeflash/cli_cmds/console.py

@@ -141,6 +141,15 @@ def __init__(self) -> None:
 @contextmanager
 def test_files_progress_bar(total: int, description: str) -> Generator[tuple[Progress, TaskID], None, None]:
     """Progress bar for test files."""
+    if is_LSP_enabled():
+
+        class DummyProgress:
+            def advance(self, *args: object, **kwargs: object) -> None:
+                pass
+
+        yield DummyProgress(), 0  # type: ignore[return-value]
+        return
+
     with Progress(
         SpinnerColumn(next(spinners)),
         TextColumn("[progress.description]{task.description}"),

codeflash/code_utils/coverage_utils.py

@@ -68,7 +68,10 @@ def generate_candidates(source_code_path: Path) -> set[str]:
 
 
 def prepare_coverage_files() -> tuple[Path, Path]:
-    """Prepare coverage configuration and output files."""
+    """Prepare coverage configuration and output files.
+
+    Returns tuple of (coverage_database_file, coverage_config_file).
+    """
     coverage_database_file = get_run_tmp_file(Path(".coverage"))
     coveragercfile = get_run_tmp_file(Path(".coveragerc"))
     coveragerc_content = f"[run]\n branch = True\ndata_file={coverage_database_file}\n"

codeflash/code_utils/git_worktree_utils.py

@@ -1,15 +1,20 @@
 from __future__ import annotations
 
 import configparser
+import contextlib
+import shutil
 import subprocess
+import sys
 import tempfile
 import time
 from pathlib import Path
-from typing import Optional
+from typing import TYPE_CHECKING, Any, Optional
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 import git
 
-from codeflash.cli_cmds.console import logger
 from codeflash.code_utils.compat import codeflash_cache_dir
 from codeflash.code_utils.git_utils import check_running_in_git_repo, git_root_dir
 
@@ -53,7 +58,6 @@ def create_worktree_snapshot_commit(worktree_dir: Path, commit_message: str) ->
 
 def create_detached_worktree(module_root: Path) -> Optional[Path]:
     if not check_running_in_git_repo(module_root):
-        logger.warning("Module is not in a git repository. Skipping worktree creation.")
         return None
     git_root = git_root_dir()
     current_time_str = time.strftime("%Y%m%d-%H%M%S")
@@ -71,7 +75,6 @@ def create_detached_worktree(module_root: Path) -> Optional[Path]:
     )
 
     if not uni_diff_text.strip():
-        logger.info("!lsp|No uncommitted changes to copy to worktree.")
         return worktree_dir
 
     # Write the diff to a temporary file
@@ -89,18 +92,190 @@ def create_detached_worktree(module_root: Path) -> Optional[Path]:
                 check=True,
             )
             create_worktree_snapshot_commit(worktree_dir, "Initial Snapshot")
-        except subprocess.CalledProcessError as e:
-            logger.error(f"Failed to apply patch to worktree: {e}")
+        except subprocess.CalledProcessError:
+            pass
 
         return worktree_dir
 
 
 def remove_worktree(worktree_dir: Path) -> None:
+    """Remove a git worktree with robust error handling for Windows file locking issues.
+
+    This function handles Windows-specific issues where files may be locked by processes,
+    causing 'Permission denied' errors. It implements retry logic with exponential backoff
+    and falls back to manual directory removal if git worktree remove fails.
+
+    Args:
+        worktree_dir: Path to the worktree directory to remove
+
+    """
+    if not worktree_dir or not worktree_dir.exists():
+        return
+
+    is_windows = sys.platform == "win32"
+    max_retries = 3 if is_windows else 1
+    retry_delay = 0.5  # Start with 500ms delay
+
+    # Try to get the repository and git root for worktree removal
     try:
         repository = git.Repo(worktree_dir, search_parent_directories=True)
-        repository.git.worktree("remove", "--force", worktree_dir)
     except Exception:
-        logger.exception(f"Failed to remove worktree: {worktree_dir}")
+        # If we can't access the repository, try manual cleanup
+        _manual_cleanup_worktree_directory(worktree_dir)
+        return
+
+    # Attempt to remove worktree using git command with retries
+    for attempt in range(max_retries):
+        try:
+            repository.git.worktree("remove", "--force", str(worktree_dir))
+            return  # noqa: TRY300
+        except git.exc.GitCommandError as e:
+            error_msg = str(e).lower()
+            is_permission_error = "permission denied" in error_msg or "access is denied" in error_msg
+
+            if is_permission_error and attempt < max_retries - 1:
+                # On Windows, file locks may be temporary - retry with exponential backoff
+                time.sleep(retry_delay)
+                retry_delay *= 2  # Exponential backoff
+            else:
+                # Last attempt failed or non-permission error
+                break
+        except Exception:
+            break
+
+    # Fallback: Try to remove worktree entry from git, then manually delete directory
+    with contextlib.suppress(Exception):
+        # Try to prune the worktree entry from git (this doesn't delete the directory)
+        # Use git worktree prune to remove stale entries
+        repository.git.worktree("prune")
+
+    # Manually remove the directory (always attempt, even if prune failed)
+    with contextlib.suppress(Exception):
+        _manual_cleanup_worktree_directory(worktree_dir)
+
+
+def _manual_cleanup_worktree_directory(worktree_dir: Path) -> None:
+    """Manually remove a worktree directory, handling Windows file locking issues.
+
+    This is a fallback method when git worktree remove fails. It uses shutil.rmtree
+    with custom error handling for Windows-specific issues.
+
+    SAFETY: This function includes multiple safeguards to prevent accidental deletion:
+    - Only deletes directories under the worktree_dirs cache location
+    - Verifies the path is a worktree directory (not the original repo)
+    - Uses resolve() to normalize paths and prevent path traversal attacks
+
+    Args:
+        worktree_dir: Path to the worktree directory to remove
+
+    """
+    if not worktree_dir or not worktree_dir.exists():
+        return
+
+    # Validate paths for safety
+    if not _validate_worktree_path_safety(worktree_dir):
+        return
+
+    # Attempt removal with retries on Windows
+    is_windows = sys.platform == "win32"
+    max_retries = 3 if is_windows else 1
+    retry_delay = 0.5
+
+    for attempt in range(max_retries):
+        attempt_num = attempt + 1
+
+        if attempt_num > 1:
+            time.sleep(retry_delay)
+            retry_delay *= 2  # Exponential backoff
+
+        # On Windows, use custom error handler to remove read-only attributes on the fly
+        # This is more efficient than pre-scanning the entire directory tree
+        error_handler = _create_windows_rmtree_error_handler() if is_windows else None
+
+        try:
+            if is_windows and error_handler:
+                shutil.rmtree(worktree_dir, onerror=error_handler)
+            else:
+                shutil.rmtree(worktree_dir, ignore_errors=True)
+
+            # Brief wait on Windows to allow file handles to be released
+            if is_windows:
+                wait_time = 0.3 if attempt_num < max_retries else 0.1
+                time.sleep(wait_time)
+
+            # Check if removal was successful
+            if not worktree_dir.exists():
+                return
+
+        except Exception:  # noqa: S110
+            pass
+
+
+def _validate_worktree_path_safety(worktree_dir: Path) -> bool:
+    """Validate that a path is safe to delete (must be under worktree_dirs).
+
+    Args:
+        worktree_dir: Path to validate
+
+    Returns:
+        True if the path is safe to delete, False otherwise
+
+    """
+    # SAFETY CHECK 1: Resolve paths to absolute, normalized paths
+    try:
+        worktree_dir_resolved = worktree_dir.resolve()
+        worktree_dirs_resolved = worktree_dirs.resolve()
+    except (OSError, ValueError):
+        return False
+
+    # SAFETY CHECK 2: Ensure worktree_dir is a subdirectory of worktree_dirs
+    try:
+        # Use relative_to to check if path is under worktree_dirs
+        worktree_dir_resolved.relative_to(worktree_dirs_resolved)
+    except ValueError:
+        return False
+
+    # SAFETY CHECK 3: Ensure it's not the worktree_dirs root itself
+    return worktree_dir_resolved != worktree_dirs_resolved
+
+
+def _create_windows_rmtree_error_handler() -> Callable[
+    [Callable[[str], None], str, tuple[type[BaseException], BaseException, Any]], None
+]:
+    """Create an error handler for shutil.rmtree that handles Windows-specific issues.
+
+    This handler attempts to remove read-only attributes when encountering permission errors.
+
+    Returns:
+        A callable error handler for shutil.rmtree's onerror parameter
+
+    """
+
+    def handle_remove_error(
+        func: Callable[[str], None], path: str, exc_info: tuple[type[BaseException], BaseException, Any]
+    ) -> None:
+        """Error handler for shutil.rmtree on Windows.
+
+        Attempts to remove read-only attributes and retry the operation.
+        """
+        # Get the exception type
+        _exc_type, exc_value, _exc_traceback = exc_info
+
+        # Only handle permission errors
+        if not isinstance(exc_value, (PermissionError, OSError)):
+            return
+
+        try:
+            # Try to change file permissions to make it writable
+            # Using permissive mask (0o777) is intentional for Windows file cleanup
+            Path(path).chmod(0o777)
+            # Retry the failed operation
+            func(path)
+        except Exception:  # noqa: S110
+            # If it still fails, silently ignore (file is truly locked)
+            pass
+
+    return handle_remove_error
 
 
 def create_diff_patch_from_worktree(
@@ -110,7 +285,6 @@ def create_diff_patch_from_worktree(
     uni_diff_text = repository.git.diff(None, "HEAD", *files, ignore_blank_lines=True, ignore_space_at_eol=True)
 
     if not uni_diff_text:
-        logger.warning("No changes found in worktree.")
         return None
 
     if not uni_diff_text.endswith("\n"):