ai-agent/backend/gitdiff.py

"""Git commit + diff extraction for the conversation diff-view.

Three things live here:

* ``list_commits`` — recent commits (with per-commit ``+add/-del`` numstat) for a
  *project*, a *service*, or the whole superproject.
* ``commit_diff`` / ``aggregate_diff`` — the structured, per-file hunk diff of a
  single commit, or the net diff across a set of commits. The output is a
  GitHub-style hunk model (unchanged regions between hunks are *collapsed* — the
  frontend renders the gaps as expandable bars).
* ``conversation_commits`` — discover which commits a *conversation* produced, by
  scanning its tagged project/service repos over the conversation's active time
  window (refined by any still-live worktree branch). The result is cached in the
  DB (``conversation_diffs``) keyed by a fingerprint of the repo HEADs.

All git runs read-only against the mounted repos. A "repo token" identifies a
repo in URLs: ``project:<slug>`` | ``service:<slug>`` | ``super:_``.
"""
from __future__ import annotations

import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path

from projects import PROJECTS_DIR, REPO_DIR, IGNORE_DIRS, ROOT_SLUG

# git's empty-tree object — used as the "before" side when diffing a root commit.
_EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
_US = "\x1f"  # unit separator for --format fields
# Per-file line cap so a single monster file can't blow up a diff response.
_MAX_FILE_LINES = 4000
# How far outside a conversation's [startedAt, endedAt] window to still count a
# commit as "its own" (commits often land a beat after the last transcript line).
_WINDOW_BUFFER = timedelta(minutes=8)


# --------------------------------------------------------------------------- #
# low-level git
# --------------------------------------------------------------------------- #
def _git(args: list[str], cwd: Path, timeout: int = 20) -> str | None:
    cmd = ["git", "-c", "safe.directory=*", "-C", str(cwd), *args]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    except (OSError, subprocess.SubprocessError):
        return None
    return out.stdout if out.returncode == 0 else None


def _safe_slug(slug: str) -> str | None:
    name = Path(slug).name
    if not name or name.startswith(".") or name in IGNORE_DIRS:
        return None
    return name


def resolve_repo(token: str) -> tuple[Path, list[str], str] | None:
    """Map a repo token → ``(cwd, pathspec, label)`` to run git in, or None.

    * ``project:<slug>`` → the project's own ``.git`` (no pathspec), else the
      superproject scoped to ``projects/<slug>``.
    * ``service:<slug>`` → superproject scoped to ``services/<slug>``.
    * ``super:_`` → the whole superproject.
    """
    kind, _, rest = token.partition(":")
    if kind == "super":
        return REPO_DIR, [], "homelab"
    slug = _safe_slug(rest)
    if not slug:
        return None
    if kind == "project":
        # The root project *is* the superproject — no projects/<slug> to scope to.
        if slug == ROOT_SLUG and not (PROJECTS_DIR / slug).is_dir():
            return REPO_DIR, [], slug
        entry = (PROJECTS_DIR / slug).resolve()
        try:
            entry.relative_to(PROJECTS_DIR)
        except ValueError:
            return None
        if (entry / ".git").exists():
            return entry, [], slug
        return REPO_DIR, ["--", f"projects/{slug}"], slug
    if kind == "service":
        return REPO_DIR, ["--", f"services/{slug}"], slug
    return None


# --------------------------------------------------------------------------- #
# commit listing (with numstat)
# --------------------------------------------------------------------------- #
def _parse_numstat_log(out: str | None) -> list[dict]:
    """Parse ``git log --numstat --format=<sha US short US iso US author US subj>``.

    Returns commit dicts with aggregated ``additions``/``deletions``/``files``.
    """
    commits: list[dict] = []
    cur: dict | None = None
    for line in (out or "").splitlines():
        if _US in line:
            parts = line.split(_US)
            if len(parts) >= 5:
                cur = {
                    "sha": parts[0],
                    "short": parts[1],
                    "date": parts[2],
                    "author": parts[3],
                    "subject": _US.join(parts[4:]),
                    "additions": 0,
                    "deletions": 0,
                    "files": 0,
                }
                commits.append(cur)
                continue
        if cur is None:
            continue
        cols = line.split("\t")
        if len(cols) == 3:
            add, dele, _path = cols
            cur["files"] += 1
            if add.isdigit():
                cur["additions"] += int(add)
            if dele.isdigit():
                cur["deletions"] += int(dele)
    return commits


