Function force_remove_path

ci/util/lock_handler.py:211–289

Force remove a path by killing processes that have it locked. Args: path: Path to remove (file or directory) max_retries: Maximum number of retry attempts Returns: True if path was successfully removed

(path: Path, max_retries: int = 3)

def force_remove_path(path: Path, max_retries: int = 3) -> bool:
    """Force remove a path by killing processes that have it locked.

    Args:
        path: Path to remove (file or directory)
        max_retries: Maximum number of retry attempts

    Returns:
        True if path was successfully removed
    """
    import shutil

    if not path.exists():
        return True

    print(f"Attempting to remove locked path: {path}")

    for attempt in range(max_retries):
        try:
            # Try to remove directly first
            if path.is_dir():
                shutil.rmtree(path)
            else:
                path.unlink()

            print(f"✓ Successfully removed: {path}")
            return True

        except (OSError, PermissionError) as e:
            print(f"Attempt {attempt + 1}/{max_retries}: Failed to remove {path}: {e}")

            if attempt < max_retries - 1:
                # Find and kill locking processes
                print("  Searching for processes locking the path...")
                locking_pids = find_processes_locking_path(path)

                if locking_pids:
                    print(f"  Found {len(locking_pids)} process(es) locking the path")
                    kill_processes(locking_pids, force=True)
                else:
                    print("  No locking processes found")

                    # On Windows, try using system commands as fallback
                    if platform.system() == "Windows":
                        print("  Trying Windows-specific removal...")
                        try:
                            if path.is_dir():
                                result = subprocess.run(
                                    ["cmd", "/c", "rmdir", "/S", "/Q", str(path)],
                                    capture_output=True,
                                    text=True,
                                    timeout=30,
                                )
                            else:
                                result = subprocess.run(
                                    ["cmd", "/c", "del", "/F", "/Q", str(path)],
                                    capture_output=True,
                                    text=True,