From 7054995f66179d192cb2a759990f145f3c7f0935 Mon Sep 17 00:00:00 2001
From: Vichitra Attri <mevichitra@gmail.com>
Date: Wed, 14 Oct 2026 12:44:08 +0000
Subject: [PATCH] feat(research): add controlled transformation experiment for
 finding identity

Real histories rarely edit the flagged line and introduce few new findings, so
they cannot show which kinds of change break which identity schemes (RQ2).

transforms.py rewrites one Python file around one finding and returns the new
text plus an exact old-line to new-line map (ground truth by construction):
reformat, insert_top, insert_below, rename_local, rename_function,
move_function, change_literal, extract_helper, duplicate_above and
duplicate_below, using libcst for source-preserving renames. An optional
ai_refactor asks a language model (model and key from flags or environment)
to refactor the enclosing function, caches replies on disk, uses the line diff
as ground truth and flags every case for manual review.

The new 'transform' subcommand samples files with findings, applies every
transformation, scans before and after, and writes the invariance matrix
(transformation x scheme, false-new and hidden rates) as CSV and Markdown,
with an optional matplotlib heatmap. Extra dependencies live in
research/finding_identity/requirements.txt.

Signed-off-by: Vichitra Attri <mevichitra@gmail.com>
---
 research/finding_identity/.gitignore          |   1 +
 research/finding_identity/README.md           | 100 ++
 research/finding_identity/cli.py              |   6 +
 research/finding_identity/experiment.py       | 457 +++++++++
 research/finding_identity/requirements.txt    |   6 +
 .../finding_identity/tests/test_experiment.py | 151 +++
 .../finding_identity/tests/test_transforms.py | 260 +++++
 research/finding_identity/transform_cli.py    | 104 ++
 research/finding_identity/transforms.py       | 922 ++++++++++++++++++
 9 files changed, 2007 insertions(+)
 create mode 100644 research/finding_identity/experiment.py
 create mode 100644 research/finding_identity/requirements.txt
 create mode 100644 research/finding_identity/tests/test_experiment.py
 create mode 100644 research/finding_identity/tests/test_transforms.py
 create mode 100644 research/finding_identity/transform_cli.py
 create mode 100644 research/finding_identity/transforms.py

diff --git a/research/finding_identity/.gitignore b/research/finding_identity/.gitignore
index 351a656..ec761da 100644
--- a/research/finding_identity/.gitignore
+++ b/research/finding_identity/.gitignore
@@ -1,2 +1,3 @@
 .repos/
 results/
