"""Clone service for repository cloning and dirty state checking.""" import logging import os import subprocess from pathlib import Path logger = logging.getLogger(__name__) def clone_repository( remote_url: str, ssh_key_path: str | None, instance_dir: str, branch: str = "main", ) -> str: """Clone a git repository into the instance directory. Args: remote_url: Git remote URL (SSH or HTTPS) ssh_key_path: Path to SSH private key for authentication (optional) instance_dir: Path to instance directory branch: Branch to clone (default: main) Returns: Path to the cloned repository """ clone_path = Path(instance_dir) / "repo-clone" clone_path.mkdir(parents=True, exist_ok=True) env = os.environ.copy() if ssh_key_path: # Use SSH key for cloning env["GIT_SSH_COMMAND"] = f"ssh -i {ssh_key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null" cmd = [ "git", "clone", "--branch", branch, "--single-branch", remote_url, str(clone_path), ] logger.debug("Cloning repository %s (branch: %s) into %s", remote_url, branch, clone_path) result = subprocess.run( cmd, capture_output=True, text=True, env=env, timeout=300, ) if result.returncode != 0: logger.error("Git clone failed: %s", result.stderr) raise RuntimeError(f"Failed to clone repository: {result.stderr}") logger.debug("Successfully cloned repository into %s", clone_path) return str(clone_path) def check_dirty_state(clone_path: str) -> tuple[bool, list[str]]: """Check for uncommitted changes in a cloned repository. Args: clone_path: Path to the cloned repository Returns: Tuple of (is_dirty, list_of_changed_files) """ result = subprocess.run( ["git", "-C", clone_path, "status", "--short"], capture_output=True, text=True, ) if result.returncode != 0: logger.warning("Failed to check git status: %s", result.stderr) return False, [] changed_files = [line.strip() for line in result.stdout.split("\n") if line.strip()] is_dirty = len(changed_files) > 0 return is_dirty, changed_files def remove_clone_directory(instance_dir: str) -> None: """Remove the cloned repository from the instance directory. Args: instance_dir: Path to instance directory """ clone_path = Path(instance_dir) / "repo-clone" if clone_path.exists(): import shutil shutil.rmtree(clone_path) logger.debug("Removed clone directory: %s", clone_path)