minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

Enhance mechanical-refactor-verify skill with a whole-chain verifier, new relocation primitives, and generator inference (#30585)

Browse Source
This commit is contained in:
fzyzcjy
2026-07-14 16:45:48 +08:00
committed by GitHub
parent 0fe2dbd42c
commit a5a71c6c26
30 changed files with 4326 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/skills/mechanical-refactor-verify/scripts/mechanical_refactor_proof_generator.py
+540 -22
View File
@@ -8,18 +8,22 @@ imports were repathed, and the symmetric module-level import diff each file gain
(realised directly with add_import / remove_imported_name, since an import diff is always
whitelisted).
``recipe_to_script`` emits a standalone ``repro_scripts/<sha>.py`` (importing only the
reproduce util); running it reproduces the commit and diffs it byte-for-byte.
reproduce util); running it reproduces the commit, diffs it byte-for-byte, and exits
non-zero unless the diff is empty (PASS).
``generate_range`` writes a whole folder (scripts + output.log + output.html) for a range.
Handles a method moved onto an existing class (call sites lowered), a method moved to a
module-level free function (call sites requalified), a free-function-source move to an
existing module (callers repath their import), and a new-file extract -- where the prep
existing module (callers repath their import), a new-file extract -- where the prep
commit staged the whole module body (scaffolding plus def) as a trailing block in the
source, so the move cuts that tail into the new file (extract_to_new_module). A rename or a
statement-level reorder relocates no def and is reported unsupported. Runnable directly:
source, so the move cuts that tail into the new file (extract_to_new_module) -- and an
intra-file inline-block extract-function (a new helper whose verbatim body is a block carved
from a sibling function, that block replaced by a call). A rename or a statement-level
reorder relocates no def and is reported unsupported. Runnable directly:
python3 mechanical_refactor_proof_generator.py <commit>
python3 mechanical_refactor_proof_generator.py <base>..<tip> --match -move: --out DIR
python3 mechanical_refactor_proof_generator.py <base>..<tip> \
--match '(?<!_)mechanical_provable' --out DIR
"""
import ast
@@ -119,6 +123,30 @@ def _enclosing_class_of_def(tree: ast.AST, name: str) -> str | None:
return None
def _delegate_stub_attr(tree: ast.AST, name: str) -> tuple[str, str] | None:
"""The component attribute a forwarding stub ``def name``: ``return self.<attr>.<m>(...)``
delegates through, with the forwarded method name -- None when no such stub exists.
"""
for node in ast.walk(tree):
if not (
isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name == name
):
continue
if (
len(node.body) == 1
and isinstance(node.body[0], ast.Return)
and isinstance(node.body[0].value, ast.Call)
and isinstance(node.body[0].value.func, ast.Attribute)
and isinstance(node.body[0].value.func.value, ast.Attribute)
and isinstance(node.body[0].value.func.value.value, ast.Name)
and node.body[0].value.func.value.value.id == "self"
):
return (node.body[0].value.func.value.attr, node.body[0].value.func.attr)
return None
return None
def _nested_in_function(tree: ast.AST, name: str) -> bool:
target = rr._find_def(tree, name)
if target is None:
@@ -270,7 +298,9 @@ class Recipe:
target: str
supported: bool = True
moves: list = field(default_factory=list)
assign_moves: list = field(default_factory=list)
extracts: list = field(default_factory=list)
extract_functions: list = field(default_factory=list)
scatter_extracts: list = field(default_factory=list)
lowerings: list = field(default_factory=list)
repaths: list = field(default_factory=list)
@@ -420,6 +450,83 @@ def _next_sibling_def_name(
return None
def _next_sibling_assign_or_def(dst_tree: ast.AST, name: str) -> str | None:
"""The name of the top-level statement (def/class/single-Name assign) that immediately
follows the assignment ``name`` in the destination, or None when it is last."""
def stmt_name(node: ast.AST) -> str | None:
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
return node.name
if (
isinstance(node, ast.Assign)
and len(node.targets) == 1
and isinstance(node.targets[0], ast.Name)
):
return node.targets[0].id
if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
return node.target.id
return None
named = [(stmt_name(n), n) for n in getattr(dst_tree, "body", []) if stmt_name(n)]
for i, (nm, _) in enumerate(named):
if nm == name:
return named[i + 1][0] if i + 1 < len(named) else None
return None
def _stmt_symbol_name(node: ast.AST) -> str | None:
"""The name a top-level statement defines -- a def/class name, or a single-Name
assignment target -- else None (an ``if TYPE_CHECKING:`` guard, a tuple assign, ...).
"""
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
return node.name
if (
isinstance(node, ast.Assign)
and len(node.targets) == 1
and isinstance(node.targets[0], ast.Name)
):
return node.targets[0].id
if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
return node.target.id
return None
def _module_move_anchor(
dst_tree: ast.AST, name: str, into_class: str | None
) -> tuple[str | None, str | None]:
"""The ``(before, after)`` anchor for reinserting the moved def ``name``. Normally
``before=<next sibling def>``. But when a module-level def lands immediately above an
unnameable statement (e.g. an ``if TYPE_CHECKING:`` guard) with a nameable statement
immediately above it, a ``before`` anchor would resolve to the next def *past* that block
and overshoot, so anchor with ``after=<preceding symbol>`` instead."""
before = _next_sibling_def_name(dst_tree, name, into_class)
if into_class is not None:
return before, None
body = list(getattr(dst_tree, "body", []))
idx = next(
(
i
for i, n in enumerate(body)
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
and n.name == name
),
None,
)
if idx is None:
return before, None
following = body[idx + 1] if idx + 1 < len(body) else None
next_is_named_def = isinstance(
following, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
)
if following is None or next_is_named_def:
return before, None
preceding = body[idx - 1] if idx > 0 else None
prev_name = _stmt_symbol_name(preceding) if preceding is not None else None
if prev_name is None:
return before, None
return None, prev_name
def _symbols_form_tail(src_text: str, symbols: list[str]) -> bool:
"""Whether ``symbols`` sit at the end of the source as a contiguous block of defs/classes
and the scaffolding leading into them -- the trailing block a prep commit stages for a
@@ -484,6 +591,183 @@ def _scatter_extract_layout(dst_after: str, symbols: list[str]) -> dict | None:
return {"header": header, "order": [node.name for node in sym_nodes]}
def _iter_defs_with_container(
tree: ast.AST,
) -> list[tuple[str | None, ast.AST]]:
"""(container_class_name_or_None, def_node) for every module-level function and every
method one class deep -- the two nesting depths an extract_function helper can land at.
"""
out: list[tuple[str | None, ast.AST]] = []
for node in getattr(tree, "body", []):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
out.append((None, node))
elif isinstance(node, ast.ClassDef):
for child in node.body:
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
out.append((node.name, child))
return out
def _statements_parse(lines: list[str]) -> bool:
"""Whether ``lines`` (dedented to their own minimum indent) parse as complete Python
statements -- used to keep a prefix/suffix split from cutting through the middle of a
multi-line statement."""
text = "".join(lines)
if not text.strip():
return True
indents = [len(ln) - len(ln.lstrip(" ")) for ln in lines if ln.strip()]
dedented = rr.dedent(text, min(indents)) if indents else text
try:
ast.parse(dedented)
return True
except SyntaxError:
return False
def _common_prefix_suffix(a: list[str], b: list[str]) -> tuple[int, int]:
"""Longest common leading and trailing run of identical lines between two line lists,
kept non-overlapping -- isolates the single contiguous region where they differ. The
greedy match is then shrunk (suffix first, then prefix) until the differing middle of
*both* lists parses as complete statements, so a boundary line the removed block and its
replacement happen to share (e.g. a lone ``)``) is not absorbed mid-statement."""
prefix = 0
while prefix < len(a) and prefix < len(b) and a[prefix] == b[prefix]:
prefix += 1
suffix = 0
while (
suffix < len(a) - prefix
and suffix < len(b) - prefix
and a[-1 - suffix] == b[-1 - suffix]
):
suffix += 1
while suffix > 0 and not (
_statements_parse(a[prefix : len(a) - suffix])
and _statements_parse(b[prefix : len(b) - suffix])
):
suffix -= 1
while prefix > 0 and not (
_statements_parse(a[prefix : len(a) - suffix])
and _statements_parse(b[prefix : len(b) - suffix])
):
prefix -= 1
return prefix, suffix
def _call_names_in(node: ast.AST) -> set[str]:
"""Names invoked as ``self.<name>(...)`` or ``<name>(...)`` anywhere under ``node``."""
names: set[str] = set()
for sub in ast.walk(node):
if isinstance(sub, ast.Call):
if isinstance(sub.func, ast.Attribute):
names.add(sub.func.attr)
elif isinstance(sub.func, ast.Name):
names.add(sub.func.id)
return names
def _infer_extract_functions(
recipe: Recipe, files: dict[str, dict], commit: str, root: str
) -> None:
"""Infer intra-file extract_function ops: a new helper ``H`` whose body is a verbatim block
cut from another function ``F`` of the same file, with ``F``'s block replaced by a call to
``H`` (optionally an ``lhs = self.H(...)`` assignment mirrored by a ``return lhs`` the helper
appends). The relocated body is byte-checked; only the header/call/return are authored. A
body whose reindent does not reconstruct the helper (a bundled edit) yields no op, so the
residual surfaces it instead of a false pass."""
for path, f in files.items():
if f.get("new") or f.get("deleted"):
continue
before_text = _git_output(["show", f"{commit}^:{path}"], root)
after_text = _git_output(["show", f"{commit}:{path}"], root)
try:
before_tree = ast.parse(before_text)
after_tree = ast.parse(after_text)
except SyntaxError:
continue
before_lines = rr._split_keepends(before_text)
after_lines = rr._split_keepends(after_text)
before_defs = _iter_defs_with_container(before_tree)
before_keys = {(c, n.name) for c, n in before_defs}
for container, helper in _iter_defs_with_container(after_tree):
if (container, helper.name) in before_keys:
continue
if not helper.body:
continue
# The signature is the def header only (through its colon), not everything up to
# the first statement -- a leading comment sits between them and belongs to the
# extracted body, not the authored signature.
helper_text = "".join(after_lines[helper.lineno - 1 : helper.end_lineno])
header_len = rr._def_header_end(helper_text)
header_text = "".join(
after_lines[helper.lineno - 1 : helper.lineno - 1 + header_len]
)
# F is the one sibling function that changed and now calls the helper.
candidates = []
for cont, node in before_defs:
after_node = next(
(
n
for c, n in _iter_defs_with_container(after_tree)
if c == cont and n.name == node.name
),
None,
)
if after_node is None or node.name == helper.name:
continue
b_lines = before_lines[node.lineno - 1 : node.end_lineno]
a_lines = after_lines[after_node.lineno - 1 : after_node.end_lineno]
if b_lines == a_lines:
continue
if helper.name not in _call_names_in(after_node):
continue
candidates.append((b_lines, a_lines))
if len(candidates) != 1:
continue
f_before, f_after = candidates[0]
prefix, suffix = _common_prefix_suffix(f_before, f_after)
block = f_before[prefix : len(f_before) - suffix]
call_lines = f_after[prefix : len(f_after) - suffix]
if not block or not call_lines:
continue
body_indent = len(block[0]) - len(block[0].lstrip(" "))
body_text = "".join(block)
# Detect the authored `return <name>` structurally, by statement count -- the
# formatter reflows lines differently at the helper's shallower indent, so a
# byte comparison of the reindented body would spuriously fail; the repro's
# byte-diff (which runs the formatter) is the real arbiter.
try:
block_stmts = ast.parse(rr.dedent(body_text, body_indent)).body
except SyntaxError:
continue
helper_stmts = helper.body
return_text: str | None = None
if len(helper_stmts) == len(block_stmts) + 1 and isinstance(
helper_stmts[-1], ast.Return
):
ret = helper_stmts[-1]
return_text = "".join(
after_lines[ret.lineno - 1 : ret.end_lineno]
).strip("\n")
elif len(helper_stmts) != len(block_stmts):
continue
recipe.extract_functions.append(
{
"src": path,
"dst": path,
"name": helper.name,
"signature": header_text,
"body": body_text,
"body_indent": body_indent,
"call": "".join(call_lines),
"return_text": return_text,
"into_class": container,
"before": _next_sibling_def_name(
after_tree, helper.name, container
),
}
)
def infer_recipe(commit: str, root: str) -> Recipe:
"""Infer a faithful relocation recipe for a move commit from its diff + before-state.
@@ -508,6 +792,9 @@ def infer_recipe(commit: str, root: str) -> Recipe:
if (m := re.match(r"\s*(?:async\s+)?def\s+(\w+)", ln))
}
def class_names(lines: list[str]) -> set[str]:
return {m.group(1) for ln in lines if (m := re.match(r"class\s+(\w+)", ln))}
new_files = {p for p, f in files.items() if f["new"]}
# A new file is a staged module body cut from one source: its top-level defs and classes
@@ -595,28 +882,106 @@ def infer_recipe(commit: str, root: str) -> Recipe:
}
)
# A move whose destination already exists becomes a move_symbol (the def relocated in
# order); a moved class to an existing file is left unsupported (move_symbol moves defs).
all_removed = [ln for f in files.values() for ln in f["removed"]]
all_added = [ln for f in files.values() for ln in f["added"]]
# A top-level class relocated between existing files moves as one block: move_symbol
# cuts the ClassDef; its methods are excluded from the per-def loop below.
moved_classes: set[str] = set()
for cname in sorted(class_names(all_removed) & class_names(all_added)):
csrc = next(
(p for p, f in files.items() if cname in class_names(f["removed"])), None
)
cdst = next(
(p for p, f in files.items() if cname in class_names(f["added"])), None
)
if csrc is None or cdst is None or csrc == cdst or cdst in new_files:
continue
cdst_tree = ast.parse(_git_output(["show", f"{commit}:{cdst}"], root))
cdst_def = rr._find_def(cdst_tree, cname) or next(
(
n
for n in ast.walk(cdst_tree)
if isinstance(n, ast.ClassDef) and n.name == cname
),
None,
)
moved_classes.add(cname)
recipe.moves.append(
{
"name": cname,
"src": csrc,
"dst": cdst,
"into_class": None,
"from_class": None,
"dedent": 0,
"dst_order": cdst_def.lineno if cdst_def else 0,
"before": _next_sibling_def_name(cdst_tree, cname, None),
"drop_self_annotation": False,
}
)
# A move whose destination already exists becomes a move_symbol (the def relocated in
# order).
for name in sorted(def_names(all_removed) & def_names(all_added)):
src = next(
(p for p, f in files.items() if name in def_names(f["removed"])), None
)
dst = next((p for p, f in files.items() if name in def_names(f["added"])), None)
if src is None or dst is None or src == dst or dst in new_files:
dst = next(
(p for p, f in files.items() if name in def_names(f["added"]) and p != src),
None,
)
# A def cut and re-added within the same file (no other file gained it) is an
# in-file reorder -- a move_symbol whose src and dst are that file. A signature or
# body edit that happens to touch the def line is not a faithful move, but the
# reproduction's byte-diff surfaces it as a residual, so this never false-passes.
if dst is None and src is not None and name in def_names(files[src]["added"]):
dst = src
if src is None or dst is None or dst in new_files:
continue
src_before = _git_output(["show", f"{commit}^:{src}"], root)
if _nested_in_function(ast.parse(src_before), name):
recipe.notes.append(f"skip {name}: nested function (moves with parent)")
continue
src_tree = ast.parse(src_before)
src_class = _enclosing_class_of_def(src_tree, name)
dst_tree = ast.parse(_git_output(["show", f"{commit}:{dst}"], root))
into_class = _enclosing_class_of_def(dst_tree, name)
dst_def = rr._find_def(dst_tree, name)
src_indent = _def_indent(files[src]["removed"], name) or 0
dst_indent = _def_indent(files[dst]["added"], name) or 0
src_indent = _def_indent(files[src]["removed"], name)
dst_indent = _def_indent(files[dst]["added"], name)
# The diff's def-line indentation says which same-named def actually moved: a
# column-0 cut is the module-level def even when a class method shares its name.
src_class = None if src_indent == 0 else _enclosing_class_of_def(src_tree, name)
if src_class in moved_classes:
recipe.notes.append(f"skip {name}: method of relocated class {src_class}")
continue
into_class = (
None if dst_indent == 0 else _enclosing_class_of_def(dst_tree, name)
)
try:
src_def = rr._find_unique_def(
src_tree, name, from_class=src_class, where=src
)
dst_def = rr._find_unique_def(
dst_tree, name, from_class=into_class, where=dst
)
except AssertionError as exc:
recipe.supported = False
recipe.notes.append(f"{name}: cannot disambiguate moved def ({exc})")
continue
src_indent = src_indent or 0
dst_indent = dst_indent or 0
# A same-named def re-added to the source is a forwarding delegate the move
# leaves behind: a body of exactly `return self.<attr>.<name>(...)` names the
# component attribute move_symbol authors the stub through.
leave_delegate = None
delegate_name = None
if name in def_names(files[src]["added"]):
src_after_tree = ast.parse(_git_output(["show", f"{commit}:{src}"], root))
stub = _delegate_stub_attr(src_after_tree, name)
if stub is not None:
leave_delegate, forwarded = stub
if forwarded != name:
delegate_name = forwarded
move_before, move_after = _module_move_anchor(dst_tree, name, into_class)
recipe.moves.append(
{
"name": name,
@@ -626,10 +991,11 @@ def infer_recipe(commit: str, root: str) -> Recipe:
"from_class": src_class,
"dedent": src_indent - dst_indent,
"dst_order": dst_def.lineno if dst_def else 0,
"before": _next_sibling_def_name(dst_tree, name, into_class),
"drop_self_annotation": _self_annotation_dropped(
rr._find_def(src_tree, name), dst_def
),
"before": move_before,
"after": move_after,
"drop_self_annotation": _self_annotation_dropped(src_def, dst_def),
"leave_delegate": leave_delegate,
"delegate_name": delegate_name,
}
)
if src_class is not None:
@@ -648,6 +1014,110 @@ def infer_recipe(commit: str, root: str) -> Recipe:
recipe, files, name=name, src=src, dst=dst, commit=commit, root=root
)
# A method whose signature line never changed leaves no def-line in the removed set:
# the body was replaced by a forwarding stub in place while the full body landed in
# another file. Detect it from the destination's added def + the source's after-state
# delegate stub.
for name in sorted(def_names(all_added) - def_names(all_removed)):
dst = next((p for p, f in files.items() if name in def_names(f["added"])), None)
if dst is None or dst in new_files:
continue
src = None
stub_info = None
for p in files:
if p == dst:
continue
try:
after_tree = ast.parse(_git_output(["show", f"{commit}:{p}"], root))
before_tree = ast.parse(_git_output(["show", f"{commit}^:{p}"], root))
except Exception:
continue
stub = _delegate_stub_attr(after_tree, name)
if stub is None:
continue
full_before = rr._find_def(before_tree, name)
if full_before is None or _delegate_stub_attr(before_tree, name):
continue
src, stub_info = p, stub
break
if src is None:
continue
src_before = _git_output(["show", f"{commit}^:{src}"], root)
src_tree = ast.parse(src_before)
dst_tree = ast.parse(_git_output(["show", f"{commit}:{dst}"], root))
src_class = _enclosing_class_of_def(src_tree, name)
dst_indent = _def_indent(files[dst]["added"], name)
into_class = (
None if dst_indent == 0 else _enclosing_class_of_def(dst_tree, name)
)
try:
src_def = rr._find_unique_def(
src_tree, name, from_class=src_class, where=src
)
dst_def = rr._find_unique_def(
dst_tree, name, from_class=into_class, where=dst
)
except AssertionError as exc:
recipe.supported = False
recipe.notes.append(f"{name}: cannot disambiguate moved def ({exc})")
continue
leave_delegate, forwarded = stub_info
move_before, move_after = _module_move_anchor(dst_tree, name, into_class)
recipe.moves.append(
{
"name": name,
"src": src,
"dst": dst,
"into_class": into_class,
"from_class": src_class,
"dedent": (src_def.col_offset or 0) - (dst_indent or 0),
"dst_order": dst_def.lineno if dst_def else 0,
"before": move_before,
"after": move_after,
"drop_self_annotation": _self_annotation_dropped(src_def, dst_def),
"leave_delegate": leave_delegate,
"delegate_name": forwarded if forwarded != name else None,
}
)
# A module-level constant that vanished from one changed file and appeared in another
# relocated with the moved code: realise it as a move_assign.
changed_paths = [p for p in files if p.endswith(".py")]
texts_before: dict = {}
texts_after: dict = {}
for p in changed_paths:
try:
texts_before[p] = (
"" if files[p]["new"] else _git_output(["show", f"{commit}^:{p}"], root)
)
texts_after[p] = _git_output(["show", f"{commit}:{p}"], root)
except Exception:
continue
for p_src in changed_paths:
if p_src not in texts_before:
continue
lost = _module_assign_names(texts_before[p_src]) - _module_assign_names(
texts_after.get(p_src, "")
)
if not lost:
continue
for p_dst in changed_paths:
if p_dst == p_src or p_dst in new_files or p_dst not in texts_after:
continue
gained = _module_assign_names(texts_after[p_dst]) - _module_assign_names(
texts_before.get(p_dst, "")
)
for cname in sorted(lost & gained):
dst_tree = ast.parse(texts_after[p_dst])
recipe.assign_moves.append(
{
"name": cname,
"src": p_src,
"dst": p_dst,
"before": _next_sibling_assign_or_def(dst_tree, cname),
}
)
# Module-level imports a file gained or lost are realised directly from the symmetric
# base<->target diff: a gained name is added (the destination needs the moved code's
# imports, or a caller of a moved free function gains one), a lost name is removed. An
@@ -677,6 +1147,11 @@ def infer_recipe(commit: str, root: str) -> Recipe:
if key not in before_tc:
recipe.typechecking_additions.append({"path": path, "text": stmt})
# An intra-file helper carved out of a sibling function's body (its block replaced by a
# call) is an extract_function -- inferred only when no cross-file move already explains it.
if not recipe.moves:
_infer_extract_functions(recipe, files, commit, root)
# A move source the commit deletes (its defs all relocated, leaving only scaffolding) is
# removed after the moves; move_symbol only cuts defs, it does not delete the emptied file.
move_srcs = {mv["src"] for mv in recipe.moves}
@@ -684,7 +1159,12 @@ def infer_recipe(commit: str, root: str) -> Recipe:
if f.get("deleted") and path in move_srcs:
recipe.deletes.append(path)
if not recipe.moves and not recipe.extracts and not recipe.scatter_extracts:
if (
not recipe.moves
and not recipe.extracts
and not recipe.scatter_extracts
and not recipe.extract_functions
):
recipe.supported = False
if not recipe.notes:
recipe.notes.append(
@@ -745,6 +1225,34 @@ def _recipe_ops(recipe: Recipe) -> list:
"dedent": mv["dedent"],
"drop_self_annotation": mv["drop_self_annotation"],
"before": mv.get("before"),
"after": mv.get("after"),
"leave_delegate": mv.get("leave_delegate"),
"delegate_name": mv.get("delegate_name"),
},
)
)
for am in recipe.assign_moves:
ops.append(
(
"move_assign",
(am["name"],),
{"src": am["src"], "dst": am["dst"], "before": am.get("before")},
)
)
for ex in recipe.extract_functions:
ops.append(
(
"extract_function",
(ex["src"], ex["dst"]),
{
"name": ex["name"],
"signature": ex["signature"],
"body": ex["body"],
"body_indent": ex["body_indent"],
"call": ex["call"],
"return_text": ex["return_text"],
"into_class": ex["into_class"],
"before": ex["before"],
},
)
)
@@ -817,7 +1325,7 @@ def recipe_to_script(recipe: Recipe, subject: str) -> str:
for method, args, kwargs in _recipe_ops(recipe):
rendered = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()]
lines.append(f"r.{method}(" + ", ".join(rendered) + ")")
lines += ["r.run()", ""]
lines += ["residual = r.run()", "sys.exit(1 if residual else 0)", ""]
return "\n".join(lines)
@@ -863,7 +1371,12 @@ def generate_range(
recipe = infer_recipe(commit, root)
script = recipe_to_script(recipe, subject)
(scripts_dir / f"{commit[:9]}.py").write_text(script)
relocates = bool(recipe.moves or recipe.extracts or recipe.scatter_extracts)
relocates = bool(
recipe.moves
or recipe.extracts
or recipe.scatter_extracts
or recipe.extract_functions
)
supported = recipe.supported and relocates
notes = recipe.notes
if supported:
@@ -1022,7 +1535,12 @@ def _main(argv: list[str]) -> int:
recipe, _git_output(["log", "-1", "--format=%s", target], root)
)
)
relocates = bool(recipe.moves or recipe.extracts or recipe.scatter_extracts)
relocates = bool(
recipe.moves
or recipe.extracts
or recipe.scatter_extracts
or recipe.extract_functions
)
if not (recipe.supported and relocates):
print("UNSUPPORTED: " + "; ".join(recipe.notes), file=sys.stderr)
return 1
.claude/skills/mechanical-refactor-verify/scripts/mechanical_refactor_reproduction_cli.py
+537
View File
@@ -0,0 +1,537 @@
"""Verify a whole mechanical-refactor chain: classification, proofs, and a full report.
Every commit in ``base..branch`` must classify itself by carrying exactly one of the two
words ``mechanical_provable`` or ``non_mechanical_provable`` anywhere in its message (the
rest of the message format is free). Every ``mechanical_provable`` commit must ship a
proof script in the proof folder (``<proof>/repro_scripts/<sha-prefix>.py`` or a flat
``<proof>/<sha-prefix>.py``), and running the proof must PASS -- reproduce the commit
byte-for-byte. A ``non_mechanical_provable`` commit carries no machine proof and is left
to human review.
The run prints a markdown report, writes it into the proof folder (``chain_report.md``),
and exits 0 iff the whole chain verifies. Normative contract: spec-reproduction-cli.md.
python3 mechanical_refactor_reproduction_cli.py \
--base <base-commit> --branch <pr-branch-name> --proof path/to/proof/folder
"""
import argparse
import hashlib
import json
import re
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
_DEFAULT_JOBS = 3
_PASSED_CACHE_FILENAME = "mechanical_refactor_passed_proofs.json"
_CACHED_PASS_DETAIL = "reused this machine's earlier PASS (--skip-passed)"
KIND_MECHANICAL = "mechanical_provable"
KIND_NON_MECHANICAL = "non_mechanical_provable"
VERDICT_PASS = "PASS"
VERDICT_FAIL = "FAIL"
VERDICT_MISSING_PROOF = "MISSING_PROOF"
VERDICT_AMBIGUOUS_PROOF = "AMBIGUOUS_PROOF"
VERDICT_HUMAN_REVIEW = "HUMAN_REVIEW"
VERDICT_UNCLASSIFIED = "UNCLASSIFIED"
VERDICT_AMBIGUOUS_KIND = "AMBIGUOUS_KIND"
_OK_VERDICTS = (VERDICT_PASS, VERDICT_HUMAN_REVIEW)
# The words are matched standalone: delimited by any non-[0-9A-Za-z_] character or the
# string boundary, so `non_mechanical_provable` never also counts as the bare word.
_KIND_WORD_RE = re.compile(
r"(?<![0-9A-Za-z_])(non_)?mechanical_provable(?![0-9A-Za-z_])"
)
# The arbiter's verdict line (Repro.run / verify_mechanical_refactor both print `PASS:`).
_PASS_LINE_RE = re.compile(r"^PASS:", re.MULTILINE)
_MIN_PROOF_STEM_LEN = 7
_REPORT_FILENAME = "chain_report.md"
_FAIL_OUTPUT_TAIL_LINES = 60
class ChainVerificationError(Exception):
"""A setup problem (bad refs, non-linear range, missing proof folder): exit code 2."""
@dataclass(frozen=True)
class CommitVerdict:
sha: str
subject: str
kind: "str | None"
verdict: str
detail: str = ""
cached: bool = False
@property
def ok(self) -> bool:
return self.verdict in _OK_VERDICTS
@dataclass(frozen=True)
class _PendingProof:
sha: str
subject: str
kind: str
script: Path
@dataclass(frozen=True)
class ChainResult:
base: str
branch: str
base_sha: str
branch_sha: str
proof_dir: Path
verdicts: "list[CommitVerdict]" = field(default_factory=list)
@property
def passed(self) -> bool:
return bool(self.verdicts) and all(v.ok for v in self.verdicts)
def main(argv: "list[str]") -> int:
parser = argparse.ArgumentParser(
description="Verify a whole mechanical-refactor chain against its proof folder."
)
parser.add_argument("--base", required=True, help="base commit of the chain")
parser.add_argument("--branch", required=True, help="PR branch name (chain tip)")
parser.add_argument("--proof", required=True, help="proof folder path")
parser.add_argument("--repo-root", default=None, help="repo root (default: cwd's)")
parser.add_argument(
"--report",
default=None,
help=f"report file path (default: <proof>/{_REPORT_FILENAME})",
)
parser.add_argument(
"--jobs",
type=int,
default=_DEFAULT_JOBS,
help=f"max concurrent proof runs (default {_DEFAULT_JOBS})",
)
parser.add_argument(
"--skip-passed",
action="store_true",
help="reuse this machine's earlier PASS verdicts for unchanged proofs",
)
args = parser.parse_args(argv)
try:
result = verify_chain(
base=args.base,
branch=args.branch,
proof=Path(args.proof),
repo_root=args.repo_root,
jobs=args.jobs,
skip_passed=args.skip_passed,
)
except ChainVerificationError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
report = render_report(result)
report_path = (
Path(args.report) if args.report else result.proof_dir / _REPORT_FILENAME
)
report_path.write_text(report)
print(report)
print(f"report written to: {report_path}")
return 0 if result.passed else 1
def verify_chain(
*,
base: str,
branch: str,
proof: Path,
repo_root: "str | None" = None,
jobs: int = _DEFAULT_JOBS,
skip_passed: bool = False,
) -> ChainResult:
"""Classify every commit in ``base..branch`` and run every provable commit's proof.
Classification and proof resolution are sequential (cheap); the proof runs execute
concurrently, up to ``jobs`` at a time — safe because each proof works in its own
throwaway worktree. The verdict list keeps chain order. With ``skip_passed``, a
pending proof whose (sha, script hash, utils hash) triple this machine already ran to
a PASS is reused instead of re-executed; every fresh PASS is recorded either way."""
root = repo_root or _repo_root()
if not proof.is_dir():
raise ChainVerificationError(f"proof folder does not exist: {proof}")
base_sha = _rev_parse(base, root)
branch_sha = _rev_parse(branch, root)
commits = _linear_commits(base_sha=base_sha, branch_sha=branch_sha, root=root)
resolved: "list[CommitVerdict | _PendingProof]" = []
for sha in commits:
subject = _git_output(["log", "-1", "--format=%s", sha], root).strip()
message = _git_output(["log", "-1", "--format=%B", sha], root)
resolved.append(
_resolve_commit(sha=sha, subject=subject, message=message, proof=proof)
)
cache_path = _passed_cache_path(root)
cache = _load_passed_cache(cache_path)
if skip_passed:
resolved = [_reuse_cached_pass(item, cache=cache) for item in resolved]
pending_by_sha = {
item.sha: item for item in resolved if isinstance(item, _PendingProof)
}
verdicts: "list[CommitVerdict]" = _run_pending_proofs(
resolved=resolved, root=root, jobs=jobs
)
_record_passes(
cache=cache,
cache_path=cache_path,
verdicts=verdicts,
pending_by_sha=pending_by_sha,
)
return ChainResult(
base=base,
branch=branch,
base_sha=base_sha,
branch_sha=branch_sha,
proof_dir=proof,
verdicts=verdicts,
)
def render_report(result: ChainResult) -> str:
"""The full chain report as markdown: header, per-commit table, failure details."""
n_mech = sum(1 for v in result.verdicts if v.kind == KIND_MECHANICAL)
n_non_mech = sum(1 for v in result.verdicts if v.kind == KIND_NON_MECHANICAL)
n_unclassified = sum(1 for v in result.verdicts if v.kind is None)
n_pass = sum(1 for v in result.verdicts if v.verdict == VERDICT_PASS)
n_cached = sum(1 for v in result.verdicts if v.cached)
lines = [
"# Mechanical refactor chain report",
"",
f"- base: `{result.base}` (`{result.base_sha[:12]}`)",
f"- branch: `{result.branch}` (`{result.branch_sha[:12]}`)",
f"- proof folder: `{result.proof_dir}`",
f"- chain verdict: **{'PASS' if result.passed else 'FAIL'}**",
f"- commits: {len(result.verdicts)} total — {n_mech} {KIND_MECHANICAL}, "
f"{n_non_mech} {KIND_NON_MECHANICAL}, {n_unclassified} classification error(s)",
f"- proofs: {n_pass}/{n_mech} PASS",
*(
[f"- reused from the passed-proof cache (--skip-passed): {n_cached}"]
if n_cached
else []
),
"",
"| # | commit | kind | verdict | subject |",
"|---|--------|------|---------|---------|",
]
for i, v in enumerate(result.verdicts, start=1):
kind = v.kind or "?"
subject = v.subject.replace("|", "\\|")
lines.append(f"| {i} | `{v.sha[:9]}` | {kind} | {v.verdict} | {subject} |")
failures = [v for v in result.verdicts if not v.ok]
if failures:
lines += ["", "## Failure details"]
for v in failures:
lines += [
"",
f"### `{v.sha[:9]}` — {v.verdict}",
"",
v.detail or "(no detail)",
]
return "\n".join(lines) + "\n"
def _reuse_cached_pass(
item: "CommitVerdict | _PendingProof", *, cache: dict
) -> "CommitVerdict | _PendingProof":
"""Turn a pending proof into a cached PASS verdict on an exact cache-key match."""
if not isinstance(item, _PendingProof):
return item
if cache.get("passed", {}).get(item.sha) != _proof_cache_key(item.script):
return item
print(f"proof {item.sha[:9]} {VERDICT_PASS} (cached)", flush=True)
return CommitVerdict(
sha=item.sha,
subject=item.subject,
kind=item.kind,
verdict=VERDICT_PASS,
detail=_CACHED_PASS_DETAIL,
cached=True,
)
def _record_passes(
*,
cache: dict,
cache_path: Path,
verdicts: "list[CommitVerdict]",
pending_by_sha: "dict[str, _PendingProof]",
) -> None:
"""Record every freshly-run PASS into the cache (a FAIL is never recorded)."""
fresh = [
v
for v in verdicts
if v.verdict == VERDICT_PASS and not v.cached and v.sha in pending_by_sha
]
if not fresh:
return
for v in fresh:
cache.setdefault("passed", {})[v.sha] = _proof_cache_key(
pending_by_sha[v.sha].script
)
try:
cache_path.write_text(json.dumps(cache, indent=2, sort_keys=True) + "\n")
except OSError as exc:
print(f"note: could not write passed-proof cache {cache_path}: {exc}")
def _proof_cache_key(script: Path) -> "dict[str, str]":
"""The cache key parts beyond the sha: hashes of the script and its utils module."""
utils_sha256 = ""
for directory in (script.parent, script.parent.parent):
utils = directory / "mechanical_refactor_reproduction_utils.py"
if utils.is_file():
utils_sha256 = hashlib.sha256(utils.read_bytes()).hexdigest()
break
return {
"script_sha256": hashlib.sha256(script.read_bytes()).hexdigest(),
"utils_sha256": utils_sha256,
}
def _passed_cache_path(root: str) -> Path:
common_dir = _git_output(["rev-parse", "--git-common-dir"], root).strip()
common = Path(common_dir)
if not common.is_absolute():
common = Path(root) / common
return common / _PASSED_CACHE_FILENAME
def _load_passed_cache(path: Path) -> dict:
"""The cache is best-effort: missing, corrupt, or unreadable means empty."""
try:
data = json.loads(path.read_text())
except (OSError, ValueError):
return {"passed": {}}
if not isinstance(data, dict) or not isinstance(data.get("passed"), dict):
return {"passed": {}}
return data
def _run_pending_proofs(
*, resolved: "list[CommitVerdict | _PendingProof]", root: str, jobs: int
) -> "list[CommitVerdict]":
"""Execute the pending proofs on a bounded thread pool; keep chain order."""
pending = [
(i, item) for i, item in enumerate(resolved) if isinstance(item, _PendingProof)
]
finished: "dict[int, CommitVerdict]" = {}
if pending:
with ThreadPoolExecutor(max_workers=max(1, jobs)) as pool:
futures = {
i: pool.submit(_proof_verdict, item, root=root) for i, item in pending
}
for i, future in futures.items():
finished[i] = future.result()
return [
finished[i] if isinstance(item, _PendingProof) else item
for i, item in enumerate(resolved)
]
def _proof_verdict(pending: _PendingProof, *, root: str) -> CommitVerdict:
passed, output = _run_proof(script=pending.script, root=root)
if passed:
verdict = CommitVerdict(
sha=pending.sha,
subject=pending.subject,
kind=pending.kind,
verdict=VERDICT_PASS,
detail="",
)
else:
tail = "\n".join(output.splitlines()[-_FAIL_OUTPUT_TAIL_LINES:])
verdict = CommitVerdict(
sha=pending.sha,
subject=pending.subject,
kind=pending.kind,
verdict=VERDICT_FAIL,
detail=(
f"proof `{pending.script}` did not PASS; output tail:\n\n"
f"```\n{tail}\n```"
),
)
print(f"proof {pending.sha[:9]} {verdict.verdict}", flush=True)
return verdict
def _resolve_commit(
*, sha: str, subject: str, message: str, proof: Path
) -> "CommitVerdict | _PendingProof":
kind, classification_error = _classify(message)
if kind is None:
return CommitVerdict(
sha=sha,
subject=subject,
kind=None,
verdict=classification_error,
detail=(
f"the commit message must contain exactly one of the words "
f"`{KIND_MECHANICAL}` or `{KIND_NON_MECHANICAL}`"
),
)
if kind == KIND_NON_MECHANICAL:
return CommitVerdict(
sha=sha,
subject=subject,
kind=kind,
verdict=VERDICT_HUMAN_REVIEW,
detail="declared non_mechanical_provable: no machine proof, review by hand",
)
scripts = _find_proof_scripts(proof=proof, sha=sha)
if not scripts:
return CommitVerdict(
sha=sha,
subject=subject,
kind=kind,
verdict=VERDICT_MISSING_PROOF,
detail=(
f"no proof script found; searched `{proof / 'repro_scripts'}` and "
f"`{proof}` for `<sha-prefix>.py` (>= {_MIN_PROOF_STEM_LEN} hex chars)"
),
)
if len(scripts) > 1:
listing = ", ".join(f"`{p}`" for p in scripts)
return CommitVerdict(
sha=sha,
subject=subject,
kind=kind,
verdict=VERDICT_AMBIGUOUS_PROOF,
detail=f"multiple proof scripts match this commit: {listing}",
)
return _PendingProof(sha=sha, subject=subject, kind=kind, script=scripts[0])
def _classify(message: str) -> "tuple[str | None, str]":
"""The commit's declared kind, or (None, error-verdict) when the word rule is broken.
Exactly one of the two words must appear (any number of times, but only one of the
two): zero occurrences is UNCLASSIFIED, both words present is AMBIGUOUS_KIND."""
kinds = {
KIND_NON_MECHANICAL if match.group(1) else KIND_MECHANICAL
for match in _KIND_WORD_RE.finditer(message)
}
if not kinds:
return None, VERDICT_UNCLASSIFIED
if len(kinds) > 1:
return None, VERDICT_AMBIGUOUS_KIND
return kinds.pop(), ""
def _find_proof_scripts(*, proof: Path, sha: str) -> "list[Path]":
"""Proof scripts naming this commit: a ``<sha-prefix>.py`` (lowercase hex, >= 7 chars)
under ``<proof>/repro_scripts/`` or flat in ``<proof>/``."""
found: "list[Path]" = []
for directory in (proof / "repro_scripts", proof):
if not directory.is_dir():
continue
for path in sorted(directory.glob("*.py")):
stem = path.stem
is_sha_prefix = (
len(stem) >= _MIN_PROOF_STEM_LEN
and all(c in "0123456789abcdef" for c in stem)
and sha.startswith(stem)
)
if is_sha_prefix:
found.append(path)
return found
def _run_proof(*, script: Path, root: str) -> "tuple[bool, str]":
"""Run one proof script from the repo root. A PASS is exit code 0 AND the arbiter's
``PASS:`` verdict line on stdout (an old-style script that exits 0 with a residual is
therefore still a FAIL)."""
result = subprocess.run(
[sys.executable, str(script.resolve())],
cwd=root,
capture_output=True,
text=True,
)
output = result.stdout + result.stderr
passed = result.returncode == 0 and bool(_PASS_LINE_RE.search(result.stdout))
return passed, output
def _linear_commits(*, base_sha: str, branch_sha: str, root: str) -> "list[str]":
if not _is_ancestor(base_sha=base_sha, branch_sha=branch_sha, root=root):
raise ChainVerificationError(
f"base {base_sha[:12]} is not an ancestor of branch {branch_sha[:12]}"
)
commits = _git_output(
["rev-list", "--reverse", f"{base_sha}..{branch_sha}"], root
).split()
if not commits:
raise ChainVerificationError(
f"no commits in {base_sha[:12]}..{branch_sha[:12]}"
)
merges = [
sha
for sha in commits
if len(_git_output(["rev-list", "--parents", "-n", "1", sha], root).split()) > 2
]
if merges:
listing = ", ".join(sha[:9] for sha in merges)
raise ChainVerificationError(
f"the chain must be linear, but it contains merge commit(s): {listing}"
)
return commits
def _is_ancestor(*, base_sha: str, branch_sha: str, root: str) -> bool:
result = subprocess.run(
["git", "merge-base", "--is-ancestor", base_sha, branch_sha],
cwd=root,
capture_output=True,
)
return result.returncode == 0
def _rev_parse(ref: str, root: str) -> str:
result = subprocess.run(
["git", "rev-parse", "--verify", f"{ref}^{{commit}}"],
cwd=root,
capture_output=True,
text=True,
)
if result.returncode != 0:
raise ChainVerificationError(f"cannot resolve {ref!r}: {result.stderr.strip()}")
return result.stdout.strip()
def _git_output(args: "list[str]", root: str) -> str:
result = subprocess.run(
["git", *args], cwd=root, capture_output=True, text=True, check=True
)
return result.stdout
def _repo_root() -> str:
return subprocess.run(
["git", "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
.claude/skills/mechanical-refactor-verify/scripts/mechanical_refactor_reproduction_utils.py
+361 -38
View File
@@ -150,9 +150,10 @@ def _find_def(tree: ast.AST, name: str) -> ast.AST | None:
def _find_unique_def(
tree: ast.AST, name: str, *, from_class: str | None = None, where: str
) -> ast.AST:
"""Resolve ``def name`` and refuse ambiguity: with same-named defs in scope the
first-match lookup could silently cut the wrong body, so the caller must scope the
search with ``from_class``."""
"""Resolve ``def name`` (or ``class name``) and refuse ambiguity: with same-named defs in
scope the first-match lookup could silently cut the wrong body, so the caller must scope
the search with ``from_class``."""
definition = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
root: ast.AST = tree
if from_class is not None:
cls = _find_class(tree, from_class)
@@ -162,16 +163,14 @@ def _find_unique_def(
top_level = [
node
for node in tree.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name == name
if isinstance(node, definition) and node.name == name
]
if len(top_level) == 1:
return top_level[0]
matches = [
node
for node in ast.walk(root)
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name == name
if isinstance(node, definition) and node.name == name
]
assert matches, f"{name} not found in {where}"
assert (
@@ -209,6 +208,19 @@ def _def_span(node: ast.AST) -> tuple[int, int]:
return start, node.end_lineno
def _symbol_named(node: ast.AST, name: str) -> bool:
"""Whether a top-level statement defines the symbol ``name`` -- a def/class by its name,
or a module-level assignment by one of its target names (so ``_is_hip = is_hip()`` is
found by ``_is_hip``)."""
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
return node.name == name
if isinstance(node, ast.AnnAssign):
return isinstance(node.target, ast.Name) and node.target.id == name
if isinstance(node, ast.Assign):
return any(isinstance(t, ast.Name) and t.id == name for t in node.targets)
return False
def _byte_slice(line: str, start: int | None, end: int | None) -> str:
"""Slice a line by UTF-8 byte offsets -- ast col_offsets count bytes, not characters."""
return line.encode("utf-8")[start:end].decode("utf-8")
@@ -340,13 +352,20 @@ def _multiline_string_interior_lines(top_level_text: str) -> set[int]:
def _audit_extract_header(
header: str, removed_assigns: dict[str, str | None], where: str
header: str,
removed_assigns: dict[str, str | None],
where: str,
rederivable: dict[str, str | None] | None = None,
) -> None:
"""Refuse header content the extraction cannot vouch for. The header of a scattered
extraction is authored text reproduced from the target commit, so anything beyond
imports, a TYPE_CHECKING import block, a logger, or a byte-equivalent copy of an
assignment deleted from the source would let arbitrary new code ride into the new
imports, a TYPE_CHECKING import block, a logger, a byte-equivalent copy of an
assignment deleted from the source (``removed_assigns``), or a byte-equivalent copy of
a module constant that *survives* in the source (``rederivable`` -- re-derived
boilerplate such as ``_is_hip = is_hip()``, provably not fiction because the same
statement still exists in the source) would let arbitrary new code ride into the new
module under a PASS verdict."""
rederivable = rederivable or {}
header_assigned: set[str] = set()
for stmt in ast.parse(header).body:
if isinstance(stmt, (ast.Import, ast.ImportFrom)):
@@ -374,6 +393,10 @@ def _audit_extract_header(
):
header_assigned.update(names)
continue
if names and all(
n in rederivable and rederivable[n] == value_src for n in names
):
continue
raise AssertionError(
f"unverifiable header statement in {where}: {ast.unparse(stmt)!r} is "
f"neither scaffolding nor a relocated source assignment"
@@ -455,6 +478,44 @@ class Repro:
self.ops.append(op)
return self
def route_call_sites_through_field(
self, name: str, *, field: str, paths: list[str], owner: str | None = None
) -> "Repro":
"""Rewrite ``<recv>.name(args)`` to ``<recv>.field.name(args)`` -- the method moved
onto a collaborator reached through ``self.field``, so its callers route through that
field. With ``owner`` given, only calls whose receiver text equals ``owner`` are
rewritten. A call already routed through ``field`` is skipped, so the pass converges.
"""
def op(root: Path) -> None:
for rel in paths:
path = root / rel
def predicate(node: ast.Call) -> bool:
return (
isinstance(node.func, ast.Attribute)
and node.func.attr == name
and not (
isinstance(node.func.value, ast.Attribute)
and node.func.value.attr == field
)
and (owner is None or ast.unparse(node.func.value) == owner)
)
def rewrite(text: str, node: ast.Call) -> str:
call_src = _node_slice(text, node)
func_src = _node_slice(text, node.func)
receiver_src = _node_slice(text, node.func.value)
return receiver_src + f".{field}.{name}" + call_src[len(func_src) :]
_write_source(
path,
_rewrite_matching_calls(_read_source(path), predicate, rewrite),
)
self.ops.append(op)
return self
def remove_import(
self, rel: str, import_text: str, *, in_function: str | None = None
) -> "Repro":
@@ -546,7 +607,13 @@ class Repro:
return self
def remove_imported_name(
self, rel: str, *, module: str | None, name: str, asname: str | None = None
self,
rel: str,
*,
module: str | None,
name: str,
asname: str | None = None,
keep_exploded: bool = False,
) -> "Repro":
"""Drop a single imported ``name`` from a module-level import: from a ``from module
import a, b`` keep the rest and drop only ``name``; when it was the sole name -- or for
@@ -554,6 +621,12 @@ class Repro:
home changed, so an importer that no longer references it loses exactly that name; the
import sorter rewrites the surviving line. An import diff is always whitelisted, so this
realises a lost name directly instead of relying on the formatter to prune it.
Removing down to a single surviving name collapses the import to one line by default
(the common case). Pass ``keep_exploded`` when the target kept the sole survivor
exploded (its magic trailing comma preserved): the alias line is then merely deleted
so the surviving name keeps its comma and the formatter leaves the import multi-line.
The choice is the commit author's and cannot be inferred from the source.
"""
def alias_text(alias: ast.alias) -> str:
@@ -584,16 +657,26 @@ class Repro:
edits.append((node.lineno, node.end_lineno, None))
continue
stmt_lines = lines[node.lineno - 1 : node.end_lineno]
if any("#" in ln for ln in stmt_lines):
own = dropped_alias.lineno
own_line = lines[own - 1]
assert own_line.strip().rstrip(",").strip() == alias_text(
dropped_alias
), (
own = dropped_alias.lineno
own_line = lines[own - 1]
on_own_line = own_line.strip().rstrip(",").strip() == alias_text(
dropped_alias
)
has_comments = any("#" in ln for ln in stmt_lines)
# Preserve the exploded form -- delete just this alias's line -- when the
# import stays multi-line: 2+ surviving names keep the magic trailing comma
# exploded, and an import carrying comments must not be rebuilt (a rebuild
# would drop them). Both match how the target was edited (a flat rebuild
# would drop the magic comma and collapse an import the target left
# multi-line). A lone surviving name collapses to one line by default, unless
# keep_exploded says the target preserved the magic comma for it too.
if on_own_line and (len(kept) >= 2 or has_comments or keep_exploded):
edits.append((own, own, None))
elif has_comments and not on_own_line:
raise AssertionError(
f"cannot drop {name!r}: it shares a line with other text and "
f"the import holds comments that a rebuild would delete"
)
edits.append((own, own, None))
else:
keyword = "import " if module is None else f"from {module} import "
rebuilt = keyword + ", ".join(alias_text(a) for a in kept) + nl
@@ -609,15 +692,81 @@ class Repro:
self.ops.append(op)
return self
def add_import(self, rel: str, import_stmt: str) -> "Repro":
def add_imported_name(
self, rel: str, *, module: str, name: str, asname: str | None = None
) -> "Repro":
"""Add a single ``name`` to an existing module-level ``from module import a, b`` --
the dual of ``remove_imported_name``. A relocated symbol gains a new importer that
already imports other names from the same module, so the target extends that line
rather than adding a fresh statement (which the sorter would not merge across an
intervening non-import statement). The import sorter rewrites the surviving line; an
import carrying comments is refused, since a rebuild would drop them."""
def alias_text(target_name: str, target_asname: str | None) -> str:
return target_name + (f" as {target_asname}" if target_asname else "")
def op(root: Path) -> None:
path = root / rel
lines = _split_keepends(_read_source(path))
nl = _newline_style("".join(lines))
for node in ast.parse("".join(lines)).body:
if not isinstance(node, ast.ImportFrom):
continue
if "." * node.level + (node.module or "") != module:
continue
stmt_lines = lines[node.lineno - 1 : node.end_lineno]
if any("#" in ln for ln in stmt_lines):
raise AssertionError(
f"cannot add {name!r} to the import from {module!r} in {rel}: "
f"it holds comments that a rebuild would delete"
)
existing = [alias_text(a.name, a.asname) for a in node.names]
added = alias_text(name, asname)
assert (
added not in existing
), f"{name!r} already imported from {module!r} in {rel}"
rebuilt = f"from {module} import " + ", ".join(existing + [added]) + nl
lines[node.lineno - 1 : node.end_lineno] = [rebuilt]
_write_source(path, "".join(lines))
return
raise AssertionError(f"no `from {module} import` statement in {rel}")
self.ops.append(op)
return self
def add_import(
self, rel: str, import_stmt: str, *, after: str | None = None
) -> "Repro":
"""Append an import after the last top-level import; the formatter's import sorter
places it (so the exact insertion point does not matter)."""
places it (so the exact insertion point does not matter). When ``after`` is given,
insert immediately after the top-level import statement whose source text contains
that substring instead -- needed for a file whose imports are split into separate
isort sections by an intervening statement (e.g. ``_is_hip = is_hip()``), where the
sorter will not carry the new import across the boundary into the intended block.
"""
def op(root: Path) -> None:
path = root / rel
lines = _split_keepends(_read_source(path))
nl = _newline_style("".join(lines))
body = ast.parse("".join(lines)).body
if after is not None:
anchor = None
for node in body:
if isinstance(
node, (ast.Import, ast.ImportFrom)
) and after in "".join(lines[node.lineno - 1 : node.end_lineno]):
anchor = node
break
if anchor is None:
raise AssertionError(
f"no top-level import containing {after!r} in {rel}"
)
at = anchor.end_lineno
_write_source(
path, "".join(lines[:at] + [import_stmt + nl] + lines[at:])
)
return
last = 0
if (
body
@@ -639,24 +788,53 @@ class Repro:
def add_typechecking_import(self, rel: str, import_stmt: str) -> "Repro":
"""Append ``import_stmt`` inside the file's ``if TYPE_CHECKING:`` block -- a moved
definition whose annotations reference a type needs that type imported there. The
import sorter orders the block, so the exact insertion point does not matter."""
import sorter orders the block, so the exact insertion point does not matter. A lone
``pass`` placeholder (the block's only statement) is dropped: populating an empty
``TYPE_CHECKING`` block makes its placeholder redundant, so the target removes it.
With no existing block, one is created after the trailing module import -- the
destination gains the guard together with its first import.
"""
def op(root: Path) -> None:
path = root / rel
lines = _split_keepends(_read_source(path))
nl = _newline_style("".join(lines))
for node in ast.parse("".join(lines)).body:
if isinstance(node, ast.If) and ast.unparse(node.test) in (
"TYPE_CHECKING",
"typing.TYPE_CHECKING",
):
indent = " " * node.body[0].col_offset
at = node.body[-1].end_lineno
lines.insert(
at, indent + import_stmt + _newline_style("".join(lines))
lone_pass = len(node.body) == 1 and isinstance(
node.body[0], ast.Pass
)
if lone_pass:
placeholder = node.body[0]
lines[placeholder.lineno - 1 : placeholder.end_lineno] = [
indent + import_stmt + nl
]
else:
lines.insert(
node.body[-1].end_lineno, indent + import_stmt + nl
)
_write_source(path, "".join(lines))
return
raise AssertionError(f"no `if TYPE_CHECKING:` block in {rel}")
tree = ast.parse("".join(lines))
imports = [
node
for node in tree.body
if isinstance(node, (ast.Import, ast.ImportFrom))
]
assert (
imports
), f"no imports to anchor a new `if TYPE_CHECKING:` block in {rel}"
insert_at = imports[-1].end_lineno
lines[insert_at:insert_at] = [
nl,
"if TYPE_CHECKING:" + nl,
" " + import_stmt + nl,
]
_write_source(path, "".join(lines))
self.ops.append(op)
return self
@@ -697,6 +875,80 @@ class Repro:
self.ops.append(op)
return self
def move_assign(
self,
name: str,
*,
src: str,
dst: str,
before: str | None = None,
) -> "Repro":
"""Cut the module-level assignment binding ``name`` from ``src`` and paste it verbatim
into ``dst`` at module level -- a module constant relocated together with the code
that reads it. Pasted immediately above the top-level statement named ``before``
when given, else after the last top-level import."""
def op(root: Path) -> None:
src_path = root / src
dst_path = root / dst
src_lines = _split_keepends(_read_source(src_path))
node = None
for cand in ast.parse("".join(src_lines)).body:
if (
isinstance(cand, ast.Assign)
and len(cand.targets) == 1
and isinstance(cand.targets[0], ast.Name)
and cand.targets[0].id == name
) or (
isinstance(cand, ast.AnnAssign)
and isinstance(cand.target, ast.Name)
and cand.target.id == name
):
node = cand
assert node is not None, f"module assignment {name} not found in {src}"
block = "".join(src_lines[node.lineno - 1 : node.end_lineno])
_write_source(
src_path,
"".join(src_lines[: node.lineno - 1] + src_lines[node.end_lineno :]),
)
dst_lines = _split_keepends(_read_source(dst_path))
dst_nl = _newline_style("".join(dst_lines))
dst_tree = ast.parse("".join(dst_lines))
at = None
if before is not None:
for cand in dst_tree.body:
cand_name = getattr(cand, "name", None) or (
cand.targets[0].id
if isinstance(cand, ast.Assign)
and len(cand.targets) == 1
and isinstance(cand.targets[0], ast.Name)
else None
)
if cand_name == before:
at = (
min(
[d.lineno for d in getattr(cand, "decorator_list", [])],
default=cand.lineno,
)
- 1
)
break
assert at is not None, f"before={before!r} not found in {dst}"
dst_lines[at:at] = [block, dst_nl]
else:
imports = [
n
for n in dst_tree.body
if isinstance(n, (ast.Import, ast.ImportFrom))
]
at = imports[-1].end_lineno if imports else 0
dst_lines[at:at] = [dst_nl, block]
_write_source(dst_path, "".join(dst_lines))
self.ops.append(op)
return self
def move_symbol(
self,
name: str,
@@ -708,16 +960,23 @@ class Repro:
dedent: int = 0,
drop_self_annotation: bool = False,
before: str | None = None,
after: str | None = None,
leave_delegate: str | None = None,
delegate_name: str | None = None,
) -> "Repro":
"""Cut ``def name`` (with decorators) from ``src`` and paste it into ``dst`` --
immediately above the sibling def ``before`` when given (so the relocated def lands in
the chain's order), else at the end of ``into_class`` (or module level when None) --
dropping a move decorator and dedenting by ``dedent``. When ``drop_self_annotation``,
the moved method's ``self: Target`` annotation is dropped (redundant inside the class).
The body is moved verbatim; the formatter normalises the surrounding blank lines.
the chain's order), immediately below the top-level symbol ``after`` when given (a
sibling def/class or a module-level assignment target -- used to land the def just
before a following ``if TYPE_CHECKING:`` guard, which is not a nameable anchor), else
at the end of ``into_class`` (or module level when None) -- dropping a move decorator
and dedenting by ``dedent``. When ``drop_self_annotation``, the moved method's
``self: Target`` annotation is dropped (redundant inside the class). The body is moved
verbatim; the formatter normalises the surrounding blank lines.
"""
assert (
before is None or after is None
), "move_symbol: before and after are mutually exclusive"
def op(root: Path) -> None:
src_path = root / src
@@ -731,10 +990,20 @@ class Repro:
block = src_lines[start - 1 : end]
decorator_lines = node.lineno - start
if leave_delegate is not None:
assert not any(
ln.strip() in _MOVE_DECORATORS for ln in block[:decorator_lines]
), f"leave_delegate on a {_MOVE_DECORATORS} method has no self to forward"
args = node.args
has_move_decorator = any(
ln.strip() in _MOVE_DECORATORS for ln in block[:decorator_lines]
)
arg_list = args.posonlyargs + args.args
self_annotated = (
bool(arg_list)
and arg_list[0].arg == "self"
and arg_list[0].annotation is not None
)
assert not has_move_decorator or self_annotated, (
f"leave_delegate on a {_MOVE_DECORATORS} method has no self to "
"forward (a de-self'd staticmethod must annotate its self param)"
)
parts = [p.arg for p in args.posonlyargs + args.args if p.arg != "self"]
if args.vararg is not None:
parts.append(f"*{args.vararg.arg}")
@@ -750,7 +1019,27 @@ class Repro:
- 1
+ _def_header_end("".join(src_lines[node.lineno - 1 : end]))
)
signature = src_lines[start - 1 : header_end]
sig_start = node.lineno - 1 if has_move_decorator else start - 1
signature_text = "".join(src_lines[sig_start:header_end])
# The stub drops the self annotation only when it names the class the
# def moved into (now redundant); an unrelated annotation (a mixin's
# `self: ModelRunner`) is part of the surviving header and stays.
ann = arg_list[0].annotation if self_annotated else None
ann_name = None
if isinstance(ann, ast.Name):
ann_name = ann.id
elif isinstance(ann, ast.Constant) and isinstance(ann.value, str):
ann_name = ann.value.split(".")[-1]
elif isinstance(ann, ast.Attribute):
ann_name = ann.attr
if self_annotated and ann_name == into_class:
sig_indent = len(signature_text) - len(signature_text.lstrip(" "))
parsable = signature_text + " " * sig_indent + " pass" + src_nl
stripped = _drop_self_annotation(parsable, name)
assert stripped.endswith(" " * sig_indent + " pass" + src_nl)
signature_text = stripped[
: -len(" " * sig_indent + " pass" + src_nl)
]
body_indent = " " * node.body[0].col_offset
returning = (
"return await"
@@ -761,7 +1050,7 @@ class Repro:
f"{body_indent}{returning} self.{leave_delegate}."
f"{delegate_name or name}({', '.join(parts)})" + src_nl
)
delegate = "".join(signature) + forward
delegate = signature_text + forward
_write_source(
src_path,
"".join(src_lines[: start - 1] + [delegate] + src_lines[end:]),
@@ -810,7 +1099,18 @@ class Repro:
None,
)
assert target is not None, f"before={before!r} not found in {dst}"
if target is not None:
if after is not None:
anchor = next(
(n for n in container if _symbol_named(n, after)),
None,
)
assert anchor is not None, f"after={after!r} not found in {dst}"
at = anchor.end_lineno
_write_source(
dst_path,
"".join(dst_lines[:at] + [dst_nl, method_text] + dst_lines[at:]),
)
elif target is not None:
at = _def_span(target)[0] - 1
_write_source(
dst_path,
@@ -992,8 +1292,23 @@ class Repro:
assert (
found_assigns == dropped
), f"{dropped - found_assigns} not assigned in {src}"
rederivable: dict[str, str | None] = {}
for node in tree.body:
targets = (
node.targets
if isinstance(node, ast.Assign)
else [node.target] if isinstance(node, ast.AnnAssign) else []
)
names = [t.id for t in targets if isinstance(t, ast.Name)]
if not names or set(names) & dropped:
continue
value_src = ast.unparse(node.value) if node.value is not None else None
for kept_name in names:
rederivable[kept_name] = value_src
if header.strip() or removed_assigns:
_audit_extract_header(header, removed_assigns, where=dst)
_audit_extract_header(
header, removed_assigns, where=dst, rederivable=rederivable
)
cuts = [(start, end, None) for start, end in spans.values()]
cuts += [(start, end, None) for start, end in assign_spans]
cuts += assign_rewrites
@@ -1130,8 +1445,15 @@ class Repro:
def delete_file(self, path: str) -> "Repro":
"""Delete a source module that its symbols' relocation left empty (the chain deletes
the leftover scaffolding-only file). Run after the moves that empty it. Refuses a
file that still holds anything beyond a docstring, imports, or a TYPE_CHECKING
block -- deleting live code is not a relocation."""
file that still holds anything beyond a docstring, imports, a TYPE_CHECKING block, or
a bare module ``logger`` -- deleting live code is not a relocation."""
def is_module_logger(stmt: ast.stmt) -> bool:
return (
isinstance(stmt, ast.Assign)
and stmt.value is not None
and ast.unparse(stmt.value) == "logging.getLogger(__name__)"
)
def op(root: Path) -> None:
target = root / path
@@ -1152,6 +1474,7 @@ class Repro:
and ast.unparse(stmt.test)
in ("TYPE_CHECKING", "typing.TYPE_CHECKING")
)
or is_module_logger(stmt)
)
]
assert not leftover, (
.claude/skills/mechanical-refactor-verify/scripts/tests/proof_generator/test_infer_extract_functions.py
+322
View File
@@ -0,0 +1,322 @@
import subprocess
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from generator_testlib import _commit, _git, _write # noqa: F401
from mechanical_refactor_proof_generator import (
build_repro,
infer_recipe,
recipe_to_script,
)
def test_infer_extract_function_with_returned_local(repo: Path) -> None:
"""A block ending in ``pool = make(...)`` carved into a helper that returns ``pool`` infers
an extract_function whose body is the verbatim block and whose return_text is authored.
"""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" base = self.setup()\n"
" if self.flag:\n"
" x = self.a\n"
" y = x + n\n"
" pool = make(\n"
" a=x,\n"
" b=y,\n"
" )\n"
" return pool\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" base = self.setup()\n"
" if self.flag:\n"
" pool = self._build_pool(n=n)\n"
" return pool\n"
"\n"
" def _build_pool(self, *, n):\n"
" x = self.a\n"
" y = x + n\n"
" pool = make(\n"
" a=x,\n"
" b=y,\n"
" )\n"
" return pool\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
_commit(repo, "extract _build_pool from dispatch")
recipe = infer_recipe("HEAD", str(repo))
assert recipe.supported
assert recipe.moves == []
assert len(recipe.extract_functions) == 1
ex = recipe.extract_functions[0]
assert ex["name"] == "_build_pool"
assert ex["src"] == "kv.py" and ex["dst"] == "kv.py"
assert ex["into_class"] == "C"
assert ex["before"] == "keep"
assert ex["body_indent"] == 12
assert ex["body"] == (
" x = self.a\n"
" y = x + n\n"
" pool = make(\n"
" a=x,\n"
" b=y,\n"
" )\n"
)
assert ex["call"] == " pool = self._build_pool(n=n)\n"
assert ex["return_text"] == " return pool"
assert ex["signature"] == " def _build_pool(self, *, n):\n"
def test_infer_extract_function_keeps_leading_comment_in_body(repo: Path) -> None:
"""A block whose first line is a comment extracts with that comment in the body, not
absorbed into the authored signature (which is the def header through its colon only).
"""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" if self.flag:\n"
" # pick the pool class for this backend\n"
" cls = PoolA\n"
" pool = cls(n)\n"
" return pool\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" if self.flag:\n"
" pool = self._build_pool(n=n)\n"
" return pool\n"
"\n"
" def _build_pool(self, *, n):\n"
" # pick the pool class for this backend\n"
" cls = PoolA\n"
" pool = cls(n)\n"
" return pool\n"
)
},
)
commit = _commit(repo, "extract _build_pool with a leading comment")
recipe = infer_recipe(commit, str(repo))
assert len(recipe.extract_functions) == 1
ex = recipe.extract_functions[0]
assert ex["signature"] == " def _build_pool(self, *, n):\n"
assert ex["body"].lstrip().startswith("# pick the pool class")
assert build_repro(recipe, repo_root=str(repo)).run() == ""
def test_infer_extract_function_no_return_text_when_body_is_whole_helper(
repo: Path,
) -> None:
"""When the helper body reproduces the source block with no trailing return, return_text
is None (the block is a side-effecting statement sequence, not a value producer)."""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def run(self):\n"
" self.pre()\n"
" self.log(1)\n"
" self.log(2)\n"
" self.post()\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def run(self):\n"
" self.pre()\n"
" self._emit()\n"
" self.post()\n"
"\n"
" def _emit(self):\n"
" self.log(1)\n"
" self.log(2)\n"
)
},
)
_commit(repo, "extract _emit from run")
recipe = infer_recipe("HEAD", str(repo))
assert recipe.supported
assert len(recipe.extract_functions) == 1
ex = recipe.extract_functions[0]
assert ex["name"] == "_emit"
assert ex["return_text"] is None
assert ex["call"] == " self._emit()\n"
def test_infer_extract_function_edited_body_does_not_pass(repo: Path) -> None:
"""A helper whose body was edited (not a verbatim cut) never yields a false pass: the
reproduction's byte-diff surfaces the bundled change as a non-empty residual."""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def run(self):\n"
" self.pre()\n"
" self.log(1)\n"
" self.post()\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def run(self):\n"
" self.pre()\n"
" self._emit()\n"
" self.post()\n"
"\n"
" def _emit(self):\n"
" self.log(2)\n"
)
},
)
commit = _commit(repo, "extract _emit but change the arg")
recipe = infer_recipe(commit, str(repo))
residual = build_repro(recipe, repo_root=str(repo)).run()
assert residual != ""
def test_infer_extract_function_when_block_and_call_share_closing_paren(
repo: Path,
) -> None:
"""The removed block and its replacement call both end in a lone ``)``; the prefix/suffix
split must not absorb that shared line, or the extracted body loses its final line.
"""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" if self.flag:\n"
" pool = make_pool(\n"
" a=n,\n"
" b=self.b,\n"
" )\n"
" return pool\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" if self.flag:\n"
" pool = self._build_pool(\n"
" n=n,\n"
" )\n"
" return pool\n"
"\n"
" def _build_pool(self, *, n):\n"
" pool = make_pool(\n"
" a=n,\n"
" b=self.b,\n"
" )\n"
" return pool\n"
)
},
)
commit = _commit(repo, "extract _build_pool")
recipe = infer_recipe(commit, str(repo))
assert len(recipe.extract_functions) == 1
ex = recipe.extract_functions[0]
assert ex["body"].rstrip().endswith(")")
assert ex["return_text"] == " return pool"
assert build_repro(recipe, repo_root=str(repo)).run() == ""
def test_emitted_script_passes_on_extract_function(repo: Path, tmp_path: Path) -> None:
"""The recipe for an extract_function reproduces the commit byte-for-byte (bare repo, no
formatter) so build_repro returns an empty residual."""
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" base = self.setup()\n"
" if self.flag:\n"
" x = self.a\n"
" pool = make(\n"
" a=x,\n"
" )\n"
" return pool\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" base = self.setup()\n"
" if self.flag:\n"
" pool = self._build_pool(n=n)\n"
" return pool\n"
"\n"
" def _build_pool(self, *, n):\n"
" x = self.a\n"
" pool = make(\n"
" a=x,\n"
" )\n"
" return pool\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
commit = _commit(repo, "extract _build_pool from dispatch")
recipe = infer_recipe(commit, str(repo))
residual = build_repro(recipe, repo_root=str(repo)).run()
assert residual == "", residual
assert "extract_function" in recipe_to_script(recipe, "extract")
.claude/skills/mechanical-refactor-verify/scripts/tests/proof_generator/test_infer_moves.py
+280
View File
@@ -38,6 +38,56 @@ def test_infer_recipe_method_onto_class(repo: Path) -> None:
assert recipe.import_additions == []
def test_infer_recipe_move_before_typechecking_uses_after_anchor(repo: Path) -> None:
"""A module-level def relocated to land just above an ``if TYPE_CHECKING:`` guard cannot
be anchored with before= (the next def sits past the guard), so the recipe anchors it with
after=<the preceding assignment>."""
_write(
repo,
**{
"model.py": (
"def keep():\n return 0\n\n\ndef helper(x):\n return x + 1\n"
),
"util.py": (
"from u import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"if TYPE_CHECKING:\n"
" from m import Thing\n"
),
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": "def keep():\n return 0\n",
"util.py": (
"from u import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"\n"
"def helper(x):\n"
" return x + 1\n"
"\n"
"\n"
"if TYPE_CHECKING:\n"
" from m import Thing\n"
),
},
)
_commit(repo, "move helper above the TYPE_CHECKING guard")
recipe = infer_recipe("HEAD", str(repo))
assert recipe.supported
assert len(recipe.moves) == 1
move = recipe.moves[0]
assert move["name"] == "helper" and move["dst"] == "util.py"
assert move["before"] is None
assert move["after"] == "_is_hip"
def test_infer_recipe_free_function_move_uses_requalify(repo: Path) -> None:
"""A move to a module-level free function dedents and requalifies the call site
(drops the qualifier), rather than lowering a receiver."""
@@ -288,3 +338,233 @@ def test_infer_recipe_records_the_source_class_for_disambiguation(repo: Path) ->
assert [mv["from_class"] for mv in recipe.moves] == ["M"]
script = recipe_to_script(recipe, "move M.foo onto C")
assert "from_class='M'" in script
def test_infer_recipe_module_level_def_shadowed_by_method_name(repo: Path) -> None:
"""A column-0 cut resolves to the module-level def even when a method shares its name."""
_write(
repo,
**{
"model.py": (
"def foo(*, x):\n"
" return x + 1\n"
"\n"
"\n"
"class M:\n"
" def foo(self):\n"
" return foo(x=self.x)\n"
),
"util.py": "def keep():\n return 1\n",
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": (
"from util import foo\n"
"\n"
"\n"
"class M:\n"
" def foo(self):\n"
" return foo(x=self.x)\n"
),
"util.py": (
"def keep():\n"
" return 1\n"
"\n"
"\n"
"def foo(*, x):\n"
" return x + 1\n"
),
},
)
commit = _commit(repo, "move module-level foo to util")
recipe = infer_recipe(commit, str(repo))
assert recipe.supported
assert [mv["name"] for mv in recipe.moves] == ["foo"]
assert recipe.moves[0]["from_class"] is None
assert recipe.moves[0]["into_class"] is None
def test_infer_recipe_class_move_between_existing_files(repo: Path) -> None:
"""A top-level class relocated to an existing module moves whole; its methods do not."""
_write(
repo,
**{
"model.py": (
"class Payload:\n"
" def get(self):\n"
" return 1\n"
"\n"
"\n"
"def stay():\n"
" return 2\n"
),
"comp.py": "def keep():\n return 3\n",
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": "def stay():\n return 2\n",
"comp.py": (
"def keep():\n"
" return 3\n"
"\n"
"\n"
"class Payload:\n"
" def get(self):\n"
" return 1\n"
),
},
)
commit = _commit(repo, "move Payload to comp")
recipe = infer_recipe(commit, str(repo))
assert recipe.supported
assert [mv["name"] for mv in recipe.moves] == ["Payload"]
assert recipe.moves[0]["from_class"] is None
assert recipe.moves[0]["into_class"] is None
def test_infer_recipe_move_leaving_a_forwarding_delegate(repo: Path) -> None:
"""A same-named stub re-added to the source infers leave_delegate on the move."""
_write(
repo,
**{
"model.py": (
"class M:\n" " def work(self, x):\n" " return x + 1\n"
),
"comp.py": "class C:\n def keep(self):\n return 1\n",
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": (
"class M:\n"
" def work(self, x):\n"
" return self.comp.work(x)\n"
),
"comp.py": (
"class C:\n"
" def keep(self):\n"
" return 1\n"
"\n"
" def work(self, x):\n"
" return x + 1\n"
),
},
)
commit = _commit(repo, "move M.work onto C, leaving a delegate")
recipe = infer_recipe(commit, str(repo))
assert recipe.supported
assert [mv["name"] for mv in recipe.moves] == ["work"]
assert recipe.moves[0]["dst"] == "comp.py"
assert recipe.moves[0]["leave_delegate"] == "comp"
assert recipe.moves[0]["delegate_name"] is None
script = recipe_to_script(recipe, "move with delegate")
assert "leave_delegate='comp'" in script
def test_infer_recipe_constant_relocated_with_the_move(repo: Path) -> None:
"""A module constant that vanished from the source and appeared in the existing
destination becomes a move_assign."""
_write(
repo,
**{
"model.py": (
"RATIO = 3\n"
"\n"
"\n"
"def work(x):\n"
" return x * RATIO\n"
"\n"
"\n"
"def stay():\n"
" return 1\n"
),
"comp.py": "import os\n\n\ndef keep():\n return 2\n",
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": "def stay():\n return 1\n",
"comp.py": (
"import os\n"
"\n"
"RATIO = 3\n"
"\n"
"\n"
"def keep():\n"
" return 2\n"
"\n"
"\n"
"def work(x):\n"
" return x * RATIO\n"
),
},
)
commit = _commit(repo, "move work + RATIO to comp")
recipe = infer_recipe(commit, str(repo))
assert recipe.supported
assert [am["name"] for am in recipe.assign_moves] == ["RATIO"]
script = recipe_to_script(recipe, "move with constant")
assert "move_assign" in script
def test_infer_recipe_in_file_method_reorder(repo: Path) -> None:
"""A method cut and re-inserted elsewhere in the same class (no other file gains it) infers
an in-file move_symbol (src == dst) anchored above its new next sibling."""
_write(
repo,
**{
"m.py": (
"class C:\n"
" def a(self):\n"
" return 1\n"
"\n"
" def b(self):\n"
" return 2\n"
"\n"
" def c(self):\n"
" return 3\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"m.py": (
"class C:\n"
" def c(self):\n"
" return 3\n"
"\n"
" def a(self):\n"
" return 1\n"
"\n"
" def b(self):\n"
" return 2\n"
)
},
)
commit = _commit(repo, "move c above a")
recipe = infer_recipe(commit, str(repo))
assert recipe.supported
assert len(recipe.moves) == 1
mv = recipe.moves[0]
assert mv["name"] == "c" and mv["src"] == "m.py" and mv["dst"] == "m.py"
assert mv["into_class"] == "C" and mv["before"] == "a"
.claude/skills/mechanical-refactor-verify/scripts/tests/proof_generator/test_main_single_commit.py
+71
View File
@@ -0,0 +1,71 @@
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from generator_testlib import _commit, _write # noqa: F401
from mechanical_refactor_proof_generator import _main
def _extract_function_commit(repo: Path) -> str:
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" x = self.a\n"
" y = x + n\n"
" return y\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
_commit(repo, "base")
_write(
repo,
**{
"kv.py": (
"class C:\n"
" def dispatch(self, n):\n"
" y = self._combine(n=n)\n"
" return y\n"
"\n"
" def _combine(self, *, n):\n"
" x = self.a\n"
" y = x + n\n"
" return y\n"
"\n"
" def keep(self):\n"
" return 0\n"
)
},
)
return _commit(repo, "extract _combine from dispatch")
def test_single_commit_extract_function_reproduces_instead_of_unsupported(
repo: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A pure intra-file extract_function commit run in single-commit mode reproduces (exit 0),
not UNSUPPORTED -- the relocates check must count extract_functions like the range path.
"""
sha = _extract_function_commit(repo)
monkeypatch.chdir(repo)
assert _main([sha]) == 0
def test_single_commit_pure_rename_is_unsupported(
repo: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A commit that relocates no definition (a bare rename) stays UNSUPPORTED with exit 1."""
_write(repo, **{"m.py": "def foo():\n return 1\n"})
_commit(repo, "base")
_write(repo, **{"m.py": "def bar():\n return 1\n"})
sha = _commit(repo, "rename foo to bar")
monkeypatch.chdir(repo)
assert _main([sha]) == 1
.claude/skills/mechanical-refactor-verify/scripts/tests/proof_generator/test_script_and_diff.py
+140 -1
View File
@@ -25,7 +25,8 @@ def test_recipe_to_script_is_self_contained_and_ordered(repo: Path) -> None:
script = recipe_to_script(infer_recipe("HEAD", str(repo)), "move foo onto C")
assert "from mechanical_refactor_reproduction_utils import Repro" in script
assert script.index("lower_call_sites") < script.index("move_symbol")
assert "r.run()" in script
assert "residual = r.run()" in script
assert "sys.exit(1 if residual else 0)" in script
# importing nothing else from the skill keeps the script auditable in isolation
assert "mechanical_refactor_verify_utils" not in script
assert "mechanical_refactor_proof_generator" not in script
@@ -40,6 +41,144 @@ def test_recipe_to_script_orders_import_ops_after_moves(repo: Path) -> None:
assert script.index("move_symbol") < script.index("add_import")
def _emit_runnable_script(repo: Path, out: Path, commit: str, subject: str) -> Path:
"""Write the emitted script plus its util dependency into a proof-folder layout."""
scripts_dir = out / "repro_scripts"
scripts_dir.mkdir(parents=True, exist_ok=True)
utils_src = Path(__file__).resolve().parents[2] / (
"mechanical_refactor_reproduction_utils.py"
)
(out / "mechanical_refactor_reproduction_utils.py").write_text(
utils_src.read_text()
)
script = recipe_to_script(infer_recipe(commit, str(repo)), subject)
script_path = scripts_dir / f"{commit[:9]}.py"
script_path.write_text(script)
return script_path
def test_emitted_script_exits_zero_on_faithful_commit(
repo: Path, tmp_path: Path
) -> None:
"""Running the emitted script on a clean move exits 0 and prints the PASS verdict."""
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\ndef resolve(m):\n return m\n",
"util.py": "import os\n",
"caller.py": (
"from model import resolve\n\n\ndef run(m):\n return resolve(m)\n"
),
},
)
_commit(repo, "base")
# The after-state is the primitives' exact output (this bare repo has no formatter
# to absorb the cut's leftover blank lines, unlike a pre-commit-clean real repo).
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\n",
"util.py": "import os\n\ndef resolve(m):\n return m\n",
"caller.py": (
"from util import resolve\n\n\ndef run(m):\n return resolve(m)\n"
),
},
)
commit = _commit(repo, "move resolve to util")
script_path = _emit_runnable_script(repo, tmp_path / "out", commit, "move")
result = subprocess.run(
[sys.executable, str(script_path)], cwd=repo, capture_output=True, text=True
)
assert result.returncode == 0, result.stdout + result.stderr
assert "PASS" in result.stdout
def test_emitted_script_exits_nonzero_on_bundled_change(
repo: Path, tmp_path: Path
) -> None:
"""A commit bundling a non-move change makes the emitted script exit non-zero."""
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\ndef resolve(m):\n return m\n",
"util.py": "import os\n",
},
)
_commit(repo, "base")
_write(
repo,
**{
"model.py": "def keep():\n return 99\n",
"util.py": "import os\n\n\ndef resolve(m):\n return m\n",
},
)
commit = _commit(repo, "move resolve AND change keep")
script_path = _emit_runnable_script(repo, tmp_path / "out", commit, "dirty move")
result = subprocess.run(
[sys.executable, str(script_path)], cwd=repo, capture_output=True, text=True
)
assert result.returncode == 1, result.stdout + result.stderr
assert "RESIDUAL" in result.stdout
def test_emitted_script_passes_on_move_above_typechecking_guard(
repo: Path, tmp_path: Path
) -> None:
"""A module-level def relocated to just above an ``if TYPE_CHECKING:`` guard reproduces
via an inferred after= anchor and the emitted script exits 0."""
_write(
repo,
**{
"model.py": (
"def keep():\n return 0\n\n\ndef helper(x):\n return x + 1\n"
),
"util.py": (
"from u import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"if TYPE_CHECKING:\n"
" from m import Thing\n"
),
},
)
_commit(repo, "base")
# After-state = the primitive's exact output (bare repo, no formatter to absorb blanks).
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\n",
"util.py": (
"from u import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"def helper(x):\n"
" return x + 1\n"
"\n"
"if TYPE_CHECKING:\n"
" from m import Thing\n"
),
},
)
commit = _commit(repo, "move helper above the TYPE_CHECKING guard")
script_path = _emit_runnable_script(
repo, tmp_path / "out", commit, "after-anchor move"
)
result = subprocess.run(
[sys.executable, str(script_path)], cwd=repo, capture_output=True, text=True
)
assert result.returncode == 0, result.stdout + result.stderr
assert "PASS" in result.stdout
assert "after='_is_hip'" in script_path.read_text()
def test_per_file_diff_keeps_content_lines_starting_with_plus_signs(repo: Path) -> None:
"""An added content line beginning with '++' is collected, not mistaken for a header."""
from mechanical_refactor_proof_generator import _per_file_diff
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/cli_testlib.py
+62
View File
@@ -0,0 +1,62 @@
import subprocess
from pathlib import Path
_PASSING_PROOF = (
"import sys\n"
'print("PASS: reproduces the commit byte-for-byte.")\n'
"sys.exit(0)\n"
)
_FAILING_PROOF = (
"import sys\n" 'print("RESIDUAL (2 lines):\\n+x\\n-y")\n' "sys.exit(1)\n"
)
def _git(repo: Path, *args: str) -> str:
return subprocess.run(
["git", *args], cwd=repo, check=True, capture_output=True, text=True
).stdout.strip()
def _write(repo: Path, **files: "str | None") -> None:
for name, content in files.items():
path = repo / name.replace("__", "/")
if content is None:
path.unlink()
else:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
def _commit(repo: Path, message: str) -> str:
_git(repo, "add", "-A")
_git(repo, "commit", "-q", "-m", message)
return _git(repo, "rev-parse", "HEAD")
def _chain(repo: Path, messages: "list[str]") -> "tuple[str, list[str]]":
"""A base commit plus one single-file commit per message, on a `chain` branch.
Returns (base_sha, commit_shas)."""
_write(repo, **{"seed.py": "SEED = 0\n"})
base = _commit(repo, "base")
_git(repo, "switch", "-q", "-c", "chain")
shas: "list[str]" = []
for i, message in enumerate(messages):
_write(repo, **{f"file_{i}.py": f"VALUE = {i}\n"})
shas.append(_commit(repo, message))
return base, shas
def _write_stub_proof(
proof_dir: Path,
sha: str,
*,
passing: bool = True,
flat: bool = False,
stem_len: int = 9,
) -> Path:
"""A stand-in proof script printing the arbiter's verdict line and exiting to match."""
directory = proof_dir if flat else proof_dir / "repro_scripts"
directory.mkdir(parents=True, exist_ok=True)
path = directory / f"{sha[:stem_len]}.py"
path.write_text(_PASSING_PROOF if passing else _FAILING_PROOF)
return path
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/conftest.py
+20
View File
@@ -0,0 +1,20 @@
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from cli_testlib import _git
@pytest.fixture
def repo(tmp_path: Path) -> Path:
root = tmp_path / "repo"
root.mkdir()
_git(root, "init", "-q", "-b", "main")
_git(root, "config", "user.email", "test@example.com")
_git(root, "config", "user.name", "test")
_git(root, "config", "commit.gpgsign", "false")
return root
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/test_classification.py
+123
View File
@@ -0,0 +1,123 @@
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cli_testlib import _chain, _write_stub_proof
from mechanical_refactor_reproduction_cli import (
KIND_MECHANICAL,
KIND_NON_MECHANICAL,
VERDICT_AMBIGUOUS_KIND,
VERDICT_HUMAN_REVIEW,
VERDICT_PASS,
VERDICT_UNCLASSIFIED,
verify_chain,
)
def _single_verdict(repo: Path, tmp_path: Path, message: str, *, with_proof: bool):
proof = tmp_path / "proof"
proof.mkdir(exist_ok=True)
base, shas = _chain(repo, [message])
if with_proof:
_write_stub_proof(proof, shas[0])
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
return result.verdicts[0]
def test_mechanical_provable_word_classifies_the_commit_as_mechanical(
repo: Path, tmp_path: Path
) -> None:
"""A message carrying mechanical_provable is classified mechanical and needs a proof."""
verdict = _single_verdict(
repo, tmp_path, "grp(step,mechanical_provable): move foo", with_proof=True
)
assert verdict.kind == KIND_MECHANICAL
assert verdict.verdict == VERDICT_PASS
def test_non_mechanical_provable_word_is_not_double_counted_as_the_bare_word(
repo: Path, tmp_path: Path
) -> None:
"""non_mechanical_provable classifies as non-mechanical, not as both words at once."""
verdict = _single_verdict(
repo,
tmp_path,
"grp(step,non_mechanical_provable): rework foo",
with_proof=False,
)
assert verdict.kind == KIND_NON_MECHANICAL
assert verdict.verdict == VERDICT_HUMAN_REVIEW
def test_message_without_either_word_is_unclassified_and_fails_the_chain(
repo: Path, tmp_path: Path
) -> None:
"""A commit missing both words gets UNCLASSIFIED and the chain does not pass."""
proof = tmp_path / "proof"
proof.mkdir()
base, _ = _chain(repo, ["plain subject with no kind word"])
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_UNCLASSIFIED
assert result.verdicts[0].kind is None
assert not result.passed
def test_message_with_both_words_is_ambiguous(repo: Path, tmp_path: Path) -> None:
"""A commit declaring both kinds gets AMBIGUOUS_KIND and fails the chain."""
verdict = _single_verdict(
repo,
tmp_path,
"subject mechanical_provable\n\nbody also says non_mechanical_provable",
with_proof=True,
)
assert verdict.verdict == VERDICT_AMBIGUOUS_KIND
assert verdict.kind is None
def test_kind_word_must_stand_alone_not_as_a_substring(
repo: Path, tmp_path: Path
) -> None:
"""xmechanical_provable / mechanical_provable_x do not count as the standalone word."""
verdict = _single_verdict(
repo,
tmp_path,
"xmechanical_provable and mechanical_provable_x only",
with_proof=False,
)
assert verdict.verdict == VERDICT_UNCLASSIFIED
def test_kind_word_delimited_by_punctuation_counts(repo: Path, tmp_path: Path) -> None:
"""The word inside punctuation, e.g. (step,mechanical_provable), is a valid match."""
verdict = _single_verdict(
repo, tmp_path, "grp(step,mechanical_provable): move", with_proof=True
)
assert verdict.kind == KIND_MECHANICAL
def test_repeating_the_same_kind_word_is_accepted(repo: Path, tmp_path: Path) -> None:
"""Multiple occurrences of one kind word still classify unambiguously."""
verdict = _single_verdict(
repo,
tmp_path,
"mechanical_provable move\n\nthis commit is mechanical_provable",
with_proof=True,
)
assert verdict.kind == KIND_MECHANICAL
assert verdict.verdict == VERDICT_PASS
def test_kind_word_in_the_body_counts_when_subject_is_free_form(
repo: Path, tmp_path: Path
) -> None:
"""Classification scans the whole message, so a body-only word is enough."""
verdict = _single_verdict(
repo,
tmp_path,
"Move resolve to util\n\nKind: non_mechanical_provable",
with_proof=False,
)
assert verdict.kind == KIND_NON_MECHANICAL
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/test_proof_discovery.py
+89
View File
@@ -0,0 +1,89 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cli_testlib import _chain, _write_stub_proof
from mechanical_refactor_reproduction_cli import (
VERDICT_AMBIGUOUS_PROOF,
VERDICT_MISSING_PROOF,
VERDICT_PASS,
verify_chain,
)
_MSG = "mechanical_provable: move foo"
def _run_single(repo: Path, proof: Path):
base, shas = _chain(repo, [_MSG])
return shas[0], verify_chain(
base=base, branch="chain", proof=proof, repo_root=str(repo)
)
def test_proof_is_found_under_repro_scripts_by_sha_prefix(
repo: Path, tmp_path: Path
) -> None:
"""The generator layout repro_scripts/<sha9>.py resolves to the commit's proof."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0], stem_len=9)
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_PASS
def test_proof_is_found_flat_in_the_proof_folder_by_full_sha(
repo: Path, tmp_path: Path
) -> None:
"""A flat <proof>/<full-sha>.py layout is also accepted."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0], flat=True, stem_len=40)
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_PASS
def test_provable_commit_without_a_proof_script_is_missing_proof(
repo: Path, tmp_path: Path
) -> None:
"""A mechanical_provable commit with no matching script fails as MISSING_PROOF."""
proof = tmp_path / "proof"
proof.mkdir()
sha, result = _run_single(repo, proof)
assert result.verdicts[0].verdict == VERDICT_MISSING_PROOF
assert not result.passed
def test_unrelated_and_non_hex_scripts_do_not_match(repo: Path, tmp_path: Path) -> None:
"""Scripts named for another sha or with a non-hex stem are not this commit's proof."""
proof = tmp_path / "proof"
scripts = proof / "repro_scripts"
scripts.mkdir(parents=True)
(scripts / "0123456789abcdef.py").write_text("raise SystemExit(1)\n")
(scripts / "not_a_sha.py").write_text("raise SystemExit(1)\n")
sha, result = _run_single(repo, proof)
assert result.verdicts[0].verdict == VERDICT_MISSING_PROOF
def test_two_scripts_matching_one_commit_are_ambiguous(
repo: Path, tmp_path: Path
) -> None:
"""A commit matched by both a nested and a flat script fails as AMBIGUOUS_PROOF."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0], stem_len=9)
_write_stub_proof(proof, shas[0], flat=True, stem_len=12)
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_AMBIGUOUS_PROOF
assert not result.passed
def test_short_hex_stem_below_minimum_length_is_ignored(
repo: Path, tmp_path: Path
) -> None:
"""A 6-char hex stem is too short to name a commit and is not treated as a proof."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0], stem_len=6)
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_MISSING_PROOF
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/test_report.py
+95
View File
@@ -0,0 +1,95 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cli_testlib import _chain, _write_stub_proof
from mechanical_refactor_reproduction_cli import main, render_report, verify_chain
def _mixed_chain_result(repo: Path, tmp_path: Path):
proof = tmp_path / "proof"
base, shas = _chain(
repo,
[
"mechanical_provable: move foo",
"non_mechanical_provable: rework bar",
"mechanical_provable: move baz",
],
)
_write_stub_proof(proof, shas[0])
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
return proof, base, shas, result
def test_report_has_header_table_row_per_commit_and_chain_verdict(
repo: Path, tmp_path: Path
) -> None:
"""The report carries base/branch/proof, one table row per commit, and the verdict."""
proof, base, shas, result = _mixed_chain_result(repo, tmp_path)
report = render_report(result)
assert "# Mechanical refactor chain report" in report
assert f"`{base[:12]}`" in report
assert "chain verdict: **FAIL**" in report
assert "3 total — 2 mechanical_provable, 1 non_mechanical_provable" in report
for sha in shas:
assert f"`{sha[:9]}`" in report
assert "| mechanical_provable | PASS |" in report
assert "| non_mechanical_provable | HUMAN_REVIEW |" in report
assert "| mechanical_provable | MISSING_PROOF |" in report
def test_report_lists_failure_details_for_each_non_ok_commit(
repo: Path, tmp_path: Path
) -> None:
"""Every non-ok commit gets a failure-details section with its explanation."""
proof, base, shas, result = _mixed_chain_result(repo, tmp_path)
report = render_report(result)
assert "## Failure details" in report
assert f"### `{shas[2][:9]}` — MISSING_PROOF" in report
assert "no proof script found" in report
def test_passing_report_has_no_failure_details_section(
repo: Path, tmp_path: Path
) -> None:
"""A fully verified chain renders a PASS report without a failure section."""
proof = tmp_path / "proof"
base, shas = _chain(repo, ["mechanical_provable: move foo"])
_write_stub_proof(proof, shas[0])
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
report = render_report(result)
assert "chain verdict: **PASS**" in report
assert "proofs: 1/1 PASS" in report
assert "## Failure details" not in report
def test_main_writes_the_report_into_the_proof_folder_by_default(
repo: Path, tmp_path: Path, capsys
) -> None:
"""main prints the report and writes <proof>/chain_report.md (or --report PATH)."""
proof = tmp_path / "proof"
base, shas = _chain(repo, ["mechanical_provable: move foo"])
_write_stub_proof(proof, shas[0])
args = [
"--base",
base,
"--branch",
"chain",
"--proof",
str(proof),
"--repo-root",
str(repo),
]
assert main(args) == 0
default_report = proof / "chain_report.md"
assert "chain verdict: **PASS**" in default_report.read_text()
assert "chain verdict: **PASS**" in capsys.readouterr().out
custom = tmp_path / "custom_report.md"
assert main([*args, "--report", str(custom)]) == 0
assert "chain verdict: **PASS**" in custom.read_text()
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/test_skip_passed.py
+179
View File
@@ -0,0 +1,179 @@
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cli_testlib import _chain, _write_stub_proof
from mechanical_refactor_reproduction_cli import (
VERDICT_FAIL,
VERDICT_PASS,
main,
render_report,
verify_chain,
)
_MSG = "mechanical_provable: move foo"
def _counting_proof(script: Path, counter: Path, *, passing: bool = True) -> None:
"""Make the stub proof bump a run counter so re-execution is observable."""
verdict = (
'print("PASS: reproduces the commit byte-for-byte.")\nsys.exit(0)\n'
if passing
else 'print("RESIDUAL (1 lines):\\n+x")\nsys.exit(1)\n'
)
script.write_text(
"import sys\n"
f"counter = __import__('pathlib').Path({str(counter)!r})\n"
"runs = int(counter.read_text()) if counter.exists() else 0\n"
"counter.write_text(str(runs + 1))\n" + verdict
)
def _cache_file(repo: Path) -> Path:
return repo / ".git" / "mechanical_refactor_passed_proofs.json"
def test_skip_passed_reuses_an_unchanged_pass_without_rerunning(
repo: Path, tmp_path: Path
) -> None:
"""A PASS recorded on the first run is reused: the proof does not execute again."""
proof = tmp_path / "proof"
counter = tmp_path / "runs"
base, shas = _chain(repo, [_MSG])
_counting_proof(_write_stub_proof(proof, shas[0]), counter)
args = dict(base=base, branch="chain", proof=proof, repo_root=str(repo))
first = verify_chain(**args)
assert first.verdicts[0].verdict == VERDICT_PASS
assert counter.read_text() == "1"
second = verify_chain(**args, skip_passed=True)
assert second.verdicts[0].verdict == VERDICT_PASS
assert second.verdicts[0].cached
assert counter.read_text() == "1"
assert second.passed
def test_without_the_flag_the_proof_always_reruns(repo: Path, tmp_path: Path) -> None:
"""The cache is recorded on every run but consulted only under skip_passed."""
proof = tmp_path / "proof"
counter = tmp_path / "runs"
base, shas = _chain(repo, [_MSG])
_counting_proof(_write_stub_proof(proof, shas[0]), counter)
args = dict(base=base, branch="chain", proof=proof, repo_root=str(repo))
verify_chain(**args)
result = verify_chain(**args)
assert counter.read_text() == "2"
assert not result.verdicts[0].cached
def test_editing_the_proof_script_invalidates_the_cache(
repo: Path, tmp_path: Path
) -> None:
"""A changed script hash misses the cache, so the edited (failing) proof reruns."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
script = _write_stub_proof(proof, shas[0])
verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
script.write_text('print("RESIDUAL (1 lines):\\n+x")\nraise SystemExit(1)\n')
result = verify_chain(
base=base, branch="chain", proof=proof, repo_root=str(repo), skip_passed=True
)
assert result.verdicts[0].verdict == VERDICT_FAIL
def test_editing_the_utils_copy_invalidates_the_cache(
repo: Path, tmp_path: Path
) -> None:
"""The utils module next to the scripts is part of the key: editing it forces a rerun."""
proof = tmp_path / "proof"
counter = tmp_path / "runs"
utils = proof / "mechanical_refactor_reproduction_utils.py"
base, shas = _chain(repo, [_MSG])
_counting_proof(_write_stub_proof(proof, shas[0]), counter)
utils.parent.mkdir(parents=True, exist_ok=True)
utils.write_text("ENGINE = 1\n")
args = dict(base=base, branch="chain", proof=proof, repo_root=str(repo))
verify_chain(**args)
utils.write_text("ENGINE = 2\n")
result = verify_chain(**args, skip_passed=True)
assert counter.read_text() == "2"
assert not result.verdicts[0].cached
def test_a_fail_is_never_recorded_in_the_cache(repo: Path, tmp_path: Path) -> None:
"""Only PASS verdicts enter the cache; a failing proof leaves no entry for its sha."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0], passing=False)
verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
cache = _cache_file(repo)
assert not cache.exists() or shas[0] not in json.loads(cache.read_text())["passed"]
def test_corrupt_cache_file_is_treated_as_empty(repo: Path, tmp_path: Path) -> None:
"""A garbage cache file never crashes the walk; the proof simply runs."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0])
_cache_file(repo).write_text("{not json")
result = verify_chain(
base=base, branch="chain", proof=proof, repo_root=str(repo), skip_passed=True
)
assert result.verdicts[0].verdict == VERDICT_PASS
assert not result.verdicts[0].cached
def test_cache_lives_in_the_git_common_dir_and_records_the_pass(
repo: Path, tmp_path: Path
) -> None:
"""A PASS writes the (sha, script hash, utils hash) entry under .git/."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0])
verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
entry = json.loads(_cache_file(repo).read_text())["passed"][shas[0]]
assert set(entry) == {"script_sha256", "utils_sha256"}
assert len(entry["script_sha256"]) == 64
def test_report_counts_reused_proofs_and_main_accepts_the_flag(
repo: Path, tmp_path: Path
) -> None:
"""The report carries the reused count and --skip-passed works through main."""
proof = tmp_path / "proof"
base, shas = _chain(repo, [_MSG])
_write_stub_proof(proof, shas[0])
cli_args = [
"--base",
base,
"--branch",
"chain",
"--proof",
str(proof),
"--repo-root",
str(repo),
]
assert main(cli_args) == 0
result = verify_chain(
base=base, branch="chain", proof=proof, repo_root=str(repo), skip_passed=True
)
report = render_report(result)
assert "reused from the passed-proof cache (--skip-passed): 1" in report
assert main([*cli_args, "--skip-passed"]) == 0
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_cli/test_verify_chain.py
+219
View File
@@ -0,0 +1,219 @@
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cli_testlib import _chain, _commit, _git, _write, _write_stub_proof
from mechanical_refactor_proof_generator import generate_range
from mechanical_refactor_reproduction_cli import (
VERDICT_FAIL,
VERDICT_HUMAN_REVIEW,
VERDICT_PASS,
ChainVerificationError,
main,
verify_chain,
)
def test_chain_of_proved_and_declared_commits_passes(
repo: Path, tmp_path: Path
) -> None:
"""A proved mechanical commit plus a declared non-mechanical one verifies as PASS."""
proof = tmp_path / "proof"
base, shas = _chain(
repo,
["mechanical_provable: move foo", "non_mechanical_provable: rework bar"],
)
_write_stub_proof(proof, shas[0])
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert [v.verdict for v in result.verdicts] == [VERDICT_PASS, VERDICT_HUMAN_REVIEW]
assert result.passed
def test_failing_proof_fails_the_commit_and_the_chain(
repo: Path, tmp_path: Path
) -> None:
"""A proof that exits non-zero yields FAIL with the output tail in the detail."""
proof = tmp_path / "proof"
base, shas = _chain(repo, ["mechanical_provable: move foo"])
_write_stub_proof(proof, shas[0], passing=False)
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_FAIL
assert "RESIDUAL" in result.verdicts[0].detail
assert not result.passed
def test_proof_exiting_zero_without_a_pass_line_is_a_fail(
repo: Path, tmp_path: Path
) -> None:
"""PASS needs exit 0 AND the PASS: verdict line, so a residual under exit 0 fails."""
proof = tmp_path / "proof"
base, shas = _chain(repo, ["mechanical_provable: move foo"])
script = _write_stub_proof(proof, shas[0])
script.write_text('print("RESIDUAL (1 lines):\\n+x")\n')
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].verdict == VERDICT_FAIL
def test_main_exit_codes_reflect_the_chain_verdict(repo: Path, tmp_path: Path) -> None:
"""main returns 0 for a verified chain and 1 once an unverifiable commit appears."""
proof = tmp_path / "proof"
base, shas = _chain(repo, ["mechanical_provable: move"])
_write_stub_proof(proof, shas[0])
args = [
"--base",
base,
"--branch",
"chain",
"--proof",
str(proof),
"--repo-root",
str(repo),
]
assert main(args) == 0
_git(repo, "commit", "-q", "--allow-empty", "-m", "plain subject with no kind word")
assert main(args) == 1
def test_unresolvable_refs_and_missing_proof_folder_are_setup_errors(
repo: Path, tmp_path: Path
) -> None:
"""Bad --base/--branch/--proof inputs raise ChainVerificationError (exit code 2)."""
proof = tmp_path / "proof"
proof.mkdir()
base, _ = _chain(repo, ["mechanical_provable: move"])
with pytest.raises(ChainVerificationError):
verify_chain(
base=base, branch="no-such-branch", proof=proof, repo_root=str(repo)
)
with pytest.raises(ChainVerificationError):
verify_chain(
base=base,
branch="chain",
proof=tmp_path / "missing",
repo_root=str(repo),
)
assert (
main(
[
"--base",
base,
"--branch",
"no-such-branch",
"--proof",
str(proof),
"--repo-root",
str(repo),
]
)
== 2
)
def test_non_ancestor_base_and_empty_range_are_setup_errors(
repo: Path, tmp_path: Path
) -> None:
"""A base off the branch or an empty base..branch range refuses to verify."""
proof = tmp_path / "proof"
proof.mkdir()
base, shas = _chain(repo, ["mechanical_provable: move"])
_git(repo, "switch", "-q", "main")
_write(repo, **{"other.py": "OTHER = 1\n"})
off_branch = _commit(repo, "unrelated main-side commit")
with pytest.raises(ChainVerificationError):
verify_chain(base=off_branch, branch="chain", proof=proof, repo_root=str(repo))
with pytest.raises(ChainVerificationError):
verify_chain(base=shas[0], branch=shas[0], proof=proof, repo_root=str(repo))
def test_merge_commit_in_the_chain_is_a_setup_error(repo: Path, tmp_path: Path) -> None:
"""A non-linear chain (contains a merge commit) refuses to verify."""
proof = tmp_path / "proof"
proof.mkdir()
base, _ = _chain(repo, ["mechanical_provable: move"])
_git(repo, "switch", "-q", "main")
_write(repo, **{"other.py": "OTHER = 1\n"})
_commit(repo, "mechanical_provable: main-side")
_git(repo, "switch", "-q", "chain")
_git(repo, "merge", "-q", "--no-ff", "-m", "non_mechanical_provable: merge", "main")
with pytest.raises(ChainVerificationError):
verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
def test_proofs_run_concurrently_up_to_jobs(repo: Path, tmp_path: Path) -> None:
"""With jobs>=2 a proof that waits on a sibling proof's sentinel still completes."""
proof = tmp_path / "proof"
sentinel = tmp_path / "sentinel"
base, shas = _chain(
repo, ["mechanical_provable: move a", "mechanical_provable: move b"]
)
waiter = _write_stub_proof(proof, shas[0])
waiter.write_text(
"import sys, time\n"
f"deadline = time.monotonic() + 30\n"
f"while not __import__('pathlib').Path({str(sentinel)!r}).exists():\n"
" if time.monotonic() > deadline:\n"
" sys.exit(1)\n"
" time.sleep(0.05)\n"
'print("PASS: reproduces the commit byte-for-byte.")\n'
"sys.exit(0)\n"
)
creator = _write_stub_proof(proof, shas[1])
creator.write_text(
"import sys\n"
f"__import__('pathlib').Path({str(sentinel)!r}).write_text('go')\n"
'print("PASS: reproduces the commit byte-for-byte.")\n'
"sys.exit(0)\n"
)
result = verify_chain(
base=base, branch="chain", proof=proof, repo_root=str(repo), jobs=2
)
assert [v.verdict for v in result.verdicts] == [VERDICT_PASS, VERDICT_PASS]
assert [v.sha for v in result.verdicts] == shas
assert result.passed
def test_end_to_end_with_a_generated_proof_folder(repo: Path, tmp_path: Path) -> None:
"""A real move commit proved by generate_range verifies through the CLI end-to-end."""
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\ndef resolve(m):\n return m\n",
"util.py": "import os\n",
},
)
base = _commit(repo, "base")
_git(repo, "switch", "-q", "-c", "chain")
# The after-state is the primitives' exact output (this bare repo has no formatter
# to absorb the cut's leftover blank lines, unlike a pre-commit-clean real repo).
_write(
repo,
**{
"model.py": "def keep():\n return 0\n\n\n",
"util.py": "import os\n\ndef resolve(m):\n return m\n",
},
)
move_sha = _commit(repo, "mechanical_provable: move resolve to util")
proof = tmp_path / "proof"
generate_range(f"{base}..chain", out_dir=str(proof), repo_root=str(repo))
result = verify_chain(base=base, branch="chain", proof=proof, repo_root=str(repo))
assert result.verdicts[0].sha == move_sha
assert result.verdicts[0].verdict == VERDICT_PASS
assert result.passed
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_add_imports.py
+145 -3
View File
@@ -34,6 +34,49 @@ def test_add_import_appends_after_last_top_level_import(tmp_path: Path) -> None:
).read_text() == "import os\nimport sys\nfrom pkg import Thing\n\nx = 1\n"
# --- add_imported_name ---------------------------------------------------------
def test_add_imported_name_extends_a_single_line_import(tmp_path: Path) -> None:
"""A new name is appended to an existing from-import on the same statement."""
(tmp_path / "m.py").write_text("from pkg import a\n\nx = 1\n")
r = Repro("b", "t").add_imported_name("m.py", module="pkg", name="b")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "from pkg import a, b\n\nx = 1\n"
def test_add_imported_name_carries_an_asname(tmp_path: Path) -> None:
"""The added name keeps its `as` alias."""
(tmp_path / "m.py").write_text("from pkg import a\n")
r = Repro("b", "t").add_imported_name("m.py", module="pkg", name="b", asname="c")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "from pkg import a, b as c\n"
def test_add_imported_name_refuses_a_commented_import(tmp_path: Path) -> None:
"""An import carrying comments is refused, since a rebuild would drop them."""
(tmp_path / "m.py").write_text("from pkg import (\n a, # keep\n)\n")
r = Repro("b", "t").add_imported_name("m.py", module="pkg", name="b")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
def test_add_imported_name_rejects_a_name_already_present(tmp_path: Path) -> None:
"""Adding a name the import already has fails loudly."""
(tmp_path / "m.py").write_text("from pkg import a, b\n")
r = Repro("b", "t").add_imported_name("m.py", module="pkg", name="b")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
def test_add_imported_name_raises_without_a_matching_import(tmp_path: Path) -> None:
"""A file lacking a `from module import` for the module fails loudly."""
(tmp_path / "m.py").write_text("from other import a\n")
r = Repro("b", "t").add_imported_name("m.py", module="pkg", name="b")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
# --- repath_import / add_typechecking_import -----------------------------------
@@ -64,6 +107,33 @@ def test_add_typechecking_import_inserts_in_block(tmp_path: Path) -> None:
)
def test_add_typechecking_import_creates_missing_block(tmp_path: Path) -> None:
"""With no TYPE_CHECKING block, one is created after the trailing module import."""
(tmp_path / "m.py").write_text(
"from typing import TYPE_CHECKING\n"
"\n"
"from a import X\n"
"\n"
"\n"
"def f():\n"
" pass\n"
)
r = Repro("b", "t").add_typechecking_import("m.py", "from b import Y")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == (
"from typing import TYPE_CHECKING\n"
"\n"
"from a import X\n"
"\n"
"if TYPE_CHECKING:\n"
" from b import Y\n"
"\n"
"\n"
"def f():\n"
" pass\n"
)
def test_add_import_into_an_empty_file(tmp_path: Path) -> None:
"""Adding an import to an empty file writes just the statement."""
(tmp_path / "m.py").write_text("")
@@ -134,9 +204,81 @@ def test_add_typechecking_import_after_a_multiline_final_import(tmp_path: Path)
)
def test_add_typechecking_import_raises_without_a_block(tmp_path: Path) -> None:
"""A file lacking a TYPE_CHECKING block fails loudly."""
(tmp_path / "m.py").write_text("import os\n\nx = 1\n")
def test_add_typechecking_import_raises_without_imports(tmp_path: Path) -> None:
"""A file with no imports cannot anchor a new TYPE_CHECKING block and fails loudly."""
(tmp_path / "m.py").write_text("x = 1\n")
r = Repro("b", "t").add_typechecking_import("m.py", "from b import Y")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
def test_add_typechecking_import_drops_a_lone_pass_placeholder(tmp_path: Path) -> None:
"""Populating a `pass`-only TYPE_CHECKING block replaces the placeholder."""
(tmp_path / "m.py").write_text(
"from typing import TYPE_CHECKING\n"
"\n"
"if TYPE_CHECKING:\n"
" pass\n"
"\n"
"x = 1\n"
)
r = Repro("b", "t").add_typechecking_import("m.py", "from b import Y")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == (
"from typing import TYPE_CHECKING\n"
"\n"
"if TYPE_CHECKING:\n"
" from b import Y\n"
"\n"
"x = 1\n"
)
def test_add_typechecking_import_keeps_a_pass_that_is_not_alone(tmp_path: Path) -> None:
"""A `pass` beside a real import is left untouched; only the new import is appended."""
(tmp_path / "m.py").write_text(
"from typing import TYPE_CHECKING\n"
"\n"
"if TYPE_CHECKING:\n"
" from a import X\n"
" pass\n"
"\n"
"x = 1\n"
)
r = Repro("b", "t").add_typechecking_import("m.py", "from b import Y")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == (
"from typing import TYPE_CHECKING\n"
"\n"
"if TYPE_CHECKING:\n"
" from a import X\n"
" pass\n"
" from b import Y\n"
"\n"
"x = 1\n"
)
# --- add_import(after=...) -----------------------------------------------------
def test_add_import_after_anchors_into_a_split_import_block(tmp_path: Path) -> None:
"""With `after`, the import lands right after the named import -- needed when a
statement splits the imports into separate blocks and the default (after the last
import) would land in the wrong block."""
(tmp_path / "m.py").write_text(
"import os\n\n_flag = os.getpid()\n\nfrom pkg import a\n\nx = 1\n"
)
r = Repro("b", "t").add_import("m.py", "from new import Thing", after="import os")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == (
"import os\nfrom new import Thing\n\n_flag = os.getpid()\n\nfrom pkg import a\n\nx = 1\n"
)
def test_add_import_after_raises_when_anchor_absent(tmp_path: Path) -> None:
"""An `after` substring that matches no top-level import raises."""
(tmp_path / "m.py").write_text("import os\n\nx = 1\n")
r = Repro("b", "t").add_import("m.py", "from new import Thing", after="import nope")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_call_sites.py
+32
View File
@@ -79,6 +79,38 @@ def test_requalify_call_sites_drops_the_qualifier(tmp_path: Path) -> None:
assert (tmp_path / "m.py").read_text() == "y = bar(a, b)\n"
def test_route_call_sites_through_field_inserts_the_field(tmp_path: Path) -> None:
"""recv.bar(a) becomes recv.updater.bar(a) when bar moves onto a collaborator field."""
(tmp_path / "m.py").write_text("y = self.worker.runner.bar(a)\n")
r = Repro("b", "t").route_call_sites_through_field(
"bar", field="updater", paths=["m.py"]
)
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "y = self.worker.runner.updater.bar(a)\n"
def test_route_call_sites_through_field_skips_an_already_routed_call(
tmp_path: Path,
) -> None:
"""A call already going through the field is left alone, so the pass converges."""
(tmp_path / "m.py").write_text("y = self.runner.updater.bar(a)\n")
r = Repro("b", "t").route_call_sites_through_field(
"bar", field="updater", paths=["m.py"]
)
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "y = self.runner.updater.bar(a)\n"
def test_route_call_sites_through_field_honors_owner_filter(tmp_path: Path) -> None:
"""With owner set, only calls on that exact receiver are routed through the field."""
(tmp_path / "m.py").write_text("a = x.bar(1)\nb = y.bar(2)\n")
r = Repro("b", "t").route_call_sites_through_field(
"bar", field="updater", paths=["m.py"], owner="x"
)
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "a = x.updater.bar(1)\nb = y.bar(2)\n"
# --- adversarial audit: call-site rewrites ---------------------------------------
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_delete_file.py
+19
View File
@@ -46,4 +46,23 @@ def test_delete_file_on_a_missing_path_is_a_no_op(tmp_path: Path) -> None:
assert not (tmp_path / "nope.py").exists()
def test_delete_file_allows_a_bare_module_logger(tmp_path: Path) -> None:
"""A leftover module holding only imports and a `logger` is deletable scaffolding."""
(tmp_path / "gone.py").write_text(
"import logging\n\nlogger = logging.getLogger(__name__)\n"
)
r = Repro("b", "t").delete_file("gone.py")
_apply(r, tmp_path)
assert not (tmp_path / "gone.py").exists()
def test_delete_file_still_refuses_a_non_logger_assignment(tmp_path: Path) -> None:
"""A leftover module-level assignment other than a logger blocks deletion."""
(tmp_path / "live.py").write_text("CONFIG = {'a': 1}\n")
r = Repro("b", "t").delete_file("live.py")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
assert (tmp_path / "live.py").exists()
# --- adversarial audit: extract_function -----------------------------------------
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_extract_symbols_to_new_module.py
+65
View File
@@ -144,6 +144,30 @@ def test_extract_symbols_to_new_module_asserts_unknown_drop_assign(
_apply(r, tmp_path)
def test_extract_symbols_to_new_module_header_accepts_a_typechecking_block(
tmp_path: Path,
) -> None:
"""An authored header may carry an `if TYPE_CHECKING:` import block; the audit accepts it
and the block is reproduced verbatim in the new module."""
(tmp_path / "src.py").write_text("def moved(x):\n return x\n")
header = (
"from __future__ import annotations\n"
"\n"
"from typing import TYPE_CHECKING\n"
"\n"
"if TYPE_CHECKING:\n"
" from other import Thing\n"
"\n"
)
r = Repro("b", "t").extract_symbols_to_new_module(
"src.py", "new.py", symbols=["moved"], header=header, order=["moved"]
)
_apply(r, tmp_path)
new_out = (tmp_path / "new.py").read_text()
assert "if TYPE_CHECKING:\n from other import Thing\n" in new_out
assert "def moved(x):\n return x\n" in new_out
# --- extract_function ----------------------------------------------------------
@@ -197,3 +221,44 @@ def test_extract_symbols_drop_assigns_preserves_other_targets_of_chained_assign(
)
_apply(r, tmp_path)
assert "B" in (tmp_path / "src.py").read_text()
def test_extract_symbols_to_new_module_allows_a_rederived_surviving_constant(
tmp_path: Path,
) -> None:
"""A header constant that also survives verbatim in the source (re-derived boilerplate,
e.g. `_is_hip = is_hip()`) is allowed: it is provably not fiction because the same
statement remains in the source."""
(tmp_path / "src.py").write_text(
"from pkg import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"\n"
"def moved():\n"
" return _is_hip\n"
)
header = "from pkg import is_hip\n\n_is_hip = is_hip()\n"
r = Repro("b", "t").extract_symbols_to_new_module(
"src.py", "new.py", symbols=["moved"], header=header, order=["moved"]
)
_apply(r, tmp_path)
assert "_is_hip = is_hip()" in (tmp_path / "src.py").read_text()
assert "_is_hip = is_hip()" in (tmp_path / "new.py").read_text()
def test_extract_symbols_to_new_module_rejects_a_fictional_header_constant(
tmp_path: Path,
) -> None:
"""A header constant that is neither dropped from nor surviving in the source is fiction
and raises: the audit refuses code the extraction cannot vouch for."""
(tmp_path / "src.py").write_text("def moved():\n return 1\n")
r = Repro("b", "t").extract_symbols_to_new_module(
"src.py",
"new.py",
symbols=["moved"],
header="_fake = evil()\n",
order=["moved"],
)
with pytest.raises(AssertionError):
_apply(r, tmp_path)
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_move_assign.py
+69
View File
@@ -0,0 +1,69 @@
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from mechanical_refactor_reproduction_utils import Repro
from reproduction_testlib import _apply # noqa: F401
def test_move_assign_relocates_a_module_constant(tmp_path: Path) -> None:
"""The assignment is cut verbatim from the source and lands after the destination's imports."""
(tmp_path / "src.py").write_text(
"import os\n\nLIMIT = 480 # seconds\n\n\ndef stay():\n return LIMIT\n"
)
(tmp_path / "dst.py").write_text("import sys\n\n\ndef keep():\n return 1\n")
r = Repro("b", "t").move_assign("LIMIT", src="src.py", dst="dst.py")
_apply(r, tmp_path)
assert "LIMIT" not in (tmp_path / "src.py").read_text().split("def stay")[0]
assert (tmp_path / "dst.py").read_text() == (
"import sys\n"
"\n"
"LIMIT = 480 # seconds\n"
"\n"
"\n"
"def keep():\n"
" return 1\n"
)
def test_move_assign_pastes_above_the_named_sibling(tmp_path: Path) -> None:
"""With before=, the constant lands immediately above the named top-level statement."""
(tmp_path / "src.py").write_text("RATIO = 3\n")
(tmp_path / "dst.py").write_text("def first():\n return 1\n")
r = Repro("b", "t").move_assign("RATIO", src="src.py", dst="dst.py", before="first")
_apply(r, tmp_path)
assert (tmp_path / "dst.py").read_text() == (
"RATIO = 3\n\ndef first():\n return 1\n"
)
def test_move_assign_relocates_an_annotated_constant(tmp_path: Path) -> None:
"""An annotated module constant (AnnAssign) is cut verbatim with its annotation intact."""
(tmp_path / "src.py").write_text(
"import os\n\nLIMIT: int = 480\n\n\ndef stay():\n return LIMIT\n"
)
(tmp_path / "dst.py").write_text("import sys\n\n\ndef keep():\n return 1\n")
r = Repro("b", "t").move_assign("LIMIT", src="src.py", dst="dst.py")
_apply(r, tmp_path)
assert "LIMIT" not in (tmp_path / "src.py").read_text().split("def stay")[0]
assert (tmp_path / "dst.py").read_text() == (
"import sys\n"
"\n"
"LIMIT: int = 480\n"
"\n"
"\n"
"def keep():\n"
" return 1\n"
)
def test_move_assign_missing_source_raises(tmp_path: Path) -> None:
"""A name with no module-level assignment in the source fails loudly."""
(tmp_path / "src.py").write_text("x = 1\n")
(tmp_path / "dst.py").write_text("import os\n")
r = Repro("b", "t").move_assign("MISSING", src="src.py", dst="dst.py")
with pytest.raises(AssertionError):
_apply(r, tmp_path)
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_move_symbol.py
+87
View File
@@ -321,6 +321,76 @@ def test_move_symbol_rejects_ambiguous_duplicate_names(tmp_path: Path) -> None:
_apply(r, tmp_path)
def test_move_symbol_after_inserts_below_named_function(tmp_path: Path) -> None:
"""With after=, the relocated def lands immediately below that sibling def."""
(tmp_path / "src.py").write_text("def moved():\n return 1\n")
(tmp_path / "dst.py").write_text(
"def first():\n return 0\n\n\ndef last():\n return 2\n"
)
r = Repro("b", "t").move_symbol(
"moved", src="src.py", dst="dst.py", into_class=None, after="first"
)
_apply(r, tmp_path)
dst_out = (tmp_path / "dst.py").read_text()
assert (
dst_out.index("def first")
< dst_out.index("def moved")
< dst_out.index("def last")
)
def test_move_symbol_after_assign_lands_before_typechecking_guard(
tmp_path: Path,
) -> None:
"""after= anchors on a module-level assignment target, landing the def just below it and
above a following ``if TYPE_CHECKING:`` guard (which is not a nameable anchor)."""
(tmp_path / "src.py").write_text("def helper(x):\n return x + 1\n")
(tmp_path / "dst.py").write_text(
"from u import is_hip\n"
"\n"
"_is_hip = is_hip()\n"
"\n"
"if TYPE_CHECKING:\n"
" from m import Thing\n"
)
r = Repro("b", "t").move_symbol(
"helper", src="src.py", dst="dst.py", into_class=None, after="_is_hip"
)
_apply(r, tmp_path)
dst_out = (tmp_path / "dst.py").read_text()
assert (
dst_out.index("_is_hip = is_hip()")
< dst_out.index("def helper")
< dst_out.index("if TYPE_CHECKING:")
)
def test_move_symbol_before_and_after_are_mutually_exclusive(tmp_path: Path) -> None:
"""Passing both before= and after= is rejected up front."""
(tmp_path / "src.py").write_text("def moved():\n return 1\n")
(tmp_path / "dst.py").write_text("def z():\n return 0\n")
with pytest.raises(AssertionError):
Repro("b", "t").move_symbol(
"moved",
src="src.py",
dst="dst.py",
into_class=None,
before="z",
after="z",
)
def test_move_symbol_asserts_when_after_symbol_missing(tmp_path: Path) -> None:
"""An after= anchor absent from the destination must raise, not fall back to append."""
(tmp_path / "src.py").write_text("def moved():\n return 1\n")
(tmp_path / "dst.py").write_text("def z():\n return 0\n")
r = Repro("b", "t").move_symbol(
"moved", src="src.py", dst="dst.py", into_class=None, after="NO_SUCH_SYMBOL"
)
with pytest.raises(AssertionError):
_apply(r, tmp_path)
def test_move_symbol_asserts_when_before_sibling_missing(tmp_path: Path) -> None:
"""A before= anchor absent from the destination must raise, not fall back to append."""
(tmp_path / "src.py").write_text("def moved():\n return 1\n")
@@ -357,4 +427,21 @@ def test_move_symbol_negative_dedent_indents_into_the_class(tmp_path: Path) -> N
assert " def helper(x):\n return x\n" in (tmp_path / "dst.py").read_text()
def test_move_symbol_relocates_a_top_level_class(tmp_path: Path) -> None:
"""move_symbol relocates a whole top-level class (with its methods) verbatim."""
(tmp_path / "src.py").write_text(
"x = 1\n\n\nclass Widget:\n def get(self, rank):\n return rank\n"
)
(tmp_path / "dst.py").write_text("y = 2\n")
r = Repro("b", "t").move_symbol(
"Widget", src="src.py", dst="dst.py", into_class=None
)
_apply(r, tmp_path)
assert "class Widget:" not in (tmp_path / "src.py").read_text()
assert (
"class Widget:\n def get(self, rank):\n return rank\n"
in (tmp_path / "dst.py").read_text()
)
# --- adversarial audit: leave_delegate stubs -------------------------------------
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_move_symbol_delegate.py
+118
View File
@@ -193,3 +193,121 @@ def test_move_symbol_async_leave_delegate_awaits_the_forwarded_call(
)
_apply(r, tmp_path)
assert "return await self.cfg.compute(n)" in (tmp_path / "src.py").read_text()
def test_move_symbol_leave_delegate_on_self_annotated_staticmethod(
tmp_path: Path,
) -> None:
"""A de-self'd staticmethod (self: Target) moves into Target; the stub drops the
decorator and the self annotation."""
(tmp_path / "src.py").write_text(
"class Runner:\n"
" @staticmethod\n"
" def work(self: Comp, n: int) -> int:\n"
" return n + self.base\n"
)
(tmp_path / "dst.py").write_text(
"class Comp:\n def existing(self):\n return 0\n"
)
r = Repro("b", "t").move_symbol(
"work",
src="src.py",
dst="dst.py",
into_class="Comp",
from_class="Runner",
drop_self_annotation=True,
leave_delegate="comp",
)
_apply(r, tmp_path)
assert (tmp_path / "src.py").read_text() == (
"class Runner:\n"
" def work(self, n: int) -> int:\n"
" return self.comp.work(n)\n"
)
assert (tmp_path / "dst.py").read_text() == (
"class Comp:\n"
" def existing(self):\n"
" return 0\n"
"\n"
" def work(self, n: int) -> int:\n"
" return n + self.base\n"
)
def test_move_symbol_leave_delegate_keeps_unrelated_self_annotation(
tmp_path: Path,
) -> None:
"""A self annotation naming a class other than the destination survives in the stub."""
(tmp_path / "src.py").write_text(
"class Mixin:\n"
" def work(self: Runner, n: int) -> int:\n"
" return n + self.base\n"
)
(tmp_path / "dst.py").write_text(
"class Comp:\n def existing(self):\n return 0\n"
)
r = Repro("b", "t").move_symbol(
"work",
src="src.py",
dst="dst.py",
into_class="Comp",
from_class="Mixin",
drop_self_annotation=True,
leave_delegate="comp",
)
_apply(r, tmp_path)
assert (tmp_path / "src.py").read_text() == (
"class Mixin:\n"
" def work(self: Runner, n: int) -> int:\n"
" return self.comp.work(n)\n"
)
def test_move_symbol_delegate_name_forwards_to_the_renamed_collaborator_method(
tmp_path: Path,
) -> None:
"""delegate_name makes the stub call a differently-named method on the collaborator."""
(tmp_path / "src.py").write_text(
"class Mixin:\n"
" def compute(self, n: int) -> int:\n"
" return n + self.cfg.base\n"
)
(tmp_path / "dst.py").write_text(
"class Cfg:\n def existing(self):\n return 0\n"
)
r = Repro("b", "t").move_symbol(
"compute",
src="src.py",
dst="dst.py",
into_class="Cfg",
leave_delegate="cfg",
delegate_name="compute_impl",
)
_apply(r, tmp_path)
assert "return self.cfg.compute_impl(n)" in (tmp_path / "src.py").read_text()
assert "def compute(self, n: int) -> int:" in (tmp_path / "dst.py").read_text()
def test_move_symbol_leave_delegate_on_unannotated_staticmethod_raises(
tmp_path: Path,
) -> None:
"""A staticmethod with no self: Target annotation has no receiver to forward through, so
leave_delegate refuses rather than author a bogus self.<field>.<name>(...) stub."""
(tmp_path / "src.py").write_text(
"class Runner:\n"
" @staticmethod\n"
" def work(x: int) -> int:\n"
" return x + 1\n"
)
(tmp_path / "dst.py").write_text(
"class Comp:\n def existing(self):\n return 0\n"
)
r = Repro("b", "t").move_symbol(
"work",
src="src.py",
dst="dst.py",
into_class="Comp",
leave_delegate="comp",
)
with pytest.raises(AssertionError):
_apply(r, tmp_path)
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_remove_import.py
+15
View File
@@ -118,6 +118,21 @@ def test_remove_import_leaves_other_statements_on_a_semicolon_line(
assert "import sys" in out and "print(sys.path)" in out
def test_remove_import_trailing_on_a_semicolon_line_leaves_no_dangling_separator(
tmp_path: Path,
) -> None:
"""Removing the trailing import on a semicolon-joined line drops the dangling ';' too
(a trailing space may remain for the formatter to strip, but the separator is gone).
"""
(tmp_path / "m.py").write_text("import sys; import os\nprint(sys.path)\n")
r = Repro("b", "t").remove_import("m.py", "import os")
_apply(r, tmp_path)
out = (tmp_path / "m.py").read_text()
assert ";" not in out
assert "import os" not in out
assert "import sys" in out and "print(sys.path)" in out
def test_remove_import_does_not_overmatch_a_submodule_import(tmp_path: Path) -> None:
"""Removing 'import os' must not also remove 'import os.path'."""
(tmp_path / "m.py").write_text("import os\nimport os.path\nprint(os.path.sep)\n")
.claude/skills/mechanical-refactor-verify/scripts/tests/reproduction_utils/test_remove_imported_name.py
+53 -3
View File
@@ -75,16 +75,49 @@ def test_remove_imported_name_asserts_when_absent(tmp_path: Path) -> None:
# --- add_import ----------------------------------------------------------------
def test_remove_imported_name_collapses_a_multiline_import_to_one_line(
def test_remove_imported_name_preserves_the_multiline_form(
tmp_path: Path,
) -> None:
"""Pruning a name from a parenthesized import rebuilds it as a single sorted-later line."""
"""Pruning a name from an exploded import deletes only that line, so the parens and the
magic trailing comma survive and the formatter keeps it multi-line (a flat rebuild would
collapse an import the target left multi-line)."""
(tmp_path / "m.py").write_text(
"from pkg import (\n a,\n moved,\n b,\n)\n\nx = a + b\n"
)
r = Repro("b", "t").remove_imported_name("m.py", module="pkg", name="moved")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "from pkg import a, b\n\nx = a + b\n"
assert (
tmp_path / "m.py"
).read_text() == "from pkg import (\n a,\n b,\n)\n\nx = a + b\n"
def test_remove_imported_name_multiline_down_to_one_collapses(
tmp_path: Path,
) -> None:
"""Pruning an exploded import down to a single surviving name collapses it to one line:
the formatter does not keep a lone name exploded, so a preserved-multiline form would
not match the target."""
(tmp_path / "m.py").write_text(
"from pkg import (\n moved,\n a,\n)\n\nx = a\n"
)
r = Repro("b", "t").remove_imported_name("m.py", module="pkg", name="moved")
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "from pkg import a\n\nx = a\n"
def test_remove_imported_name_down_to_one_with_a_comment_stays_exploded(
tmp_path: Path,
) -> None:
"""A lone survivor that carries a comment stays exploded (a rebuild would drop the
comment); only its own line is deleted."""
(tmp_path / "m.py").write_text(
"from pkg import (\n moved,\n a, # keep me\n)\n\nx = a\n"
)
r = Repro("b", "t").remove_imported_name("m.py", module="pkg", name="moved")
_apply(r, tmp_path)
assert (
tmp_path / "m.py"
).read_text() == "from pkg import (\n a, # keep me\n)\n\nx = a\n"
def test_remove_imported_name_matches_a_relative_module(tmp_path: Path) -> None:
@@ -111,3 +144,20 @@ def test_remove_imported_name_preserves_comments_in_a_multiline_import(
r = Repro("b", "t").remove_imported_name("m.py", module="pkg", name="moved")
_apply(r, tmp_path)
assert "# used by frobnicator" in (tmp_path / "m.py").read_text()
def test_remove_imported_name_keep_exploded_holds_a_lone_survivor_multiline(
tmp_path: Path,
) -> None:
"""With keep_exploded, pruning down to a single survivor deletes only the removed line,
so the survivor keeps its magic trailing comma and the import stays multi-line (the
author's choice, which the source cannot reveal). A regenerating impl would collapse.
"""
(tmp_path / "m.py").write_text(
"from pkg import (\n moved,\n a,\n)\n\nx = a\n"
)
r = Repro("b", "t").remove_imported_name(
"m.py", module="pkg", name="moved", keep_exploded=True
)
_apply(r, tmp_path)
assert (tmp_path / "m.py").read_text() == "from pkg import (\n a,\n)\n\nx = a\n"