Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions engine/skills/make-pr/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,10 @@ Declare exactly one review unit that matches the dominant changed paths:
| `corpus/skills/` | `corpus-lesson` |
| `product/skills/` | `product-skill` |

One review unit per PR; neutral files (`docs/`, repo-root `tests/`) ride
along. See [docs/ecosystem.md](../../../docs/ecosystem.md). A change whose
One review unit per PR. Repo-root `tests/` and the `docs/ecosystem.md`
inventory row ride along; every other doc, including the root `README.md`,
is its own `docs` unit. `drafter.config.json` is the only source of these
rules, and preflight reads it rather than a copy. A change whose
code reads another unit's output is two stacked PRs, producer first -- not
one PR with the coupling explained in Slice Rationale.

Expand Down
136 changes: 107 additions & 29 deletions engine/skills/make-pr/scripts/preflight.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,14 @@
unchecked, not clean.

Exit 0: one review unit, every gate passed. Exit 1: mixed units or a gate
failed. Exit 2: usage / no diff.
failed. Exit 2: usage / no diff. Exit 3: the review-unit rules could not be read.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
Expand All @@ -35,38 +37,105 @@
_HERE = os.path.abspath(__file__)
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(_HERE), "..", "..", "..", ".."))

# Mirrors drafter.config.json classification.pathRules (first match wins):
# corpus/skills -> corpus-lesson, product/skills -> product-skill,
# engine|scripts|.github + install.sh/drafter.config.json -> engine-runtime.
# Repo-root tests/ and docs/ are proof/docs units that co-locate, so they are
# neutral here.
UNIT_RULES = (
("corpus/skills/", "corpus-lesson"),
("product/skills/", "product-skill"),
("engine/", "engine-runtime"),
("scripts/", "engine-runtime"),
(".github/", "engine-runtime"),
)
ROOT_ENGINE_FILES = {"install.sh", "drafter.config.json"}


def review_unit_for(path: str) -> str | None:
if path in ROOT_ENGINE_FILES:
return "engine-runtime"
if path == "scripts/skill_test_debt_allowlist.txt":
return None
for prefix, unit in UNIT_RULES:
if path.startswith(prefix):
return unit
return None
DEFAULT_CONFIG = os.path.join(REPO_ROOT, "drafter.config.json")
UNCHECKED_EXIT = 3


class UnitRulesUnreadable(Exception):
pass


def expand_braces(pattern: str) -> list[str]:
match = re.search(r"\{([^{}]*)\}", pattern)
if not match:
return [pattern]
head, tail = pattern[: match.start()], pattern[match.end():]
return [p for option in match.group(1).split(",") for p in expand_braces(head + option + tail)]


def segment_regex(segment: str) -> str:
body = "".join(
"[^/]*" if ch == "*" else "[^/]" if ch == "?" else re.escape(ch) for ch in segment
)
return body if segment.startswith(".") else "(?!\\.)" + body


def classify(paths: list[str]) -> dict:
def glob_regex(pattern: str) -> re.Pattern:
parts = pattern.split("/")
out = ""
for i, part in enumerate(parts):
last = i == len(parts) - 1
if part == "**":
out += "(?:(?!\\.)[^/]*(?:/(?!\\.)[^/]*)*)?" if last else "(?:(?!\\.)[^/]*/)*"
else:
out += segment_regex(part) + ("" if last else "/")
return re.compile(out + "\\Z")


def read_config_text(config_path: str | None) -> tuple[str, str]:
if config_path is None and os.path.isfile(DEFAULT_CONFIG):
config_path = DEFAULT_CONFIG
if config_path is not None:
try:
with open(config_path, encoding="utf-8") as handle:
return config_path, handle.read()
except OSError as exc:
raise UnitRulesUnreadable(f"{config_path}: {exc}") from exc
label = "origin/main:drafter.config.json"
res = subprocess.run(["git", "show", label], capture_output=True, text=True)
if res.returncode != 0:
raise UnitRulesUnreadable(f"no drafter.config.json beside preflight and git show {label} failed: {res.stderr.strip()}")
return label, res.stdout


def load_unit_rules(config_path: str | None) -> dict:
config_path, text = read_config_text(config_path)
try:
config = json.loads(text)
units = config["taxonomy"]["units"]
path_rules = config["classification"]["pathRules"]
except (ValueError, KeyError, TypeError) as exc:
raise UnitRulesUnreadable(f"{config_path}: {exc}") from exc
compiled = []
for rule in path_rules:
if "pathGlob" in rule:
regexes = [glob_regex(p) for p in expand_braces(rule["pathGlob"])]
compiled.append((lambda path, basename, rx=regexes: any(r.match(path) for r in rx), rule["unit"]))
elif "basenamePattern" in rule:
rx = re.compile(rule["basenamePattern"])
compiled.append((lambda path, basename, rx=rx: bool(rx.search(basename)), rule["unit"]))
else:
raise UnitRulesUnreadable(f"{config_path}: path rule {rule.get('id')} has no pathGlob or basenamePattern")
return {
"rules": compiled,
"productUnits": {u["id"] for u in units if u.get("isProductUnit")},
"coLocatingUnits": {u["id"] for u in units if u.get("coLocatesWithProductUnits")},
}


