145 lines
5.8 KiB
Python
145 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Deterministic diff signals for the review gauntlet.
|
|
|
|
Pure git — no LLM judgment. Every signal is computed from the changed-file
|
|
list and the added lines of the diff. Failures never sink the gauntlet:
|
|
on any error the script reports `signals_error` and marks the signals
|
|
unavailable — downstream (build_items / default_lanes) treats that as
|
|
UNKNOWN surface and degrades WIDER (security on; probe when a recipe
|
|
exists), on top of caller context and forced lanes.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import time
|
|
|
|
def load_state():
|
|
if path := os.environ.get("GRAPH_STATE_FILE"):
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
return json.loads(os.environ.get("GRAPH_STATE", "{}"))
|
|
|
|
|
|
state = load_state()
|
|
proj = os.path.expanduser(
|
|
(state.get("project_dir_in") or "").strip() or state.get("project_dir") or "."
|
|
)
|
|
spec = (state.get("diff_spec") or "worktree").strip()
|
|
spec_raw = spec
|
|
# The parse stage should emit a bare revision token, but callers write prose
|
|
# and extraction drifts ("run get_diff --base origin/main" once reached
|
|
# `git diff` verbatim, killing the signals — and with them the security and
|
|
# probe lanes). Deterministically recover a usable spec before giving up.
|
|
_TOKEN = r"[\w./~^@{}-]+"
|
|
if spec not in ("", "worktree") and not re.fullmatch(
|
|
rf"{_TOKEN}(\.\.\.?{_TOKEN})?", spec
|
|
):
|
|
if m := re.search(rf"({_TOKEN}\.\.\.?{_TOKEN})", spec):
|
|
spec = m.group(1) # a revision range buried in prose
|
|
elif m := re.search(rf"--base[= ]({_TOKEN})", spec):
|
|
spec = f"{m.group(1)}...HEAD" # the get_diff idiom
|
|
|
|
|
|
def git(*args):
|
|
r = subprocess.run(
|
|
["git", "-C", proj, *args], capture_output=True, text=True, timeout=90
|
|
)
|
|
if r.returncode != 0:
|
|
raise RuntimeError((r.stderr or r.stdout).strip()[:400])
|
|
return r.stdout
|
|
|
|
|
|
out = {"project_dir": proj}
|
|
|
|
# Deterministic plan_context recovery: the parse LLM proved a lossy channel
|
|
# for verbatim spec text (observed replacing ~4 KB of pasted criteria with
|
|
# file paths + a summary sentence). When the raw prompt carries the callers'
|
|
# labeled plan block, cut it out verbatim and override the extraction — but
|
|
# only when the recovered block is LONGER, so a good extraction is never
|
|
# degraded. The adversary lane fails closed on whatever goes missing here.
|
|
_prompt = state.get("initial_prompt")
|
|
if isinstance(_prompt, str) and _prompt:
|
|
_m = re.search(
|
|
r"Plan / acceptance criteria:\s*(.+?)"
|
|
r"(?=\n\s*(?:Local-run recipe|Forced lanes|Rigor:|Security posture:|Extra context)\b|\Z)",
|
|
_prompt,
|
|
re.S | re.I,
|
|
)
|
|
if _m:
|
|
_block = _m.group(1).strip()
|
|
if len(_block) > len((state.get("plan_context") or "").strip()):
|
|
out["plan_context"] = _block
|
|
# retry_gate measures its wall-clock budget from this stamp; it is set
|
|
# before the git work so a failed diff still starts the clock.
|
|
if not state.get("gauntlet_started_at"):
|
|
out["gauntlet_started_at"] = time.time()
|
|
try:
|
|
if spec in ("", "worktree"):
|
|
names = git("diff", "--name-only", "HEAD")
|
|
difftext = git("diff", "HEAD")
|
|
else:
|
|
names = git("diff", "--name-only", spec)
|
|
difftext = git("diff", spec)
|
|
files = [f.strip() for f in names.splitlines() if f.strip()]
|
|
added = "\n".join(
|
|
l for l in difftext.splitlines() if l.startswith("+") and not l.startswith("+++")
|
|
)
|
|
|
|
def any_path(pattern):
|
|
return any(re.search(pattern, f, re.I) for f in files)
|
|
|
|
out.update(
|
|
{
|
|
"changed_files": files[:200],
|
|
"file_count": len(files),
|
|
"touches_proto": any_path(r"\.proto$|buf\.(yaml|gen\.yaml)$"),
|
|
"touches_migrations": any_path(r"migrations?/|\.sql$"),
|
|
# auth(?!or) admits auth/authn/authz/oauth but not "author".
|
|
# Over-firing here is the safe direction: it only ADDS a lane.
|
|
"touches_auth": any_path(
|
|
r"auth(?!or)|login|logout|token|session|permission|rbac|\bacl\b|credential|secret"
|
|
)
|
|
or bool(
|
|
re.search(
|
|
r"auth(?!or)|password|token|secret|credential|jwt|session",
|
|
added,
|
|
re.I,
|
|
)
|
|
),
|
|
"touches_deps": any_path(
|
|
r"go\.(mod|sum)$|package(-lock)?\.json$|yarn\.lock$|pnpm-lock\.yaml$"
|
|
r"|Cargo\.(toml|lock)$|requirements[^/]*\.txt$|pyproject\.toml$"
|
|
r"|Gemfile(\.lock)?$|pom\.xml$|build\.gradle|composer\.(json|lock)$"
|
|
),
|
|
"touches_exec": bool(
|
|
re.search(
|
|
r"exec\.Command|subprocess|os/exec|Runtime\.getRuntime"
|
|
r"|shell_exec|\beval\(|\bsystem\(|child_process",
|
|
added,
|
|
)
|
|
),
|
|
"consumer_surface": any_path(
|
|
r"routes?|handlers?|\bapi\b|endpoints?|\bcmd/|\bcli\b|\.proto$"
|
|
r"|openapi|swagger|controllers?|graphql"
|
|
),
|
|
"docs_only": bool(files)
|
|
and all(re.search(r"\.(md|rst|txt|adoc)$", f, re.I) for f in files),
|
|
"signals_summary": f"{len(files)} file(s) changed (diff: {spec})"
|
|
+ (f" — spec normalized from {spec_raw!r}" if spec != spec_raw else ""),
|
|
}
|
|
)
|
|
except Exception as e: # noqa: BLE001 — degraded signals must not sink the gauntlet
|
|
what = f"diff spec {spec!r}" + (
|
|
f", normalized from {spec_raw!r}" if spec != spec_raw else ""
|
|
)
|
|
out["signals_error"] = (
|
|
f"NOTE: diff signals could not be computed ({what}: {e}); surface UNKNOWN — "
|
|
"lane selection degrades WIDER (security lane added; probe eligible when a "
|
|
"local-run recipe exists), plus caller context and forced lanes."
|
|
)
|
|
out["signals_summary"] = "signals unavailable"
|
|
|
|
print(json.dumps(out))
|