File size: 5,535 Bytes
c27193b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """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",
]
|