Skill: Preview commit plan

#!/usr/bin/env python3
"""Render a commit plan (JSON) as byte-aligned ASCII tables.

Usage:
    python3 render_plan.py plan.json
    python3 render_plan.py < plan.json

Input schema (JSON):
    {
      "commits": [
        {
          "message":  "feat(scope): subject",        # required
          "details":  "body prose",                  # optional -> "(none)"
          "breaking": "BREAKING CHANGE footer text", # optional
          "warn":     "why this commit is large",    # optional; auto-added
                                                     # when > 500 lines change
          "files": [                                 # required, >= 1
            {"status": "A", "path": "src/x.ts", "add": 10, "del": 2},
            {"status": "M", "path": "logo.png", "binary": true},
            {"status": "M", "path": "y.ts", "add": 5, "del": 1,
             "staged": false},                       # "~" marker after status
            {"status": "R", "path": "new.ts", "from": "old.ts",
             "add": 0, "del": 0}                     # "from": the old path;
                                                     # write_commit_script.py
                                                     # needs it
          ]
        }
      ]
    }

Tables print to stdout; lint notes and the alignment verdict print to
stderr. Exit is nonzero if the plan is invalid or any table fails the
byte-alignment / ASCII-only self-check.
"""

import json
import re
import sys
import textwrap

STATUS_ORDER = {"A": 0, "M": 1, "D": 2, "R": 3}
TYPES = "feat|fix|docs|style|refactor|perf|test|deps|build|ci|chore"
SUBJECT_RE = re.compile(r"^(" + TYPES + r")(\([a-z0-9./-]+\))?(!)?: [a-z0-9]")
MIN_PROSE_W = 65   # floor for the detail-table value column
PATH_CAP = 64      # longer paths are left-truncated, filename preserved
WARN_LINES = 500


def die(msg):
    """Print a fatal error to stderr and exit nonzero.

    Args:
        msg: The error message.
    """
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(1)


def load_plan(path=None):
    """Read and validate the plan JSON from path, argv[1] or stdin."""
    if path is None and len(sys.argv) > 1:
        path = sys.argv[1]
    try:
        if path:
            with open(path) as f:
                plan = json.load(f)
        else:
            plan = json.load(sys.stdin)
    except (OSError, json.JSONDecodeError) as e:
        die(f"cannot read plan JSON: {e}")
    return validate_plan(plan)


def validate_plan(plan):
    """Return the commits of the plan, or exit if the plan is invalid."""
    commits = plan.get("commits")
    if not isinstance(commits, list) or not commits:
        die('plan must have a non-empty "commits" array')
    for i, c in enumerate(commits, 1):
        if not c.get("message"):
            die(f'commit {i}: missing "message"')
        files = c.get("files")
        if not isinstance(files, list) or not files:
            die(f'commit {i}: missing or empty "files"')
        for f in files:
            st = f.get("status")
            if st not in STATUS_ORDER:
                die(f'commit {i}: file status must be one of A/M/D/R, got {st!r}')
            if not f.get("path"):
                die(f'commit {i}: file entry missing "path"')
            if not f.get("binary") and ("add" not in f or "del" not in f):
                die(f'commit {i}: {f["path"]}: needs "add" and "del" (or "binary": true)')
    return commits


def center_r(text, width):
    """Center text with right-biased padding (matches the spec examples)."""
    left = (width - len(text) + 1) // 2
    return (" " * left + text).ljust(width)


def fit_path(path, width):
    """Left-truncate a path to width, keeping the filename end visible."""
    if len(path) <= width:
        return path.ljust(width)
    return "..." + path[-(width - 3):]


def lint(commits):
    """Emit non-fatal style notes about commit messages to stderr."""
    for i, c in enumerate(commits, 1):
        subject = c["message"]
        if len(subject) > 50:
            print(f"LINT: commit {i} subject is {len(subject)} chars (keep under 50)",
                  file=sys.stderr)
        if not SUBJECT_RE.match(subject):
            print(f"LINT: commit {i} subject does not look like Conventional Commits: "
                  f"{subject!r}", file=sys.stderr)
        if subject.rstrip().endswith("."):
            print(f"LINT: commit {i} subject ends with a period", file=sys.stderr)


