diff --git a/engine/skills/make-pr/scripts/preflight.py b/engine/skills/make-pr/scripts/preflight.py index b659e488..138fbdd6 100644 --- a/engine/skills/make-pr/scripts/preflight.py +++ b/engine/skills/make-pr/scripts/preflight.py @@ -90,12 +90,19 @@ def gates_for(paths: list[str], base: str | None = None) -> list[list[str]]: check_skill_test_coverage.py is diff-aware: without the slice refs it defaults to origin/main and can report ok for a slice it never compared, which is a vacuous pass. So it gets --base/--head whenever `base` is real. + + check_codify_has_code.py is diff-aware the same way: with no refs it falls + back to origin/main, so on a stacked slice a sibling's code can satisfy + this slice's prose. Found by scripts/plan_preflight.py on its first run. """ cmds: list[list[str]] = [] if touches_rule_prose(paths): # thrash-reflect-automate: a codified invariant needs code enforcing it. # Pass --allow-prose-only by hand (and say so in the PR) for a docs-only change. - cmds.append(["python3", "scripts/check_codify_has_code.py"]) + cmds.append( + ["python3", "scripts/check_codify_has_code.py"] + + (["--base", base] if base is not None else []) + ) if base is not None: cmds.append(["python3", "scripts/check_no_dated_provenance.py", "--base", base]) for hook in touched_hooks(paths): diff --git a/engine/skills/make-pr/tests/test_preflight.py b/engine/skills/make-pr/tests/test_preflight.py index 8506b13a..a646e1af 100644 --- a/engine/skills/make-pr/tests/test_preflight.py +++ b/engine/skills/make-pr/tests/test_preflight.py @@ -104,6 +104,25 @@ def test_coverage_gate_omits_refs_when_there_is_no_base(self): coverage = [c for c in cmds if "check_skill_test_coverage.py" in " ".join(c)] self.assertEqual(coverage, [["python3", "scripts/check_skill_test_coverage.py"]]) + def test_codify_gate_carries_the_slice_base(self): + """Like the coverage gate, codify-has-code is diff-aware: with no refs it + defaults to origin/main, so on a stacked slice a sibling's code can + satisfy a prose-only slice's rule. Repro'd on a two-slice stack: the + gate exits 0 at origin/main and 1 at the real slice base.""" + cmds = pf.gates_for(PR89, base="origin/main") + codify = [c for c in cmds if "check_codify_has_code.py" in " ".join(c)] + self.assertEqual(len(codify), 1, cmds) + self.assertEqual( + codify[0], + ["python3", "scripts/check_codify_has_code.py", "--base", "origin/main"], + ) + + def test_codify_gate_omits_the_base_when_there_is_none(self): + """Under --paths there is no real ref; the flag must be absent, not 'None'.""" + cmds = pf.gates_for(PR89, base=None) + codify = [c for c in cmds if "check_codify_has_code.py" in " ".join(c)] + self.assertEqual(codify, [["python3", "scripts/check_codify_has_code.py"]]) + def test_gates_for_rule_prose_with_base_includes_dated_provenance_check(self): self.assertIn( ["python3", "scripts/check_no_dated_provenance.py", "--base", "origin/main"], diff --git a/scripts/plan_preflight.py b/scripts/plan_preflight.py new file mode 100755 index 00000000..4361d8ad --- /dev/null +++ b/scripts/plan_preflight.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +"""Answer plan-first Steps 2-4 for paths that do not exist yet. + +make-pr's preflight.py asks "what does this diff trip?" and needs the work +already written. This asks "what will these paths trip?" from a list of files +you intend to touch, so a re-split costs a line edit instead of a rebase. + +Three outputs, one per plan-first step: + +- Step 2, slices: the review unit per planned path, and an explicit split when + more than one unit appears. `preflight.py` fails on a mixed diff; finding + that out here is the entire point. +- Step 3, gates at their scope: the gates those paths trip, each marked + ref-aware or whole-tree. A ref-aware gate invoked with no refs falls back to + its own default and can report clean on work it never compared -- the + vacuous pass this column exists to prevent. +- Step 4, base: whether the named base is current with its remote, and + whether any ancestor in a stack has already merged. + +The unit table and gate list are imported from preflight.py rather than +restated, so the two cannot disagree (principle-bind-to-named-inventory). +Ref-awareness is derived by reading each gate's own argparse flags, not from a +list kept here -- a gate that gains --base starts being reported without an +edit to this file. + + python3 scripts/plan_preflight.py --paths corpus/skills/x/SKILL.md scripts/y.py + python3 scripts/plan_preflight.py --paths ... --base origin/main + python3 scripts/plan_preflight.py --paths ... --json +""" +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "engine" / "skills" / "make-pr" / "scripts")) +import preflight as pf # noqa: E402 + +REF_FLAG_RE = re.compile(r'"--(base|head)"') + + +def ref_flags(gate_path: Path) -> list[str]: + """Which ref flags this gate's own argparse accepts. [] for whole-tree.""" + try: + text = gate_path.read_text(encoding="utf-8") + except OSError: + return [] + return sorted({m.group(1) for m in REF_FLAG_RE.finditer(text)}) + + +def slices_for(paths: list[str]) -> dict: + """Review unit per path, plus the split when units are mixed.""" + info = pf.classify(paths) + units = info["units"] + return { + "units": {u: sorted(ps) for u, ps in sorted(units.items())}, + "neutral": sorted(info["neutral"]), + "mixed": len(units) > 1, + "slice_count": max(len(units), 1), + } + + +def gates_for_plan(paths: list[str], base: str | None) -> list[dict]: + """Each gate the planned paths trip, with its scope contract.""" + out: list[dict] = [] + for cmd in pf.gates_for(paths, base=base): + script = next((a for a in cmd if a.endswith(".py")), None) + flags = ref_flags(REPO_ROOT / script) if script else [] + passed = [a for a in cmd if a.startswith("--")] + out.append( + { + "command": " ".join(cmd), + "script": script, + "accepts_refs": flags, + "scoped": bool(flags) and any(f"--{f}" in passed for f in flags), + "whole_tree": not flags, + } + ) + return out + + +def base_status(base: str) -> dict: + """Is the planned base current, and has anything under it already merged?""" + + def git(*args: str) -> str | None: + try: + r = subprocess.run( + ["git", "-C", str(REPO_ROOT), *args], + capture_output=True, text=True, check=True, + ) + return r.stdout.strip() + except (OSError, subprocess.CalledProcessError): + return None + + resolved = git("rev-parse", "--short", base) + if resolved is None: + return {"ref": base, "resolved": None, "current": None, + "note": "cannot resolve; check the ref name"} + remote = base if base.startswith("origin/") else f"origin/{base}" + remote_sha = git("rev-parse", "--short", remote) + if remote_sha is None: + return {"ref": base, "resolved": resolved, "current": None, + "note": f"no remote counterpart ({remote}); local-only base"} + behind = git("rev-list", "--count", f"{base}..{remote}") + return { + "ref": base, + "resolved": resolved, + "remote": remote, + "remote_resolved": remote_sha, + "current": behind == "0", + "behind_by": int(behind) if behind and behind.isdigit() else None, + "note": "current" if behind == "0" else f"behind {remote} by {behind} commit(s); rebase before planning", + } + + +def render(plan: dict) -> str: + lines: list[str] = [] + s = plan["slices"] + lines.append(f"Step 2 slices: {s['slice_count']}") + for unit, ps in s["units"].items(): + lines.append(f" {unit:<16}{len(ps)} file(s)") + for p in ps: + lines.append(f" {p}") + if s["neutral"]: + lines.append(f" {'neutral':<16}{len(s['neutral'])} file(s): {', '.join(s['neutral'])}") + if s["mixed"]: + lines.append(" SPLIT REQUIRED: preflight.py fails on a mixed-unit diff.") + lines.append(" One slice per unit above, ordered evidence-before-change.") + + lines.append(f"\nStep 3 gates: {len(plan['gates'])}") + for g in plan["gates"]: + if g["whole_tree"]: + mark = "whole-tree" + elif g["scoped"]: + mark = "scoped ok" + else: + mark = "UNSCOPED: may pass vacuously" + lines.append(f" [{mark:<26}] {g['command']}") + unscoped = [g for g in plan["gates"] if not g["whole_tree"] and not g["scoped"]] + if unscoped: + flags = sorted({f"--{f}" for g in unscoped for f in g["accepts_refs"]}) + lines.append(f" {len(unscoped)} gate(s) accept {', '.join(flags)} but were planned without them.") + lines.append(" Pass the slice refs, or the gate compares something other than your slice.") + + b = plan["base"] + lines.append(f"\nStep 4 base: {b['ref']} ({b.get('resolved') or 'unresolved'}) -- {b['note']}") + + if plan["unknowns"]: + lines.append("\nUnknowns this script cannot answer:") + for u in plan["unknowns"]: + lines.append(f" - {u}") + return "\n".join(lines) + + +UNKNOWNS = [ + "Whether each unit is really one reviewable claim, or two sharing a path prefix.", + "Whether a gate not listed here runs in CI but not in preflight.", + "Whether an upstream slice will squash-merge, which rewrites your base mid-stack.", +] + + +def build_plan(paths: list[str], base: str) -> dict: + return { + "paths": sorted(paths), + "slices": slices_for(paths), + "gates": gates_for_plan(paths, base), + "base": base_status(base), + "unknowns": UNKNOWNS, + } + + +def main(argv: list[str] | None = None) -> int: + """Render the plan, and exit 1 when the plan as stated would be rejected + later: a mixed-unit slice, a ref-aware gate planned with no refs, or a + stale base. Advisory-by-exit-code so a planner can gate on it; it never + edits anything.""" + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--paths", nargs="+", required=True, help="repo-relative paths you plan to touch") + ap.add_argument("--base", default="origin/main", help="ref the slices are planned against") + ap.add_argument("--json", action="store_true", help="machine-readable plan") + args = ap.parse_args(argv) + + plan = build_plan(args.paths, args.base) + print(json.dumps(plan, indent=2) if args.json else render(plan)) + + if plan["slices"]["mixed"]: + return 1 + if any(not g["whole_tree"] and not g["scoped"] for g in plan["gates"]): + return 1 + if plan["base"].get("current") is False: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_plan_preflight.py b/tests/test_plan_preflight.py new file mode 100644 index 00000000..d4d159a8 --- /dev/null +++ b/tests/test_plan_preflight.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +"""Tests for scripts/plan_preflight.py. + +The codify-has-code case is the real one: this lever's first run against a +realistic plan reported `check_codify_has_code.py` as unscoped, and reading +scripts/check_codify_has_code.py:91 confirmed it defaults to origin/main while +preflight invoked it with no refs. That is the same vacuous-pass class already +fixed for the coverage gate, found in a sibling nobody had checked. +""" +from __future__ import annotations + +import subprocess +import sys +import unittest +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "scripts")) +import plan_preflight as pp # noqa: E402 + +SCRIPT = REPO / "scripts" / "plan_preflight.py" + +CORPUS_ONLY = ["corpus/skills/principle-x/SKILL.md", "corpus/skills/principle-x/tests/fires_example.md"] +MIXED = ["corpus/skills/principle-x/SKILL.md", "scripts/check_x.py"] +HOOK_SLICE = ["engine/hooks/verdict-flip-watch/detect.py"] + + +class TestSlices(unittest.TestCase): + def test_single_unit_is_one_slice_and_not_mixed(self): + s = pp.slices_for(CORPUS_ONLY) + self.assertEqual(list(s["units"]), ["corpus-lesson"]) + self.assertFalse(s["mixed"]) + self.assertEqual(s["slice_count"], 1) + + def test_corpus_plus_engine_is_flagged_mixed(self): + """preflight.py fails this diff; the plan should say so first.""" + s = pp.slices_for(MIXED) + self.assertTrue(s["mixed"]) + self.assertEqual(s["slice_count"], 2) + self.assertEqual(sorted(s["units"]), ["corpus-lesson", "engine-runtime"]) + + def test_repo_root_tests_are_neutral_not_a_unit(self): + s = pp.slices_for(CORPUS_ONLY + ["tests/test_x.py"]) + self.assertEqual(s["neutral"], ["tests/test_x.py"]) + self.assertFalse(s["mixed"]) + + +class TestRefFlagDerivation(unittest.TestCase): + """Derived from each gate's own argparse, never a list kept here.""" + + def test_coverage_gate_accepts_base_and_head(self): + flags = pp.ref_flags(REPO / "scripts" / "check_skill_test_coverage.py") + self.assertEqual(flags, ["base", "head"]) + + def test_codify_gate_accepts_base(self): + self.assertEqual(pp.ref_flags(REPO / "scripts" / "check_codify_has_code.py"), ["base"]) + + def test_whole_tree_gate_accepts_neither(self): + self.assertEqual(pp.ref_flags(REPO / "scripts" / "check_ecosystem_boundaries.py"), []) + + def test_missing_file_is_treated_as_whole_tree(self): + self.assertEqual(pp.ref_flags(REPO / "scripts" / "no_such_gate.py"), []) + + +class TestGateScoping(unittest.TestCase): + def test_ref_aware_gate_with_a_base_is_scoped(self): + gates = pp.gates_for_plan(CORPUS_ONLY, base="origin/main") + codify = next(g for g in gates if "check_codify_has_code" in g["command"]) + self.assertTrue(codify["scoped"], codify) + self.assertFalse(codify["whole_tree"]) + + def test_ref_aware_gate_without_a_base_is_unscoped(self): + """No base (the --paths case) means no gate can be scoped.""" + gates = pp.gates_for_plan(CORPUS_ONLY, base=None) + ref_aware = [g for g in gates if g["accepts_refs"]] + self.assertTrue(ref_aware, "expected at least one ref-aware gate") + self.assertTrue(all(not g["scoped"] for g in ref_aware)) + + def test_whole_tree_gate_is_never_reported_unscoped(self): + for g in pp.gates_for_plan(CORPUS_ONLY, base="origin/main"): + if g["whole_tree"]: + self.assertFalse(g["scoped"]) + self.assertEqual(g["accepts_refs"], []) + + def test_hook_slice_pulls_in_its_hook_coverage_gate(self): + cmds = " ".join(g["command"] for g in pp.gates_for_plan(HOOK_SLICE, base="origin/main")) + self.assertIn("check_hook_test_coverage.py", cmds) + + +class TestPreflightPassesRefsToBothDiffAwareGates(unittest.TestCase): + """Regression for the defect this lever found on its first run.""" + + def test_codify_gate_receives_the_slice_base(self): + sys.path.insert(0, str(REPO / "engine" / "skills" / "make-pr" / "scripts")) + import preflight as pf + + cmds = pf.gates_for(["corpus/skills/principle-x/SKILL.md"], base="origin/main") + codify = [c for c in cmds if "check_codify_has_code.py" in " ".join(c)] + self.assertEqual(len(codify), 1, cmds) + self.assertIn("--base", codify[0]) + self.assertIn("origin/main", codify[0]) + + def test_codify_gate_omits_refs_when_there_is_no_base(self): + sys.path.insert(0, str(REPO / "engine" / "skills" / "make-pr" / "scripts")) + import preflight as pf + + cmds = pf.gates_for(["corpus/skills/principle-x/SKILL.md"], base=None) + codify = [c for c in cmds if "check_codify_has_code.py" in " ".join(c)] + self.assertEqual(codify, [["python3", "scripts/check_codify_has_code.py"]]) + + +class TestBaseStatus(unittest.TestCase): + def test_origin_main_resolves_and_reports_current(self): + b = pp.base_status("origin/main") + self.assertIsNotNone(b["resolved"]) + self.assertTrue(b["current"], b) + + def test_unresolvable_ref_says_so_instead_of_claiming_current(self): + b = pp.base_status("origin/definitely-not-a-real-ref-xyz") + self.assertIsNone(b["resolved"]) + self.assertIsNone(b["current"]) + self.assertIn("cannot resolve", b["note"]) + + +class TestCli(unittest.TestCase): + def _run(self, *args: str): + return subprocess.run( + [sys.executable, str(SCRIPT), *args], capture_output=True, text=True, cwd=REPO + ) + + def test_mixed_units_exit_1_and_say_split_required(self): + res = self._run("--paths", *MIXED, "--base", "origin/main") + self.assertEqual(res.returncode, 1, res.stdout) + self.assertIn("SPLIT REQUIRED", res.stdout) + + def test_json_output_carries_every_section(self): + res = self._run("--paths", *CORPUS_ONLY, "--base", "origin/main", "--json") + import json + + plan = json.loads(res.stdout) + for key in ("paths", "slices", "gates", "base", "unknowns"): + self.assertIn(key, plan) + + def test_unknowns_are_always_reported(self): + """A plan with no unknowns has not been examined.""" + res = self._run("--paths", *CORPUS_ONLY, "--base", "origin/main") + self.assertIn("Unknowns this script cannot answer", res.stdout) + + def test_paths_is_required(self): + self.assertEqual(self._run().returncode, 2) + + +if __name__ == "__main__": + unittest.main()