def review_units_for(path: str, rules: dict) -> list[str]:
path = path.replace("\\", "/")
basename = path.rsplit("/", 1)[-1]
for matches, unit in rules["rules"]:
if matches(path, basename):
return list(unit)
return []


def classify(paths: list[str], config_path: str | None = None) -> dict:
rules = load_unit_rules(config_path)
per_path = {p: review_units_for(p, rules) for p in paths}
present = {u for found in per_path.values() for u in found}
has_product_unit = bool(present & rules["productUnits"])
ride_along = rules["coLocatingUnits"] if has_product_unit else set()
units: dict[str, list[str]] = {}
neutral: list[str] = []
for p in paths:
u = review_unit_for(p)
(units.setdefault(u, []) if u else neutral).append(p)
found = [u for u in per_path[p] if u not in ride_along]
if not found:
neutral.append(p)
for u in found:
units.setdefault(u, []).append(p)
return {"units": units, "neutral": neutral}


Expand Down Expand Up @@ -179,13 +248,22 @@ def main(argv: list[str] | None = None) -> int:
ap.add_argument("--paths", nargs="*", help="classify these paths instead of reading git")
ap.add_argument("--dry-run", action="store_true", help="print the plan, do not run gates")
ap.add_argument("--body-file", help="the PR description to check for claims about the repo's past")
ap.add_argument(
"--config", default=None,
help="drafter.config.json holding the review-unit rules (default: beside this script, else origin/main's copy)",
)
args = ap.parse_args(argv)