def summary_table(commits):
    """Build the one-row-per-commit summary table as a list of lines."""
    msg_w = max(len("Commit message"), max(len(c["message"]) for c in commits))
    files_w = 5
    plus_w = max(4, len(str(sum(c["adds"] for c in commits))))
    del_w = max(4, len(str(sum(c["dels"] for c in commits))))

    def row(cells):
        return "| " + " | ".join(cells) + " |"

    sep = "+" + "+".join("-" * (w + 2) for w in (3, msg_w, files_w, plus_w, del_w)) + "+"
    out = [sep,
           row(["#".ljust(3), "Commit message".ljust(msg_w), "Files".rjust(files_w),
                center_r("+", plus_w), center_r("-", del_w)]),
           sep]
    for i, c in enumerate(commits, 1):
        out.append(row([str(i).ljust(3), c["message"].ljust(msg_w),
                        str(len(c["files"])).rjust(files_w),
                        str(c["adds"]).rjust(plus_w), str(c["dels"]).rjust(del_w)]))
    out.append(sep)
    merged = "TOTAL".ljust(3 + 3 + msg_w)  # spans the "#" and message columns
    out.append(row([merged, str(sum(len(c["files"]) for c in commits)).rjust(files_w),
                    str(sum(c["adds"] for c in commits)).rjust(plus_w),
                    str(sum(c["dels"] for c in commits)).rjust(del_w)]))
    out.append(sep)
    return out


def detail_table(c, i, total, label_w):
    """Build one commit's detail table as a list of lines."""
    num_w = max(4, max((len(f"+{f.get('add', 0)}") for f in c["files"]), default=4),
                max((len(f"-{f.get('del', 0)}") for f in c["files"]), default=4))
    path_w = min(PATH_CAP, max(len(f["path"]) for f in c["files"]))
    val_w = max(MIN_PROSE_W, len(c["message"]), 3 + path_w + 2 + num_w + 2 + num_w)
    path_w = val_w - 3 - 2 - num_w - 2 - num_w  # widen paths into any slack

    sep = "+" + "-" * (label_w + 2) + "+" + "-" * (val_w + 2) + "+"
    out = [sep]

    def row(label, text):
        out.append("| " + label.ljust(label_w) + " | " + text.ljust(val_w) + " |")

    def prose(label, text):
        for j, line in enumerate(textwrap.wrap(text, val_w) or [text]):
            row(label if j == 0 else "", line)

    row("Commit", f"{i} of {total}")
    out.append(sep)
    prose("Message", c["message"])
    out.append(sep)
    prose("Details", c.get("details") or "(none)")
    out.append(sep)
    if c.get("breaking"):
        prose("BREAKING", "BREAKING CHANGE: " + c["breaking"])
        out.append(sep)
    warn = c.get("warn")
    total_lines = c["adds"] + c["dels"]
    if warn is None and total_lines > WARN_LINES:
        warn = (f"This commit changes {total_lines} lines (over the "
                f"{WARN_LINES}-line guideline) - consider splitting.")
    if warn:
        prose("WARN", warn)
        out.append(sep)
    for j, f in enumerate(c["files"]):
        marker = " " if f.get("staged", True) else "~"
        if f.get("binary"):
            cell = f["status"] + marker + " " + fit_path(f["path"], path_w) \
                + "  " + "(binary)".rjust(num_w * 2 + 2)
        else:
            cell = f["status"] + marker + " " + fit_path(f["path"], path_w) \
                + "  " + f"+{f['add']}".rjust(num_w) + "  " + f"-{f['del']}".rjust(num_w)
        row("Files" if j == 0 else "", cell)
    unit = "file" if len(c["files"]) == 1 else "files"
    row("", f"-- {len(c['files'])} {unit}, +{c['adds']} -{c['dels']}")
    out.append(sep)
    return out


def self_check(blocks):
    """Verify every table is ASCII-only and byte-aligned; exit 1 if not."""
    ok = True
    for b, block in enumerate(blocks, 1):
        widths = {len(line) for line in block}
        if len(widths) != 1:
            ok = False
            print(f"MISALIGNED: table {b} has row widths {sorted(widths)}", file=sys.stderr)
        bad = {ch for line in block for ch in line if ord(ch) > 127}
        if bad:
            ok = False
            print(f"NON-ASCII in table {b}: {sorted(bad)}", file=sys.stderr)
    print("ALIGNMENT:", "OK" if ok else "FAILED", file=sys.stderr)
    if not ok:
        sys.exit(1)


def main():
    """Load, lint, render, and self-check the plan."""
    commits = load_plan()
    for c in commits:
        c["files"].sort(key=lambda f: (STATUS_ORDER[f["status"]], f["path"]))
        c["adds"] = sum(f.get("add", 0) for f in c["files"])
        c["dels"] = sum(f.get("del", 0) for f in c["files"])
    lint(commits)
    label_w = 8 if any(c.get("breaking") for c in commits) else 7
    blocks = [summary_table(commits)]
    blocks += [detail_table(c, i, len(commits), label_w)
               for i, c in enumerate(commits, 1)]
    print("\n\n".join("\n".join(b) for b in blocks))
    self_check(blocks)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Write an executable bash script that makes the commits of a plan.

Usage:
    python3 write_commit_script.py plan.json [--repo DIR] [--name NAME] [--force]

