| |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import re |
| import subprocess |
| import sys |
| from dataclasses import dataclass, field |
| from enum import Enum |
| from pathlib import Path |
|
|
|
|
| class Status(str, Enum): |
| PASS = "PASS" |
| PARK = "PARK" |
| OWNER_DECISION = "OWNER_DECISION" |
| EVIDENCE_REQUIRED = "EVIDENCE_REQUIRED" |
| BLOCK = "BLOCK" |
|
|
|
|
| BLOCKING_STATUSES = {Status.BLOCK, Status.EVIDENCE_REQUIRED, Status.OWNER_DECISION} |
|
|
| RISKY_PATH_PATTERNS = ( |
| re.compile(r"(^|/)\.env($|[./])"), |
| re.compile(r"(^|/)\.env\.[^/]+$"), |
| re.compile(r"(^|/)(secret|secrets|credentials|private|token|tokens)(/|$)", re.I), |
| re.compile(r"(^|/)infra(/|$)", re.I), |
| re.compile(r"(^|/)migrations?(/|$)", re.I), |
| re.compile(r"(^|/)\.github/workflows(/|$)", re.I), |
| re.compile(r"(^|/)wrangler\.jsonc?$", re.I), |
| re.compile(r"(^|/)terraform(/|$)", re.I), |
| ) |
|
|
| CODE_PATH_SUFFIXES = ( |
| ".py", |
| ".js", |
| ".jsx", |
| ".ts", |
| ".tsx", |
| ".go", |
| ".rs", |
| ".java", |
| ".kt", |
| ".c", |
| ".cc", |
| ".cpp", |
| ".h", |
| ".hpp", |
| ".cs", |
| ".php", |
| ".rb", |
| ".swift", |
| ".scala", |
| ".sh", |
| ) |
|
|
| TEST_EVIDENCE_PATTERNS = ( |
| re.compile(r"\b(test|tests|pytest|unittest|vitest|jest|npm test|pnpm test|uv run pytest|cargo test|go test|ruff|lint|typecheck|tsc|mypy)\b", re.I), |
| re.compile(r"\b(passed|pass|green|success|successful|ok|0 failures?|0 failed)\b", re.I), |
| ) |
|
|
| OWNER_APPROVAL_PATTERNS = ( |
| re.compile(r"owner (approval|approved|freigabe)", re.I), |
| re.compile(r"samuel (approved|freigegeben|hat freigegeben)", re.I), |
| re.compile(r"\bapproved-by:\s*\S+", re.I), |
| ) |
|
|
| PARK_PATTERNS = ( |
| re.compile(r"\b(follow[- ]?up|nice[- ]?to[- ]?have|future work|later|park|parking lot|wishlist)\b", re.I), |
| re.compile(r"\b(spaeter|später|irgendwann|nicht jetzt|wishlist|idee)\b", re.I), |
| ) |
|
|
|
|
| @dataclass |
| class GateFinding: |
| status: Status |
| reason: str |
| evidence: str = "" |
|
|
|
|
| @dataclass |
| class GateResult: |
| status: Status |
| findings: list[GateFinding] = field(default_factory=list) |
| changed_files: list[str] = field(default_factory=list) |
| pr_title: str = "" |
| pr_body: str = "" |
|
|
| @property |
| def should_block(self) -> bool: |
| return self.status in BLOCKING_STATUSES |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser(description="Run the deterministic HAI PR gate.") |
| parser.add_argument("--base", default=None, help="Base ref/commit for git diff, e.g. origin/main.") |
| parser.add_argument("--event", default=os.environ.get("GITHUB_EVENT_PATH"), help="GitHub event JSON path.") |
| parser.add_argument("--summary", default=os.environ.get("GITHUB_STEP_SUMMARY"), help="Markdown summary path.") |
| parser.add_argument("--files", default=None, help="Optional newline-delimited changed files fixture.") |
| args = parser.parse_args(argv) |
|
|
| changed_files = load_changed_files(args.files, args.base) |
| title, body = load_pr_text(args.event) |
| result = evaluate(changed_files=changed_files, pr_title=title, pr_body=body) |
| summary = render_owner_packet(result) |
| write_summary(summary, args.summary) |
| print(summary) |
| return 1 if result.should_block else 0 |
|
|
|
|
| def evaluate(*, changed_files: list[str], pr_title: str = "", pr_body: str = "") -> GateResult: |
| body = pr_body.strip() |
| findings: list[GateFinding] = [] |
|
|
| risky_files = [path for path in changed_files if is_risky_path(path)] |
| code_files = [path for path in changed_files if is_code_path(path)] |
| parked_items = find_parked_items(body) |
| has_owner_approval = contains_owner_approval(body) |
| has_test_evidence = contains_test_evidence(body) |
|
|
| if risky_files and not has_owner_approval: |
| findings.append( |
| GateFinding( |
| Status.BLOCK, |
| "Risky files changed without explicit owner approval.", |
| ", ".join(risky_files[:8]), |
| ) |
| ) |
|
|
| if len(changed_files) > 12 and not has_owner_approval: |
| findings.append( |
| GateFinding( |
| Status.OWNER_DECISION, |
| "Large PR scope needs an explicit owner decision.", |
| f"{len(changed_files)} changed files", |
| ) |
| ) |
|
|
| if code_files and not has_test_evidence: |
| findings.append( |
| GateFinding( |
| Status.EVIDENCE_REQUIRED, |
| "Code changed but PR body does not show concrete test/check evidence.", |
| ", ".join(code_files[:8]), |
| ) |
| ) |
|
|
| if parked_items: |
| findings.append( |
| GateFinding( |
| Status.PARK, |
| "Follow-up or future-work content should stay parked, not expand this PR.", |
| parked_items[0], |
| ) |
| ) |
|
|
| if not body: |
| findings.append( |
| GateFinding( |
| Status.EVIDENCE_REQUIRED, |
| "PR body is empty; HAI requires goal, verification, risk, and owner-decision state.", |
| "empty PR body", |
| ) |
| ) |
|
|
| status = strongest_status(findings) |
| return GateResult( |
| status=status, |
| findings=findings, |
| changed_files=changed_files, |
| pr_title=pr_title, |
| pr_body=pr_body, |
| ) |
|
|
|
|
| def strongest_status(findings: list[GateFinding]) -> Status: |
| priority = { |
| Status.BLOCK: 0, |
| Status.EVIDENCE_REQUIRED: 1, |
| Status.OWNER_DECISION: 2, |
| Status.PARK: 3, |
| Status.PASS: 4, |
| } |
| if not findings: |
| return Status.PASS |
| return sorted((finding.status for finding in findings), key=lambda status: priority[status])[0] |
|
|
|
|
| def is_risky_path(path: str) -> bool: |
| normalized = path.strip().replace("\\", "/") |
| return any(pattern.search(normalized) for pattern in RISKY_PATH_PATTERNS) |
|
|
|
|
| def is_code_path(path: str) -> bool: |
| normalized = path.strip().lower() |
| return normalized.endswith(CODE_PATH_SUFFIXES) |
|
|
|
|
| def contains_test_evidence(text: str) -> bool: |
| if not text.strip(): |
| return False |
| return all(pattern.search(text) for pattern in TEST_EVIDENCE_PATTERNS) |
|
|
|
|
| def contains_owner_approval(text: str) -> bool: |
| return any(pattern.search(text) for pattern in OWNER_APPROVAL_PATTERNS) |
|
|
|
|
| def find_parked_items(text: str) -> list[str]: |
| lines = [] |
| for raw_line in text.splitlines(): |
| stripped = raw_line.strip() |
| if stripped.startswith("#"): |
| continue |
| lines.append(stripped.strip("-* >\t ")) |
| return [line for line in lines if line and any(pattern.search(line) for pattern in PARK_PATTERNS)] |
|
|
|
|
| def render_owner_packet(result: GateResult) -> str: |
| primary = primary_finding(result) |
| status = result.status.value |
| recommendation = recommendation_for(result.status) |
| changed = "\n".join(f"- `{path}`" for path in result.changed_files[:20]) or "- No changed files detected." |
| extra_count = max(0, len(result.changed_files) - 20) |
| if extra_count: |
| changed += f"\n- ...and {extra_count} more" |
|
|
| findings = "\n".join( |
| f"- **{finding.status.value}**: {finding.reason}" |
| + (f" Evidence: `{finding.evidence}`" if finding.evidence else "") |
| for finding in result.findings |
| ) |
| if not findings: |
| findings = "- **PASS**: No blocking HAI risk detected." |
|
|
| decision = owner_decision_for(result.status, primary) |
|
|
| return "\n".join( |
| [ |
| "## HAI Owner Packet", |
| "", |
| f"Status: **{status}**", |
| "", |
| "### One Decision", |
| decision, |
| "", |
| "### Recommendation", |
| recommendation, |
| "", |
| "### Findings", |
| findings, |
| "", |
| "### Changed Files", |
| changed, |
| "", |
| "### Gate Result", |
| "Merge is blocked until this gate passes." if result.should_block else "Gate passed; merge is not blocked by HAI.", |
| "", |
| ] |
| ) |
|
|
|
|
| def primary_finding(result: GateResult) -> GateFinding | None: |
| if not result.findings: |
| return None |
| priority = { |
| Status.BLOCK: 0, |
| Status.EVIDENCE_REQUIRED: 1, |
| Status.OWNER_DECISION: 2, |
| Status.PARK: 3, |
| Status.PASS: 4, |
| } |
| return sorted(result.findings, key=lambda finding: priority[finding.status])[0] |
|
|
|
|
| def owner_decision_for(status: Status, primary: GateFinding | None) -> str: |
| if status == Status.PASS: |
| return "No owner decision required by this gate." |
| if status == Status.PARK: |
| return "Keep parked items out of this PR unless Samuel explicitly reopens scope." |
| if status == Status.BLOCK: |
| return "Do not merge. Samuel must approve or remove the risky change." |
| if status == Status.EVIDENCE_REQUIRED: |
| return "Do not merge. Agent must add concrete test/check evidence." |
| if status == Status.OWNER_DECISION: |
| return "Do not merge. Samuel must choose whether this scope is allowed." |
| if primary: |
| return primary.reason |
| return "Do not merge until HAI status is clear." |
|
|
|
|
| def recommendation_for(status: Status) -> str: |
| return { |
| Status.PASS: "Proceed with normal review.", |
| Status.PARK: "Leave follow-up ideas parked and keep the PR scoped.", |
| Status.BLOCK: "Stop and reduce risk before review continues.", |
| Status.EVIDENCE_REQUIRED: "Request tests, command output, or reproducible validation before review continues.", |
| Status.OWNER_DECISION: "Ask Samuel for a single explicit scope decision before further agent work.", |
| }[status] |
|
|
|
|
| def load_pr_text(event_path: str | None) -> tuple[str, str]: |
| if not event_path: |
| return "", "" |
| path = Path(event_path) |
| if not path.exists(): |
| return "", "" |
| raw = json.loads(path.read_text(encoding="utf-8")) |
| pr = raw.get("pull_request") or {} |
| return str(pr.get("title") or ""), str(pr.get("body") or "") |
|
|
|
|
| def load_changed_files(files_path: str | None, base: str | None) -> list[str]: |
| if files_path: |
| return [ |
| line.strip() |
| for line in Path(files_path).read_text(encoding="utf-8").splitlines() |
| if line.strip() |
| ] |
| if not base: |
| return [] |
| try: |
| proc = subprocess.run( |
| ["git", "diff", "--name-only", f"{base}...HEAD"], |
| check=True, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| ) |
| except subprocess.CalledProcessError as exc: |
| fallback = subprocess.run( |
| ["git", "diff", "--name-only", base], |
| check=False, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| ) |
| if fallback.returncode != 0: |
| raise SystemExit( |
| f"Could not compute changed files against {base}: {exc.stderr or fallback.stderr}" |
| ) from exc |
| proc = fallback |
| return [line.strip() for line in proc.stdout.splitlines() if line.strip()] |
|
|
|
|
| def write_summary(summary: str, summary_path: str | None) -> None: |
| if not summary_path: |
| return |
| Path(summary_path).write_text(summary, encoding="utf-8") |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