paths = args.paths if args.paths is not None else changed_paths(args.base)
if not paths:
print("no changed files vs " + args.base, file=sys.stderr)
return 2
info = classify(paths)
try:
info = classify(paths, args.config)
except UnitRulesUnreadable as exc:
print(f"fail unchecked review units: {exc}")
print("fail preflight: fix the above before gh pr create")
return UNCHECKED_EXIT
units = info["units"]
for unit, files in sorted(units.items()):
print(f"unit {unit}: {len(files)} file(s)")
Expand Down
80 changes: 78 additions & 2 deletions engine/skills/make-pr/tests/test_preflight.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
sets of PRs in this repo (e.g. #89 visual-proof, the 2026-09-01 hook slices)."""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
Expand Down Expand Up @@ -39,9 +42,13 @@ def test_engine_and_neutral_paths(self):
self.assertEqual(info["neutral"], ["tests/test_install.py"])

def test_scripts_and_install_sh_are_engine_runtime_like_drafter_config(self):
info = pf.classify(["scripts/check_codify_has_code.py", "install.sh", ".github/workflows/ci.yml", "docs/x.md"])
info = pf.classify(["scripts/check_codify_has_code.py", "install.sh", ".github/workflows/ci.yml", "docs/ecosystem.md"])
self.assertEqual(set(info["units"]), {"engine-runtime"})
self.assertEqual(info["neutral"], ["docs/x.md"])
self.assertEqual(info["neutral"], ["docs/ecosystem.md"])

def test_docs_other_than_the_inventory_are_their_own_unit(self):
info = pf.classify(["engine/hooks/demo/detect.py", "docs/guide.md"])
self.assertEqual(set(info["units"]), {"engine-runtime", "docs"})

def test_gates_for_hook_slice_run_hook_and_skill_checks(self):
cmds = pf.gates_for(HOOK_SLICE)
Expand Down Expand Up @@ -204,6 +211,75 @@ def test_fails_engine_and_product_skill_mix_and_names_the_split(self):
self.assertIn("split engine-runtime: engine/hooks/playbook-router/detect.py", res.stdout)
self.assertIn("split product-skill: product/skills/ship-a-detector/SKILL.md", res.stdout)

def test_fails_a_hook_that_also_edits_the_root_readme_and_names_the_split(self):
pr506 = [
"README.md",
"docs/ecosystem.md",
"engine/hooks/handoff-needs-smoke-test/detect.py",
"engine/hooks/handoff-needs-smoke-test/tests/test_hooks.py",
"install.sh",
"tests/test_install.py",
]
res = subprocess.run([sys.executable, SCRIPT, "--dry-run", "--paths"] + pr506, capture_output=True, text=True)
self.assertEqual(res.returncode, 1, res.stdout)
self.assertIn("split docs: README.md", res.stdout)
self.assertIn("split engine-runtime: engine/hooks/handoff-needs-smoke-test/detect.py", res.stdout)

def test_a_hook_with_its_inventory_row_and_install_test_passes(self):
res = subprocess.run(
[sys.executable, SCRIPT, "--dry-run", "--paths", "engine/hooks/demo/detect.py", "docs/ecosystem.md", "tests/test_install.py"],
capture_output=True, text=True,
)
self.assertEqual(res.returncode, 0, res.stdout)
self.assertIn("declare Review Unit: engine-runtime", res.stdout)

def test_unreadable_unit_rules_fail_as_unchecked_not_pass(self):
res = subprocess.run(
[sys.executable, SCRIPT, "--dry-run", "--config", "/nonexistent/drafter.config.json", "--paths"] + PR89,
capture_output=True, text=True,
)
self.assertEqual(res.returncode, pf.UNCHECKED_EXIT, res.stdout + res.stderr)
self.assertIn("unchecked review units", res.stdout)
self.assertNotIn("ok preflight passed", res.stdout)

def test_a_copy_outside_the_repo_classifies_with_an_explicit_config(self):
with tempfile.TemporaryDirectory() as tmp:
copy = os.path.join(tmp, "preflight.py")
shutil.copy2(SCRIPT, copy)
res = subprocess.run(
[sys.executable, copy, "--dry-run", "--config", os.path.join(pf.REPO_ROOT, "drafter.config.json"),
"--paths", "engine/hooks/demo/detect.py", "README.md"],
capture_output=True, text=True, cwd=tmp,
)
self.assertEqual(res.returncode, 1, res.stdout + res.stderr)
self.assertIn("split docs: README.md", res.stdout)


class TestRulesMatchDrafterCore(unittest.TestCase):
def test_every_tracked_path_gets_the_same_units_as_the_pr_body_checker(self):
paths = subprocess.run(
["git", "-C", pf.REPO_ROOT, "ls-files", "-z"], capture_output=True, text=True, check=True,
).stdout.split("\0")
paths = [p for p in paths if p] + [
".mergify.yml", "package-lock.json", "a/b/yarn.lock", "tsconfig.base.json", "e2e/foo/bar.ts",
"x/.hidden/y.md", ".github/workflows/ci.yml", "corpus/skills/a/tests/t.py", "docs/guide.md",
]
script = (
"import { classifyReviewUnitsForPath, loadDrafterConfig } from '@neko-catpital-labs/drafter-core';"
"const paths = JSON.parse(await new Response(process.stdin).text());"
"const config = await loadDrafterConfig({});"
"console.log(JSON.stringify(paths.map((p) => classifyReviewUnitsForPath(p, config))));"
)
res = subprocess.run(
["node", "--input-type=module", "-e", script], input=json.dumps(paths),
capture_output=True, text=True, cwd=pf.REPO_ROOT,
)
self.assertEqual(res.returncode, 0, "drafter-core could not run, so parity is unchecked (run npm ci): " + res.stderr)
expected = json.loads(res.stdout)
rules = pf.load_unit_rules(pf.DEFAULT_CONFIG)
differ = [(p, pf.review_units_for(p, rules), e) for p, e in zip(paths, expected) if pf.review_units_for(p, rules) != e]
self.assertEqual(differ, [])

def test_passes_single_unit_dry_run(self):
res = subprocess.run([sys.executable, SCRIPT, "--dry-run", "--paths"] + PR89, capture_output=True, text=True)
self.assertEqual(res.returncode, 0, res.stdout)
Expand Down
2 changes: 1 addition & 1 deletion scripts/check_ecosystem_boundaries.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@

PATH_STRING_NOT_RUNTIME_IMPORT_EXCEPTIONS = (
("auto-pr/detect.py", "RELEVANT_PREFIXES"),
("make-pr/scripts/preflight.py", "UNIT_RULES"),
("make-pr/scripts/preflight.py", "PROSE_RULE_PREFIXES"),
("make-pr/tests/test_preflight.py", "PR89"),
)

Expand Down
7 changes: 4 additions & 3 deletions tests/test_git_template_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from git_test_repo import disable_background_maintenance, init_repo # noqa: E402

PREFLIGHT = "engine/skills/make-pr/scripts/preflight.py"
UNIT_RULES = "drafter.config.json"
TRACKED_HOOK = "scripts/git-hooks/pre-push"
TEMPLATE_HOOK = "scripts/git-hooks/template-pre-push"
INSTALLER = REPO / "scripts/install-git-template.sh"
Expand Down Expand Up @@ -186,7 +187,7 @@ def clone_and_branch(self, remote_files: tuple[str, ...], url: str) -> Sandbox:
return box

def test_catstack_clone_refuses_mixed_push(self):
box = self.clone_and_branch((PREFLIGHT, TRACKED_HOOK, TEMPLATE_HOOK), CATSTACK_URL)
box = self.clone_and_branch((UNIT_RULES, PREFLIGHT, TRACKED_HOOK, TEMPLATE_HOOK), CATSTACK_URL)
res = box.push("mixed")
self.assertNotEqual(res.returncode, 0, res.stderr)
self.assertIn("more than one review unit", res.stderr)
Expand All @@ -195,15 +196,15 @@ def test_catstack_clone_refuses_mixed_push(self):
self.assertEqual(os.listdir(box.hook_tmp), [])

def test_other_repo_clone_pushes_mixed_branch(self):
box = self.clone_and_branch((PREFLIGHT, TRACKED_HOOK, TEMPLATE_HOOK), OTHER_URL)
box = self.clone_and_branch((UNIT_RULES, PREFLIGHT, TRACKED_HOOK, TEMPLATE_HOOK), OTHER_URL)
res = box.push("mixed")
self.assertEqual(res.returncode, 0, res.stderr)
self.assertNotIn("review unit", res.stderr)
self.assertTrue(box.remote_has("mixed"))
self.assertEqual(os.listdir(box.hook_tmp), [])

def test_catstack_clone_without_tracked_hook_is_refused_as_unchecked(self):
box = self.clone_and_branch((PREFLIGHT, TEMPLATE_HOOK), CATSTACK_URL)
box = self.clone_and_branch((UNIT_RULES, PREFLIGHT, TEMPLATE_HOOK), CATSTACK_URL)
res = box.push("mixed")
self.assertNotEqual(res.returncode, 0, res.stderr)
self.assertIn(UNCHECKED_LINE, res.stderr)
Expand Down
14 changes: 14 additions & 0 deletions tests/test_pre_push_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from git_test_repo import init_repo # noqa: E402

SHIPPED = (
"drafter.config.json",
"engine/skills/make-pr/scripts/preflight.py",
"scripts/git-hooks/pre-push",
"scripts/install-git-hooks.sh",
Expand Down Expand Up @@ -208,6 +209,19 @@ def test_missing_origin_main_on_a_pr_branch_is_refused_as_unchecked(self):
self.assertIn(UNCHECKED_LINE, res.stderr)
self.assertFalse(box.remote_has("single"))

def test_origin_main_without_unit_rules_is_refused_as_unchecked(self):
box = Sandbox(self.root, gh="base:main")
box.git("rm", "-q", "drafter.config.json")
box.git("commit", "-q", "-m", "drop unit rules")
self.assertEqual(box.push("--no-verify", "main").returncode, 0)
box.git("fetch", "-q", "origin")
box.branch("single", SINGLE)
res = box.push("single")
self.assertNotEqual(res.returncode, 0, res.stderr)
self.assertIn("unchecked review units", res.stderr)
self.assertIn("pre-push: UNCHECKED: preflight exited 3 for refs/heads/single", res.stderr)
self.assertFalse(box.remote_has("single"))

def test_branch_deletion_push_is_accepted(self):
box = Sandbox(self.root, gh="base:main")
box.branch("mixed", MIXED)
Expand Down
9 changes: 9 additions & 0 deletions tests/test_validate_pr_body_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from git_test_repo import init_repo # noqa: E402

COPIED = (
"drafter.config.json",
"scripts/validate-pr-body-local.mjs",
"engine/skills/make-pr/scripts/preflight.py",
"engine/skills/draft-pr/scripts/validate-pr-body.mjs",
Expand Down Expand Up @@ -86,6 +87,14 @@ def test_one_unit_without_drafter_core_prints_unchecked(self):
self.assertEqual(res.returncode, 0, res.stdout + res.stderr)
self.assertIn(UNCHECKED, res.stdout)

def test_checkout_without_unit_rules_fails_as_unchecked(self):
self._git("rm", "-q", "drafter.config.json")
self._commit("engine/hooks/x/detect.py")
res = self._run("--body-file", str(self.body), "--base", "main")
self.assertEqual(res.returncode, 1, res.stdout + res.stderr)
self.assertIn("unchecked review units", res.stdout)
self.assertIn("preflight.py exited 3", res.stderr)

def test_missing_body_file_is_usage_error(self):
self._commit("engine/hooks/x/detect.py")
res = self._run("--base", "main")
Expand Down
Loading