Fusion
063a839790
- Add clone_mode and branch fields to tool_instances - Add ssh_key_id to git_repositories for per-repo SSH key assignment - Implement host-side git cloning with branch selection (default: main) - Mount SSH keys into containers for git operations in clone mode - Add dirty state check on clone-mode instance deletion with confirmation - Update SessionsPage with mount/clone selector, branch input, SSH key display - Add SSH key selector to repository creation form - Add dirty delete confirmation modal with changed files list - Update API schemas and endpoints for new fields - Sync delta specs to main specs (git-repo, tool-instances, repo-clone-mode) - Archive completed OpenSpec change: repo-clone-mode-with-ssh - Document git requirement for custom tool types Quality gates: Frontend typecheck and build passed OpenSpec: repo-clone-mode-with-ssh archived with all tasks complete
98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
"""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.info("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.info("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.info("Removed clone directory: %s", clone_path)
|