core/git_operations.py

"""
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-<timestamp>``).
* **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 <base_sha>`` 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": <git.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-<timestamp>`` 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: <task_name>``. 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-<ts>`` off that baseline.
        4. Commit staged changes with ``AI-Patch: <task_name>``.
        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",
]