def list_commits(
    token: str,
    *,
    limit: int = 30,
    since: str | None = None,
    until: str | None = None,
    revs: list[str] | None = None,
) -> list[dict]:
    """Recent commits for a repo token, each tagged with that ``repo`` token."""
    r = resolve_repo(token)
    if not r:
        return []
    cwd, pathspec, _label = r
    fmt = _US.join(["%H", "%h", "%cI", "%an", "%s"])
    args = ["log", f"--max-count={limit}", f"--format={fmt}", "--numstat"]
    if since:
        args.append(f"--since={since}")
    if until:
        args.append(f"--until={until}")
    if revs:
        args += revs
    args += pathspec
    commits = _parse_numstat_log(_git(args, cwd))
    for c in commits:
        c["repo"] = token
    return commits


# --------------------------------------------------------------------------- #
# unified-diff → structured hunk model
# --------------------------------------------------------------------------- #
def _parse_unified_diff(text: str) -> list[dict]:
    """Parse ``git show``/``git diff`` output into per-file hunk models."""
    files: list[dict] = []
    cur: dict | None = None
    hunk: dict | None = None
    old_no = new_no = 0

    def flush_hunk():
        nonlocal hunk
        hunk = None

    for raw in (text or "").splitlines():
        if raw.startswith("diff --git "):
            cur = {
                "path": "", "oldPath": None, "status": "modified",
                "additions": 0, "deletions": 0, "binary": False,
                "truncated": False, "hunks": [],
            }
            files.append(cur)
            flush_hunk()
            # a/<old> b/<new> — a decent default before ---/+++ refine it
            try:
                a, b = raw[len("diff --git "):].split(" b/", 1)
                cur["path"] = b.strip()
                cur["oldPath"] = a[2:].strip() if a.startswith("a/") else a.strip()
            except ValueError:
                pass
            continue
        if cur is None:
            continue
        if raw.startswith("new file mode"):
            cur["status"] = "added"
            continue
        if raw.startswith("deleted file mode"):
            cur["status"] = "deleted"
            continue
        if raw.startswith("rename from "):
            cur["status"] = "renamed"
            cur["oldPath"] = raw[len("rename from "):].strip()
            continue
        if raw.startswith("rename to "):
            cur["status"] = "renamed"
            cur["path"] = raw[len("rename to "):].strip()
            continue
        if raw.startswith("Binary files"):
            cur["binary"] = True
            continue
        if raw.startswith("--- "):
            p = raw[4:].strip()
            if p != "/dev/null":
                cur["oldPath"] = p[2:] if p.startswith("a/") else p
            continue
        if raw.startswith("+++ "):
            p = raw[4:].strip()
            if p != "/dev/null":
                cur["path"] = p[2:] if p.startswith("b/") else p
            continue
        if raw.startswith("@@"):
            # @@ -oldStart,oldLines +newStart,newLines @@ section heading
            try:
                spec = raw.split("@@")[1].strip()
                old_part, new_part = spec.split(" ")
                os_, ol = (old_part[1:].split(",") + ["1"])[:2]
                ns_, nl = (new_part[1:].split(",") + ["1"])[:2]
                old_no, new_no = int(os_), int(ns_)
                hunk = {
                    "header": raw.strip(),
                    "oldStart": old_no, "oldLines": int(ol),
                    "newStart": new_no, "newLines": int(nl),
                    "lines": [],
                }
                cur["hunks"].append(hunk)
            except (ValueError, IndexError):
                hunk = None
            continue
        if hunk is None:
            continue
        if len(cur["hunks"]) and len(hunk["lines"]) >= _MAX_FILE_LINES:
            cur["truncated"] = True
            continue
        tag = raw[:1]
        body = raw[1:]
        if tag == "+":
            hunk["lines"].append({"type": "add", "oldNo": None, "newNo": new_no, "text": body})
            cur["additions"] += 1
            new_no += 1
        elif tag == "-":
            hunk["lines"].append({"type": "del", "oldNo": old_no, "newNo": None, "text": body})
            cur["deletions"] += 1
            old_no += 1
        elif tag == "\\":
            continue  # "\ No newline at end of file"
        else:  # context (leading space, or an empty line)
            hunk["lines"].append({"type": "ctx", "oldNo": old_no, "newNo": new_no, "text": body})
            old_no += 1
            new_no += 1
    return files


# --------------------------------------------------------------------------- #
# single-commit + aggregate diffs
# --------------------------------------------------------------------------- #
def _commit_meta(cwd: Path, sha: str) -> dict | None:
    fmt = _US.join(["%H", "%h", "%cI", "%an", "%s"])
    out = _git(["show", "-s", f"--format={fmt}", sha], cwd)
    if not out:
        return None
    parts = out.strip().split(_US)
    if len(parts) < 5:
        return None
    return {"sha": parts[0], "short": parts[1], "date": parts[2],
            "author": parts[3], "subject": _US.join(parts[4:])}


