repo-manager/src/repo_manager/provenance.py

"""Install and report on coding-assistant commit provenance."""

from __future__ import annotations

import re
import subprocess
from collections import defaultdict
from pathlib import Path
from typing import Any

import yaml

_TRAILER_RE = re.compile(r"^(Assistant(?:-Model|-Process|-Session)?):\s*(.+)$", re.MULTILINE)


def install_hook(hooks_path: Path) -> dict[str, Any]:
    """Configure the user's Git installation to use the governed hook directory."""
    hooks_path = hooks_path.expanduser().resolve()
    hook = hooks_path / "prepare-commit-msg"
    if not hook.is_file():
        return {"ok": False, "error": f"missing hook: {hook}"}
    if not hook.stat().st_mode & 0o111:
        return {"ok": False, "error": f"hook is not executable: {hook}"}
    completed = subprocess.run(
        ["git", "config", "--global", "core.hooksPath", str(hooks_path)],
        check=False,
        capture_output=True,
        text=True,
    )
    if completed.returncode:
        return {"ok": False, "error": completed.stderr.strip() or "git config failed"}
    return {"ok": True, "hooks_path": str(hooks_path), "hook": str(hook)}


def _cutover(repo_root: Path) -> str | None:
    path = repo_root / "config" / "assistant-provenance.yaml"
    if not path.is_file():
        return None
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    value = data.get("cutover_commit") if isinstance(data, dict) else None
    return str(value) if value else None


def _commits(repo_root: Path, rev: str, max_count: int | None) -> list[dict[str, Any]]:
    command = ["git", "log", "--reverse", "--format=%H%x1f%aI%x1f%an%x1f%ae%x1f%B%x1e"]
    if max_count:
        command.append(f"--max-count={max_count}")
    command.append(rev)
    completed = subprocess.run(command, cwd=repo_root, check=True, capture_output=True, text=True)
    commits: list[dict[str, Any]] = []
    for raw in completed.stdout.split("\x1e"):
        raw = raw.strip("\n")
        if not raw:
            continue
        parts = raw.split("\x1f", 4)
        if len(parts) != 5:
            continue
        sha, authored_at, author, email, body = parts
        trailers = {key: value.strip() for key, value in _TRAILER_RE.findall(body)}
        commits.append(
            {
                "sha": sha,
                "authored_at": authored_at,
                "author": author,
                "email": email,
                "assistant": trailers.get("Assistant"),
                "model": trailers.get("Assistant-Model"),
                "process": trailers.get("Assistant-Process"),
                "session": trailers.get("Assistant-Session"),
            }
        )
    return commits


def assistant_report(repo_root: Path, *, rev: str = "HEAD", max_count: int | None = None) -> dict[str, Any]:
    """Derive assistant activity and interleaved-session signals from Git alone."""
    repo_root = repo_root.resolve()
    commits = _commits(repo_root, rev, max_count)
    cutover = _cutover(repo_root)
    cutover_seen = False
    assistants: dict[str, dict[str, Any]] = defaultdict(
        lambda: {"commit_count": 0, "models": set(), "sessions": set(), "processes": set()}
    )
    unattributed = {"before_cutover": 0, "after_cutover": 0, "known_automation": 0}
    session_sequence: list[str] = []

    for commit in commits:
        if cutover and commit["sha"].startswith(cutover):
            cutover_seen = True
        assistant = commit["assistant"]
        if assistant:
            entry = assistants[assistant]
            entry["commit_count"] += 1
            for field, bucket in (("model", "models"), ("session", "sessions"), ("process", "processes")):
                if commit[field]:
                    entry[bucket].add(commit[field])
            if commit["session"]:
                session_sequence.append(commit["session"])
        elif commit["author"] == "custodian-sync" or commit["email"] == "custodian-sync@railiance.local":
            unattributed["known_automation"] += 1
        elif cutover_seen:
            unattributed["after_cutover"] += 1
        else:
            unattributed["before_cutover"] += 1

    interleaved: set[tuple[str, str]] = set()
    positions: dict[str, list[int]] = defaultdict(list)
    for position, session in enumerate(session_sequence):
        positions[session].append(position)
    sessions = sorted(positions)
    for index, first in enumerate(sessions):
        for second in sessions[index + 1 :]:
            merged = [session_sequence[pos] for pos in sorted(positions[first] + positions[second])]
            compressed = [value for pos, value in enumerate(merged) if pos == 0 or value != merged[pos - 1]]
            if len(compressed) >= 3:
                interleaved.add((first, second))

    return {
        "ok": True,
        "repo": str(repo_root),
        "revision": rev,
        "cutover_commit": cutover,
        "commit_count": len(commits),
        "assistants": {
            name: {
                "commit_count": item["commit_count"],
                "models": sorted(item["models"]),
                "sessions": sorted(item["sessions"]),
                "processes": sorted(item["processes"]),
            }
            for name, item in sorted(assistants.items())
        },
        "interleaved_sessions": [list(pair) for pair in sorted(interleaved)],
        "unattributed": unattributed,
    }