""" core/git_operations.py — Git-Native Operations layer (v4.7). Every successful patch is committed to a temporary local branch so we can inspect what changed, and — critically — instantly ``rollback`` if the subsequent test / lint stages fail. Design decisions ================ * **Zero global state** — every call takes a ``project_root`` and returns a self-describing dict. The pipeline can attach the dict to ``state`` and the UI can render it directly. * **Optional dependency** — ``GitPython`` is imported lazily so the rest of DevAI Studio still boots on machines without git. All functions degrade to ``{"ok": False, "reason": "..."}`` when git or GitPython is missing. * **Read-only origin** — we NEVER touch ``origin/*`` or push. All work happens on a local, sandboxed branch (default: ``devai/patch-``). * **Safety net** — the caller can always find the pre-patch commit SHA at ``state["git_ops"]["base_sha"]`` for a manual ``git reset --hard``. Public API ---------- * :func:`is_git_available` — quick health check. * :func:`init_or_open_repo` — open existing repo or ``git init``. * :func:`create_patch_branch` — create + checkout ``devai/patch-*``. * :func:`commit_patch` — stage all + commit with AI-Patch message; returns commit SHA. * :func:`rollback_to_base` — ``git reset --hard `` on the current branch (safe: only on the patch branch). * :func:`git_status_summary` — human-friendly status snapshot. * :func:`run_git_patch_workflow` — one-shot helper the pipeline uses. None of the helpers raise on git errors; they all return structured dicts so the calling pipeline stage can log-and-continue cleanly. """ from __future__ import annotations import os import shutil import subprocess from datetime import datetime from typing import Any, Dict, List, Optional DEFAULT_AUTHOR_NAME = "DevAI Studio" DEFAULT_AUTHOR_EMAIL = "devai-studio@localhost" BRANCH_PREFIX = "devai/patch-" # --------------------------------------------------------------------------- # Environment probing # --------------------------------------------------------------------------- def _has_git_binary() -> bool: return shutil.which("git") is not None def _import_gitpython(): # -> module or None """Import GitPython lazily; return module or None on failure.""" try: import git # type: ignore return git except Exception: return None def is_git_available() -> Dict[str, Any]: """Return a diagnostic dict — safe to expose in ``system_info``.""" git_mod = _import_gitpython() return { "git_binary": _has_git_binary(), "gitpython_installed": git_mod is not None, "gitpython_version": getattr(git_mod, "__version__", None) if git_mod else None, "available": bool(_has_git_binary() and git_mod is not None), } # --------------------------------------------------------------------------- # Low-level helpers # --------------------------------------------------------------------------- def _fallback_run(project_root: str, args: List[str]) -> Dict[str, Any]: """Last-resort raw-git shell call (used when GitPython is missing).""" if not _has_git_binary(): return {"ok": False, "reason": "git binary not found"} try: out = subprocess.run( ["git"] + args, cwd=project_root, capture_output=True, text=True, timeout=30, ) return { "ok": out.returncode == 0, "stdout": out.stdout.strip(), "stderr": out.stderr.strip(), "returncode": out.returncode, } except Exception as exc: return {"ok": False, "reason": f"subprocess error: {exc}"} def _ensure_author_config(repo) -> None: """Guarantee user.name/user.email are set so commits don't blow up.""" try: with repo.config_writer() as cw: try: repo.config_reader().get_value("user", "name") except Exception: cw.set_value("user", "name", DEFAULT_AUTHOR_NAME) try: repo.config_reader().get_value("user", "email") except Exception: cw.set_value("user", "email", DEFAULT_AUTHOR_EMAIL) except Exception: # Non-fatal — commits may still fail, and the caller will surface it. pass # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def init_or_open_repo(project_root: str) -> Dict[str, Any]: """ Return ``{"ok": True, "repo": , "initialised": bool, "head": "..."}`` on success, or ``{"ok": False, "reason": "..."}`` otherwise. * If ``project_root`` is already a git repo, we open it. * Otherwise, we run ``git init`` and stage-commit the current tree as the pre-patch baseline (so ``rollback_to_base`` has something to revert to). """ git_mod = _import_gitpython() if git_mod is None or not _has_git_binary(): return { "ok": False, "reason": "GitPython or git binary not installed", "gitpython": git_mod is not None, "git_binary": _has_git_binary(), } if not project_root or not os.path.isdir(project_root): return {"ok": False, "reason": f"project_root does not exist: {project_root}"} try: # Try opening as an existing repo first. try: repo = git_mod.Repo(project_root, search_parent_directories=False) initialised = False except Exception: # ``git init`` + baseline commit. repo = git_mod.Repo.init(project_root) initialised = True _ensure_author_config(repo) try: repo.git.add(A=True) if repo.is_dirty(untracked_files=True): repo.index.commit("DevAI Studio: pre-patch baseline snapshot") except Exception as exc: # An empty repo with no files is still fine — record it. return { "ok": True, "repo": repo, "initialised": True, "head": None, "warning": f"baseline commit skipped: {exc}", } _ensure_author_config(repo) head_sha = None try: head_sha = repo.head.commit.hexsha except Exception: # Repo with no commits yet. try: repo.git.add(A=True) if repo.is_dirty(untracked_files=True): repo.index.commit("DevAI Studio: pre-patch baseline snapshot") head_sha = repo.head.commit.hexsha except Exception: head_sha = None return { "ok": True, "repo": repo, "initialised": initialised, "head": head_sha, } except Exception as exc: return {"ok": False, "reason": f"init/open failed: {exc}"} def create_patch_branch( repo, base_sha: Optional[str] = None, branch_name: Optional[str] = None, ) -> Dict[str, Any]: """Create + checkout ``devai/patch-`` off ``base_sha``.""" if repo is None: return {"ok": False, "reason": "repo is None"} try: _ensure_author_config(repo) branch = branch_name or f"{BRANCH_PREFIX}{datetime.now().strftime('%Y%m%dT%H%M%S')}" # If base_sha is missing, fall back to current HEAD. try: starting_ref = base_sha or repo.head.commit.hexsha except Exception: starting_ref = None if starting_ref: new_branch = repo.create_head(branch, starting_ref) else: new_branch = repo.create_head(branch) new_branch.checkout() return { "ok": True, "branch": branch, "base_sha": starting_ref, } except Exception as exc: return {"ok": False, "reason": f"branch create failed: {exc}"} def commit_patch( repo, task_name: str, modified_files: Optional[List[str]] = None, ) -> Dict[str, Any]: """ Stage everything and commit with a message like ``AI-Patch: ``. Returns the commit SHA. """ if repo is None: return {"ok": False, "reason": "repo is None"} label = (task_name or "unnamed task").strip()[:120] try: _ensure_author_config(repo) repo.git.add(A=True) # Nothing changed? Return early — the caller can decide. if not repo.is_dirty(untracked_files=True): return { "ok": True, "skipped": True, "reason": "no changes to commit", "sha": repo.head.commit.hexsha, } commit_msg = f"AI-Patch: {label}" if modified_files: preview = ", ".join(sorted(modified_files)[:6]) more = f" (+{len(modified_files) - 6} more)" if len(modified_files) > 6 else "" commit_msg += f"\n\nFiles: {preview}{more}" commit = repo.index.commit(commit_msg) return { "ok": True, "skipped": False, "sha": commit.hexsha, "message": commit_msg.splitlines()[0], } except Exception as exc: return {"ok": False, "reason": f"commit failed: {exc}"} def rollback_to_base(repo, base_sha: str) -> Dict[str, Any]: """ Hard-reset the current branch to ``base_sha`` — used when tests fail and the user hits the Git Rollback button. """ if repo is None: return {"ok": False, "reason": "repo is None"} if not base_sha: return {"ok": False, "reason": "base_sha not provided"} try: current_branch = None try: current_branch = repo.active_branch.name except Exception: pass # Refuse to rollback anything that doesn't look like our own branch. if current_branch and not current_branch.startswith(BRANCH_PREFIX): # Still allow it if the caller explicitly opts in via env. if os.environ.get("DEVAI_ALLOW_ROLLBACK_ANY_BRANCH", "").lower() not in ("1", "true", "yes"): return { "ok": False, "reason": ( f"refusing to rollback branch '{current_branch}' — " f"only branches starting with '{BRANCH_PREFIX}' are auto-reset. " "Set DEVAI_ALLOW_ROLLBACK_ANY_BRANCH=1 to override." ), } repo.git.reset("--hard", base_sha) # Clean untracked cruft too so the tree matches the snapshot exactly. try: repo.git.clean("-fd") except Exception: pass return { "ok": True, "branch": current_branch, "reset_to": base_sha, "new_head": repo.head.commit.hexsha, } except Exception as exc: return {"ok": False, "reason": f"rollback failed: {exc}"} def git_status_summary(repo) -> Dict[str, Any]: """Compact snapshot for the UI badge / timeline.""" if repo is None: return {"ok": False, "reason": "repo is None"} try: head_sha = None try: head_sha = repo.head.commit.hexsha[:8] except Exception: pass branch = None try: branch = repo.active_branch.name except Exception: pass dirty = False try: dirty = repo.is_dirty(untracked_files=True) except Exception: pass return { "ok": True, "branch": branch, "head_sha": head_sha, "dirty": dirty, } except Exception as exc: return {"ok": False, "reason": str(exc)} # --------------------------------------------------------------------------- # One-shot workflow used by the supervisor / run_pipeline # --------------------------------------------------------------------------- def run_git_patch_workflow( project_root: str, task_name: str, modified_files: Optional[List[str]] = None, ) -> Dict[str, Any]: """ End-to-end helper: 1. Open (or init) the repo. 2. Record baseline SHA. 3. Create ``devai/patch-`` off that baseline. 4. Commit staged changes with ``AI-Patch: ``. 5. Return everything the UI + rollback flow needs. Never raises — all failures come back as ``{"ok": False, "reason": ...}``. """ result: Dict[str, Any] = { "ok": False, "available": False, "project_root": project_root, "task_name": task_name, "modified_files": list(modified_files or []), "created_branch": None, "base_sha": None, "patch_sha": None, "commit_message": None, "reason": None, } availability = is_git_available() result["available"] = availability["available"] if not availability["available"]: result["reason"] = "GitPython or git binary not installed" result["availability"] = availability return result open_res = init_or_open_repo(project_root) if not open_res.get("ok"): result["reason"] = open_res.get("reason") return result repo = open_res["repo"] baseline = open_res.get("head") branch_res = create_patch_branch(repo, base_sha=baseline) if not branch_res.get("ok"): result["reason"] = branch_res.get("reason") return result result["created_branch"] = branch_res["branch"] result["base_sha"] = branch_res.get("base_sha") commit_res = commit_patch(repo, task_name=task_name, modified_files=modified_files) if not commit_res.get("ok"): result["reason"] = commit_res.get("reason") return result result["ok"] = True result["patch_sha"] = commit_res.get("sha") result["commit_message"] = commit_res.get("message") result["skipped_empty_commit"] = bool(commit_res.get("skipped")) result["status"] = git_status_summary(repo) return result def rollback_project(project_root: str, base_sha: str) -> Dict[str, Any]: """Public rollback used by the UI button.""" if not _has_git_binary() or _import_gitpython() is None: # Fallback: raw ``git reset`` via shell. return _fallback_run(project_root, ["reset", "--hard", base_sha]) open_res = init_or_open_repo(project_root) if not open_res.get("ok"): return open_res return rollback_to_base(open_res["repo"], base_sha) __all__ = [ "BRANCH_PREFIX", "is_git_available", "init_or_open_repo", "create_patch_branch", "commit_patch", "rollback_to_base", "rollback_project", "git_status_summary", "run_git_patch_workflow", ]