The plan JSON is the same file that render_plan.py reads. A rename entry
("status": "R") must also give the old path in "from".

The script goes into the root of the repo (--repo, default: the repo of
the current directory) as NAME (default: commit-plan.sh). The tool
overwrites a file only if an earlier run of this tool wrote it, or if
--force is given. It prints the path of the script to stdout.

The script itself makes the commits only when the user runs it. Its
--dry-run option does all the checks in a temporary index and changes
nothing.
"""

import argparse
import os
import shlex
import subprocess
import sys
import textwrap

from render_plan import die, load_plan

MARKER = "# Generated by the preview-commit-plan skill."
BODY_W = 72

TEMPLATE = r"""#!/usr/bin/env bash
@MARKER@
#
# Makes the @COUNT@ commits of the plan below, in order. The plan puts the
# commits in dependency order, so each commit builds.
#
# The script first takes a snapshot of the staged changes in a temporary
# index, and adds the unstaged files of the plan (status with "~") to it.
# Each commit gets the content of its files from that snapshot, so later
# edits in the working tree do not go into the commits. The script does
# not change the working tree.
#
# Usage: ./@NAME@ [--dry-run]
#   --dry-run  Do the checks and show the commits. Change nothing.
set -euo pipefail
export GIT_LITERAL_PATHSPECS=1

cd "$(dirname "$0")"

me=${0##*/}
# The plan is for this HEAD. Empty means a repo with no commits.
expected_head=@HEAD@
dry_run=false
case ${1:-} in
  "") ;;
  --dry-run) dry_run=true ;;
  *)
    echo "usage: $me [--dry-run]" >&2
    exit 2
    ;;
esac

die() {
  echo "$me: $*" >&2
  exit 1
}

head=$(git rev-parse -q --verify HEAD || true)
if [[ $head != "$expected_head" ]]; then
  die "HEAD is ${head:-unborn}, but the plan is for ${expected_head:-an unborn HEAD}. Make the plan again."
fi
for state in MERGE_HEAD CHERRY_PICK_HEAD REVERT_HEAD rebase-merge rebase-apply; do
  if [[ -e $(git rev-parse --git-path "$state") ]]; then
    die "a git operation is in progress ($state). Finish or abort it first."
  fi
done

# One entry for each file: the commit number, the status and the path.
# A "~" after the status marks an unstaged file. "R-from" is the old path
# of a rename.
files=(
@FILES@
)

# The message of each commit, in commit order.
messages=(
@MESSAGES@
)