def commit_diff(token: str, sha: str) -> dict | None:
    """Structured per-file hunk diff of one commit within a repo token."""
    r = resolve_repo(token)
    if not r:
        return None
    cwd, pathspec, label = r
    if not _valid_rev(sha):
        return None
    meta = _commit_meta(cwd, sha)
    if not meta:
        return None
    patch = _git(["show", "--format=", "--no-color", "-U3", "-M", sha, *pathspec], cwd)
    files = _parse_unified_diff(patch or "")
    for f in files:
        f["repo"] = token
    add = sum(f["additions"] for f in files)
    dele = sum(f["deletions"] for f in files)
    ref = dict(meta, additions=add, deletions=dele, files=len(files), repo=token)
    return {
        "title": meta["subject"],
        "subtitle": f"{meta['short']} · {label}",
        "files": files, "additions": add, "deletions": dele,
        "commits": [ref], "approx": False,
    }


def _valid_rev(rev: str) -> bool:
    return bool(rev) and all(c.isalnum() or c in "._-^~/" for c in rev) and len(rev) <= 80


# Cap a blob at ~12 MB so a huge tracked binary can't blow up the response.
_MAX_BLOB_BYTES = 12 * 1024 * 1024


def _git_bytes(args: list[str], cwd: Path, timeout: int = 20) -> bytes | None:
    """Like ``_git`` but returns raw stdout bytes (for binary blobs)."""
    cmd = ["git", "-c", "safe.directory=*", "-C", str(cwd), *args]
    try:
        out = subprocess.run(cmd, capture_output=True, timeout=timeout)
    except (OSError, subprocess.SubprocessError):
        return None
    return out.stdout if out.returncode == 0 else None


def blob_bytes(token: str, path: str, *, sha: str | None = None) -> bytes | None:
    """Raw bytes of ``path`` within a repo token.

    With ``sha`` → the file as it was at that rev (``git show <rev>:<path>``);
    without → the current working-tree file. ``path`` is repo-relative to the
    token's git cwd (the same paths the diff model carries). Returns None if the
    file is missing, escapes the repo, or is larger than ``_MAX_BLOB_BYTES``.
    """
    r = resolve_repo(token)
    if not r:
        return None
    cwd, _pathspec, _label = r
    # Reject path traversal / absolute paths — everything must stay under cwd.
    rel = Path(path)
    if rel.is_absolute() or ".." in rel.parts or not path:
        return None
    if sha is not None:
        if not _valid_rev(sha):
            return None
        data = _git_bytes(["show", f"{sha}:{path}"], cwd)
        if data is None:
            return None
    else:
        target = (cwd / rel).resolve()
        try:
            target.relative_to(cwd.resolve())
        except ValueError:
            return None
        if not target.is_file():
            return None
        try:
            data = target.read_bytes()
        except OSError:
            return None
    if len(data) > _MAX_BLOB_BYTES:
        return None
    return data


def _parent_rev(cwd: Path, sha: str) -> str:
    """``<sha>^`` if it has a parent, else git's empty tree (root commit)."""
    if _git(["rev-parse", "--verify", "--quiet", f"{sha}^"], cwd):
        return f"{sha}^"
    return _EMPTY_TREE


def aggregate_diff(token: str, shas: list[str], *, title: str, subtitle: str) -> dict | None:
    """Net diff across a contiguous run of commits in ONE repo (oldest → newest).

    ``shas`` must be ordered newest-first (as ``list_commits`` returns). We diff
    ``<oldest>^..<newest>`` scoped to the repo's pathspec.
    """
    r = resolve_repo(token)
    if not r or not shas:
        return None
    cwd, pathspec, _label = r
    ordered = [s for s in shas if _valid_rev(s)]
    if not ordered:
        return None
    newest, oldest = ordered[0], ordered[-1]
    base = _parent_rev(cwd, oldest)
    patch = _git(["diff", "--no-color", "-U3", "-M", f"{base}..{newest}", *pathspec], cwd)
    files = _parse_unified_diff(patch or "")
    for f in files:
        f["repo"] = token
    add = sum(f["additions"] for f in files)
    dele = sum(f["deletions"] for f in files)
    return {"title": title, "subtitle": subtitle, "files": files,
            "additions": add, "deletions": dele, "approx": True}


