mindxtrain/research/guard.py

"""Git editable-allowlist fence + dedup (disciplines 1 and 3).

The fence is what converts the search trail into a sequence of re-checkable git
commits — exactly what `receipt_emit._git_sha` already reads into the manifest. Three
rules, all authoritative (a git failure aborts the attempt, never silently passes):

1. **Commit-before-measure** — the declared editable must be committed; no
   staged/unstaged edits left at measure time.
2. **Editable fence** — nothing in the working tree may be dirty *outside* the
   editable allowlist. The agent touches one file; nothing else moves.
3. **Dedup-by-commit** — refuse to re-measure a commit already scored for this
   researcher (looked up against the ledger).

Pure stdlib; base-install importable. The status parser uses
``--porcelain=v1 -z --untracked-files=all`` and handles rename/copy two-field records
and exotic filenames — the naive ``line[3:].strip()`` approach is wrong on renames,
quoting, untracked-dir collapse, and spaced names.
"""

from __future__ import annotations

import os
import subprocess
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from mindxtrain.research.ledger import Ledger


class ResearchAbort(RuntimeError):
    """Raised when an attempt violates the git fence or dedup rule."""


def _git(*args: str, cwd: Path) -> str:
    """Authoritative git wrapper — RAISES on nonzero/OSError.

    Same subprocess shape as `receipt_emit._git_sha` (capture, text, 5s timeout) but
    raises `ResearchAbort` instead of returning "" — the fence must be trustworthy.
    """
    try:
        out = subprocess.run(
            ["git", *args],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=5.0,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise ResearchAbort(f"git {' '.join(args)} failed: {exc}") from exc
    if out.returncode != 0:
        raise ResearchAbort(
            f"git {' '.join(args)} failed (exit {out.returncode}): {out.stderr.strip()}"
        )
    return out.stdout


def _rel_posix(path: str | Path, root: Path) -> str:
    """Normalise any path to a repo-root-relative POSIX string for comparison.

    Porcelain emits root-relative POSIX already; `editable` may be cwd-relative or
    absolute. Resolving both to the same frame is what stops rule 2 false-positiving
    on every attempt.
    """
    p = Path(path)
    if p.is_absolute():
        p = Path(os.path.relpath(p, root))
    return PurePosixPath(os.path.normpath(p.as_posix())).as_posix()


def _status_entries(root: Path) -> list[tuple[str, str]]:
    """Parse `git status --porcelain=v1 -z --untracked-files=all` into (XY, path).

    Records are NUL-separated. A rename/copy (X or Y in {R, C}) is two NUL fields —
    destination first, then source (the `-z` field order is reversed vs the human
    format); both are yielded so a rename *out of* the allowlist is still caught.
    `--untracked-files=all` expands untracked directories to individual files so the
    fence can match them.
    """
    raw = _git("status", "--porcelain=v1", "-z", "--untracked-files=all", cwd=root)
    tokens = raw.split("\0")
    entries: list[tuple[str, str]] = []
    i, n = 0, len(tokens)
    while i < n:
        tok = tokens[i]
        if len(tok) < 4:  # empty trailing token or malformed; skip defensively
            i += 1
            continue
        xy, path = tok[:2], tok[3:]
        entries.append((xy, path))
        if xy[0] in ("R", "C") or xy[1] in ("R", "C"):
            i += 1
            if i < n and tokens[i]:
                entries.append((xy, tokens[i]))  # rename/copy source path
        i += 1
    return entries


def assert_clean_and_committed(editable: list[Path], root: Path) -> str:
    """Enforce rules 1 + 2; return the short-12 commit hash of HEAD.

    Raises `ResearchAbort` if the working tree is dirty outside the editable
    allowlist (rule 2) or if the editable itself still has uncommitted changes
    (rule 1). On success the tree is clean and the editable's edit lives in HEAD.
    """
    root = Path(root)
    allow = {_rel_posix(p, root) for p in editable}

    stray = sorted({_rel_posix(path, root) for _, path in _status_entries(root)} - allow)
    if stray:
        raise ResearchAbort(f"working tree dirty outside editable allowlist: {stray}")

    editable_dirty = _git(
        "status", "--porcelain", "-z", "--", *[str(p) for p in editable], cwd=root
    )
    if editable_dirty.strip("\0").strip():
        raise ResearchAbort("commit the editable file(s) before measuring")

    return _git("rev-parse", "--short=12", "HEAD", cwd=root).strip()


def assert_new_commit(commit: str, researcher: str, ledger: Ledger) -> None:
    """Dedup gate (rule 3): abort if this researcher already scored `commit`."""
    if ledger.has_commit(researcher, commit):
        raise ResearchAbort(f"commit {commit} already scored for researcher {researcher!r}")


def repo_root(cwd: Path | None = None) -> Path:
    """Absolute path of the enclosing git work tree; raises if not in a repo."""
    out = _git("rev-parse", "--show-toplevel", cwd=cwd or Path.cwd())
    return Path(out.strip())


def git(*args: str, cwd: Path) -> str:
    """Public authoritative git call (raises on failure). Used by the search loop."""
    return _git(*args, cwd=cwd)


__all__ = [
    "ResearchAbort",
    "assert_clean_and_committed",
    "assert_new_commit",
    "git",
    "repo_root",
]