+.cache/
diff --git a/research/finding_identity/README.md b/research/finding_identity/README.md
index 6c915c5..89c86b1 100644
--- a/research/finding_identity/README.md
+++ b/research/finding_identity/README.md
@@ -81,6 +81,104 @@ Tests:
 python -m pytest research/finding_identity/tests
 ```
 
+## Controlled transformation experiment (RQ2)
+
+Real histories rarely edit the flagged line itself and introduce few new findings, so
+they cannot say *which kinds of change break which schemes*. The `transform` subcommand
+answers that with a controlled experiment: it samples Python files with findings,
+rewrites each one with a set of transformations whose effect on line identity is known
+by construction, scans before and after, and scores every scheme.
+
+| Transformation | What it does | Behaviour |
+|---|---|---|
+| `reformat` | `black` at line length 79 (`--reformat-line-length`; `ruff format` if black is missing) | kept |
+| `insert_top` | a comment and an unused import below the docstring and `__future__` imports | kept |
+| `insert_below` | a comment line right below the flagged statement (GitHub's hash reads 100 non-whitespace characters forward) | kept |
+| `rename_local` | renames a local variable (not a parameter) used on the flagged line, in its whole function | kept |
+| `rename_function` | renames the enclosing function or method and its references in the file (`self.m`, `cls.m`, `Class.m` for methods) | kept in the file |
+| `move_function` | swaps the enclosing function with the adjacent function in the same body | kept |
+| `change_literal` | appends `_v2` to a string literal on the flagged line, or adds one to a number | changed |
+| `extract_helper` | moves the flagged statement into a new module-level function; reads become parameters, assignments are returned | kept |
+| `duplicate_above` / `duplicate_below` | copies the flagged statement directly above / below itself: a genuinely new finding | changed |
+| `ai_refactor` | asks a language model to refactor the enclosing function without changing its behaviour (optional) | checked by hand |
+
+Each transformation returns the new text and an exact old-line to new-line map, which
+is the ground truth. `reformat` pairs the syntax trees before and after (black keeps the
+tree), so when it splits a line the finding is followed to the line its node starts on.
+`ai_refactor` is the exception: its ground truth is the line diff of `linemap.py`, so
+its cases are flagged `needs_review` and written to `review.jsonl` with the old and new
+function for a manual check.
+
+Renames, extraction and moves are rewritten with `libcst` (source preserving) and only
+where they are safe within the file: names bound once, no `global`/`nonlocal`/`locals()`,
+no special methods, no overridden method names, no statements that `yield`, `await` or
+use `super()`. When a transformation does not fit a target finding it tries the next
+finding in the file (in a seeded random order); if none fits, the file counts as "n/a"
+for that transformation, with the reason. Other modules are not updated by renames.
+
+### Running it
+
+```bash
+pip install -r research/finding_identity/requirements.txt
+
+python -m research.finding_identity transform https://github.com/pallets/flask --samples 200 --heatmap
+python -m research.finding_identity transform https://github.com/DefectDojo/django-DefectDojo \
+    --ref dev --label defectdojo --samples 200 --heatmap
+
+# A subset of transformations, on a local directory
+python -m research.finding_identity transform path/to/src --transforms rename_local,extract_helper
+```
+
+`--samples` is the number of files with findings to use; `--seed` fixes the sample and
+the target findings. Results go to `results/transform-<label>/` (git-ignored):
+
+- `matrix.csv`: one row per transformation and scheme (the invariance matrix);
+- `matrix.md`: the same as a Markdown table (false-new % / hidden %);
+- `matrix.png`: heatmaps of the false-new and hidden rates (`--heatmap`, needs matplotlib);
+- `cases.jsonl`: one line per applied transformation, with the target, its outcome under
+  each scheme and the introduced findings;
+- `summary.json`: all counts, including "n/a" reasons and findings lost by the change.
+
+### Invariance matrix
+
+Two scopes are reported. The **target** scope is the paper's matrix: per transformation,
+
+- **false-new rate** = target findings a scheme reports as new / target findings still
+  detected after the change (`target_known`);
+- **hidden rate** = introduced findings a scheme matches to an old finding / introduced
+  findings (only the `duplicate` transformations introduce findings);
+- **mispair rate** = target findings matched to the wrong old finding / `target_known`;
+- **collision rate** = findings sharing a key with another finding in the transformed file.
+
+The **file** scope (`file_*` columns) scores every finding in the transformed file, as
+a CI baseline would see the change.
+
+A target finding that the change makes undetectable is counted as `target_lost` (with
+its rule in `summary.json`) and the next finding in the file is tried; lost cases are
+not scheme failures and do not enter the matrix. `extract_helper` loses most loop rules
+this way (`CG-PERF-002`, `CG-PERF-005`): outside its loop the statement is no longer a
+finding.
+
+The duplicate transformations show how matching order matters: schemes pair equal keys
+in line order, so a copy *above* the finding is matched to the old finding (hidden) and
+the original is reported as new; GitHub's hash, which reads forward, is fooled by a copy
+*below* instead.
+
+### Optional AI refactor
+
+`ai_refactor` uses the Anthropic API (`pip install anthropic`). No model is hard-coded:
+
+```bash
+export ANTHROPIC_API_KEY=...                 # or --ai-api-key
+export FINDING_IDENTITY_AI_MODEL=<model id>  # or --ai-model
+python -m research.finding_identity transform <repo> --transforms all --samples 50
+```
+
+`--transforms all` includes `ai_refactor` only when a model is set. Every reply is
+cached in `.cache/ai/` (git-ignored) under a hash of the model and prompt, so a rerun
+with the same cache reproduces the edits; `--ai-offline` uses the cache only. Replies
+that do not parse or that rename the function count as "n/a".
+
 ## Known limitations
 
 - The ground truth is a line diff. A finding whose anchor line is rewritten beyond the
@@ -89,3 +187,5 @@ python -m pytest research/finding_identity/tests
   form feeds and other rare separators, so in files containing them the `codeguard` key can
   hash a neighbouring line. The harness measures the tool as it is.
 - Taint findings depend on several lines; the ground truth tracks only the reported line.
+- The transformations are applied one at a time to one finding each; real commits mix
+  them. `change_literal` and the `duplicate` transformations change behaviour on purpose.
diff --git a/research/finding_identity/cli.py b/research/finding_identity/cli.py
index ce0f51c..3e67808 100644
--- a/research/finding_identity/cli.py
+++ b/research/finding_identity/cli.py
@@ -5,6 +5,7 @@
 import argparse
 from pathlib import Path
 
+from . import transform_cli
 from .gitrepo import GitRepo
 from .replay import replay
 from .scan import Scanner, rule_ids_for
@@ -36,7 +37,12 @@ def main(argv: list[str] | None = None) -> None:
     sp.add_argument("results", nargs="+", type=Path)
     sp.add_argument("--csv", type=Path, help="also write the rows to this CSV file")
 
+    transform_cli.add_parser(sub)
+
     args = parser.parse_args(argv)
+    if args.command == "transform":
+        transform_cli.run(args)
+        return
     if args.command == "summarize":
         summarize(args.results, args.csv)
         return
diff --git a/research/finding_identity/experiment.py b/research/finding_identity/experiment.py
new file mode 100644
index 0000000..2f5f51e
--- /dev/null
+++ b/research/finding_identity/experiment.py
@@ -0,0 +1,457 @@
+"""Controlled transformation experiment (RQ2): which kinds of change break which schemes?
+
+Python files with findings are sampled from a repository. For each file and each
+transformation (see :mod:`transforms`), a target finding is picked, the file is
+rewritten around it, both versions are scanned, and every scheme is scored against the
+line map the transformation returns. The result is an *invariance matrix*:
+transformation x scheme, with false-new and hidden rates.
+
+Two scopes are reported:
+
+- ``target``: the finding the transformation was aimed at (did it keep its identity?)
+  plus any finding the transformation introduced (was it hidden?);
+- ``file``: every finding in the transformed file, as a CI baseline would see it.
+"""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import json
+import random
+import sys
+import time
+from collections import Counter, defaultdict
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import codeguard
+
+from .matching import OUTCOMES, Observation, collisions, evaluate
+from .scan import Scanner, is_candidate
+from .schemes import split_lines
+from .transforms import Edit, NotApplicable, Source, Transform
+
+GT_KEYS = ("old", "new", "known", "known_equal", "known_edited", "introduced", "removed")
+#: Target findings tried per file and transformation before giving up ("n/a").
+MAX_TRIES = 50
+_SKIP_DIRS = {".git", ".hg", ".tox", ".venv", "venv", "node_modules", "__pycache__"}
+
+
+@dataclass
+class Target:
+    rule: str
+    line: int
+
+
+def _empty(schemes: Sequence[str]) -> dict[str, Any]:
+    return {
+        "cases": 0,
+        "not_applicable": Counter(),
+        "broken": 0,
+        "target_lost": 0,
+        "target_lost_rules": Counter(),
+        "target_known": 0,
+        "needs_review": 0,
+        "gt": dict.fromkeys(GT_KEYS, 0),
+        "file": {s: dict.fromkeys(OUTCOMES, 0) for s in schemes},
+        "target": {s: dict.fromkeys(OUTCOMES, 0) for s in schemes},
+        "colliding": dict.fromkeys(schemes, 0),
+        "observations": 0,
+    }
+
+
+def python_files(root: Path) -> list[str]:
+    """Repository-relative paths of the Python files worth scanning, sorted."""
+    out = []
+    for path in root.rglob("*.py"):
+        rel = path.relative_to(root)
+        if _SKIP_DIRS.intersection(rel.parts[:-1]) or not path.is_file():
+            continue
+        if is_candidate(rel.as_posix()):
+            out.append(rel.as_posix())
+    return sorted(out)
+
+
+def ground_truth_map(
+    edit: Edit, old_text: str, new_text: str, old: Sequence[Observation], new: Sequence[Observation]
+) -> dict[int, tuple[int, str]]:
+    """The edit's line map in the form :func:`matching.evaluate` takes.
+
+    Where reformatting split an old line, the old line maps to the candidate new line
+    that carries a finding of the same rule. A line is "equal" when its text is
+    unchanged, else "edited".
+    """
+    old_lines, new_lines = split_lines(old_text), split_lines(new_text)
+    new_rules: dict[int, set[str]] = defaultdict(set)
+    for n in new:
+        new_rules[n.line].add(n.rule)
+    old_rules: dict[int, set[str]] = defaultdict(set)
+    for o in old:
+        old_rules[o.line].add(o.rule)
+    out: dict[int, tuple[int, str]] = {}
+    for old_line, new_line in edit.line_map.items():
+        for candidate in edit.alternatives.get(old_line, ()):
+            if old_rules[old_line] & new_rules[candidate]:
+                new_line = candidate
+                break
+        same = old_lines[old_line - 1] == new_lines[new_line - 1]
+        out[old_line] = (new_line, "equal" if same else "edited")
+    return out
+
+
+class Experiment:
+    def __init__(
+        self,
+        root: Path,
+        transforms: dict[str, Transform],
+        scanner: Scanner,
+        schemes: Sequence[str],
+        *,
+        seed: int = 0,
+    ) -> None:
+        self.root = root
+        self.transforms = transforms
+        self.scanner = scanner
+        self.schemes = tuple(schemes)
+        self.seed = seed
+        self.totals = {name: _empty(schemes) for name in transforms}
+        self.cases: list[dict[str, Any]] = []
+        self.reviews: list[dict[str, Any]] = []
+        self.files_with_findings = 0
+
+    def _scan(self, path: str, text: str) -> list[Observation] | None:
+        data = text.encode("utf-8")
+        result = self.scanner.scan(hashlib.sha256(data).hexdigest(), path, data)
+        return None if result is None else result.observations
+
+    def run(self, samples: int, *, progress: bool = True) -> None:
+        files = python_files(self.root)
+        random.Random(self.seed).shuffle(files)  # noqa: S311  (reproducible sampling)
+        started = time.monotonic()
+        for path in files:
+            if self.files_with_findings >= samples:
+                break
+            try:
+                text = (self.root / path).read_bytes().decode("utf-8")
+                src = Source(text)
+            except (UnicodeDecodeError, SyntaxError, ValueError, RecursionError):
+                continue
+            old = self._scan(path, text)
+            if not old:
+                continue
+            self.files_with_findings += 1
+            self.run_file(path, src, old)
+            if progress and self.files_with_findings % 10 == 0:
+                print(
+                    f"  {self.files_with_findings}/{samples} files, "
+                    f"{time.monotonic() - started:.0f}s",
+                    file=sys.stderr,
+                )
+
+    def run_file(self, path: str, src: Source, old: list[Observation]) -> None:
+        rng = random.Random(f"{self.seed}:{path}")  # noqa: S311
+        order = sorted({(o.rule, o.line) for o in old})
+        rng.shuffle(order)
+        for name, transform in self.transforms.items():
+            totals = self.totals[name]
+            reasons: Counter[str] = Counter()
+            for rule, line in order[:MAX_TRIES]:
+                try:
+                    edit = transform(src, line)
+                except NotApplicable as exc:
+                    reasons[str(exc)] += 1
+                    continue
+                status = self._score(name, path, src, old, Target(rule, line), edit)
+                if status == "scored":
+                    break
+                reasons[status] += 1  # target lost or output broken: try the next finding
+            else:
+                # Record why the transformation fit no finding in this file (one reason).
+                totals["not_applicable"][reasons.most_common(1)[0][0] if reasons else "?"] += 1
+
+    def _score(
+        self,
+        name: str,
+        path: str,
+        src: Source,
+        old: list[Observation],
+        target: Target,
+        edit: Edit,
+    ) -> str:
+        """Score one applied transformation; returns the case status.
+
+        Only a case whose target finding is still detected afterwards ("scored") enters
+        the matrix. A target the change makes undetectable ("target_lost", e.g. a loop
+        rule after the statement leaves the loop) is counted and logged, not scored.
+        """
+        totals = self.totals[name]
+        case: dict[str, Any] = {"path": path, "transform": name, "rule": target.rule}
+        case["line"] = target.line
+        try:
+            Source(edit.text)
+        except (SyntaxError, ValueError):
+            totals["broken"] += 1
+            self.cases.append({**case, "status": "broken"})
+            return "broken"
+        new = self._scan(path, edit.text)
+        if new is None:
+            totals["broken"] += 1
+            self.cases.append({**case, "status": "unscannable"})
+            return "broken"
+        line_map = ground_truth_map(edit, src.text, edit.text, old, new)
+        result = evaluate(old, new, line_map, self.schemes, keep_records=True)
+        key = ("known", target.rule, target.line)
+        tracked = next(
+            (r for r in result.records if (r["truth"], r["rule"], r["old_line"]) == key), None
+        )
+        case["new_line"] = line_map.get(target.line, (None,))[0]
+        if tracked is None:
+            totals["target_lost"] += 1
+            totals["target_lost_rules"][target.rule] += 1
+            self.cases.append({**case, "status": "target_lost"})
+            return "target lost"
+        totals["cases"] += 1
+        totals["needs_review"] += int(edit.needs_review)
+        for k, v in result.gt.items():
+            totals["gt"][k] += v
+        for scheme, counts in result.by_scheme.items():
+            for k in OUTCOMES:
+                totals["file"][scheme][k] += counts[k]
+        for scheme, n in collisions(new, self.schemes).items():
+            totals["colliding"][scheme] += n
+        totals["observations"] += len(new)
+
+        for rec in result.records:
+            if rec["truth"] == "new":
+                for scheme, outcome in rec["outcomes"].items():  # type: ignore[attr-defined]
+                    totals["target"][scheme][outcome] += 1
+        totals["target_known"] += 1
+        for scheme, outcome in tracked["outcomes"].items():  # type: ignore[attr-defined]
+            totals["target"][scheme][outcome] += 1
+        case.update(
+            status="scored",
+            gt=result.gt,
+            target_outcomes=tracked["outcomes"],
+            introduced=[r for r in result.records if r["truth"] == "new"],
+            needs_review=edit.needs_review,
+        )
+        self.cases.append(case)
+        if edit.review:
+            self.reviews.append({**case, **edit.review})
+        return "scored"
+
+
+# ---------------------------------------------------------------------------
+# Rates and output
+# ---------------------------------------------------------------------------
+
+
+def _rate(num: int, den: int) -> float | None:
+    return num / den if den else None
+
+
+def matrix_rows(totals: dict[str, dict[str, Any]], schemes: Sequence[str]) -> list[dict[str, Any]]:
+    """One row per transformation x scheme."""
+    rows = []
+    for name, t in totals.items():
+        gt = t["gt"]
+        for scheme in schemes:
+            tc, fc = t["target"][scheme], t["file"][scheme]
+            introduced = tc["false_known"] + tc["new_ok"]
+            rows.append(
+                {
+                    "transform": name,
+                    "scheme": scheme,
+                    "cases": t["cases"],
+                    "target_known": t["target_known"],
+                    "target_lost": t["target_lost"],
+                    "false_new": tc["false_new"],
+                    "false_new_rate": _rate(tc["false_new"], t["target_known"]),
+                    "mispaired": tc["mispaired"],
+                    "mispair_rate": _rate(tc["mispaired"], t["target_known"]),
+                    "introduced": introduced,
+                    "hidden": tc["false_known"],
+                    "hidden_rate": _rate(tc["false_known"], introduced),
+                    "file_known": gt["known"],
+                    "file_false_new_rate": _rate(fc["false_new"], gt["known"]),
+                    "file_mispair_rate": _rate(fc["mispaired"], gt["known"]),
+                    "collision_rate": _rate(t["colliding"][scheme], t["observations"]),
+                }
+            )
+    return rows
+
+
+def _pct(x: float | None) -> str:
+    return "  n/a" if x is None else f"{100 * x:5.1f}"
+
+
+def print_matrix(rows: Sequence[dict[str, Any]], schemes: Sequence[str]) -> None:
+    by = {(r["transform"], r["scheme"]): r for r in rows}
+    names = list(dict.fromkeys(r["transform"] for r in rows))
+    for title, key in (
+        ("false-new % (target finding reported as new)", "false_new_rate"),
+        ("hidden % (introduced finding matched to an old one)", "hidden_rate"),
+    ):
+        print(f"\n{title}")
+        head = f"{'transform':<16} {'n':>4} " + " ".join(f"{s:>11}" for s in schemes)
+        print(head)
+        print("-" * len(head))
+        for name in names:
+            n = by[(name, schemes[0])]["target_known" if key == "false_new_rate" else "introduced"]
+            cells = " ".join(f"{_pct(by[(name, s)][key]):>11}" for s in schemes)
+            print(f"{name:<16} {n:>4} {cells}")
+
+
+def write_csv(rows: Sequence[dict[str, Any]], path: Path) -> None:
+    with path.open("w", newline="", encoding="utf-8") as fh:
+        writer = csv.DictWriter(fh, fieldnames=list(rows[0]))
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def markdown_matrix(rows: Sequence[dict[str, Any]], schemes: Sequence[str]) -> str:
+    """The matrix as a Markdown table: false-new % / hidden % per cell."""
+    by = {(r["transform"], r["scheme"]): r for r in rows}
+    names = list(dict.fromkeys(r["transform"] for r in rows))
+    out = ["| transform | n | " + " | ".join(schemes) + " |"]
+    out.append("|---|---:|" + "---:|" * len(schemes))
+    for name in names:
+        first = by[(name, schemes[0])]
+        cells = []
+        for s in schemes:
+            r = by[(name, s)]
+            cell = _pct(r["false_new_rate"]).strip()
+            if r["introduced"]:
+                cell += f" / {_pct(r['hidden_rate']).strip()}"
+            cells.append(cell)
+        out.append(f"| {name} | {first['target_known']} | " + " | ".join(cells) + " |")
+    return "\n".join(out) + "\n"
+
+
+#: One-hue light-to-dark ramp for rates (0 = lightest), plus text and n/a colours.
+SEQUENTIAL_BLUE = ("#cde2fb", "#86b6ef", "#3987e5", "#1c5cab", "#0d366b")
+NA_FILL, INK, INK_MUTED = "#f0efec", "#1f2328", "#6e7781"
+
+
+def heatmap(rows: Sequence[dict[str, Any]], schemes: Sequence[str], path: Path) -> bool:
+    """Draw the matrix as two heatmaps (false-new and hidden). False without matplotlib."""
+    try:
+        import matplotlib
+
+        matplotlib.use("Agg")
+        import matplotlib.pyplot as plt
+        import numpy as np
+        from matplotlib.colors import LinearSegmentedColormap
+    except ImportError:
+        return False
+    cmap = LinearSegmentedColormap.from_list("rate", SEQUENTIAL_BLUE)
+    cmap.set_bad(NA_FILL)
+    by = {(r["transform"], r["scheme"]): r for r in rows}
+    names = list(dict.fromkeys(r["transform"] for r in rows))
+    hidden_names = [n for n in names if by[(n, schemes[0])]["introduced"]]
+    panels = [("False-new rate (target finding)", "false_new_rate", "target_known", names)]
+    if hidden_names:
+        panels.append(
+            ("Hidden rate (introduced finding)", "hidden_rate", "introduced", hidden_names)
+        )
+    heights = [len(p[3]) for p in panels]
+    fig, axes = plt.subplots(
+        len(panels),
+        1,
+        figsize=(1.1 * len(schemes) + 2.6, 0.42 * sum(heights) + 1.6 * len(panels)),
+        gridspec_kw={"height_ratios": heights},
+        squeeze=False,
+    )
+    for ax, (title, key, count, ylabels) in zip(axes[:, 0], panels, strict=True):
+        values = np.array(
+            [
+                [np.nan if by[(n, s)][key] is None else by[(n, s)][key] for s in schemes]
+                for n in ylabels
+            ],
+            dtype=float,
+        )
+        image = ax.imshow(np.ma.masked_invalid(values), cmap=cmap, vmin=0, vmax=1, aspect="auto")
+        ax.set_xticks(range(len(schemes)), schemes, rotation=30, ha="right")
+        ax.set_yticks(
+            range(len(ylabels)), [f"{n} (n={by[(n, schemes[0])][count]})" for n in ylabels]
+        )
+        ax.set_title(title, loc="left", fontsize=11)
+        ax.tick_params(length=0)
+        for spine in ax.spines.values():
+            spine.set_visible(False)
+        for i in range(len(ylabels)):
+            for j in range(len(schemes)):
+                v = values[i, j]
+                label = "n/a" if np.isnan(v) else f"{100 * v:.0f}"
+                dark = not np.isnan(v) and v > 0.5
+                ink = "white" if dark else (INK_MUTED if np.isnan(v) else INK)
+                ax.text(j, i, label, ha="center", va="center", fontsize=8, color=ink)
+        fig.colorbar(image, ax=ax, fraction=0.025, pad=0.02, format=lambda x, _: f"{x:.0%}")
+    fig.tight_layout()
+    fig.savefig(path, dpi=160)
+    plt.close(fig)
+    return True
+
+
+def run_experiment(
+    root: Path,
+    *,
+    label: str,
+    transforms: dict[str, Transform],
+    scanner: Scanner,
+    schemes: Sequence[str],
+    samples: int,
+    seed: int,
+    out_dir: Path,
+    meta: dict[str, Any] | None = None,
+    draw_heatmap: bool = False,
+) -> dict[str, Any]:
+    out_dir.mkdir(parents=True, exist_ok=True)
+    started = time.monotonic()
+    exp = Experiment(root, transforms, scanner, schemes, seed=seed)
+    exp.run(samples)
+    rows = matrix_rows(exp.totals, schemes)
+    summary = {
+        "label": label,
+        "root": str(root),
+        "samples": samples,
+        "seed": seed,
+        "files_with_findings": exp.files_with_findings,
+        "transforms": list(transforms),
+        "schemes": list(schemes),
+        "codeguard_version": codeguard.__version__,
+        "created": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "elapsed_seconds": round(time.monotonic() - started, 1),
+        **(meta or {}),
+        "totals": exp.totals,
+    }
+    (out_dir / "summary.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
+    with (out_dir / "cases.jsonl").open("w", encoding="utf-8") as fh:
+        for case in exp.cases:
+            fh.write(json.dumps(case) + "\n")
+    if exp.reviews:
+        with (out_dir / "review.jsonl").open("w", encoding="utf-8") as fh:
+            for case in exp.reviews:
+                fh.write(json.dumps(case) + "\n")
+    if rows:
+        write_csv(rows, out_dir / "matrix.csv")
+        (out_dir / "matrix.md").write_text(markdown_matrix(rows, schemes), encoding="utf-8")
+    print(f"\n{label}: {exp.files_with_findings} files with findings")
+    print(f"{'transform':<16} {'cases':>5} {'lost':>5} {'n/a':>5} {'broken':>6}  top n/a reason")
+    for name, t in exp.totals.items():
+        na = sum(t["not_applicable"].values())
+        reason = t["not_applicable"].most_common(1)[0][0] if na else ""
+        counts = f"{t['cases']:>5} {t['target_lost']:>5} {na:>5} {t['broken']:>6}"
+        print(f"{name:<16} {counts}  {reason}")
+    if rows:
+        print_matrix(rows, schemes)
+    if draw_heatmap and rows:
+        if heatmap(rows, schemes, out_dir / "matrix.png"):
+            print(f"\nheatmap: {out_dir / 'matrix.png'}")
+        else:
+            print("\nheatmap skipped: matplotlib is not installed", file=sys.stderr)
+    return summary
diff --git a/research/finding_identity/requirements.txt b/research/finding_identity/requirements.txt
new file mode 100644
index 0000000..08b563a
--- /dev/null
+++ b/research/finding_identity/requirements.txt
@@ -0,0 +1,6 @@
+# Extra dependencies of the controlled transformation experiment (`transform` subcommand).
+# CodeGuard itself is installed from the repository root: pip install -e .
+libcst>=1.1
+black>=24.1        # reformat; falls back to `ruff format` when absent
+matplotlib>=3.8    # optional: --heatmap
+anthropic>=0.40    # optional: ai_refactor
diff --git a/research/finding_identity/tests/test_experiment.py b/research/finding_identity/tests/test_experiment.py
new file mode 100644
index 0000000..fa09f23
--- /dev/null
+++ b/research/finding_identity/tests/test_experiment.py
@@ -0,0 +1,151 @@
+"""End to end: run the transformation experiment on a small fixture directory."""
+
+from __future__ import annotations
+
+import csv
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+pytest.importorskip("libcst")
+
+from research.finding_identity.cli import main
+from research.finding_identity.experiment import ground_truth_map, run_experiment
+from research.finding_identity.matching import Observation
+from research.finding_identity.scan import Scanner
+from research.finding_identity.schemes import ALL_SCHEMES
+from research.finding_identity.transforms import Edit, deterministic_transforms
+
+HANDLERS = '''\
+"""Request handlers."""
+import subprocess
+
+
+def parse(expr):
+    value = eval(expr)
+    return value
+
+
+def run(cmd):
+    code = subprocess.call(cmd, shell=True)
+    return code
+'''
+MODELS = """\
+class Model:
+    def load(self, blob):
+        data = eval(blob)
+        return data
+
+
+def unused():
+    return 1
+"""
+NO_FINDINGS = "def ok():\n    return 1\n"
+
+
+def _fixture(root: Path) -> None:
+    (root / "pkg").mkdir(parents=True)
+    (root / "pkg" / "handlers.py").write_text(HANDLERS)
+    (root / "pkg" / "models.py").write_text(MODELS)
+    (root / "pkg" / "clean.py").write_text(NO_FINDINGS)
+    (root / "pkg" / "broken.py").write_text("def broken(:\n")
+
+
+def _rows(out: Path) -> dict[tuple[str, str], dict[str, Any]]:
+    with (out / "matrix.csv").open() as fh:
+        return {(r["transform"], r["scheme"]): r for r in csv.DictReader(fh)}
+
+
+def test_invariance_matrix(tmp_path: Path) -> None:
+    _fixture(tmp_path / "src")
+    out = tmp_path / "out"
+    summary = run_experiment(
+        tmp_path / "src",
+        label="fixture",
+        transforms=deterministic_transforms(),
+        scanner=Scanner(["CG-SEC-003", "CG-SEC-005"], ALL_SCHEMES),
+        schemes=ALL_SCHEMES,
+        samples=10,
+        seed=1,
+        out_dir=out,
+    )
+    assert summary["files_with_findings"] == 2
+    rows = _rows(out)
+
+    def rate(transform: str, scheme: str, key: str = "false_new_rate") -> float:
+        return float(rows[(transform, scheme)][key])
+
+    # Lines inserted at the top move every finding: only the line number breaks.
+    assert rate("insert_top", "line") == 1.0
+    assert all(rate("insert_top", s) == 0.0 for s in ALL_SCHEMES if s != "line")
+    # A comment right below the finding changes the text GitHub's hash reads.
+    assert rate("insert_below", "github") == 1.0 and rate("insert_below", "codeguard") == 0.0
+    # Renaming the enclosing function breaks exactly the schemes that include the scope.
+    assert rate("rename_function", "codeguard") == 1.0
+    assert rate("rename_function", "cg_raw") == 1.0
+    assert rate("rename_function", "cg_no_scope") == 0.0
+    assert rate("rename_function", "line_text") == 0.0
+    # Extracting into a helper changes the scope and the position, not the statement.
+    assert rate("extract_helper", "cg_no_scope") == 0.0
+    assert rate("extract_helper", "codeguard") == 1.0
+    # A copy below the finding takes over its GitHub hash: the new finding is hidden.
+    assert rate("duplicate_below", "github", "hidden_rate") == 1.0
+    assert rate("duplicate_below", "codeguard", "hidden_rate") == 0.0
+    # A copy above it is paired first by every key scheme and by the line number.
+    assert rate("duplicate_above", "line", "hidden_rate") == 1.0
+    assert rate("duplicate_above", "codeguard", "hidden_rate") == 1.0
+    assert rate("duplicate_above", "github", "hidden_rate") == 0.0
+    assert float(rows[("duplicate_above", "codeguard")]["collision_rate"]) > 0
+
+    cases = [json.loads(line) for line in (out / "cases.jsonl").read_text().splitlines()]
+    assert {c["transform"] for c in cases} >= {"insert_top", "duplicate_below"}
+    assert all(c["status"] == "scored" for c in cases)
+    assert (out / "summary.json").exists() and (out / "matrix.md").exists()
+
+
+def test_cli_subcommand(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    _fixture(tmp_path / "src")
+    out = tmp_path / "out"
+    argv = ["transform", str(tmp_path / "src"), "--transforms", "insert_top,rename_local"]
+    main([*argv, "--samples", "1", "--categories", "security", "--out", str(out)])
+    assert set(row[0] for row in _rows(out)) == {"insert_top", "rename_local"}
+    assert "false-new %" in capsys.readouterr().out
+    with pytest.raises(SystemExit):
+        main(["transform", str(tmp_path / "src"), "--transforms", "ai_refactor"])
+
+
+def test_ground_truth_follows_the_finding_into_a_split_line() -> None:
+    old = [Observation("R", 1, 1, {})]
+    new = [Observation("R", 3, 5, {})]
+    edit = Edit("a\nb\nc\n", {1: 1}, alternatives={1: (1, 3)})
+    assert ground_truth_map(edit, "a\n", "a\nb\nc\n", old, new) == {1: (3, "edited")}
+
+
+LOOP = """\
+def join(items):
+    out = ""
+    for item in items:
+        out += str(item)
+    return out
+"""
+
+
+def test_a_finding_the_change_makes_undetectable_is_lost_not_scored(tmp_path: Path) -> None:
+    (tmp_path / "src").mkdir()
+    (tmp_path / "src" / "loop.py").write_text(LOOP)
+    summary = run_experiment(
+        tmp_path / "src",
+        label="loop",
+        transforms={"extract_helper": deterministic_transforms()["extract_helper"]},
+        scanner=Scanner(["CG-PERF-002"], ALL_SCHEMES),
+        schemes=ALL_SCHEMES,
+        samples=1,
+        seed=0,
+        out_dir=tmp_path / "out",
+    )
+    totals = summary["totals"]["extract_helper"]
+    # Outside its loop, `out += ...` is no longer a finding: lost, and nothing scored.
+    assert totals["target_lost"] == 1 and totals["target_lost_rules"] == {"CG-PERF-002": 1}
+    assert totals["cases"] == 0 and totals["not_applicable"] == {"target lost": 1}
diff --git a/research/finding_identity/tests/test_transforms.py b/research/finding_identity/tests/test_transforms.py
new file mode 100644
index 0000000..442f120
--- /dev/null
+++ b/research/finding_identity/tests/test_transforms.py
@@ -0,0 +1,260 @@
+"""Each transformation: valid output, exact line map, behaviour kept where it should be."""
+
+from __future__ import annotations
+
+import ast
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+pytest.importorskip("libcst")
+
+from research.finding_identity.schemes import split_lines
+from research.finding_identity.transforms import (
+    BELOW_LINE,
+    TOP_LINES,
+    AIRefactor,
+    NotApplicable,
+    Source,
+    deterministic_transforms,
+    node_line_map,
+    split_keepends,
+)
+
+SAMPLE = '''\
+"""Sample module."""
+from __future__ import annotations
+
+import subprocess
+
+SCALE = 3
+
+
+def helper(x):
+    return x * SCALE
+
+
+def compute(expr, items):
+    total = helper(len(items))
+    result = eval(expr, {"n": total})
+    for item in items:
+        result += item
+    return result
+
+
+class Box:
+    def __init__(self, base):
+        self.base = base
+
+    def get(self, expr):
+        value = eval(expr) + self.base
+        return value
+
+    def use(self):
+        return self.get("1 + 1")
+
+
+def run(cmd):
+    return subprocess.call(cmd, shell=True)
+'''
+LINES = split_lines(SAMPLE)
+
+
+def line_of(marker: str, lines: list[str] = LINES) -> int:
+    return next(i for i, text in enumerate(lines, start=1) if marker in text)
+
+
+COMPUTE = line_of("result = eval(")
+GET = line_of("value = eval(")
+TRANSFORMS = deterministic_transforms()
+#: Transformations that keep the file's behaviour (checked by running it).
+PRESERVING = (
+    "reformat",
+    "insert_top",
+    "insert_below",
+    "rename_local",
+    "rename_function",
+    "move_function",
+    "extract_helper",
+)
+#: Transformations that only move or copy lines (indentation aside).
+MOVING = ("insert_top", "insert_below", "move_function", "extract_helper", "duplicate_above",
+          "duplicate_below")  # fmt: skip
+
+
+def behaviour(text: str) -> tuple[Any, ...]:
+    ns: dict[str, Any] = {}
+    exec(compile(text, "sample.py", "exec"), ns)  # noqa: S102  (trusted test code)
+    compute = ns.get("compute") or ns["compute_renamed"]
+    box = ns["Box"](5)
+    get = getattr(box, "get", None) or box.get_renamed
+    return compute("n * 2", [1, 2]), box.use(), get("2"), ns["helper"](4)
+
+
+def apply(name: str, line: int) -> tuple[str, dict[int, int]]:
+    if name == "reformat":
+        pytest.importorskip("black")
+    edit = TRANSFORMS[name](Source(SAMPLE), line)
+    ast.parse(edit.text)
+    return edit.text, edit.line_map
+
+
+def test_split_keepends_counts_lines_like_the_parser() -> None:
+    text = "a\r\nb\rc\n\x0cd\ne"
+    assert [ln.rstrip("\r\n") for ln in split_keepends(text)] == split_lines(text)
+    assert split_keepends("x\n") == ["x\n"]
+
+
+@pytest.mark.parametrize("line", [COMPUTE, GET])
+@pytest.mark.parametrize("name", list(TRANSFORMS))
+def test_line_map_is_exact(name: str, line: int) -> None:
+    if (name, line) == ("change_literal", GET):
+        pytest.skip("no literal on the line (see test_not_applicable)")
+    text, line_map = apply(name, line)
+    new = split_lines(text)
+    if name != "reformat":
+        assert len(set(line_map.values())) == len(line_map), "two old lines on one new line"
+    if name in MOVING:
+        for old, target in line_map.items():
+            assert LINES[old - 1].strip() == new[target - 1].strip(), (old, target)
+    elif name != "reformat":
+        changed = [old for old, target in line_map.items() if LINES[old - 1] != new[target - 1]]
+        assert changed
+        if name != "rename_function":  # renames the def line, not the flagged line
+            assert line in changed
+
+
+@pytest.mark.parametrize("line", [COMPUTE, GET])
+@pytest.mark.parametrize("name", PRESERVING)
+def test_behaviour_is_preserved(name: str, line: int) -> None:
+    text, _ = apply(name, line)
+    assert text != SAMPLE
+    assert behaviour(text) == behaviour(SAMPLE)
+
+
+def test_insert_top_goes_below_future_imports() -> None:
+    text, line_map = apply("insert_top", COMPUTE)
+    new = split_lines(text)
+    assert new[2:4] == list(TOP_LINES)
+    assert line_map[2] == 2 and line_map[COMPUTE] == COMPUTE + 2
+
+
+def test_insert_below_follows_the_flagged_statement() -> None:
+    text, line_map = apply("insert_below", COMPUTE)
+    new = split_lines(text)
+    assert line_map[COMPUTE] == COMPUTE
+    assert new[COMPUTE] == "    " + BELOW_LINE
+    assert line_map[COMPUTE + 1] == COMPUTE + 2
+
+
+def test_rename_local_renames_every_use_in_the_function() -> None:
+    text, _ = apply("rename_local", COMPUTE)
+    assert text.count("result_v2") == 3
+    assert "value = eval" in text
+
+
+def test_rename_function_updates_method_calls() -> None:
+    text, _ = apply("rename_function", GET)
+    assert "def get_renamed(self, expr):" in text
+    assert 'self.get_renamed("1 + 1")' in text
+
+
+def test_move_function_swaps_with_the_previous_function() -> None:
+    text, line_map = apply("move_function", COMPUTE)
+    new = split_lines(text)
+    assert new.index("def compute(expr, items):") < new.index("def helper(x):")
+    assert new[line_map[COMPUTE] - 1] == LINES[COMPUTE - 1]
+
+
+def test_extract_helper_moves_the_statement_into_a_helper() -> None:
+    text, line_map = apply("extract_helper", GET)
+    new = split_lines(text)
+    target = line_map[GET]
+    assert new[target - 2] == "def _get_extracted(expr, self):"
+    assert new[target - 1] == "    value = eval(expr) + self.base"
+    assert target < line_of("class Box:", new)
+    assert "        value = _get_extracted(expr, self)" in new
+
+
+@pytest.mark.parametrize("name", ["duplicate_above", "duplicate_below"])
+def test_duplicate_adds_an_unmapped_copy(name: str) -> None:
+    text, line_map = apply(name, COMPUTE)
+    new = split_lines(text)
+    copies = [k for k, t in enumerate(new, start=1) if t == LINES[COMPUTE - 1]]
+    assert len(copies) == 2
+    (unmapped,) = set(copies) - set(line_map.values())
+    assert unmapped == (COMPUTE if name == "duplicate_above" else COMPUTE + 1)
+
+
+def test_change_literal_changes_a_literal_on_the_line() -> None:
+    text, line_map = apply("change_literal", COMPUTE)
+    assert split_lines(text)[line_map[COMPUTE] - 1] == '    result = eval(expr, {"n_v2": total})'
+
+
+@pytest.mark.parametrize(
+    ("name", "marker"),
+    [
+        ("rename_local", "return subprocess.call"),  # only a parameter on the line
+        ("rename_function", "self.base = base"),  # __init__
+        ("extract_helper", "SCALE = 3"),  # module level
+        ("duplicate_above", "def compute("),  # compound statement header
+        ("change_literal", "value = eval("),  # no literal
+    ],
+)
+def test_not_applicable(name: str, marker: str) -> None:
+    with pytest.raises(NotApplicable):
+        TRANSFORMS[name](Source(SAMPLE), line_of(marker))
+
+
+def test_reformat_maps_split_lines_to_every_node_start() -> None:
+    pytest.importorskip("black")
+    text = (
+        "x = foo(aaaaaaaaaaaaaaaa, bbbbbbbbbbbbbbbbbbbb, eval(ccccccccccccccccc), ddddddddddddd)\n"
+    )
+    edit = TRANSFORMS["reformat"](Source(text), 1)
+    assert split_lines(edit.text)[3].strip() == "eval(ccccccccccccccccc),"
+    assert edit.line_map == {1: 1}
+    assert 4 in edit.alternatives[1]
+    assert node_line_map(ast.parse(text), ast.parse(edit.text))[1] == edit.alternatives[1]
+
+
+REFACTORED = """\
+```python
+def get(self, expr):
+    return eval(expr) + self.base
+```"""
+
+
+def test_ai_refactor_uses_the_cache_and_a_line_diff(tmp_path: Path) -> None:
+    prompts: list[str] = []
+
+    def ask(prompt: str) -> str:
+        prompts.append(prompt)
+        return REFACTORED
+
+    first = AIRefactor("some-model", tmp_path, ask=ask)
+    edit = first(Source(SAMPLE), GET)
+    assert edit.needs_review and "def get(self, expr):" in edit.review["old"]
+    new = split_lines(edit.text)
+    assert new[GET - 1] == "        return eval(expr) + self.base"
+    # Ground truth from the line diff: the edited eval line still pairs with the old one.
+    assert edit.line_map[GET] == GET
+    assert edit.line_map[GET - 1] == GET - 1 and edit.line_map[GET + 2] == GET + 1
+    assert GET + 1 not in edit.line_map  # "return value" was deleted
+    assert behaviour(edit.text) == behaviour(SAMPLE)
+
+    cached = json.loads(next(tmp_path.glob("*.json")).read_text())
+    assert cached["model"] == "some-model" and cached["response"] == REFACTORED
+    offline = AIRefactor("some-model", tmp_path, offline=True)
+    assert offline(Source(SAMPLE), GET).text == edit.text
+    assert len(prompts) == 1 and offline.cache_hits == 1
+    with pytest.raises(NotApplicable):
+        AIRefactor("other-model", tmp_path, offline=True)(Source(SAMPLE), GET)
+
+
+def test_ai_refactor_rejects_a_different_function(tmp_path: Path) -> None:
+    ai = AIRefactor("m", tmp_path, ask=lambda _: "```python\ndef other(self):\n    pass\n```")
+    with pytest.raises(NotApplicable):
+        ai(Source(SAMPLE), GET)
diff --git a/research/finding_identity/transform_cli.py b/research/finding_identity/transform_cli.py
new file mode 100644
index 0000000..81f29d2
--- /dev/null
+++ b/research/finding_identity/transform_cli.py
@@ -0,0 +1,104 @@
+"""``python -m research.finding_identity transform``: the controlled experiment.
+
+Kept apart from :mod:`cli` and free of heavy imports, so ``replay`` and ``summarize``
+work without the experiment's extra dependencies (``requirements.txt``).
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+from typing import Any
+
+HERE = Path(__file__).parent
+MODEL_ENV = "FINDING_IDENTITY_AI_MODEL"
+KEY_ENV = "ANTHROPIC_API_KEY"
+
+
+def add_parser(sub: Any) -> None:
+    tp = sub.add_parser(
+        "transform",
+        help="apply controlled transformations and score every scheme (invariance matrix)",
+    )
+    tp.add_argument("source", help="local directory, local repository or git URL to clone")
+    tp.add_argument("--ref", help="branch or tag to clone (URLs only)")
+    tp.add_argument(
+        "--transforms",
+        default="all",
+        help="comma-separated transformations, or 'all' (ai_refactor only with a model)",
+    )
+    tp.add_argument("--samples", type=int, default=200, help="files with findings to use")
+    tp.add_argument("--seed", type=int, default=0, help="seed for sampling files and targets")
+    tp.add_argument("--label", help="name for the results (default: directory name)")
+    tp.add_argument("--out", type=Path, help="results directory")
+    tp.add_argument("--workdir", type=Path, default=HERE / ".repos", help="where to clone")
+    tp.add_argument(
+        "--categories",
+        default="security,performance,quality",
+        help="comma-separated rule categories to run",
+    )
+    tp.add_argument(
+        "--reformat-line-length", type=int, default=79, help="line length for 'reformat'"
+    )
+    tp.add_argument("--heatmap", action="store_true", help="also draw matrix.png (matplotlib)")
+    ai = tp.add_argument_group("ai_refactor")
+    ai.add_argument("--ai-model", help=f"model name (default: ${MODEL_ENV})")
+    ai.add_argument("--ai-api-key", help=f"API key (default: ${KEY_ENV})")
+    ai.add_argument(
+        "--ai-cache", type=Path, default=HERE / ".cache" / "ai", help="cached model responses"
+    )
+    ai.add_argument(
+        "--ai-offline", action="store_true", help="use cached responses only; never call the API"
+    )
+
+
+def run(args: argparse.Namespace) -> None:
+    from .experiment import run_experiment
+    from .gitrepo import GitRepo
+    from .scan import Scanner, rule_ids_for
+    from .schemes import ALL_SCHEMES
+    from .transforms import ALL_TRANSFORMS, AIRefactor, Transform, deterministic_transforms
+
+    model = args.ai_model or os.environ.get(MODEL_ENV)
+    wanted = [t.strip() for t in args.transforms.split(",") if t.strip()]
+    if wanted == ["all"]:
+        wanted = [t for t in ALL_TRANSFORMS if t != "ai_refactor" or model]
+    unknown = sorted(set(wanted) - set(ALL_TRANSFORMS))
+    if unknown:
+        raise SystemExit(f"unknown transformations: {', '.join(unknown)}")
+    available: dict[str, Transform] = dict(deterministic_transforms(args.reformat_line_length))
+    if "ai_refactor" in wanted:
+        if not model:
+            raise SystemExit(f"ai_refactor needs --ai-model or ${MODEL_ENV}")
+        available["ai_refactor"] = AIRefactor(
+            model,
+            args.ai_cache,
+            api_key=args.ai_api_key or os.environ.get(KEY_ENV),
+            offline=args.ai_offline,
+        )
+    transforms = {name: available[name] for name in wanted}
+
+    meta: dict[str, Any] = {"source": args.source, "ai_model": model if model else None}
+    local = Path(args.source)
+    if local.is_dir():
+        root = local
+    else:
+        repo = GitRepo.open_or_clone(args.source, args.workdir, depth=1, ref=args.ref)
+        root = repo.path
+        meta["head"] = repo.resolve("HEAD")
+        repo.close()
+    label = args.label or root.resolve().name
+    categories = [c.strip() for c in args.categories.split(",") if c.strip()]
+    run_experiment(
+        root,
+        label=label,
+        transforms=transforms,
+        scanner=Scanner(rule_ids_for(categories), ALL_SCHEMES),
+        schemes=ALL_SCHEMES,
+        samples=args.samples,
+        seed=args.seed,
+        out_dir=args.out or HERE / "results" / f"transform-{label}",
+        meta={**meta, "categories": categories},
+        draw_heatmap=args.heatmap,
+    )
diff --git a/research/finding_identity/transforms.py b/research/finding_identity/transforms.py
new file mode 100644
index 0000000..5259a6c
--- /dev/null
+++ b/research/finding_identity/transforms.py
@@ -0,0 +1,922 @@
+"""Source transformations with exact line maps, for the controlled experiment (RQ2).
+
+Each transformation rewrites one Python file around one finding (the *target*, given by
+its line) and returns the new text plus a map from old lines to new lines. The map is
+the ground truth by construction: a finding on old line ``l`` is the same finding as one
+with the same rule on new line ``line_map[l]``. Lines that the transformation deletes or
+rewrites beyond recognition are absent from the map.
+
+| Transformation    | What it does                                                        |
+|-------------------|---------------------------------------------------------------------|
+| reformat          | runs ``black`` (or ``ruff format``) at a narrower line length       |
+| insert_top        | inserts a comment and an import below the docstring and imports     |
+| insert_below      | inserts a comment line right below the flagged statement            |
+| rename_local      | renames a local variable used on the flagged line                   |
+| rename_function   | renames the enclosing function or method and its call sites         |
+| move_function     | swaps the enclosing function with an adjacent function              |
+| change_literal    | changes a string or number literal on the flagged line              |
+| extract_helper    | moves the flagged statement into a new module-level helper function |
+| duplicate_above   | copies the flagged statement directly above itself (a new finding)  |
+| duplicate_below   | copies the flagged statement directly below itself (a new finding)  |
+| ai_refactor       | asks a language model to refactor the enclosing function (optional) |
+
+All but ``change_literal``, the two ``duplicate`` transformations and ``ai_refactor``
+preserve the behaviour of the file itself; renames do not update other modules. A
+transformation raises :class:`NotApplicable` when the target does not fit it (for
+example, a finding at module level has no enclosing function to rename).
+"""
+
+from __future__ import annotations
+
+import ast
+import hashlib
+import importlib
+import io
+import json
+import re
+import subprocess
+import textwrap
+import tokenize
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from functools import cached_property
+from pathlib import Path
+from typing import Any, Protocol
+
+import libcst as cst
+from libcst.metadata import (
+    Assignment,
+    ClassScope,
+    CodeRange,
+    FunctionScope,
+    MetadataWrapper,
+    PositionProvider,
+    Scope,
+    ScopeProvider,
+)
+
+from .linemap import map_lines
+from .schemes import split_lines
+
+DETERMINISTIC = (
+    "reformat",
+    "insert_top",
+    "insert_below",
+    "rename_local",
+    "rename_function",
+    "move_function",
+    "change_literal",
+    "extract_helper",
+    "duplicate_above",
+    "duplicate_below",
+)
+ALL_TRANSFORMS = (*DETERMINISTIC, "ai_refactor")
+
+#: Text the insertions add. Comments and an unused import keep the file's behaviour.
+TOP_LINES = (
+    "# Added by the finding-identity experiment.",
+    "import collections as _fi_collections  # noqa: F401",
+)
+BELOW_LINE = "# Reviewed: no behaviour change."
+
+
+class NotApplicable(Exception):  # noqa: N818  (not an error: a skipped case)
+    """The transformation does not fit this target (the reason is the message)."""
+
+
+@dataclass
+class Edit:
+    """The result of one transformation."""
+
+    text: str
+    #: ``{old line: new line}`` for every line the transformation keeps (1-based).
+    line_map: dict[int, int]
+    #: Reformatting can split one old line over several new lines; the syntax nodes that
+    #: started on the old line start on these new lines (first entry = ``line_map``).
+    alternatives: dict[int, tuple[int, ...]] = field(default_factory=dict)
+    #: The ground truth is a line diff, not a construction: check these by hand.
+    needs_review: bool = False
+    #: Material for the manual check (old and new code).
+    review: dict[str, str] = field(default_factory=dict)
+
+
+class Transform(Protocol):
+    def __call__(self, src: Source, line: int) -> Edit: ...
+
+
+# ---------------------------------------------------------------------------
+# Source file with lazily computed views
+# ---------------------------------------------------------------------------
+
+_LINE_RE = re.compile(r"[^\r\n]*(?:\r\n|\r|\n)|[^\r\n]+\Z")
+_FUNC_NODES = (ast.FunctionDef, ast.AsyncFunctionDef)
+
+
+def split_keepends(text: str) -> list[str]:
+    """Lines with their endings, split on LF, CRLF and CR only (as the parser counts)."""
+    return _LINE_RE.findall(text)
+
+
+def _indent(line: str) -> str:
+    return line[: len(line) - len(line.lstrip(" \t"))]
+
+
+class Source:
+    """One version of one Python file. Raises ``SyntaxError`` if it does not parse."""
+
+    def __init__(self, text: str) -> None:
+        self.text = text
+        self.lines = split_keepends(text)
+        self.tree = ast.parse(text)
+        self.newline = "\r\n" if "\r\n" in text else "\n"
+        #: Results that do not depend on the target, keyed by transformation.
+        self.memo: dict[str, Edit | NotApplicable] = {}
+
+    def line(self, n: int) -> str:
+        """Line *n* (1-based) without its ending."""
+        return self.lines[n - 1].rstrip("\r\n") if 1 <= n <= len(self.lines) else ""
+
+    @cached_property
+    def parents(self) -> dict[ast.AST, ast.AST]:
+        return {c: p for p in ast.walk(self.tree) for c in ast.iter_child_nodes(p)}
+
+    @cached_property
+    def statements(self) -> list[ast.stmt]:
+        return [n for n in ast.walk(self.tree) if isinstance(n, ast.stmt)]
+
+    @cached_property
+    def _cst(self) -> tuple[MetadataWrapper, Mapping[cst.CSTNode, Scope | None]] | None:
+        try:
+            wrapper = MetadataWrapper(cst.parse_module(self.text))
+            return wrapper, wrapper.resolve(ScopeProvider)
+        except Exception:  # libcst raises parser errors and recursion errors
+            return None
+
+    @property
+    def wrapper(self) -> MetadataWrapper:
+        if self._cst is None:
+            raise NotApplicable("libcst cannot parse the file")
+        return self._cst[0]
+
+    @property
+    def scopes(self) -> Mapping[cst.CSTNode, Scope | None]:
+        if self._cst is None:
+            raise NotApplicable("libcst cannot parse the file")
+        return self._cst[1]
+
+    @cached_property
+    def positions(self) -> Mapping[cst.CSTNode, CodeRange]:
+        return self.wrapper.resolve(PositionProvider)
+
+    @cached_property
+    def cst_by_line(self) -> dict[int, list[cst.CSTNode]]:
+        """CST names, literals and function definitions by start line, in column order."""
+        out: dict[int, list[tuple[int, cst.CSTNode]]] = {}
+        kinds = (cst.Name, cst.SimpleString, cst.Integer, cst.Float, cst.FunctionDef)
+        for node, pos in self.positions.items():
+            if isinstance(node, cst.FunctionDef):
+                pos = self.positions[node.name]
+            if isinstance(node, kinds):
+                out.setdefault(pos.start.line, []).append((pos.start.column, node))
+        return {k: [n for _, n in sorted(v, key=lambda t: t[0])] for k, v in out.items()}
+
+    # -- syntax helpers ------------------------------------------------------
+
+    def enclosing_function(self, line: int) -> ast.FunctionDef | ast.AsyncFunctionDef:
+        """Innermost function whose ``def`` line to last line contains *line*."""
+        best: ast.FunctionDef | ast.AsyncFunctionDef | None = None
+        for node in ast.walk(self.tree):
+            if isinstance(node, _FUNC_NODES) and node.lineno <= line <= _end(node):
+                if best is None or node.lineno >= best.lineno:
+                    best = node
+        if best is None:
+            raise NotApplicable("finding is not inside a function")
+        return best
+
+    def ancestors(self, node: ast.AST) -> list[ast.AST]:
+        out = []
+        while node in self.parents:
+            node = self.parents[node]
+            out.append(node)
+        return out
+
+    def simple_statement(self, line: int) -> ast.stmt:
+        """The innermost statement containing *line*; it must be a simple statement that
+        occupies whole lines on its own (no ``;`` neighbours, no ``if x: stmt``)."""
+        containing = [s for s in self.statements if s.lineno <= line <= _end(s)]
+        if not containing:
+            raise NotApplicable("no statement on the flagged line")
+        stmt = max(containing, key=lambda s: (s.lineno, -_end(s)))
+        if hasattr(stmt, "body"):
+            raise NotApplicable("flagged line is a compound statement header")
+        if self._prefix(stmt.lineno, stmt.col_offset).strip():
+            raise NotApplicable("statement shares its first line")
+        family = set(self.ancestors(stmt)) | {stmt}
+        for other in self.statements:
+            if other not in family and other.lineno <= _end(stmt) and _end(other) >= stmt.lineno:
+                raise NotApplicable("statement shares a line with another statement")
+        return stmt
+
+    def _prefix(self, line: int, col: int) -> str:
+        """Text of *line* before byte column *col* (ast columns are UTF-8 offsets)."""
+        return self.line(line).encode("utf-8")[:col].decode("utf-8", "replace")
+
+
+def _end(node: ast.AST) -> int:
+    end = getattr(node, "end_lineno", None)
+    return int(end) if end is not None else int(getattr(node, "lineno", 0))
+
+
+def _start(node: ast.stmt) -> int:
+    """First line of a statement, including its decorators."""
+    decorators: list[ast.expr] = getattr(node, "decorator_list", [])
+    return min([node.lineno, *(d.lineno for d in decorators)])
+
+
+# ---------------------------------------------------------------------------
+# Line-splicing helpers
+# ---------------------------------------------------------------------------
+
+
+def _terminated(lines: list[str], newline: str) -> list[str]:
+    out = list(lines)
+    if out and not out[-1].endswith(("\n", "\r")):
+        out[-1] += newline
+    return out
+
+
+def _insert(src: Source, before: int, block: list[str]) -> Edit:
+    """Insert *block* (lines without endings) before old line *before* (1-based)."""
+    lines = _terminated(src.lines, src.newline)
+    new = lines[: before - 1] + [b + src.newline for b in block] + lines[before - 1 :]
+    n = len(block)
+    line_map = {k: k if k < before else k + n for k in range(1, len(lines) + 1)}
+    return Edit("".join(new), line_map)
+
+
+def _identity(src: Source, new_text: str) -> Edit:
+    if len(split_keepends(new_text)) != len(src.lines):
+        raise NotApplicable("rewrite changed the number of lines")
+    return Edit(new_text, {k: k for k in range(1, len(src.lines) + 1)})
+
+
+def _unique_name(src: Source, base: str) -> str:
+    name, k = base, 2
+    while re.search(rf"\b{re.escape(name)}\b", src.text):
+        name, k = f"{base}{k}", k + 1
+    return name
+
+
+# ---------------------------------------------------------------------------
+# reformat
+# ---------------------------------------------------------------------------
+
+
+def _format(text: str, line_length: int) -> str:
+    try:
+        black: Any = importlib.import_module("black")
+    except ImportError:
+        proc = subprocess.run(
+            ["ruff", "format", "--isolated", f"--line-length={line_length}", "-"],
+            input=text.encode("utf-8"),
+            capture_output=True,
+            check=False,
+        )
+        if proc.returncode:
+            raise NotApplicable("formatter failed") from None
+        return proc.stdout.decode("utf-8")
+    try:
+        return str(black.format_str(text, mode=black.Mode(line_length=line_length)))
+    except Exception as exc:  # black raises several unrelated exception types
+        raise NotApplicable(f"formatter failed: {type(exc).__name__}") from exc
+
+
+def node_line_map(old: ast.AST, new: ast.AST) -> dict[int, tuple[int, ...]]:
+    """Pair the syntax nodes of two equivalent trees and map each old line on which
+    nodes start to the new lines where the same nodes start (outermost first)."""
+    old_nodes = [n for n in ast.walk(old) if hasattr(n, "lineno")]
+    new_nodes = [n for n in ast.walk(new) if hasattr(n, "lineno")]
+    if len(old_nodes) != len(new_nodes) or any(
+        type(a) is not type(b) for a, b in zip(old_nodes, new_nodes, strict=True)
+    ):
+        raise NotApplicable("reformatting changed the syntax tree")
+    out: dict[int, list[int]] = {}
+    for a, b in zip(old_nodes, new_nodes, strict=True):
+        targets = out.setdefault(a.lineno, [])
+        if b.lineno not in targets:
+            targets.append(b.lineno)
+    return {k: tuple(v) for k, v in out.items()}
+
+
+def _reformat(src: Source, line_length: int) -> Edit:
+    new_text = _format(src.text, line_length)
+    if new_text == src.text:
+        raise NotApplicable("already formatted")
+    alternatives = node_line_map(src.tree, ast.parse(new_text))
+    return Edit(
+        new_text,
+        {k: v[0] for k, v in alternatives.items()},
+        {k: v for k, v in alternatives.items() if len(v) > 1},
+    )
+
+
+def make_reformat(line_length: int = 79) -> Transform:
+    """Reformat the whole file. Projects already formatted at 88 columns still change
+    at the default of 79 (the PEP 8 limit)."""
+
+    def reformat(src: Source, line: int) -> Edit:
+        key = f"reformat:{line_length}"
+        if key not in src.memo:
+            try:
+                src.memo[key] = _reformat(src, line_length)
+            except NotApplicable as exc:
+                src.memo[key] = exc
+        result = src.memo[key]
+        if isinstance(result, NotApplicable):
+            raise result
+        return result
+
+    return reformat
+
+
+# ---------------------------------------------------------------------------
+# insert_top, insert_below
+# ---------------------------------------------------------------------------
+
+_MAGIC_COMMENT = re.compile(r"^#!|^#.*coding[:=]")
+
+
+def insert_top(src: Source, line: int) -> Edit:
+    """Insert a comment and an import at the top, after the module docstring and the
+    ``from __future__`` imports (or after a shebang or encoding line)."""
+    body, last, i = src.tree.body, 0, 0
+    if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant):
+        last, i = _end(body[0]), 1
+    while i < len(body):
+        node = body[i]
+        if not (isinstance(node, ast.ImportFrom) and node.module == "__future__"):
+            break
+        last, i = _end(node), i + 1
+    if last == 0:
+        while last < min(2, len(src.lines)) and _MAGIC_COMMENT.match(src.line(last + 1)):
+            last += 1
+    return _insert(src, last + 1, list(TOP_LINES))
+
+
+def _logical_line_end(src: Source, line: int) -> int:
+    """Last physical line of the logical line that contains *line*."""
+    text = src.text.replace("\r\n", "\n").replace("\r", "\n")
+    try:
+        for tok in tokenize.generate_tokens(io.StringIO(text).readline):
+            if tok.type == tokenize.NEWLINE and tok.start[0] >= line:
+                return tok.start[0]
+    except (tokenize.TokenError, SyntaxError) as exc:
+        raise NotApplicable("tokenizer failed") from exc
+    return len(src.lines)
+
+
+def insert_below(src: Source, line: int) -> Edit:
+    """Insert a comment line right below the flagged statement (its logical line). This
+    changes the text GitHub's line hash reads (100 characters forward)."""
+    end = _logical_line_end(src, line)
+    containing = [s for s in src.statements if s.lineno <= line <= _end(s)]
+    first = max((s.lineno for s in containing), default=line)
+    indent = _indent(src.line(first))
+    for k in range(end + 1, len(src.lines) + 1):
+        if src.line(k).strip():
+            if len(_indent(src.line(k))) > len(indent):
+                indent = _indent(src.line(k))
+            break
+    return _insert(src, end + 1, [indent + BELOW_LINE])
+
+
+# ---------------------------------------------------------------------------
+# rename_local, rename_function, change_literal (libcst, source preserving)
+# ---------------------------------------------------------------------------
+
+
+class _Rewriter(cst.CSTTransformer):
+    """Give new ``value`` text to names and literals, chosen by node identity."""
+
+    def __init__(self, values: dict[int, str]) -> None:
+        super().__init__()
+        self._values = values
+
+    def leave_Name(self, original_node: cst.Name, updated_node: cst.Name) -> cst.Name:  # noqa: N802
+        value = self._values.get(id(original_node))
+        return updated_node if value is None else updated_node.with_changes(value=value)
+
+    def leave_SimpleString(  # noqa: N802
+        self, original_node: cst.SimpleString, updated_node: cst.SimpleString
+    ) -> cst.BaseExpression:
+        value = self._values.get(id(original_node))
+        return updated_node if value is None else updated_node.with_changes(value=value)
+
+    def leave_Integer(  # noqa: N802
+        self, original_node: cst.Integer, updated_node: cst.Integer
+    ) -> cst.BaseExpression:
+        value = self._values.get(id(original_node))
+        return updated_node if value is None else updated_node.with_changes(value=value)
+
+    def leave_Float(  # noqa: N802
+        self, original_node: cst.Float, updated_node: cst.Float
+    ) -> cst.BaseExpression:
+        value = self._values.get(id(original_node))
+        return updated_node if value is None else updated_node.with_changes(value=value)
+
+
+def _rewrite(src: Source, values: dict[int, str]) -> Edit:
+    return _identity(src, src.wrapper.module.visit(_Rewriter(values)).code)
+
+
+def _function_text(src: Source, func: ast.AST) -> str:
+    return "".join(src.lines[getattr(func, "lineno", 1) - 1 : _end(func)])
+
+
+def rename_local(src: Source, line: int) -> Edit:
+    """Rename a local variable (not a parameter) used on the flagged line, everywhere in
+    its function, including nested functions that read it."""
+    func = src.enclosing_function(line)
+    if re.search(r"\b(locals|vars)\(\)|\bnonlocal\b|\bglobal\b", _function_text(src, func)):
+        raise NotApplicable("function inspects or rebinds its namespace")
+    for node in src.cst_by_line.get(line, []):
+        if not isinstance(node, cst.Name):
+            continue
+        scope = src.scopes.get(node)
+        if not isinstance(scope, FunctionScope):
+            continue
+        assignments = list(scope.assignments[node.value])
+        if not assignments or not all(
+            isinstance(a, Assignment) and isinstance(a.node, cst.Name) for a in assignments
+        ):
+            continue  # not local, or bound by a parameter, import, def or class
+        targets: list[cst.CSTNode] = []
+        for a in assignments:
+            assert isinstance(a, Assignment)
+            targets.append(a.node)
+            targets.extend(r.node for r in a.references if isinstance(r.node, cst.Name))
+        new = _unique_name(src, f"{node.value}_v2")
+        return _rewrite(src, {id(t): new for t in targets})
+    raise NotApplicable("no local variable on the flagged line")
+
+
+def _cst_function(src: Source, func: ast.FunctionDef | ast.AsyncFunctionDef) -> cst.FunctionDef:
+    for node in src.cst_by_line.get(func.lineno, []):
+        if isinstance(node, cst.FunctionDef) and node.name.value == func.name:
+            return node
+    raise NotApplicable("function not found in the CST")
+
+
+def rename_function(src: Source, line: int) -> Edit:
+    """Rename the enclosing function or method and its references in the file: names
+    bound to a function, and ``self.m``, ``cls.m`` and ``Class.m`` for a method."""
+    func = src.enclosing_function(line)
+    name = func.name
+    if name.startswith("__") and name.endswith("__"):
+        raise NotApplicable("special method")
+    fdef = _cst_function(src, func)
+    scope = src.scopes.get(fdef)
+    new = _unique_name(src, f"{name}_renamed")
+    if isinstance(scope, ClassScope):
+        same_name = [n for n in ast.walk(src.tree) if isinstance(n, _FUNC_NODES) and n.name == name]
+        if len(same_name) > 1:
+            raise NotApplicable("method name defined more than once (override or property)")
+        cls = src.parents.get(func)
+        assert isinstance(cls, ast.ClassDef)
+        owners = {"self", "cls", cls.name}
+        targets: list[cst.CSTNode] = [fdef.name]
+        for node in src.positions:
+            if (
+                isinstance(node, cst.Attribute)
+                and node.attr.value == name
+                and isinstance(node.value, cst.Name)
+                and node.value.value in owners
+            ):
+                targets.append(node.attr)
+        return _rewrite(src, {id(t): new for t in targets})
+    if scope is None:
+        raise NotApplicable("no scope for the function")
+    assignments = list(scope.assignments[name])
+    only = assignments[0] if len(assignments) == 1 else None
+    if not isinstance(only, Assignment) or only.node is not fdef:
+        raise NotApplicable("function name is bound more than once")
+    targets = [fdef.name, *(r.node for r in only.references if isinstance(r.node, cst.Name))]
+    return _rewrite(src, {id(t): new for t in targets})
+
+
+_FLOAT_RE = re.compile(r"\d+\.\d*|\.\d+")
+
+
+def _changed_literal(node: cst.CSTNode) -> str | None:
+    if isinstance(node, cst.SimpleString):
+        prefix, quote = node.prefix, node.quote
+        inner = node.value[len(prefix) + len(quote) : -len(quote)]
+        return f"{prefix}{quote}{inner}_v2{quote}"
+    if isinstance(node, cst.Integer) and node.value.isdigit():
+        return str(int(node.value) + 1)
+    if isinstance(node, cst.Float) and _FLOAT_RE.fullmatch(node.value):
+        return repr(float(node.value) + 1)
+    return None
+
+
+def change_literal(src: Source, line: int) -> Edit:
+    """Change the first string literal on the flagged line (append ``_v2``), or else the
+    first number literal (add one). Literals spanning several lines are left alone."""
+    nodes = src.cst_by_line.get(line, [])
+    strings = [n for n in nodes if isinstance(n, cst.SimpleString)]
+    numbers = [n for n in nodes if isinstance(n, cst.Integer | cst.Float)]
+    for node in strings + numbers:
+        if src.positions[node].end.line != line:
+            continue
+        changed = _changed_literal(node)
+        if changed is not None:
+            return _rewrite(src, {id(node): changed})
+    raise NotApplicable("no literal on the flagged line")
+
+
+# ---------------------------------------------------------------------------
+# move_function
+# ---------------------------------------------------------------------------
+
+
+def _definition_deps(node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
+    """Names a ``def`` evaluates when it runs: decorators, defaults and annotations."""
+    parts: list[ast.AST] = [*node.decorator_list, *node.args.defaults]
+    parts += [d for d in node.args.kw_defaults if d is not None]
+    args = [*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
+    args += [a for a in (node.args.vararg, node.args.kwarg) if a is not None]
+    parts += [a.annotation for a in args if a.annotation is not None]
+    if node.returns is not None:
+        parts.append(node.returns)
+    return {n.id for p in parts for n in ast.walk(p) if isinstance(n, ast.Name)}
+
+
+def _block(src: Source, node: ast.stmt, floor: int) -> tuple[int, int]:
+    """Lines of a definition with its decorators and the comments right above it."""
+    start = _start(node)
+    while start - 1 > floor and src.line(start - 1).lstrip().startswith("#"):
+        start -= 1
+    return start, _end(node)
+
+
+def move_function(src: Source, line: int) -> Edit:
+    """Swap the enclosing function with the function defined right before it (or, if
+    that is not safe, right after it) in the same body. Comments directly above a
+    definition move with it."""
+    func = src.enclosing_function(line)
+    parent = src.parents.get(func)
+    body = next(
+        (
+            b
+            for b in (getattr(parent, f, None) for f in ("body", "orelse", "finalbody"))
+            if isinstance(b, list) and func in b
+        ),
+        None,
+    )
+    if body is None:
+        raise NotApplicable("function body not found")
+    # Comments right after a class or def header belong to the body, not to the header.
+    floor = 0 if isinstance(parent, ast.Module) else getattr(parent, "lineno", 0)
+    i = body.index(func)
+    for j in (i - 1, i + 1):
+        if not 0 <= j < len(body):
+            continue
+        a, b = (body[j], func) if j < i else (func, body[j])
+        if not (isinstance(a, _FUNC_NODES) and isinstance(b, _FUNC_NODES)):
+            continue
+        if a.name == b.name or a.name in _definition_deps(b) or b.name in _definition_deps(a):
+            continue
+        k = body.index(a)
+        a0, a1 = _block(src, a, _end(body[k - 1]) if k > 0 else floor)
+        b0, b1 = _block(src, b, a1)
+        return _swap(src, (a0, a1), (b0, b1))
+    raise NotApplicable("no adjacent function to swap with")
+
+
+def _swap(src: Source, a: tuple[int, int], b: tuple[int, int]) -> Edit:
+    (a0, a1), (b0, b1) = a, b
+    lines = _terminated(src.lines, src.newline)
+    block_a, gap, block_b = lines[a0 - 1 : a1], lines[a1 : b0 - 1], lines[b0 - 1 : b1]
+    new = lines[: a0 - 1] + block_b + gap + block_a + lines[b1:]
+    nb, ng = len(block_b), len(gap)
+    line_map: dict[int, int] = {}
+    for k in range(1, len(lines) + 1):
+        if a0 <= k <= a1:
+            line_map[k] = a0 + nb + ng + (k - a0)
+        elif a1 < k < b0:
+            line_map[k] = a0 + nb + (k - a1 - 1)
+        elif b0 <= k <= b1:
+            line_map[k] = a0 + (k - b0)
+        else:
+            line_map[k] = k
+    return Edit("".join(new), line_map)
+
+
+# ---------------------------------------------------------------------------
+# extract_helper, duplicate
+# ---------------------------------------------------------------------------
+
+_EXTRACTABLE = (
+    ast.Assign,
+    ast.AugAssign,
+    ast.AnnAssign,
+    ast.Expr,
+    ast.Return,
+    ast.Raise,
+    ast.Assert,
+)
+_NESTED_SCOPES = (
+    ast.FunctionDef,
+    ast.AsyncFunctionDef,
+    ast.ClassDef,
+    ast.Lambda,
+    ast.ListComp,
+    ast.SetComp,
+    ast.DictComp,
+    ast.GeneratorExp,
+)
+
+
+def _function_locals(func: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
+    a = func.args
+    names = {x.arg for x in [*a.posonlyargs, *a.args, *a.kwonlyargs]}
+    names |= {x.arg for x in (a.vararg, a.kwarg) if x is not None}
+    todo: list[ast.AST] = list(func.body)
+    while todo:
+        node = todo.pop()
+        if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store | ast.Del):
+            names.add(node.id)
+        elif isinstance(node, ast.alias):
+            names.add((node.asname or node.name).split(".")[0])
+        elif isinstance(node, ast.ExceptHandler) and node.name:
+            names.add(node.name)
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
+            names.add(node.name)
+        if not isinstance(node, _NESTED_SCOPES):
+            todo.extend(ast.iter_child_nodes(node))
+    return names
+
+
+def _nested_bound(stmt: ast.stmt) -> set[str]:
+    """Names bound by lambdas and comprehensions inside *stmt*."""
+    out: set[str] = set()
+    for node in ast.walk(stmt):
+        if isinstance(node, ast.Lambda):
+            a = node.args
+            out |= {x.arg for x in [*a.posonlyargs, *a.args, *a.kwonlyargs]}
+            out |= {x.arg for x in (a.vararg, a.kwarg) if x is not None}
+        elif isinstance(node, ast.comprehension):
+            out |= {n.id for n in ast.walk(node.target) if isinstance(n, ast.Name)}
+    return out
+
+
+def _check_extractable(stmt: ast.stmt) -> None:
+    if not isinstance(stmt, _EXTRACTABLE):
+        raise NotApplicable(f"cannot extract a {type(stmt).__name__} statement")
+    if isinstance(stmt, ast.AnnAssign) and stmt.value is None:
+        raise NotApplicable("bare annotation")
+    if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant):
+        raise NotApplicable("docstring")
+    for node in ast.walk(stmt):
+        if isinstance(node, ast.Yield | ast.YieldFrom | ast.Await | ast.NamedExpr):
+            raise NotApplicable("statement yields, awaits or uses :=")
+        if isinstance(node, ast.Name) and node.id in {"super", "locals", "vars", "__class__"}:
+            raise NotApplicable("statement depends on its frame")
+        name = node.id if isinstance(node, ast.Name) else getattr(node, "attr", "")
+        if isinstance(name, str) and name.startswith("__") and not name.endswith("__"):
+            raise NotApplicable("private name mangling")
+
+
+def _reindented(src: Source, stmt: ast.stmt, indent: str) -> list[str]:
+    """The statement's lines with *indent* in place of its own indentation, unless a
+    string literal spans lines (re-indenting would change the string)."""
+    lines = [src.line(k) for k in range(stmt.lineno, _end(stmt) + 1)]
+    own = _indent(lines[0])
+    multiline_string = any(
+        isinstance(n, ast.Constant | ast.JoinedStr) and _end(n) > n.lineno for n in ast.walk(stmt)
+    )
+    if multiline_string or not all(ln.startswith(own) or not ln.strip() for ln in lines):
+        return lines
+    return [indent + ln[len(own) :] if ln.strip() else ln for ln in lines]
+
+
+def extract_helper(src: Source, line: int) -> Edit:
+    """Move the flagged statement into a new module-level function placed right before
+    the top-level definition that contains it, and call it from the old place. Locals
+    the statement reads become parameters; locals it assigns are returned."""
+    stmt = src.simple_statement(line)
+    _check_extractable(stmt)
+    func = src.enclosing_function(stmt.lineno)
+    if func not in src.ancestors(stmt):
+        raise NotApplicable("statement is not in a function body")
+    chain = src.ancestors(func)
+    if not (
+        isinstance(chain[0], ast.Module)
+        or (isinstance(chain[0], ast.ClassDef) and isinstance(chain[1], ast.Module))
+    ):
+        raise NotApplicable("function is nested in another function")
+    if any(isinstance(n, ast.Global | ast.Nonlocal) for n in ast.walk(func)):
+        raise NotApplicable("function declares global or nonlocal names")
+    top = chain[-2] if len(chain) >= 2 else func
+    assert isinstance(top, ast.stmt)
+
+    local = _function_locals(func)
+    nested = _nested_bound(stmt)
+    loads: list[str] = []
+    stores: list[str] = []
+    for node in ast.walk(stmt):
+        if isinstance(node, ast.Name):
+            bucket = loads if isinstance(node.ctx, ast.Load) else stores
+            bucket.append(node.id)
+    if isinstance(stmt, ast.AugAssign) and isinstance(stmt.target, ast.Name):
+        loads.append(stmt.target.id)
+    if set(loads) & nested & local:
+        raise NotApplicable("a comprehension or lambda shadows a local")
+    params = list(dict.fromkeys(n for n in loads if n in local and n not in nested))
+    outputs = list(dict.fromkeys(n for n in stores if n not in nested))
+
+    name = _unique_name(src, f"_{func.name}_extracted")
+    unit = "\t" if _indent(src.line(stmt.lineno)).startswith("\t") else "    "
+    helper = [f"def {name}({', '.join(params)}):"]
+    helper += _reindented(src, stmt, unit)
+    if outputs:
+        helper.append(f"{unit}return {', '.join(outputs)}")
+    helper += ["", ""]
+
+    call = f"{name}({', '.join(params)})"
+    if isinstance(stmt, ast.Return):
+        call = f"return {call}"
+    elif outputs:
+        call = f"{', '.join(outputs)} = {call}"
+    call = _indent(src.line(stmt.lineno)) + call
+
+    insert_at, _ = _block(src, top, 0)
+    s0, s1 = stmt.lineno, _end(stmt)
+    lines = _terminated(src.lines, src.newline)
+    nl = src.newline
+    new = (
+        lines[: insert_at - 1]
+        + [h + nl for h in helper]
+        + lines[insert_at - 1 : s0 - 1]
+        + [call + nl]
+        + lines[s1:]
+    )
+    k, n = len(helper), s1 - s0 + 1
+    line_map: dict[int, int] = {}
+    for old in range(1, len(lines) + 1):
+        if old < insert_at:
+            line_map[old] = old
+        elif old < s0:
+            line_map[old] = old + k
+        elif old <= s1:
+            line_map[old] = insert_at + 1 + (old - s0)
+        else:
+            line_map[old] = old + k - n + 1
+    return Edit("".join(new), line_map)
+
+
+def _duplicate(src: Source, line: int, *, above: bool) -> Edit:
+    stmt = src.simple_statement(line)
+    src.enclosing_function(stmt.lineno)
+    if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant):
+        raise NotApplicable("docstring")
+    s0, s1 = stmt.lineno, _end(stmt)
+    copy = [src.line(k) for k in range(s0, s1 + 1)]
+    return _insert(src, s0 if above else s1 + 1, copy)
+
+
+def duplicate_above(src: Source, line: int) -> Edit:
+    """Copy the flagged statement directly above itself: the copy is a new finding."""
+    return _duplicate(src, line, above=True)
+
+
+def duplicate_below(src: Source, line: int) -> Edit:
+    """Copy the flagged statement directly below itself: the copy is a new finding."""
+    return _duplicate(src, line, above=False)
+
+
+# ---------------------------------------------------------------------------
+# ai_refactor
+# ---------------------------------------------------------------------------
+
+AI_PROMPT_VERSION = 1
+AI_PROMPT = """\
+Refactor this Python function without changing its behaviour. Keep its name, its
+signature and its decorators. Reply with only the complete refactored function in one
+```python code block.
+
+```python
+{code}
+```
+"""
+_CODE_BLOCK = re.compile(r"```(?:python|py)?[ \t]*\n(.*?)```", re.DOTALL)
+
+
+class AIRefactor:
+    """Ask a language model to refactor the enclosing function.
+
+    Responses are cached on disk by (model, prompt), so a rerun with the same cache
+    gives the same edits without network access. The ground truth for these edits is a
+    line diff (:func:`linemap.map_lines`), so every case is flagged for manual review.
+    """
+
+    def __init__(
+        self,
+        model: str,
+        cache_dir: Path,
+        *,
+        api_key: str | None = None,
+        offline: bool = False,
+        ask: Callable[[str], str] | None = None,
+    ) -> None:
+        self.model = model
+        self.cache_dir = cache_dir
+        self.api_key = api_key
+        self.offline = offline
+        self._ask = ask
+        self._client: Any = None
+        self.calls = 0
+        self.cache_hits = 0
+
+    def complete(self, prompt: str) -> str:
+        key = hashlib.sha256(f"{self.model}\0{prompt}".encode()).hexdigest()
+        path = self.cache_dir / f"{key}.json"
+        if path.exists():
+            self.cache_hits += 1
+            return str(json.loads(path.read_text(encoding="utf-8"))["response"])
+        if self.offline:
+            raise NotApplicable("no cached model response (offline)")
+        response = self._ask(prompt) if self._ask else self._call_api(prompt)
+        self.calls += 1
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        record = {"model": self.model, "prompt": prompt, "response": response}
+        path.write_text(json.dumps(record, indent=1) + "\n", encoding="utf-8")
+        return response
+
+    def _call_api(self, prompt: str) -> str:
+        if self._client is None:
+            try:
+                anthropic: Any = importlib.import_module("anthropic")
+            except ImportError as exc:
+                raise RuntimeError("ai_refactor needs `pip install anthropic`") from exc
+            self._client = anthropic.Anthropic(api_key=self.api_key)
+        message = self._client.messages.create(
+            model=self.model,
+            max_tokens=16000,
+            messages=[{"role": "user", "content": prompt}],
+        )
+        if message.stop_reason == "refusal":
+            raise NotApplicable("model declined")
+        return "".join(b.text for b in message.content if b.type == "text")
+
+    def __call__(self, src: Source, line: int) -> Edit:
+        func = src.enclosing_function(line)
+        f0, f1 = _start(func), _end(func)
+        old_code = "".join(src.lines[f0 - 1 : f1])
+        base = _indent(src.line(f0))
+        prompt = AI_PROMPT.format(code=textwrap.dedent(old_code).rstrip("\n"))
+        reply = self.complete(prompt)
+        match = _CODE_BLOCK.search(reply)
+        code = textwrap.dedent(match.group(1) if match else reply).strip("\n")
+        try:
+            module = ast.parse(code)
+        except SyntaxError as exc:
+            raise NotApplicable("model reply does not parse") from exc
+        if not (
+            len(module.body) == 1
+            and isinstance(module.body[0], _FUNC_NODES)
+            and module.body[0].name == func.name
+        ):
+            raise NotApplicable("model reply is not one function with the same name")
+        nl = src.newline
+        new_lines = [(base + ln if ln.strip() else ln) + nl for ln in code.split("\n")]
+        lines = _terminated(src.lines, nl)
+        new_text = "".join(lines[: f0 - 1] + new_lines + lines[f1:])
+        try:
+            ast.parse(new_text)
+        except SyntaxError as exc:
+            raise NotApplicable("spliced file does not parse") from exc
+        if new_text == src.text:
+            raise NotApplicable("model returned the function unchanged")
+        mapping = map_lines(split_lines(src.text), split_lines(new_text))
+        return Edit(
+            new_text,
+            {k: v[0] for k, v in mapping.items()},
+            needs_review=True,
+            review={"old": old_code, "new": "".join(new_lines)},
+        )
+
+
+def deterministic_transforms(reformat_line_length: int = 79) -> dict[str, Transform]:
+    return {
+        "reformat": make_reformat(reformat_line_length),
+        "insert_top": insert_top,
+        "insert_below": insert_below,
+        "rename_local": rename_local,
+        "rename_function": rename_function,
+        "move_function": move_function,
+        "change_literal": change_literal,
+        "extract_helper": extract_helper,
+        "duplicate_above": duplicate_above,
+        "duplicate_below": duplicate_below,
+    }