mechanical-refactor-verify

1"""Reproduce a whole mechanical refactor and diff it byte-for-byte against a commit.2 3You write a ``transform()`` that recreates the change; ``verify_mechanical_refactor``4checks out the base commit in a throwaway worktree, runs the transform, runs pre-commit on5the changed files, and diffs the result against the target commit. An empty diff is a PASS.6Use this for a single mechanical PR -- a relocation, a whole-file split, or a rename where a7formatter re-wraps lines, which reproduce-and-byte-diff certifies exactly.8 9The ``Repro`` builder composes faithful relocation primitives (``move_symbol``,10``extract_to_new_module``, ``extract_symbols_to_new_module``, ``extract_function``,11``lower_call_sites``, ``requalify_call_sites``, ``remove_import``, ``remove_imported_name``,12``add_import``, ``repath_import``, ``add_typechecking_import``) into a transform, so a move13that a formatter re-wrapped can be reproduced and certified. Each primitive does only a relocation-faithful14edit (it never changes logic), so a byte match after the formatter certifies the commit is15exactly that relocation. The primitives are deliberately small -- AST-located, spliced as16original source text; see spec-reproduction-utils.md.17 18This module is self-contained and needs only git and the standard library.19"""20 21import ast22import io23import re24import shlex25import subprocess26import sys27import tempfile28import tokenize29from collections.abc import Callable30from pathlib import Path31 32 33def exec_command(cmd: str, cwd: str | None = None, check: bool = True) -> str:34    print(f"  $ {cmd}", flush=True)35    result = subprocess.run(36        cmd,37        shell=True,38        cwd=cwd,39        capture_output=True,40        text=True,41    )42    if check and result.returncode != 0:43        raise RuntimeError(f"command failed: {cmd}\n{result.stderr.strip()}")44    return result.stdout.strip()45 46 47def git_add_and_commit(message: str, cwd: str) -> None:48    exec_command(f"git add -A && git commit -m {shlex.quote(message)}", cwd=cwd)49 50 51def dedent(text: str, n: int) -> str:52    """Remove exactly n leading spaces from each line."""53    lines = _split_keepends(text)54    return "".join(line[n:] if line[:n] == " " * n else line for line in lines)55 56 57def _split_keepends(text: str) -> list[str]:58    """Split into lines ending in "\\n" only -- unlike ``str.splitlines``, a form feed or59    other exotic line break stays inside its line, matching ast's line numbering."""60    parts = text.split("\n")61    lines = [part + "\n" for part in parts[:-1]]62    if parts[-1]:63        lines.append(parts[-1])64    return lines65 66 67def _read_source(path: Path) -> str:68    """Read preserving the file's line endings (no universal-newline translation), so a69    CRLF file round-trips byte-for-byte through the primitives."""70    with path.open("r", newline="") as f:71        return f.read()72 73 74def _write_source(path: Path, text: str) -> None:75    with path.open("w", newline="") as f:76        f.write(text)77 78 79def _newline_style(text: str) -> str:80    return "\r\n" if "\r\n" in text else "\n"81 82 83def verify_mechanical_refactor(84    base_commit: str,85    target_commit: str,86    transform: "Callable[[Path], None]",87) -> None:88    repo_root = exec_command("git rev-parse --show-toplevel")89    worktree_dir = tempfile.mkdtemp(prefix="verify-mechanical-")90    branch_name = f"verify-mechanical-{base_commit[:8]}"91 92    try:93        print(f"[1/4] Creating worktree at {base_commit[:8]}...")94        exec_command(95            f"git worktree add -b {branch_name} {worktree_dir} {base_commit}",96            cwd=repo_root,97        )98 99        print("[2/4] Running transformation...")100        transform(Path(worktree_dir))101 102        print("[3/4] Running pre-commit...")103        exec_command("git add -A", cwd=worktree_dir)104        changed = exec_command(105            f"git diff --cached --name-only --diff-filter=ACMR {base_commit}",106            cwd=worktree_dir,107        ).split()108        if changed:109            files = " ".join(shlex.quote(path) for path in changed)110            exec_command(111                f"pre-commit run --files {files}", cwd=worktree_dir, check=False112            )113        if exec_command("git status --porcelain", cwd=worktree_dir):114            git_add_and_commit("pre-commit fixes", cwd=worktree_dir)115 116        print(f"[4/4] Diffing against {target_commit[:8]}...")117        diff = exec_command(118            f"git diff {target_commit} -- .",119            cwd=worktree_dir,120            check=False,121        )122 123        if diff:124            print(f"\nFAIL: diff is non-empty:\n{diff}")125            sys.exit(1)126        else:127            print("\nPASS: transform reproduces the commit exactly.")128 129    finally:130        print(f"\nWorktree left at: {worktree_dir}")131        print(f"Branch: {branch_name}")132        print("To clean up manually:")133        print(f"  git worktree remove {worktree_dir} && git branch -D {branch_name}")134 135 136# Decorators a method sheds when it becomes a free function; carried on one side of a move.137_MOVE_DECORATORS = {"@staticmethod", "@classmethod"}138 139 140def _find_def(tree: ast.AST, name: str) -> ast.AST | None:141    for node in ast.walk(tree):142        if (143            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))144            and node.name == name145        ):146            return node147    return None148 149 150def _find_unique_def(151    tree: ast.AST, name: str, *, from_class: str | None = None, where: str152) -> ast.AST:153    """Resolve ``def name`` (or ``class name``) and refuse ambiguity: with same-named defs in154    scope the first-match lookup could silently cut the wrong body, so the caller must scope155    the search with ``from_class``."""156    definition = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)157    root: ast.AST = tree158    if from_class is not None:159        cls = _find_class(tree, from_class)160        assert cls is not None, f"class {from_class} not found in {where}"161        root = cls162    elif isinstance(tree, ast.Module):163        top_level = [164            node165            for node in tree.body166            if isinstance(node, definition) and node.name == name167        ]168        if len(top_level) == 1:169            return top_level[0]170    matches = [171        node172        for node in ast.walk(root)173        if isinstance(node, definition) and node.name == name174    ]175    assert matches, f"{name} not found in {where}"176    assert len(matches) == 1, (177        f"{len(matches)} defs named {name} in {where}; pass from_class to disambiguate"178    )179    return matches[0]180 181 182def _def_header_end(def_text: str) -> int:183    """1-based line (within ``def_text``, which starts at the def line) of the colon that184    opens the body. Tokenize-based, so parentheses inside string defaults do not confuse185    the bracket depth."""186    depth = 0187    for token in tokenize.generate_tokens(io.StringIO(def_text).readline):188        if token.type == tokenize.OP:189            if token.string in "([{":190                depth += 1191            elif token.string in ")]}":192                depth -= 1193            elif token.string == ":" and depth == 0:194                return token.start[0]195    raise AssertionError("no header-ending colon found")196 197 198def _find_class(tree: ast.AST, name: str) -> ast.ClassDef | None:199    for node in ast.walk(tree):200        if isinstance(node, ast.ClassDef) and node.name == name:201            return node202    return None203 204 205def _def_span(node: ast.AST) -> tuple[int, int]:206    """(first, last) 1-based line numbers of a def, including its decorators."""207    start = min([node.lineno] + [d.lineno for d in node.decorator_list])208    return start, node.end_lineno209 210 211def _symbol_named(node: ast.AST, name: str) -> bool:212    """Whether a top-level statement defines the symbol ``name`` -- a def/class by its name,213    or a module-level assignment by one of its target names (so ``_is_hip = is_hip()`` is214    found by ``_is_hip``)."""215    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):216        return node.name == name217    if isinstance(node, ast.AnnAssign):218        return isinstance(node.target, ast.Name) and node.target.id == name219    if isinstance(node, ast.Assign):220        return any(isinstance(t, ast.Name) and t.id == name for t in node.targets)221    return False222 223 224def _byte_slice(line: str, start: int | None, end: int | None) -> str:225    """Slice a line by UTF-8 byte offsets -- ast col_offsets count bytes, not characters."""226    return line.encode("utf-8")[start:end].decode("utf-8")227 228 229def _replace_span(text: str, sl: int, sc: int, el: int, ec: int, repl: str) -> str:230    """Replace the text from (sl, sc) to (el, ec) -- 1-based lines, 0-based byte columns,231    end exclusive (ast node span semantics) -- with ``repl``."""232    lines = _split_keepends(text)233    before = "".join(lines[: sl - 1]) + _byte_slice(lines[sl - 1], None, sc)234    after = _byte_slice(lines[el - 1], ec, None) + "".join(lines[el:])235    return before + repl + after236 237 238def _slice_span(text: str, sl: int, sc: int, el: int, ec: int) -> str:239    lines = _split_keepends(text)240    if sl == el:241        return _byte_slice(lines[sl - 1], sc, ec)242    return (243        _byte_slice(lines[sl - 1], sc, None)244        + "".join(lines[sl : el - 1])245        + _byte_slice(lines[el - 1], None, ec)246    )247 248 249def _node_slice(text: str, node: ast.AST) -> str:250    return _slice_span(251        text, node.lineno, node.col_offset, node.end_lineno, node.end_col_offset252    )253 254 255def _rewrite_matching_calls(256    text: str, predicate: "Callable", rewrite: "Callable"257) -> str:258    """Rewrite every call ``predicate`` accepts by splicing the original source text259    (never ``ast.unparse``, which would re-spell literals and drop comments). One call is260    rewritten per pass and the text re-parsed, so a matching call nested inside another261    match is rewritten on a later pass instead of being overwritten."""262    while True:263        node = next(264            (265                n266                for n in ast.walk(ast.parse(text))267                if isinstance(n, ast.Call) and predicate(n)268            ),269            None,270        )271        if node is None:272            return text273        text = _replace_span(274            text,275            node.lineno,276            node.col_offset,277            node.end_lineno,278            node.end_col_offset,279            rewrite(text, node),280        )281 282 283def _lowered_call_text(text: str, node: ast.Call) -> str:284    """Original call text with the leading receiver argument spliced out and made the285    call's new receiver: ``Owner.foo(recv, rest...)`` -> ``recv.foo(rest...)``. All other286    argument bytes (literal spelling, comments, the magic trailing comma) are untouched.287    """288    receiver = node.args[0]289    receiver_src = _node_slice(text, receiver)290    assert "\n" not in receiver_src and "#" not in receiver_src, (291        f"receiver {receiver_src!r} must be single-line and comment-free"292    )293    opener = _slice_span(294        text,295        node.func.end_lineno,296        node.func.end_col_offset,297        receiver.lineno,298        receiver.col_offset,299    )300    assert "#" not in opener, f"comment before the receiver in {opener!r}"301    seg = _slice_span(302        text,303        receiver.end_lineno,304        receiver.end_col_offset,305        node.end_lineno,306        node.end_col_offset,307    )308    head, comma, rest = seg.partition(",")309    assert "#" not in head, f"comment after the receiver in {head!r}"310    if comma:311        assert head.strip() == "", f"unexpected text {head!r} after the receiver"312        rest = rest.lstrip(" \t")313    else:314        assert head.strip() == ")", f"unexpected text {head!r} after the receiver"315        rest = head.lstrip(" \t")316    return f"{receiver_src}.{node.func.attr}({rest}"317 318 319def _drop_self_annotation(method_text: str, name: str) -> str:320    """Drop the type annotation from a moved method's ``self`` parameter -- relocating321    ``def foo(self: Target)`` into ``Target`` makes the annotation redundant. The body is322    otherwise untouched, so the relocation stays byte-faithful. ``method_text`` may still be323    class-indented, so it is dedented to parse and the columns are mapped back."""324    first_line = method_text.split("\n", 1)[0]325    base_indent = len(first_line) - len(first_line.lstrip(" "))326    fn = _find_def(ast.parse(dedent(method_text, base_indent)), name)327    if fn is None or not fn.args.args:328        return method_text329    first = fn.args.args[0]330    if first.arg != "self" or first.annotation is None:331        return method_text332    annotation = first.annotation333    return _replace_span(334        method_text,335        first.lineno,336        first.col_offset + base_indent + len("self"),337        annotation.end_lineno,338        annotation.end_col_offset + base_indent,339        "",340    )341 342 343def _multiline_string_interior_lines(top_level_text: str) -> set[int]:344    """1-based lines of ``top_level_text`` that lie inside a multi-line string token345    (every line after the token's opening line, including the closing-delimiter line).346    Re-indenting those lines would change the literal's value, not its layout."""347    interior: set[int] = set()348    for token in tokenize.generate_tokens(io.StringIO(top_level_text).readline):349        if token.type == tokenize.STRING and token.end[0] > token.start[0]:350            interior.update(range(token.start[0] + 1, token.end[0] + 1))351    return interior352 353 354def _audit_extract_header(355    header: str,356    removed_assigns: dict[str, str | None],357    where: str,358    rederivable: dict[str, str | None] | None = None,359) -> None:360    """Refuse header content the extraction cannot vouch for. The header of a scattered361    extraction is authored text reproduced from the target commit, so anything beyond362    imports, a TYPE_CHECKING import block, a logger, a byte-equivalent copy of an363    assignment deleted from the source (``removed_assigns``), or a byte-equivalent copy of364    a module constant that *survives* in the source (``rederivable`` -- re-derived365    boilerplate such as ``_is_hip = is_hip()``, provably not fiction because the same366    statement still exists in the source) would let arbitrary new code ride into the new367    module under a PASS verdict."""368    rederivable = rederivable or {}369    header_assigned: set[str] = set()370    for stmt in ast.parse(header).body:371        if isinstance(stmt, (ast.Import, ast.ImportFrom)):372            continue373        if (374            isinstance(stmt, ast.Expr)375            and isinstance(stmt.value, ast.Constant)376            and isinstance(stmt.value.value, str)377        ):378            continue379        if (380            isinstance(stmt, ast.If)381            and ast.unparse(stmt.test) in ("TYPE_CHECKING", "typing.TYPE_CHECKING")382            and all(isinstance(sub, (ast.Import, ast.ImportFrom)) for sub in stmt.body)383        ):384            continue385        if isinstance(stmt, (ast.Assign, ast.AnnAssign)):386            targets = stmt.targets if isinstance(stmt, ast.Assign) else [stmt.target]387            names = [x.id for x in targets if isinstance(x, ast.Name)]388            value_src = ast.unparse(stmt.value) if stmt.value is not None else None389            if value_src == "logging.getLogger(__name__)":390                continue391            if names and all(392                n in removed_assigns and removed_assigns[n] == value_src for n in names393            ):394                header_assigned.update(names)395                continue396            if names and all(397                n in rederivable and rederivable[n] == value_src for n in names398            ):399                continue400        raise AssertionError(401            f"unverifiable header statement in {where}: {ast.unparse(stmt)!r} is "402            f"neither scaffolding nor a relocated source assignment"403        )404    missing = set(removed_assigns) - header_assigned405    assert not missing, (406        f"drop_assigns {sorted(missing)} deleted from the source but not reproduced "407        f"in the header of {where}"408    )409 410 411class Repro:412    """Builds a faithful relocation transform from primitives, then reproduces a commit.413 414    Operations are recorded and applied in order to a throwaway worktree at ``base``; the415    formatter (pre-commit) runs, and the result is diffed against ``target``. ``run`` prints416    PASS on an empty diff, otherwise the residual -- exactly what the relocation does not417    account for (a bundled change, or a re-derived scaffold a human must confirm)."""418 419    def __init__(self, base: str, target: str, repo_root: str | None = None) -> None:420        self.base = base421        self.target = target422        self.repo_root = repo_root423        self.ops: list[Callable[[Path], None]] = []424 425    def lower_call_sites(self, name: str, owner: str, *, paths: list[str]) -> "Repro":426        """Rewrite ``owner.name(receiver, rest)`` to ``receiver.name(rest)`` -- a static427        call becoming an instance-method call when ``name`` moves onto a class."""428 429        def op(root: Path) -> None:430            for rel in paths:431                path = root / rel432 433                def predicate(node: ast.Call) -> bool:434                    return (435                        isinstance(node.func, ast.Attribute)436                        and node.func.attr == name437                        and bool(node.args)438                        and ast.unparse(node.func.value) == owner439                    )440 441                _write_source(442                    path,443                    _rewrite_matching_calls(444                        _read_source(path), predicate, _lowered_call_text445                    ),446                )447 448        self.ops.append(op)449        return self450 451    def requalify_call_sites(452        self, name: str, owner: str, *, paths: list[str]453    ) -> "Repro":454        """Rewrite ``owner.name(args)`` to ``name(args)`` -- dropping the qualifier when455        ``name`` moves to a module-level free function."""456 457        def op(root: Path) -> None:458            for rel in paths:459                path = root / rel460 461                def predicate(node: ast.Call) -> bool:462                    return (463                        isinstance(node.func, ast.Attribute)464                        and node.func.attr == name465                        and ast.unparse(node.func.value) == owner466                    )467 468                def rewrite(text: str, node: ast.Call) -> str:469                    call_src = _node_slice(text, node)470                    func_src = _node_slice(text, node.func)471                    return name + call_src[len(func_src) :]472 473                _write_source(474                    path,475                    _rewrite_matching_calls(_read_source(path), predicate, rewrite),476                )477 478        self.ops.append(op)479        return self480 481    def route_call_sites_through_field(482        self, name: str, *, field: str, paths: list[str], owner: str | None = None483    ) -> "Repro":484        """Rewrite ``<recv>.name(args)`` to ``<recv>.field.name(args)`` -- the method moved485        onto a collaborator reached through ``self.field``, so its callers route through that486        field. With ``owner`` given, only calls whose receiver text equals ``owner`` are487        rewritten. A call already routed through ``field`` is skipped, so the pass converges.488        """489 490        def op(root: Path) -> None:491            for rel in paths:492                path = root / rel493 494                def predicate(node: ast.Call) -> bool:495                    return (496                        isinstance(node.func, ast.Attribute)497                        and node.func.attr == name498                        and not (499                            isinstance(node.func.value, ast.Attribute)500                            and node.func.value.attr == field501                        )502                        and (owner is None or ast.unparse(node.func.value) == owner)503                    )504 505                def rewrite(text: str, node: ast.Call) -> str:506                    call_src = _node_slice(text, node)507                    func_src = _node_slice(text, node.func)508                    receiver_src = _node_slice(text, node.func.value)509                    return receiver_src + f".{field}.{name}" + call_src[len(func_src) :]510 511                _write_source(512                    path,513                    _rewrite_matching_calls(_read_source(path), predicate, rewrite),514                )515 516        self.ops.append(op)517        return self518 519    def remove_import(520        self, rel: str, import_text: str, *, in_function: str | None = None521    ) -> "Repro":522        """Remove every import statement whose text contains ``import_text`` (and a trailing523        blank), optionally scoped to one function so a same-text module-level import (e.g. a524        ``TYPE_CHECKING`` guard) is left untouched."""525 526        def op(root: Path) -> None:527            path = root / rel528            lines = _split_keepends(_read_source(path))529            tree = ast.parse("".join(lines))530            scope: tuple[int, int] | None = None531            if in_function is not None:532                fn = _find_unique_def(tree, in_function, where=rel)533                scope = (fn.lineno, fn.end_lineno)534            compound = (535                ast.FunctionDef,536                ast.AsyncFunctionDef,537                ast.ClassDef,538                ast.If,539                ast.For,540                ast.AsyncFor,541                ast.While,542                ast.With,543                ast.AsyncWith,544                ast.Try,545                ast.Match,546            )547            simple_stmt_lines: dict[int, int] = {}548            for stmt in ast.walk(tree):549                if isinstance(stmt, ast.stmt) and not isinstance(stmt, compound):550                    for lineno in range(stmt.lineno, stmt.end_lineno + 1):551                        simple_stmt_lines[lineno] = simple_stmt_lines.get(lineno, 0) + 1552            pattern = re.compile(rf"(?<![\w.]){re.escape(import_text)}(?![\w.])")553            matched: list[ast.stmt] = []554            for node in ast.walk(tree):555                if isinstance(node, (ast.Import, ast.ImportFrom)):556                    if scope is not None and not (scope[0] <= node.lineno <= scope[1]):557                        continue558                    seg = _slice_span(559                        "".join(lines),560                        node.lineno,561                        node.col_offset,562                        node.end_lineno,563                        node.end_col_offset,564                    )565                    if pattern.search(seg):566                        matched.append(node)567            assert matched, f"import {import_text!r} not found in {rel}"568            whole: list[tuple[int, int]] = []569            shared: list[ast.stmt] = []570            for node in matched:571                alone = all(572                    simple_stmt_lines[lineno] == 1573                    for lineno in range(node.lineno, node.end_lineno + 1)574                )575                if alone:576                    lo, hi = node.lineno, node.end_lineno577                    if hi < len(lines) and lines[hi].strip() == "":578                        hi += 1579                    whole.append((lo, hi))580                else:581                    shared.append(node)582            text = "".join(lines)583            for node in sorted(584                shared, key=lambda n: (n.lineno, n.col_offset), reverse=True585            ):586                text = _replace_span(587                    text,588                    node.lineno,589                    node.col_offset,590                    node.end_lineno,591                    node.end_col_offset,592                    "",593                )594                fixed = _split_keepends(text)595                joined_line = fixed[node.lineno - 1]596                cleaned = re.sub(r";\s*;", ";", joined_line)597                cleaned = re.sub(r"^(\s*);\s*", r"\1", cleaned)598                cleaned = re.sub(r"\s*;(\s*)$", r"\1", cleaned)599                fixed[node.lineno - 1] = cleaned600                text = "".join(fixed)601            lines = _split_keepends(text)602            for lo, hi in sorted(whole, reverse=True):603                del lines[lo - 1 : hi]604            _write_source(path, "".join(lines))605 606        self.ops.append(op)607        return self608 609    def remove_imported_name(610        self,611        rel: str,612        *,613        module: str | None,614        name: str,615        asname: str | None = None,616        keep_exploded: bool = False,617    ) -> "Repro":618        """Drop a single imported ``name`` from a module-level import: from a ``from module619        import a, b`` keep the rest and drop only ``name``; when it was the sole name -- or for620        a plain ``import name`` (``module=None``) -- remove the whole statement. The symbol's621        home changed, so an importer that no longer references it loses exactly that name; the622        import sorter rewrites the surviving line. An import diff is always whitelisted, so this623        realises a lost name directly instead of relying on the formatter to prune it.624 625        Removing down to a single surviving name collapses the import to one line by default626        (the common case). Pass ``keep_exploded`` when the target kept the sole survivor627        exploded (its magic trailing comma preserved): the alias line is then merely deleted628        so the surviving name keeps its comma and the formatter leaves the import multi-line.629        The choice is the commit author's and cannot be inferred from the source.630        """631 632        def alias_text(alias: ast.alias) -> str:633            return alias.name + (f" as {alias.asname}" if alias.asname else "")634 635        def op(root: Path) -> None:636            path = root / rel637            lines = _split_keepends(_read_source(path))638            nl = _newline_style("".join(lines))639            edits: list[tuple[int, int, str | None]] = []640            for node in ast.parse("".join(lines)).body:641                if module is None:642                    if not isinstance(node, ast.Import):643                        continue644                else:645                    if not isinstance(node, ast.ImportFrom):646                        continue647                    if "." * node.level + (node.module or "") != module:648                        continue649                dropped_alias = next(650                    (a for a in node.names if a.name == name and a.asname == asname),651                    None,652                )653                if dropped_alias is None:654                    continue655                kept = [a for a in node.names if a is not dropped_alias]656                if not kept:657                    edits.append((node.lineno, node.end_lineno, None))658                    continue659                stmt_lines = lines[node.lineno - 1 : node.end_lineno]660                own = dropped_alias.lineno661                own_line = lines[own - 1]662                on_own_line = own_line.strip().rstrip(",").strip() == alias_text(663                    dropped_alias664                )665                has_comments = any("#" in ln for ln in stmt_lines)666                # Preserve the exploded form -- delete just this alias's line -- when the667                # import stays multi-line: 2+ surviving names keep the magic trailing comma668                # exploded, and an import carrying comments must not be rebuilt (a rebuild669                # would drop them). Both match how the target was edited (a flat rebuild670                # would drop the magic comma and collapse an import the target left671                # multi-line). A lone surviving name collapses to one line by default, unless672                # keep_exploded says the target preserved the magic comma for it too.673                if on_own_line and (len(kept) >= 2 or has_comments or keep_exploded):674                    edits.append((own, own, None))675                elif has_comments and not on_own_line:676                    raise AssertionError(677                        f"cannot drop {name!r}: it shares a line with other text and "678                        f"the import holds comments that a rebuild would delete"679                    )680                else:681                    keyword = "import " if module is None else f"from {module} import "682                    rebuilt = keyword + ", ".join(alias_text(a) for a in kept) + nl683                    edits.append((node.lineno, node.end_lineno, rebuilt))684            assert edits, f"import of {name!r} from {module!r} not found in {rel}"685            for lo, hi, repl in sorted(edits, reverse=True):686                if repl is None:687                    del lines[lo - 1 : hi]688                else:689                    lines[lo - 1 : hi] = [repl]690            _write_source(path, "".join(lines))691 692        self.ops.append(op)693        return self694 695    def add_imported_name(696        self, rel: str, *, module: str, name: str, asname: str | None = None697    ) -> "Repro":698        """Add a single ``name`` to an existing module-level ``from module import a, b`` --699        the dual of ``remove_imported_name``. A relocated symbol gains a new importer that700        already imports other names from the same module, so the target extends that line701        rather than adding a fresh statement (which the sorter would not merge across an702        intervening non-import statement). The import sorter rewrites the surviving line; an703        import carrying comments is refused, since a rebuild would drop them."""704 705        def alias_text(target_name: str, target_asname: str | None) -> str:706            return target_name + (f" as {target_asname}" if target_asname else "")707 708        def op(root: Path) -> None:709            path = root / rel710            lines = _split_keepends(_read_source(path))711            nl = _newline_style("".join(lines))712            for node in ast.parse("".join(lines)).body:713                if not isinstance(node, ast.ImportFrom):714                    continue715                if "." * node.level + (node.module or "") != module:716                    continue717                stmt_lines = lines[node.lineno - 1 : node.end_lineno]718                if any("#" in ln for ln in stmt_lines):719                    raise AssertionError(720                        f"cannot add {name!r} to the import from {module!r} in {rel}: "721                        f"it holds comments that a rebuild would delete"722                    )723                existing = [alias_text(a.name, a.asname) for a in node.names]724                added = alias_text(name, asname)725                assert added not in existing, (726                    f"{name!r} already imported from {module!r} in {rel}"727                )728                rebuilt = f"from {module} import " + ", ".join(existing + [added]) + nl729                lines[node.lineno - 1 : node.end_lineno] = [rebuilt]730                _write_source(path, "".join(lines))731                return732            raise AssertionError(f"no `from {module} import` statement in {rel}")733 734        self.ops.append(op)735        return self736 737    def add_import(738        self, rel: str, import_stmt: str, *, after: str | None = None739    ) -> "Repro":740        """Append an import after the last top-level import; the formatter's import sorter741        places it (so the exact insertion point does not matter). When ``after`` is given,742        insert immediately after the top-level import statement whose source text contains743        that substring instead -- needed for a file whose imports are split into separate744        isort sections by an intervening statement (e.g. ``_is_hip = is_hip()``), where the745        sorter will not carry the new import across the boundary into the intended block.746        """747 748        def op(root: Path) -> None:749            path = root / rel750            lines = _split_keepends(_read_source(path))751            nl = _newline_style("".join(lines))752            body = ast.parse("".join(lines)).body753            if after is not None:754                anchor = None755                for node in body:756                    if isinstance(757                        node, (ast.Import, ast.ImportFrom)758                    ) and after in "".join(lines[node.lineno - 1 : node.end_lineno]):759                        anchor = node760                        break761                if anchor is None:762                    raise AssertionError(763                        f"no top-level import containing {after!r} in {rel}"764                    )765                at = anchor.end_lineno766                _write_source(767                    path, "".join(lines[:at] + [import_stmt + nl] + lines[at:])768                )769                return770            last = 0771            if (772                body773                and isinstance(body[0], ast.Expr)774                and isinstance(body[0].value, ast.Constant)775                and isinstance(body[0].value.value, str)776            ):777                last = body[0].end_lineno778            for node in body:779                if isinstance(node, (ast.Import, ast.ImportFrom)):780                    last = max(last, node.end_lineno)781            _write_source(782                path, "".join(lines[:last] + [import_stmt + nl] + lines[last:])783            )784 785        self.ops.append(op)786        return self787 788    def add_typechecking_import(self, rel: str, import_stmt: str) -> "Repro":789        """Append ``import_stmt`` inside the file's ``if TYPE_CHECKING:`` block -- a moved790        definition whose annotations reference a type needs that type imported there. The791        import sorter orders the block, so the exact insertion point does not matter. A lone792        ``pass`` placeholder (the block's only statement) is dropped: populating an empty793        ``TYPE_CHECKING`` block makes its placeholder redundant, so the target removes it.794        With no existing block, one is created after the trailing module import -- the795        destination gains the guard together with its first import.796        """797 798        def op(root: Path) -> None:799            path = root / rel800            lines = _split_keepends(_read_source(path))801            nl = _newline_style("".join(lines))802            for node in ast.parse("".join(lines)).body:803                if isinstance(node, ast.If) and ast.unparse(node.test) in (804                    "TYPE_CHECKING",805                    "typing.TYPE_CHECKING",806                ):807                    indent = " " * node.body[0].col_offset808                    lone_pass = len(node.body) == 1 and isinstance(809                        node.body[0], ast.Pass810                    )811                    if lone_pass:812                        placeholder = node.body[0]813                        lines[placeholder.lineno - 1 : placeholder.end_lineno] = [814                            indent + import_stmt + nl815                        ]816                    else:817                        lines.insert(818                            node.body[-1].end_lineno, indent + import_stmt + nl819                        )820                    _write_source(path, "".join(lines))821                    return822            tree = ast.parse("".join(lines))823            imports = [824                node825                for node in tree.body826                if isinstance(node, (ast.Import, ast.ImportFrom))827            ]828            assert imports, (829                f"no imports to anchor a new `if TYPE_CHECKING:` block in {rel}"830            )831            insert_at = imports[-1].end_lineno832            lines[insert_at:insert_at] = [833                nl,834                "if TYPE_CHECKING:" + nl,835                "    " + import_stmt + nl,836            ]837            _write_source(path, "".join(lines))838 839        self.ops.append(op)840        return self841 842    def repath_import(843        self, rel: str, *, old_module: str, new_module: str, name: str844    ) -> "Repro":845        """Repath every function-scoped ``from old_module import ... name ...`` to846        ``from new_module import ...`` in place -- the moved symbol's home changed, so its847        importer adjusts. Only imports nested below module level are touched; a module-level848        import is left to the import sorter via add_import / remove_import."""849 850        def op(root: Path) -> None:851            path = root / rel852            lines = _split_keepends(_read_source(path))853            tree = ast.parse("".join(lines))854            top_level = {id(node) for node in tree.body}855            changed = False856            for node in ast.walk(tree):857                if (858                    isinstance(node, ast.ImportFrom)859                    and id(node) not in top_level860                    and node.module == old_module861                    and any(alias.name == name for alias in node.names)862                ):863                    spelled = "." * node.level + (node.module or "")864                    replaced = lines[node.lineno - 1].replace(865                        f"from {spelled} import", f"from {new_module} import", 1866                    )867                    assert replaced != lines[node.lineno - 1], (868                        f"import spelling {spelled!r} not found on its line in {rel}"869                    )870                    lines[node.lineno - 1] = replaced871                    changed = True872            assert changed, f"nested import of {name} from {old_module} not in {rel}"873            _write_source(path, "".join(lines))874 875        self.ops.append(op)876        return self877 878    def move_assign(879        self,880        name: str,881        *,882        src: str,883        dst: str,884        before: str | None = None,885    ) -> "Repro":886        """Cut the module-level assignment binding ``name`` from ``src`` and paste it verbatim887        into ``dst`` at module level -- a module constant relocated together with the code888        that reads it. Pasted immediately above the top-level statement named ``before``889        when given, else after the last top-level import."""890 891        def op(root: Path) -> None:892            src_path = root / src893            dst_path = root / dst894            src_lines = _split_keepends(_read_source(src_path))895            node = None896            for cand in ast.parse("".join(src_lines)).body:897                if (898                    isinstance(cand, ast.Assign)899                    and len(cand.targets) == 1900                    and isinstance(cand.targets[0], ast.Name)901                    and cand.targets[0].id == name902                ) or (903                    isinstance(cand, ast.AnnAssign)904                    and isinstance(cand.target, ast.Name)905                    and cand.target.id == name906                ):907                    node = cand908            assert node is not None, f"module assignment {name} not found in {src}"909            block = "".join(src_lines[node.lineno - 1 : node.end_lineno])910            _write_source(911                src_path,912                "".join(src_lines[: node.lineno - 1] + src_lines[node.end_lineno :]),913            )914 915            dst_lines = _split_keepends(_read_source(dst_path))916            dst_nl = _newline_style("".join(dst_lines))917            dst_tree = ast.parse("".join(dst_lines))918            at = None919            if before is not None:920                for cand in dst_tree.body:921                    cand_name = getattr(cand, "name", None) or (922                        cand.targets[0].id923                        if isinstance(cand, ast.Assign)924                        and len(cand.targets) == 1925                        and isinstance(cand.targets[0], ast.Name)926                        else None927                    )928                    if cand_name == before:929                        at = (930                            min(931                                [d.lineno for d in getattr(cand, "decorator_list", [])],932                                default=cand.lineno,933                            )934                            - 1935                        )936                        break937                assert at is not None, f"before={before!r} not found in {dst}"938                dst_lines[at:at] = [block, dst_nl]939            else:940                imports = [941                    n942                    for n in dst_tree.body943                    if isinstance(n, (ast.Import, ast.ImportFrom))944                ]945                at = imports[-1].end_lineno if imports else 0946                dst_lines[at:at] = [dst_nl, block]947            _write_source(dst_path, "".join(dst_lines))948 949        self.ops.append(op)950        return self951 952    def move_symbol(953        self,954        name: str,955        *,956        src: str,957        dst: str,958        into_class: str | None,959        from_class: str | None = None,960        dedent: int = 0,961        drop_self_annotation: bool = False,962        before: str | None = None,963        after: str | None = None,964        leave_delegate: str | None = None,965        delegate_name: str | None = None,966    ) -> "Repro":967        """Cut ``def name`` (with decorators) from ``src`` and paste it into ``dst`` --968        immediately above the sibling def ``before`` when given (so the relocated def lands in969        the chain's order), immediately below the top-level symbol ``after`` when given (a970        sibling def/class or a module-level assignment target -- used to land the def just971        before a following ``if TYPE_CHECKING:`` guard, which is not a nameable anchor), else972        at the end of ``into_class`` (or module level when None) -- dropping a move decorator973        and dedenting by ``dedent``. When ``drop_self_annotation``, the moved method's974        ``self: Target`` annotation is dropped (redundant inside the class). The body is moved975        verbatim; the formatter normalises the surrounding blank lines.976        """977        assert before is None or after is None, (978            "move_symbol: before and after are mutually exclusive"979        )980 981        def op(root: Path) -> None:982            src_path = root / src983            dst_path = root / dst984            src_lines = _split_keepends(_read_source(src_path))985            src_nl = _newline_style("".join(src_lines))986            node = _find_unique_def(987                ast.parse("".join(src_lines)), name, from_class=from_class, where=src988            )989            start, end = _def_span(node)990            block = src_lines[start - 1 : end]991            decorator_lines = node.lineno - start992            if leave_delegate is not None:993                args = node.args994                has_move_decorator = any(995                    ln.strip() in _MOVE_DECORATORS for ln in block[:decorator_lines]996                )997                arg_list = args.posonlyargs + args.args998                self_annotated = (999                    bool(arg_list)1000                    and arg_list[0].arg == "self"1001                    and arg_list[0].annotation is not None1002                )1003                assert not has_move_decorator or self_annotated, (1004                    f"leave_delegate on a {_MOVE_DECORATORS} method has no self to "1005                    "forward (a de-self'd staticmethod must annotate its self param)"1006                )1007                parts = [p.arg for p in args.posonlyargs + args.args if p.arg != "self"]1008                if args.vararg is not None:1009                    parts.append(f"*{args.vararg.arg}")1010                parts += [f"{k.arg}={k.arg}" for k in args.kwonlyargs]1011                if args.kwarg is not None:1012                    parts.append(f"**{args.kwarg.arg}")1013                # The signature spans the def header only (def line through the line whose1014                # colon opens the body). node.body[0].lineno would skip over any leading1015                # comment/blank lines (not AST nodes), wrongly absorbing them into the1016                # delegate, so the header end is found by tokenizing the def.1017                header_end = (1018                    node.lineno1019                    - 11020                    + _def_header_end("".join(src_lines[node.lineno - 1 : end]))1021                )1022                sig_start = node.lineno - 1 if has_move_decorator else start - 11023                signature_text = "".join(src_lines[sig_start:header_end])1024                # The stub drops the self annotation only when it names the class the1025                # def moved into (now redundant); an unrelated annotation (a mixin's1026                # `self: ModelRunner`) is part of the surviving header and stays.1027                ann = arg_list[0].annotation if self_annotated else None1028                ann_name = None1029                if isinstance(ann, ast.Name):1030                    ann_name = ann.id1031                elif isinstance(ann, ast.Constant) and isinstance(ann.value, str):1032                    ann_name = ann.value.split(".")[-1]1033                elif isinstance(ann, ast.Attribute):1034                    ann_name = ann.attr1035                if self_annotated and ann_name == into_class:1036                    sig_indent = len(signature_text) - len(signature_text.lstrip(" "))1037                    parsable = signature_text + " " * sig_indent + "    pass" + src_nl1038                    stripped = _drop_self_annotation(parsable, name)1039                    assert stripped.endswith(" " * sig_indent + "    pass" + src_nl)1040                    signature_text = stripped[1041                        : -len(" " * sig_indent + "    pass" + src_nl)1042                    ]1043                body_indent = " " * node.body[0].col_offset1044                returning = (1045                    "return await"1046                    if isinstance(node, ast.AsyncFunctionDef)1047                    else "return"1048                )1049                forward = (1050                    f"{body_indent}{returning} self.{leave_delegate}."1051                    f"{delegate_name or name}({', '.join(parts)})" + src_nl1052                )1053                delegate = signature_text + forward1054                _write_source(1055                    src_path,1056                    "".join(src_lines[: start - 1] + [delegate] + src_lines[end:]),1057                )1058            else:1059                _write_source(1060                    src_path, "".join(src_lines[: start - 1] + src_lines[end:])1061                )1062 1063            kept = [1064                ln1065                for index, ln in enumerate(block)1066                if not (index < decorator_lines and ln.strip() in _MOVE_DECORATORS)1067            ]1068            if dedent > 0:1069                kept = [1070                    ln[dedent:] if ln[:dedent] == " " * dedent else ln for ln in kept1071                ]1072            elif dedent < 0:1073                pad = " " * -dedent1074                kept = [pad + ln if ln.strip() else ln for ln in kept]1075            method_text = "".join(kept)1076            if drop_self_annotation:1077                method_text = _drop_self_annotation(method_text, name)1078 1079            dst_lines = _split_keepends(_read_source(dst_path))1080            dst_nl = _newline_style("".join(dst_lines))1081            dst_tree = ast.parse("".join(dst_lines))1082            container = dst_tree.body1083            if into_class is not None:1084                cls = _find_class(dst_tree, into_class)1085                assert cls is not None, f"class {into_class} not found in {dst}"1086                container = cls.body1087            target = None1088            if before is not None:1089                target = next(1090                    (1091                        n1092                        for n in container1093                        if isinstance(1094                            n,1095                            (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef),1096                        )1097                        and n.name == before1098                    ),1099                    None,1100                )1101                assert target is not None, f"before={before!r} not found in {dst}"1102            if after is not None:1103                anchor = next(1104                    (n for n in container if _symbol_named(n, after)),1105                    None,1106                )1107                assert anchor is not None, f"after={after!r} not found in {dst}"1108                at = anchor.end_lineno1109                _write_source(1110                    dst_path,1111                    "".join(dst_lines[:at] + [dst_nl, method_text] + dst_lines[at:]),1112                )1113            elif target is not None:1114                at = _def_span(target)[0] - 11115                _write_source(1116                    dst_path,1117                    "".join(dst_lines[:at] + [method_text, dst_nl] + dst_lines[at:]),1118                )1119            else:1120                at = (1121                    container[-1].end_lineno1122                    if into_class is not None1123                    else len(dst_lines)1124                )1125                _write_source(1126                    dst_path,1127                    "".join(dst_lines[:at] + [dst_nl, method_text] + dst_lines[at:]),1128                )1129 1130        self.ops.append(op)1131        return self1132 1133    def extract_to_new_module(1134        self,1135        src: str,1136        dst: str,1137        *,1138        symbols: list[str],1139        future_import: bool = True,1140    ) -> "Repro":1141        """Cut the contiguous tail of ``src`` -- the moved ``symbols`` and the module1142        scaffolding that leads into them (imports, a ``TYPE_CHECKING`` guard, a logger,1143        module constants) -- and write it as the new module ``dst``, prepending1144        ``from __future__ import annotations`` when ``future_import``. The body is moved1145        verbatim; the formatter sorts the imports and normalises the blank lines."""1146 1147        def op(root: Path) -> None:1148            src_path = root / src1149            dst_path = root / dst1150            src_lines = _split_keepends(_read_source(src_path))1151            body = ast.parse("".join(src_lines)).body1152            wanted = set(symbols)1153 1154            def is_scaffolding(node: ast.stmt) -> bool:1155                if isinstance(node, (ast.Import, ast.ImportFrom)):1156                    return True1157                if isinstance(node, ast.If):1158                    return ast.unparse(node.test) in (1159                        "TYPE_CHECKING",1160                        "typing.TYPE_CHECKING",1161                    )1162                if isinstance(node, ast.Assign):1163                    return all(isinstance(x, ast.Name) for x in node.targets)1164                if isinstance(node, ast.AnnAssign):1165                    return isinstance(node.target, ast.Name)1166                return False1167 1168            cut = len(body)1169            while cut > 0:1170                node = body[cut - 1]1171                is_symbol = (1172                    isinstance(1173                        node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)1174                    )1175                    and node.name in wanted1176                )1177                if is_symbol or is_scaffolding(node):1178                    cut -= 11179                else:1180                    break1181            tail = body[cut:]1182            present = {1183                node.name1184                for node in tail1185                if isinstance(1186                    node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)1187                )1188            }1189            assert wanted <= present, f"{wanted - present} not in the cut tail of {src}"1190 1191            decorators = getattr(tail[0], "decorator_list", [])1192            start = min([tail[0].lineno] + [d.lineno for d in decorators])1193            block = "".join(src_lines[start - 1 :])1194            _write_source(src_path, "".join(src_lines[: start - 1]))1195 1196            nl = _newline_style(block)1197            prefix = "from __future__ import annotations" + nl if future_import else ""1198            dst_path.parent.mkdir(parents=True, exist_ok=True)1199            _write_source(dst_path, prefix + block)1200 1201        self.ops.append(op)1202        return self1203 1204    def extract_symbols_to_new_module(1205        self,1206        src: str,1207        dst: str,1208        *,1209        symbols: list[str],1210        header: str,1211        order: list[str],1212        drop_assigns: list[str] | None = None,1213    ) -> "Repro":1214        """Relocate the named top-level defs/classes from *scattered* positions in ``src`` into1215        a new module ``dst`` whose authored ``header`` (the imports, module constants, a logger,1216        a ``TYPE_CHECKING`` block -- harmless or re-derived boilerplate reproduced from the1217        target) precedes them. Unlike ``extract_to_new_module``, the symbols need not be a1218        contiguous tail: each is cut from ``src`` verbatim, so its body stays a proven1219        relocation, and the cut blocks are appended in ``order`` (their order in the target).1220        ``drop_assigns`` names module-level assignments (e.g. a ``_is_hip = is_hip()`` constant)1221        that moved into the new module's header, so they are deleted from ``src`` too -- their1222        relocated copy is reproduced in the authored ``header``. The formatter normalises the1223        spacing; the byte diff then certifies the bodies are exactly the source's, while only1224        the small header is authored."""1225 1226        def op(root: Path) -> None:1227            src_path = root / src1228            dst_path = root / dst1229            src_lines = _split_keepends(_read_source(src_path))1230            src_nl = _newline_style("".join(src_lines))1231            wanted = set(symbols)1232            dropped = set(drop_assigns or [])1233            assert set(order) == wanted, f"order {order} must permute symbols {symbols}"1234            tree = ast.parse("".join(src_lines))1235            nodes = {1236                node.name: node1237                for node in tree.body1238                if isinstance(1239                    node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)1240                )1241                and node.name in wanted1242            }1243            missing = wanted - set(nodes)1244            assert not missing, f"{missing} not top-level defs/classes in {src}"1245            spans = {name: _def_span(node) for name, node in nodes.items()}1246            blocks = {1247                name: "".join(src_lines[start - 1 : end])1248                for name, (start, end) in spans.items()1249            }1250            assign_spans: list[tuple[int, int]] = []1251            assign_rewrites: list[tuple[int, int, str]] = []1252            removed_assigns: dict[str, str] = {}1253            found_assigns: set[str] = set()1254            for node in tree.body:1255                targets = (1256                    node.targets1257                    if isinstance(node, ast.Assign)1258                    else [node.target]1259                    if isinstance(node, ast.AnnAssign)1260                    else []1261                )1262                names = {t.id for t in targets if isinstance(t, ast.Name)}1263                hit = names & dropped1264                if not hit:1265                    continue1266                assert len(names) == len(targets), (1267                    f"drop_assigns {sorted(hit)}: non-name targets in {src}"1268                )1269                value_src = ast.unparse(node.value) if node.value is not None else None1270                for dropped_name in hit:1271                    removed_assigns[dropped_name] = value_src1272                surviving = [1273                    x.id1274                    for x in targets1275                    if isinstance(x, ast.Name) and x.id not in dropped1276                ]1277                if surviving:1278                    kept_stmt = (1279                        " = ".join(surviving)1280                        + " = "1281                        + _slice_span(1282                            "".join(src_lines),1283                            node.value.lineno,1284                            node.value.col_offset,1285                            node.value.end_lineno,1286                            node.value.end_col_offset,1287                        )1288                        + src_nl1289                    )1290                    assign_rewrites.append((node.lineno, node.end_lineno, kept_stmt))1291                else:1292                    assign_spans.append((node.lineno, node.end_lineno))1293                found_assigns |= hit1294            assert found_assigns == dropped, (1295                f"{dropped - found_assigns} not assigned in {src}"1296            )1297            rederivable: dict[str, str | None] = {}1298            for node in tree.body:1299                targets = (1300                    node.targets1301                    if isinstance(node, ast.Assign)1302                    else [node.target]1303                    if isinstance(node, ast.AnnAssign)1304                    else []1305                )1306                names = [t.id for t in targets if isinstance(t, ast.Name)]1307                if not names or set(names) & dropped:1308                    continue1309                value_src = ast.unparse(node.value) if node.value is not None else None1310                for kept_name in names:1311                    rederivable[kept_name] = value_src1312            if header.strip() or removed_assigns:1313                _audit_extract_header(1314                    header, removed_assigns, where=dst, rederivable=rederivable1315                )1316            cuts = [(start, end, None) for start, end in spans.values()]1317            cuts += [(start, end, None) for start, end in assign_spans]1318            cuts += assign_rewrites1319            for start, end, repl in sorted(1320                cuts, key=lambda c: (c[0], c[1]), reverse=True1321            ):1322                if repl is None:1323                    del src_lines[start - 1 : end]1324                else:1325                    src_lines[start - 1 : end] = [repl]1326            _write_source(src_path, "".join(src_lines))1327 1328            gap = src_nl * 31329            relocated = gap.join(blocks[name].rstrip("\r\n") for name in order)1330            prefix = header.rstrip("\r\n") + gap if header.strip() else ""1331            dst_path.parent.mkdir(parents=True, exist_ok=True)1332            _write_source(dst_path, prefix + relocated + src_nl)1333 1334        self.ops.append(op)1335        return self1336 1337    def extract_function(1338        self,1339        src: str,1340        dst: str,1341        *,1342        name: str,1343        signature: str,1344        body: str,1345        body_indent: int,1346        call: str,1347        return_text: str | None = None,1348        before: str | None = None,1349        into_class: str | None = None,1350    ) -> "Repro":1351        """Extract an inline block into a new ``name`` function. The block ``body`` is cut from1352        ``src`` *verbatim* (so the byte diff certifies the function body is exactly the source's),1353        re-indented from ``body_indent`` to a function-body indent, and wrapped under the1354        authored ``signature`` (with ``return_text`` appended when given); the def is inserted1355        into ``dst`` (above the sibling ``before`` or at the end of ``into_class`` / module), and1356        the block in ``src`` is replaced by the authored ``call``.1357 1358        This is the certifiable core of an extract-function: the bulk (the relocated body) is1359        machine-checked, and only the small signature/return/call interface is authored. It is1360        faithful **only** when the body is moved unchanged -- a de-self (``self.x`` -> a1361        parameter), a control-flow restructure, or a bookkeeping consolidation must be done as a1362        separate semantic commit first, since those are not relocations (see1363        guide-split.md)."""1364 1365        def reindent(text: str, shift: int) -> str:1366            if shift == 0:1367                return text1368            interior = _multiline_string_interior_lines(dedent(text, body_indent))1369            lines = _split_keepends(text)1370            if shift < 0:1371                return "".join(1372                    (1373                        line[-shift:]1374                        if index + 1 not in interior and line[:-shift] == " " * -shift1375                        else line1376                    )1377                    for index, line in enumerate(lines)1378                )1379            pad = " " * shift1380            return "".join(1381                pad + line if line.strip() and index + 1 not in interior else line1382                for index, line in enumerate(lines)1383            )1384 1385        def op(root: Path) -> None:1386            src_path = root / src1387            src_text = _read_source(src_path)1388            assert src_text.count(body) == 1, f"block not found uniquely in {src}"1389            at = src_text.find(body)1390            assert at == 0 or src_text[at - 1] == "\n", (1391                f"block matches mid-line in {src}; it must start at a line boundary"1392            )1393            _write_source(src_path, src_text.replace(body, call, 1))1394 1395            dst_path = root / dst1396            dst_lines = _split_keepends(_read_source(dst_path))1397            dst_nl = _newline_style("".join(dst_lines))1398            sig_first = _split_keepends(signature)[0]1399            sig_indent = len(sig_first) - len(sig_first.lstrip(" "))1400            function = (1401                signature.rstrip("\r\n")1402                + dst_nl1403                + reindent(body, sig_indent + 4 - body_indent)1404            )1405            if return_text is not None:1406                function = function.rstrip("\r\n") + dst_nl + return_text1407            function = function.rstrip("\r\n") + dst_nl1408 1409            dst_tree = ast.parse("".join(dst_lines))1410            container = dst_tree.body1411            if into_class is not None:1412                cls = _find_class(dst_tree, into_class)1413                assert cls is not None, f"class {into_class} not found in {dst}"1414                container = cls.body1415            anchor = None1416            if before is not None:1417                anchor = next(1418                    (1419                        node1420                        for node in container1421                        if isinstance(1422                            node,1423                            (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef),1424                        )1425                        and node.name == before1426                    ),1427                    None,1428                )1429            if anchor is not None:1430                at = _def_span(anchor)[0] - 11431                _write_source(1432                    dst_path,1433                    "".join(dst_lines[:at] + [function, dst_nl] + dst_lines[at:]),1434                )1435            else:1436                at = (1437                    container[-1].end_lineno1438                    if into_class is not None1439                    else len(dst_lines)1440                )1441                _write_source(1442                    dst_path,1443                    "".join(dst_lines[:at] + [dst_nl, function] + dst_lines[at:]),1444                )1445 1446        self.ops.append(op)1447        return self1448 1449    def delete_file(self, path: str) -> "Repro":1450        """Delete a source module that its symbols' relocation left empty (the chain deletes1451        the leftover scaffolding-only file). Run after the moves that empty it. Refuses a1452        file that still holds anything beyond a docstring, imports, a TYPE_CHECKING block, or1453        a bare module ``logger`` -- deleting live code is not a relocation."""1454 1455        def is_module_logger(stmt: ast.stmt) -> bool:1456            return (1457                isinstance(stmt, ast.Assign)1458                and stmt.value is not None1459                and ast.unparse(stmt.value) == "logging.getLogger(__name__)"1460            )1461 1462        def op(root: Path) -> None:1463            target = root / path1464            if not target.exists():1465                return1466            leftover = [1467                ast.unparse(stmt)1468                for stmt in ast.parse(_read_source(target)).body1469                if not (1470                    isinstance(stmt, (ast.Import, ast.ImportFrom))1471                    or (1472                        isinstance(stmt, ast.Expr)1473                        and isinstance(stmt.value, ast.Constant)1474                        and isinstance(stmt.value.value, str)1475                    )1476                    or (1477                        isinstance(stmt, ast.If)1478                        and ast.unparse(stmt.test)1479                        in ("TYPE_CHECKING", "typing.TYPE_CHECKING")1480                    )1481                    or is_module_logger(stmt)1482                )1483            ]1484            assert not leftover, (1485                f"{path} still holds non-scaffolding code, refusing to delete: "1486                f"{leftover[:3]}"1487            )1488            target.unlink()1489 1490        self.ops.append(op)1491        return self1492 1493    def run(self) -> str:1494        """Apply the operations to a worktree at base, run pre-commit, diff against target.1495        Returns the residual diff ("" on a clean reproduction)."""1496        repo_root = self.repo_root or exec_command("git rev-parse --show-toplevel")1497        worktree = tempfile.mkdtemp(prefix="repro-")1498        branch = Path(worktree).name1499        try:1500            exec_command(1501                f"git worktree add -b {branch} {worktree} {self.base}", cwd=repo_root1502            )1503            for op in self.ops:1504                op(Path(worktree))1505            exec_command("git add -A", cwd=worktree)1506            changed = exec_command(1507                f"git diff --cached --name-only --diff-filter=ACMR {self.base}",1508                cwd=worktree,1509            ).split()1510            if changed:1511                files = " ".join(shlex.quote(path) for path in changed)1512                exec_command(1513                    f"pre-commit run --files {files}", cwd=worktree, check=False1514                )1515            if exec_command("git status --porcelain", cwd=worktree):1516                git_add_and_commit("repro", cwd=worktree)1517            diff = exec_command(1518                f"git diff {self.target} -- .", cwd=worktree, check=False1519            )1520            if diff:1521                print(f"\nRESIDUAL ({len(diff.splitlines())} lines):\n{diff}")1522            else:1523                print("\nPASS: reproduces the commit byte-for-byte.")1524            return diff1525        finally:1526            exec_command(1527                f"git worktree remove --force {worktree}", cwd=repo_root, check=False1528            )1529            exec_command(f"git branch -D {branch}", cwd=repo_root, check=False)1530