paths=()
unstaged=()
for e in "${files[@]}"; do
  rest=${e#* }
  st=${rest%% *}
  paths+=("${rest#* }")
  if [[ $st == *"~" ]]; then
    unstaged+=("${rest#* }")
  fi
done

tmp_dir=$(mktemp -d)
trap 'rm -rf "$tmp_dir"' EXIT
tmp_index=$tmp_dir/index
msg_file=$tmp_dir/message

# Take the snapshot in a copy of the index. The real index does not change.
real_index=$(git rev-parse --git-path index)
if [[ -e $real_index ]]; then
  cp "$real_index" "$tmp_index"
fi
staged_tree=$(GIT_INDEX_FILE=$tmp_index git write-tree) ||
  die "the index has unmerged files. Resolve them first."
if ((${#unstaged[@]})); then
  GIT_INDEX_FILE=$tmp_index git add -A -- "${unstaged[@]}" ||
    die "cannot add the unstaged files of the plan. Make the plan again."
fi
full_tree=$(GIT_INDEX_FILE=$tmp_index git write-tree)

# Check that the plan holds each changed file one time, and no other file.
if [[ -n $expected_head ]]; then
  base=$expected_head
else
  base=$(git hash-object -t tree /dev/null)
fi
planned=$(printf '%s\n' "${paths[@]}" | LC_ALL=C sort)
changed=$(git diff-tree -r -z --no-renames --name-only "$base" "$full_tree" |
  tr '\0' '\n' | LC_ALL=C sort)
if [[ -n $(uniq -d <<<"$planned") ]]; then
  die "a file is in more than one commit. Stop."
fi
if [[ $planned != "$changed" ]]; then
  echo "Only in the plan:" >&2
  comm -23 <(echo "$planned") <(echo "$changed") >&2
  echo "Only in the snapshot:" >&2
  comm -13 <(echo "$planned") <(echo "$changed") >&2
  die "the changes are not the changes of the plan. Make the plan again."
fi

if $dry_run; then
  for i in "${!messages[@]}"; do
    n=$((i + 1))
    echo "== $n of ${#messages[@]}: ${messages[i]%%$'\n'*}"
    for e in "${files[@]}"; do
      if [[ ${e%% *} == "$n" ]]; then
        echo "   ${e#* }"
      fi
    done
  done
  echo "Checks passed. Nothing changed."
  exit 0
fi

echo "Snapshot: $full_tree"

# On a failure, tell the user how to get the old index back.
trap 'echo >&2; echo "$me: stopped. To get back the index from before the script, run:" >&2; echo "  git read-tree $staged_tree" >&2' ERR

# Empty the index to HEAD. The working tree does not change.
if [[ -n $expected_head ]]; then
  git reset -q
else
  git read-tree --empty
fi

for i in "${!messages[@]}"; do
  n=$((i + 1))
  for e in "${files[@]}"; do
    if [[ ${e%% *} != "$n" ]]; then
      continue
    fi
    rest=${e#* }
    path=${rest#* }
    if git cat-file -e "$full_tree:$path" 2>/dev/null; then
      git ls-tree -z "$full_tree" -- "$path" | git update-index -z --index-info
    else
      git update-index --force-remove -- "$path"
    fi
  done
  printf '%s\n' "${messages[i]}" >"$msg_file"
  git commit -q -F "$msg_file"
  echo "$n of ${#messages[@]}: $(git log -1 --format='%h %s')"
done

if [[ $(git rev-parse 'HEAD^{tree}') != "$full_tree" ]]; then
  die "the last commit does not match the snapshot $full_tree. Check git status."
fi
echo "Done. HEAD has all the changes of the plan."
"""


def git(repo, *args):
    """Return the stdout of a git command, or None if the command fails."""
    r = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True)
    return r.stdout.strip() if r.returncode == 0 else None


def build_message(c):
    """Return the full commit message: subject, body and footer."""
    parts = [c["message"].strip()]
    for para in (c.get("details") or "").split("\n\n"):
        if para.strip():
            parts.append(textwrap.fill(" ".join(para.split()), BODY_W))
    if c.get("breaking"):
        parts.append(textwrap.fill("BREAKING CHANGE: " + " ".join(c["breaking"].split()),
                                   BODY_W))
    return "\n\n".join(parts)


def file_entries(commits):
    """Return the "N status path" entries of the script, in commit order."""
    owner = {}
    entries = []
    for n, c in enumerate(commits, 1):
        # A path can be in one commit two times: staged and unstaged parts.
        merged = {}
        for f in c["files"]:
            pairs = [(f["status"], f["path"])]
            if f["status"] == "R":
                if not f.get("from"):
                    die(f'commit {n}: rename {f["path"]} needs "from" (the old path)')
                pairs.append(("R-from", f["from"]))
            for st, path in pairs:
                if "\n" in path or path.startswith("/"):
                    die(f"commit {n}: cannot use the path {path!r}")
                if owner.get(path, n) != n:
                    die(f"{path} is in commit {owner[path]} and commit {n}. "
                        "Put each file in one commit.")
                owner[path] = n
                old = merged.get(path)
                unstaged = f.get("staged", True) is False or (old and old[1])
                merged[path] = (old[0] if old else st, unstaged)
        for path, (st, unstaged) in merged.items():
            entries.append(f"{n} {st}{'~' if unstaged else ''} {path}")
    return entries


def main():
    """Validate the plan, write the script and check its syntax."""
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("plan")
    ap.add_argument("--repo", default=".")
    ap.add_argument("--name", default="commit-plan.sh")
    ap.add_argument("--force", action="store_true")
    args = ap.parse_args()

    if "/" in args.name or not args.name.endswith(".sh"):
        die(f"--name must be a file name that ends with .sh, got {args.name!r}")
    root = git(args.repo, "rev-parse", "--show-toplevel")
    if not root:
        die(f"{args.repo} is not in a git repo")
    commits = load_plan(args.plan)
    head = git(root, "rev-parse", "-q", "--verify", "HEAD") or ""

    out = os.path.join(root, args.name)
    if os.path.exists(out) and not args.force:
        with open(out, errors="replace") as f:
            if MARKER not in f.read(4096):
                die(f"{out} exists and this tool did not write it. "
                    "Use another --name, or --force to overwrite it.")

    script = TEMPLATE
    for key, value in {
        "@MARKER@": MARKER,
        "@COUNT@": str(len(commits)),
        "@NAME@": args.name,
        "@HEAD@": head,
        "@FILES@": "\n".join("  " + shlex.quote(e) for e in file_entries(commits)),
        "@MESSAGES@": "\n".join("  " + shlex.quote(build_message(c)) for c in commits),
    }.items():
        script = script.replace(key, value)

    with open(out, "w") as f:
        f.write(script)
    os.chmod(out, 0o755)
    r = subprocess.run(["bash", "-n", out], capture_output=True, text=True)
    if r.returncode != 0:
        die(f"{out} has a bash syntax error: {r.stderr.strip()}")
    print(out)


if __name__ == "__main__":
    main()