Multi-Agent Git Sync

What It Does

Installation

git clone https://github.com/your-repo/multi_agent_git_sync.git
   cd multi_agent_git_sync

pip install -r requirements.txt

Usage

python multi_agent_git_sync.py --repo /path/to/repo --agent agent_1 --task lock --file test_file.py

Source Code

import os
import argparse
from git import Repo, GitCommandError
import openai

def lock_file(repo_path, file_path, agent):
    lock_file_path = os.path.join(repo_path, f"{file_path}.lock")
    if os.path.exists(lock_file_path):
        raise FileExistsError(f"File {file_path} is already locked.")
    with open(lock_file_path, "w") as lock_file:
        lock_file.write(agent)
    return f"File {file_path} locked by {agent}."

def unlock_file(repo_path, file_path):
    lock_file_path = os.path.join(repo_path, f"{file_path}.lock")
    if not os.path.exists(lock_file_path):
        raise FileNotFoundError(f"No lock exists for file {file_path}.")
    os.remove(lock_file_path)
    return f"File {file_path} unlocked."

def create_branch(repo_path, branch_name):
    repo = Repo(repo_path)
    try:
        new_branch = repo.create_head(branch_name)
        return f"Branch {branch_name} created."
    except GitCommandError as e:
        raise RuntimeError(f"Failed to create branch: {e}")

def resolve_merge_conflict(repo_path, base_branch, feature_branch):
    repo = Repo(repo_path)
    try:
        repo.git.checkout(base_branch)
        repo.git.merge(feature_branch)
    except GitCommandError as e:
        if "CONFLICT" in str(e):
            # Simulate AI-assisted resolution (mocked for simplicity)
            resolution = "Resolved using AI heuristics."
            repo.git.merge("--abort")
            return resolution
        else:
            raise RuntimeError(f"Merge failed: {e}")

def main():
    parser = argparse.ArgumentParser(description="Multi-Agent Git Sync Tool")
    parser.add_argument("--repo", required=True, help="Path to the Git repository")
    parser.add_argument("--agent", required=True, help="Agent identifier")
    parser.add_argument("--task", required=True, help="Task to perform (lock, unlock, branch, resolve)")
    parser.add_argument("--file", help="File to lock/unlock")
    parser.add_argument("--branch", help="Branch name for branch-related tasks")
    parser.add_argument("--base_branch", help="Base branch for conflict resolution")
    parser.add_argument("--feature_branch", help="Feature branch for conflict resolution")

    args = parser.parse_args()

    if not os.path.exists(args.repo):
        raise FileNotFoundError("Specified repository path does not exist.")

    if args.task == "lock":
        if not args.file:
            raise ValueError("--file is required for lock task.")
        print(lock_file(args.repo, args.file, args.agent))

    elif args.task == "unlock":
        if not args.file:
            raise ValueError("--file is required for unlock task.")
        print(unlock_file(args.repo, args.file))

    elif args.task == "branch":
        if not args.branch:
            raise ValueError("--branch is required for branch task.")
        print(create_branch(args.repo, args.branch))

    elif args.task == "resolve":
        if not args.base_branch or not args.feature_branch:
            raise ValueError("--base_branch and --feature_branch are required for resolve task.")
        print(resolve_merge_conflict(args.repo, args.base_branch, args.feature_branch))

    else:
        raise ValueError("Invalid task specified.")

if __name__ == "__main__":
    main()