# --------------------------------------------------------------------------- #
# conversation → commits (the "diff cache")
# --------------------------------------------------------------------------- #
def _iso_shift(iso: str | None, delta: timedelta) -> str | None:
    if not iso:
        return None
    try:
        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
    except ValueError:
        return None
    return (dt.astimezone(timezone.utc) + delta).isoformat()


def _repo_head(token: str) -> str:
    r = resolve_repo(token)
    if not r:
        return "?"
    cwd, _pathspec, _label = r
    return (_git(["rev-parse", "HEAD"], cwd) or "?").strip()


def _conversation_repos(summary: dict, meta: dict) -> tuple[list[str], list[str], bool]:
    """(project tokens, service slugs, include-superproject) for a conversation."""
    projects = sorted({*(summary.get("projectsAuto") or []), *(meta.get("projects") or [])})
    services = sorted({*(summary.get("servicesAuto") or []), *(meta.get("services") or [])})
    projects = [p for p in (_safe_slug(p) for p in projects) if p]
    services = [s for s in (_safe_slug(s) for s in services) if s]
    worktrees = summary.get("worktreesAuto") or meta.get("worktrees") or []
    committed = bool(meta.get("committed")) or bool((summary.get("lifecycleAuto") or {}).get("committed"))
    # A conversation with a worktree/commit but no tagged project/service likely
    # edited repo-root files (scripts, Makefile, docs) → scan the whole super repo.
    include_super = bool(services) or (not projects and (bool(worktrees) or committed))
    return [f"project:{p}" for p in projects], services, include_super


def _worktree_branches(summary: dict, meta: dict) -> list[str]:
    out: list[str] = []
    for wt in (summary.get("worktreesAuto") or []) + (meta.get("worktrees") or []):
        name = (wt or {}).get("name")
        if name and name not in out and _valid_rev(name):
            out.append(name)
    return out


def compute_conversation_commits(summary: dict, meta: dict) -> dict:
    """Discover a conversation's commits across its repos (uncached)."""
    started = summary.get("startedAt")
    ended = summary.get("endedAt")
    since = _iso_shift(started, -_WINDOW_BUFFER)
    until = _iso_shift(ended, _WINDOW_BUFFER)
    project_tokens, services, include_super = _conversation_repos(summary, meta)
    branches = _worktree_branches(summary, meta)

    seen: set[str] = set()
    commits: list[dict] = []

    def collect(found: list[dict]):
        for c in found:
            if c["sha"] not in seen:
                seen.add(c["sha"])
                commits.append(c)

    for token in project_tokens:
        collect(list_commits(token, since=since, until=until, limit=50))
        for br in branches:
            # commits unique to a still-live worktree branch (survives after merge)
            collect(list_commits(token, revs=[f"main..{br}"], limit=50))

    if include_super:
        # one super-repo query scoped to the tagged services (or the whole repo)
        r = resolve_repo("super:_")
        if r:
            cwd, _ps, _lbl = r
            fmt = _US.join(["%H", "%h", "%cI", "%an", "%s"])
            args = ["log", "--max-count=50", f"--format={fmt}", "--numstat"]
            if since:
                args.append(f"--since={since}")
            if until:
                args.append(f"--until={until}")
            if services:
                args += ["--"] + [f"services/{s}" for s in services]
            found = _parse_numstat_log(_git(args, cwd))
            for c in found:
                c["repo"] = "super:_"
            collect(found)

    commits.sort(key=lambda c: c["date"], reverse=True)
    return {
        "commits": commits,
        "repos": [*project_tokens, *(["super:_"] if include_super else [])],
        "filesChanged": sum(c["files"] for c in commits),
        "additions": sum(c["additions"] for c in commits),
        "deletions": sum(c["deletions"] for c in commits),
    }


def _fingerprint(summary: dict, meta: dict, repos: list[str]) -> str:
    heads = ",".join(f"{t}={_repo_head(t)}" for t in repos)
    return f"{summary.get('endedAt')}|{heads}"


def conversation_commits(store, summary: dict, meta: dict, session_id: str) -> dict:
    """Cached conversation-commits: recompute only when a repo HEAD moved."""
    project_tokens, _services, include_super = _conversation_repos(summary, meta)
    repos = [*project_tokens, *(["super:_"] if include_super else [])]
    fp = _fingerprint(summary, meta, repos)
    cached = store.get_conversation_diff(session_id) if session_id else None
    if cached and cached.get("fingerprint") == fp:
        return cached["data"]
    data = compute_conversation_commits(summary, meta)
    if session_id:
        store.set_conversation_diff(session_id, fp, data)
    return data