RedFlag/.publication/commit_voice.py

#!/usr/bin/env python3
"""Block public commit prose that is not fit to leave the private forge."""
import argparse
import json
import re
import subprocess
import sys
from datetime import date

MAX_SUBJECT = 59
MAX_PROSE = 180
MAX_SENTENCES = 2
TRAILER = re.compile(r"^(?:Source-Sha|Policy-Sha|Tree-Digest):\s+\S+$")
PRIVATE = re.compile(
    r"\b(?:10|127)(?:\.[0-9]{1,3}){2,3}\b"
    r"|\b192\.168(?:\.[0-9]{1,3}){1,2}\b"
    r"|\b172\.(?:1[6-9]|2[0-9]|3[01])(?:\.[0-9]{1,3}){1,2}\b"
    r"|\bwiuf[-a-z0-9_]*\b|\barchdev\b|/home/[a-z0-9_-]+|/root/",
    re.I,
)
ATTRIBUTION = re.compile(
    r"co-authored-by:.*(?:claude|openai|chatgpt|copilot|letta|cursor)"
    r"|generated by|generated with|ai-assisted|auto-generated by",
    re.I,
)
PROMPT = re.compile(
    r"\bcasey\b|\b(?:the\s+)?user\s+(?:said|asked|wanted|reported|told)\b"
    r"|(?:\*|_)[\"“].+?[\"”](?:\*|_)|^\s*>\s+",
    re.I | re.M | re.S,
)
RULES = {
    "subject-empty", "subject-length", "subject-case", "prose-length",
    "prose-sentences", "private-infrastructure", "model-attribution",
    "prompt-prose",
}


def public_prose(body):
    """Drop transport trailers; they are proof, not Field Notes prose."""
    lines = body.strip().splitlines()
    while lines and (not lines[-1].strip() or TRAILER.match(lines[-1].strip())):
        lines.pop()
    return "\n".join(lines).strip()


def violations(subject, body):
    prose = public_prose(body)
    message = f"{subject}\n{prose}"
    found = []
    if not subject:
        found.append(("subject-empty", "subject is empty"))
    if len(subject) > MAX_SUBJECT:
        found.append(("subject-length", f"subject is {len(subject)} characters; maximum is {MAX_SUBJECT}"))
    first_alpha = next((char for char in subject if char.isalpha()), "")
    if first_alpha and not first_alpha.islower():
        found.append(("subject-case", "subject must start lowercase"))
    if len(prose) > MAX_PROSE:
        found.append(("prose-length", f"public prose is {len(prose)} characters; maximum is {MAX_PROSE}"))
    sentences = len(re.findall(r"[.!?](?=\s|$)", prose))
    if sentences > MAX_SENTENCES:
        found.append(("prose-sentences", f"public prose has {sentences} sentences; maximum is {MAX_SENTENCES}"))
    if PRIVATE.search(message):
        found.append(("private-infrastructure", "message contains private infrastructure"))
    if ATTRIBUTION.search(message):
        found.append(("model-attribution", "message contains model attribution"))
    if PROMPT.search(message):
        found.append(("prompt-prose", "message contains prompt or private conversational prose"))
    return found


def load_allowlist(path, repository):
    with open(path) as handle:
        raw = json.load(handle)
    if raw.get("schema_version") != 2:
        raise ValueError("allowlist schema_version must be 2")
    if raw.get("repository") != repository:
        raise ValueError("allowlist repository does not match this publication")
    allowed = {}
    for item in raw.get("exceptions", []):
        item_repository = str(item.get("repository") or "")
        sha = str(item.get("sha") or "").lower()
        rule = str(item.get("rule") or "")
        reason = str(item.get("reason") or "").strip()
        reviewer = str(item.get("reviewer") or "").strip()
        reviewed = str(item.get("reviewed") or "")
        scope = str(item.get("scope") or "").strip()
        if item_repository != repository:
            raise ValueError("every exception must bind this exact repository")
        if not re.fullmatch(r"[0-9a-f]{40}", sha):
            raise ValueError("every exception needs one exact 40-character SHA")
        if rule not in RULES:
            raise ValueError(f"unknown exception rule: {rule}")
        if not reason:
            raise ValueError(f"{sha[:12]} {rule}: human reason is required")
        if not reviewer:
            raise ValueError(f"{sha[:12]} {rule}: reviewer identity is required")
        if not scope:
            raise ValueError(f"{sha[:12]} {rule}: lifetime or epoch scope is required")
        try:
            date.fromisoformat(reviewed)
        except ValueError as error:
            raise ValueError(f"{sha[:12]} {rule}: ISO review date is required") from error
        allowed[(sha, rule)] = item
    return allowed


def records(repo, revision_range):
    result = subprocess.run(
        ["git", "-C", repo, "log", "--format=%H%x1f%s%x1f%b%x1e", revision_range],
        capture_output=True, text=True,
    )
    if result.returncode:
        raise RuntimeError(result.stderr.strip() or "git log failed")
    for record in result.stdout.split("\x1e"):
        fields = record.strip("\r\n").split("\x1f", 2)
        if len(fields) == 3:
            yield fields


def check(repo, revision_range, allowlist):
    errors = []
    notes = []
    for sha, subject, body in records(repo, revision_range):
        for rule, reason in violations(subject.strip(), body.strip()):
            exception = allowlist.get((sha.lower(), rule))
            if exception:
                notes.append(f"{sha[:12]} {rule}: approved: {exception['reason']} (reviewed {exception['reviewed']})")
            else:
                errors.append(f"{sha[:12]} {rule}: {reason}")
    return errors, notes


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--repo", default=".")
    parser.add_argument("--range", required=True)
    parser.add_argument("--allowlist", required=True)
    parser.add_argument("--repository", required=True)
    args = parser.parse_args()
    try:
        errors, notes = check(args.repo, args.range, load_allowlist(args.allowlist, args.repository))
    except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as error:
        print(f"[commit-voice] {error}", file=sys.stderr)
        return 1
    for note in notes:
        print(f"[commit-voice] {note}")
    for error in errors:
        print(f"[commit-voice] {error}", file=sys.stderr)
    if errors:
        return 1
    print("[commit-voice] clean")
    return 0


if __name__ == "__main__":
    sys.exit(main())