From 47e400b52a1a07670557d4aec0ea35f0baaa4804 Mon Sep 17 00:00:00 2001
From: Vichitra Attri <mevichitra@gmail.com>
Date: Wed, 14 Oct 2026 12:19:05 +0000
Subject: [PATCH] feat(research): add finding-identity replay harness

Research tooling for a study of how well static-analysis findings keep
their identity as code changes. The harness replays a repository's
first-parent history, scans each changed file before and after every
commit with CodeGuard, and scores identity schemes against a ground
truth built from a line diff.

Schemes: line number, exact line text, GitHub code scanning's
primaryLocationLineHash (a port of codeql-action's fingerprints.ts,
verified against the original), CodeGuard's codeguard/v1 fingerprint,
and three ablations of it. Each new finding is classified as correctly
matched, mispaired, falsely new, or falsely known (hidden), and
same-file key collisions are counted.

Lives under research/ and is not part of the released package. Results
and clones are git-ignored.

Signed-off-by: Vichitra Attri <mevichitra@gmail.com>
---
 research/__init__.py                          |   1 +
 research/finding_identity/.gitignore          |   2 +
 research/finding_identity/README.md           |  91 +++++++++
 research/finding_identity/__init__.py         |   2 +
 research/finding_identity/__main__.py         |   3 +
 research/finding_identity/cli.py              |  64 ++++++
 research/finding_identity/gitrepo.py          | 119 +++++++++++
 research/finding_identity/linemap.py          |  64 ++++++
 research/finding_identity/matching.py         | 152 +++++++++++++++
 research/finding_identity/replay.py           | 152 +++++++++++++++
 research/finding_identity/scan.py             |  91 +++++++++
 research/finding_identity/schemes.py          | 184 ++++++++++++++++++
 research/finding_identity/summarize.py        | 113 +++++++++++
 research/finding_identity/tests/__init__.py   |   0
 .../finding_identity/tests/test_matching.py   |  40 ++++
 .../finding_identity/tests/test_replay.py     |  71 +++++++
 .../finding_identity/tests/test_schemes.py    |  77 ++++++++
 17 files changed, 1226 insertions(+)
 create mode 100644 research/__init__.py
 create mode 100644 research/finding_identity/.gitignore
 create mode 100644 research/finding_identity/README.md
 create mode 100644 research/finding_identity/__init__.py
 create mode 100644 research/finding_identity/__main__.py
 create mode 100644 research/finding_identity/cli.py
 create mode 100644 research/finding_identity/gitrepo.py
 create mode 100644 research/finding_identity/linemap.py
 create mode 100644 research/finding_identity/matching.py
 create mode 100644 research/finding_identity/replay.py
 create mode 100644 research/finding_identity/scan.py
 create mode 100644 research/finding_identity/schemes.py
 create mode 100644 research/finding_identity/summarize.py
 create mode 100644 research/finding_identity/tests/__init__.py
 create mode 100644 research/finding_identity/tests/test_matching.py
 create mode 100644 research/finding_identity/tests/test_replay.py
 create mode 100644 research/finding_identity/tests/test_schemes.py

diff --git a/research/__init__.py b/research/__init__.py
new file mode 100644
index 0000000..4606669
--- /dev/null
+++ b/research/__init__.py
@@ -0,0 +1 @@
+"""Research experiments built on CodeGuard (not part of the released package)."""
diff --git a/research/finding_identity/.gitignore b/research/finding_identity/.gitignore
new file mode 100644
index 0000000..351a656
--- /dev/null
+++ b/research/finding_identity/.gitignore
@@ -0,0 +1,2 @@
+.repos/
+results/
diff --git a/research/finding_identity/README.md b/research/finding_identity/README.md
new file mode 100644
index 0000000..6c915c5
--- /dev/null
+++ b/research/finding_identity/README.md
@@ -0,0 +1,91 @@
+# Finding identity: replay harness
+
+Research code for the study *How well do static-analysis findings keep their identity as
+code changes?* It is not part of the released `codeguard` package.
+
+Baselines, suppressions, "fail only on new findings" CI checks and alert de-duplication all
+rely on recognising that a finding in version *n+1* of a file is the same finding as one in
+version *n*. This harness replays real git histories, scans every changed file before and
+after each commit with CodeGuard, and scores several **identity schemes** against a
+**ground truth** built from the diff.
+
+## Schemes
+
+Every key includes the rule ID and the file path.
+
+| Scheme | Identity is built from |
+|---|---|
+| `line` | the line number |
+| `line_text` | the exact text of the flagged line |
+| `github` | GitHub code scanning's `primaryLocationLineHash` (exact port of `fingerprints.ts` from github/codeql-action, verified against the original) |
+| `codeguard` | CodeGuard's `codeguard/v1` fingerprint: enclosing function/class plus the normalized statement (literals masked, comments and whitespace removed) |
+| `cg_no_scope` | ablation: `codeguard` without the enclosing scope |
+| `cg_no_mask` | ablation: `codeguard` without masking literals |
+| `cg_raw` | ablation: `codeguard` without any normalization (scope plus raw line) |
+
+## Ground truth
+
+For each changed file, old lines are mapped to new lines with a line diff
+(`linemap.py`). Unchanged lines map exactly ("moved"). Inside a replaced block, lines are
+paired when their similarity is at least 0.6 ("edited"). An old and a new finding are the
+same finding when they have the same rule and the old line maps to the new line. Each new
+finding is then truly **known** (it existed before) or truly **new**.
+
+## Outcomes and metrics
+
+For each new finding, a scheme either matches it to an old finding or not:
+
+| Outcome | Truth | Scheme | Meaning |
+|---|---|---|---|
+| `known_ok` | known | matched to the right finding | correct |
+| `mispaired` | known | matched to a different old finding | wrong history carried over |
+| `false_new` | known | not matched | old problem reported as new: noise, broken baselines |
+| `false_known` | new | matched to an old finding | **new problem hidden** |
+| `new_ok` | new | not matched | correct |
+
+The summary reports, per scheme:
+
+- **false-new rate** = `false_new / known`, also split by moved and edited findings
+- **hidden rate** = `false_known / introduced`
+- **mispair rate** = `mispaired / known`
+- **collision rate** = findings sharing a key with another finding in the same file version,
+  over all findings in all scanned versions
+
+Only files that changed in a commit are compared. Every scheme trivially keeps the findings
+in untouched files, so including them would only inflate the counts.
+
+## Usage
+
+From the repository root, with CodeGuard installed (`pip install -e .`):
+
+```bash
+# Replay the last 200 first-parent commits of a repository (clones shallowly if a URL)
+python -m research.finding_identity replay https://github.com/pallets/flask --commits 200
+
+# Branch, label, rule categories, and one record per finding for manual labelling
+python -m research.finding_identity replay https://github.com/DefectDojo/django-DefectDojo \
+    --ref dev --commits 200 --label defectdojo --categories security,performance,quality --details
+
+# Rate tables for one or more runs; pools all runs into an "ALL" table
+python -m research.finding_identity summarize research/finding_identity/results/* --csv rates.csv
+```
+
+Clones go to `.repos/` and results to `results/<label>/`; both are git-ignored. Each result
+directory holds `summary.json`, `pairs.jsonl` (one line per commit pair) and, with
+`--details`, `findings.jsonl` (one line per compared finding with its true status and each
+scheme's outcome, for sampling and manual checks).
+
+Tests:
+
+```bash
+python -m pytest research/finding_identity/tests
+```
+
+## Known limitations
+
+- The ground truth is a line diff. A finding whose anchor line is rewritten beyond the
+  similarity threshold counts as removed plus introduced; check a sample by hand.
+- CodeGuard's fingerprint takes the statement from `str.splitlines()`, which also splits on
+  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.
diff --git a/research/finding_identity/__init__.py b/research/finding_identity/__init__.py
new file mode 100644
index 0000000..9c9b05b
--- /dev/null
+++ b/research/finding_identity/__init__.py
@@ -0,0 +1,2 @@
+"""Replay harness for the finding-identity study: how well do static-analysis findings
+keep their identity as code changes?  See README.md."""
diff --git a/research/finding_identity/__main__.py b/research/finding_identity/__main__.py
new file mode 100644
index 0000000..4e28416
--- /dev/null
+++ b/research/finding_identity/__main__.py
@@ -0,0 +1,3 @@
+from .cli import main
+
+main()
diff --git a/research/finding_identity/cli.py b/research/finding_identity/cli.py
new file mode 100644
index 0000000..ce0f51c
--- /dev/null
+++ b/research/finding_identity/cli.py
@@ -0,0 +1,64 @@
+"""Command line: ``python -m research.finding_identity {replay,summarize} ...``."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from .gitrepo import GitRepo
+from .replay import replay
+from .scan import Scanner, rule_ids_for
+from .schemes import ALL_SCHEMES
+from .summarize import summarize
+
+HERE = Path(__file__).parent
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(prog="python -m research.finding_identity")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    rp = sub.add_parser("replay", help="replay a repository's history")
+    rp.add_argument("repo", help="local repository path or a git URL to clone")
+    rp.add_argument("--ref", default="HEAD", help="branch or commit to walk back from")
+    rp.add_argument("--commits", type=int, default=200, help="number of commit pairs")
+    rp.add_argument("--label", help="name for the results (default: repository name)")
+    rp.add_argument("--out", type=Path, help="results directory")
+    rp.add_argument("--workdir", type=Path, default=HERE / ".repos", help="where to clone")
+    rp.add_argument(
+        "--categories",
+        default="security,performance,quality",
+        help="comma-separated rule categories to run",
+    )
+    rp.add_argument("--details", action="store_true", help="write one record per finding")
+
+    sp = sub.add_parser("summarize", help="print rate tables for one or more result dirs")
+    sp.add_argument("results", nargs="+", type=Path)
+    sp.add_argument("--csv", type=Path, help="also write the rows to this CSV file")
+
+    args = parser.parse_args(argv)
+    if args.command == "summarize":
+        summarize(args.results, args.csv)
+        return
+
+    ref = None if args.ref == "HEAD" else args.ref
+    repo = GitRepo.open_or_clone(args.repo, args.workdir, depth=args.commits + 2, ref=ref)
+    label = args.label or repo.path.name
+    out = args.out or HERE / "results" / label
+    categories = [c.strip() for c in args.categories.split(",") if c.strip()]
+    scanner = Scanner(rule_ids_for(categories), ALL_SCHEMES)
+    try:
+        replay(
+            repo,
+            ref=args.ref,
+            commits=args.commits,
+            scanner=scanner,
+            schemes=ALL_SCHEMES,
+            out_dir=out,
+            label=label,
+            categories=categories,
+            details=args.details,
+        )
+    finally:
+        repo.close()
+    summarize([out])
diff --git a/research/finding_identity/gitrepo.py b/research/finding_identity/gitrepo.py
new file mode 100644
index 0000000..5aaeafa
--- /dev/null
+++ b/research/finding_identity/gitrepo.py
@@ -0,0 +1,119 @@
+"""Minimal git access for the replay: history, per-commit changes, and blob contents."""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import IO
+
+_NULL_SHA = "0" * 40
+_SKIP_MODES = {"160000", "120000"}  # submodules and symlinks
+
+
+@dataclass(frozen=True)
+class Change:
+    """One file changed between two commits. A missing side is ``None``."""
+
+    status: str  # A, D, M, R (rename), C (copy), T (type change)
+    old_path: str | None
+    new_path: str | None
+    old_sha: str | None
+    new_sha: str | None
+
+
+class GitRepo:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self._batch: subprocess.Popen[bytes] | None = None
+
+    # -- construction -------------------------------------------------------
+
+    @classmethod
+    def open_or_clone(cls, spec: str, workdir: Path, *, depth: int, ref: str | None) -> GitRepo:
+        """Use *spec* as a local repository, or clone it (shallow) into *workdir*."""
+        local = Path(spec)
+        if local.is_dir() and (local / ".git").exists():
+            return cls(local)
+        name = spec.rstrip("/").removesuffix(".git").rsplit("/", 1)[-1]
+        target = workdir / name
+        if not (target / ".git").exists():
+            workdir.mkdir(parents=True, exist_ok=True)
+            args = ["git", "clone", "--quiet", "--no-tags", "--single-branch", f"--depth={depth}"]
+            if ref:
+                args += ["--branch", ref]
+            subprocess.run([*args, spec, str(target)], check=True)
+        return cls(target)
+
+    # -- queries ------------------------------------------------------------
+
+    def _git(self, *args: str) -> bytes:
+        return subprocess.run(
+            ["git", "-C", str(self.path), *args],
+            check=True,
+            capture_output=True,
+        ).stdout
+
+    def resolve(self, ref: str) -> str:
+        return self._git("rev-parse", "--verify", f"{ref}^{{commit}}").decode().strip()
+
+    def first_parent_history(self, ref: str, count: int) -> list[str]:
+        """The last *count* commits on *ref*'s first-parent line, oldest first."""
+        out = self._git("rev-list", "--first-parent", f"--max-count={count}", ref)
+        return list(reversed(out.decode().split()))
+
+    def changes(self, old: str, new: str) -> list[Change]:
+        raw = self._git("diff", "--raw", "-z", "-M", "--no-abbrev", old, new)
+        fields = raw.decode("utf-8", "surrogateescape").split("\0")
+        out: list[Change] = []
+        i = 0
+        while i < len(fields) and fields[i]:
+            meta = fields[i].lstrip(":").split()
+            old_mode, new_mode, old_sha, new_sha, status = meta[:5]
+            kind = status[0]
+            if kind in ("R", "C"):
+                old_path, new_path = fields[i + 1], fields[i + 2]
+                i += 3
+            else:
+                old_path = new_path = fields[i + 1]
+                i += 2
+            if old_mode in _SKIP_MODES or new_mode in _SKIP_MODES:
+                continue
+            out.append(
+                Change(
+                    status=kind,
+                    old_path=None if kind == "A" else old_path,
+                    new_path=None if kind == "D" else new_path,
+                    old_sha=None if old_sha == _NULL_SHA else old_sha,
+                    new_sha=None if new_sha == _NULL_SHA else new_sha,
+                )
+            )
+        return out
+
+    def blob(self, sha: str) -> bytes:
+        """Contents of blob *sha*, through one long-running ``git cat-file --batch``."""
+        if self._batch is None:
+            self._batch = subprocess.Popen(
+                ["git", "-C", str(self.path), "cat-file", "--batch"],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
+        stdin: IO[bytes] = self._batch.stdin  # type: ignore[assignment]
+        stdout: IO[bytes] = self._batch.stdout  # type: ignore[assignment]
+        stdin.write(sha.encode() + b"\n")
+        stdin.flush()
+        header = stdout.readline().decode().split()
+        if len(header) < 3 or header[1] != "blob":
+            raise KeyError(f"blob {sha} not found")
+        data = stdout.read(int(header[2]))
+        stdout.read(1)  # trailing newline
+        return data
+
+    def close(self) -> None:
+        if self._batch is not None:
+            if self._batch.stdin:
+                self._batch.stdin.close()
+            self._batch.wait()
+            if self._batch.stdout:
+                self._batch.stdout.close()
+            self._batch = None
diff --git a/research/finding_identity/linemap.py b/research/finding_identity/linemap.py
new file mode 100644
index 0000000..7215354
--- /dev/null
+++ b/research/finding_identity/linemap.py
@@ -0,0 +1,64 @@
+"""Map lines of an old file version to lines of the new version (the ground truth).
+
+Lines the diff keeps unchanged map exactly ("equal"). Inside a replaced block, old and
+new lines are paired when they are similar enough ("edited"): an order-preserving
+alignment that maximizes total similarity, counting only pairs at or above the
+threshold. Deleted lines and lines with no similar partner are not mapped.
+"""
+
+from __future__ import annotations
+
+import difflib
+from collections.abc import Sequence
+
+EDIT_THRESHOLD = 0.6
+#: Replaced blocks larger than this (old lines x new lines) are aligned position by
+#: position instead of with the full alignment, to keep the cost bounded.
+MAX_ALIGN_CELLS = 4000
+
+
+def similarity(a: str, b: str) -> float:
+    a, b = a.strip(), b.strip()
+    if not a and not b:
+        return 1.0
+    matcher = difflib.SequenceMatcher(None, a, b, autojunk=False)
+    if matcher.real_quick_ratio() < EDIT_THRESHOLD or matcher.quick_ratio() < EDIT_THRESHOLD:
+        return 0.0
+    return matcher.ratio()
+
+
+def _align(old: Sequence[str], new: Sequence[str]) -> list[tuple[int, int]]:
+    n, m = len(old), len(new)
+    if n * m > MAX_ALIGN_CELLS:
+        return [(k, k) for k in range(min(n, m)) if similarity(old[k], new[k]) >= EDIT_THRESHOLD]
+    sim = [[similarity(old[i], new[j]) for j in range(m)] for i in range(n)]
+    best = [[0.0] * (m + 1) for _ in range(n + 1)]
+    for i in range(n - 1, -1, -1):
+        for j in range(m - 1, -1, -1):
+            take = sim[i][j] + best[i + 1][j + 1] if sim[i][j] >= EDIT_THRESHOLD else -1.0
+            best[i][j] = max(take, best[i + 1][j], best[i][j + 1])
+    pairs: list[tuple[int, int]] = []
+    i = j = 0
+    while i < n and j < m:
+        if sim[i][j] >= EDIT_THRESHOLD and best[i][j] == sim[i][j] + best[i + 1][j + 1]:
+            pairs.append((i, j))
+            i, j = i + 1, j + 1
+        elif best[i][j] == best[i + 1][j]:
+            i += 1
+        else:
+            j += 1
+    return pairs
+
+
+def map_lines(old: Sequence[str], new: Sequence[str]) -> dict[int, tuple[int, str]]:
+    """``{old line: (new line, "equal" | "edited")}`` with 1-based line numbers."""
+    out: dict[int, tuple[int, str]] = {}
+    matcher = difflib.SequenceMatcher(None, old, new, autojunk=False)
+    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+        if tag == "equal":
+            for k in range(i2 - i1):
+                out[i1 + k + 1] = (j1 + k + 1, "equal")
+        elif tag == "replace":
+            for i, j in _align(old[i1:i2], new[j1:j2]):
+                out[i1 + i + 1] = (j1 + j + 1, "edited")
+    return out
diff --git a/research/finding_identity/matching.py b/research/finding_identity/matching.py
new file mode 100644
index 0000000..071a1d9
--- /dev/null
+++ b/research/finding_identity/matching.py
@@ -0,0 +1,152 @@
+"""Compare each identity scheme with the ground truth for one file across one change.
+
+Ground truth: an old finding and a new finding are the same when they have the same
+rule and the old finding's line maps (see :mod:`linemap`) to the new finding's line.
+Each new finding then has a true status, "known" (it existed before) or "new".
+
+A scheme predicts a status by matching keys: findings with equal keys are paired in
+line order. Comparing the two statuses gives, per new finding:
+
+- ``known_ok``     known, and the scheme paired it with its true predecessor
+- ``mispaired``    known, but the scheme paired it with a different old finding
+- ``false_new``    known, but the scheme reports it as new (noise, broken baselines)
+- ``false_known``  new, but the scheme matched it to an old finding (a hidden finding)
+- ``new_ok``       new, and the scheme reports it as new
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+
+OUTCOMES = ("known_ok", "mispaired", "false_new", "false_known", "new_ok")
+
+
+@dataclass(frozen=True)
+class Observation:
+    """One finding in one version of one file."""
+
+    rule: str
+    line: int
+    col: int
+    keys: dict[str, str] = field(hash=False, compare=False)
+
+
+@dataclass
+class FileResult:
+    """Outcome counts for one file across one change."""
+
+    gt: dict[str, int]
+    by_scheme: dict[str, dict[str, int]]
+    records: list[dict[str, object]]
+
+
+def ground_truth(
+    old: Sequence[Observation],
+    new: Sequence[Observation],
+    line_map: dict[int, tuple[int, str]],
+) -> dict[int, tuple[int, str]]:
+    """``{new index: (old index, "equal" | "edited")}`` for findings that persisted."""
+    by_rule_line: dict[tuple[str, int], list[int]] = defaultdict(list)
+    for j, n in enumerate(new):
+        by_rule_line[(n.rule, n.line)].append(j)
+    pairs: dict[int, tuple[int, str]] = {}
+    for i in sorted(range(len(old)), key=lambda k: (old[k].line, old[k].col)):
+        o = old[i]
+        mapped = line_map.get(o.line)
+        if mapped is None:
+            continue
+        candidates = [j for j in by_rule_line.get((o.rule, mapped[0]), []) if j not in pairs]
+        if candidates:
+            j = min(candidates, key=lambda k: abs(new[k].col - o.col))
+            pairs[j] = (i, mapped[1])
+    return pairs
+
+
+def scheme_pairs(
+    old: Sequence[Observation], new: Sequence[Observation], scheme: str
+) -> dict[int, int]:
+    """``{new index: old index}`` as the scheme would match them."""
+    old_by_key: dict[str, list[int]] = defaultdict(list)
+    new_by_key: dict[str, list[int]] = defaultdict(list)
+    for i, o in enumerate(old):
+        old_by_key[o.keys[scheme]].append(i)
+    for j, n in enumerate(new):
+        new_by_key[n.keys[scheme]].append(j)
+    pairs: dict[int, int] = {}
+    for key, new_idx in new_by_key.items():
+        old_idx = sorted(old_by_key.get(key, []), key=lambda k: (old[k].line, old[k].col))
+        new_sorted = sorted(new_idx, key=lambda k: (new[k].line, new[k].col))
+        for i, j in zip(old_idx, new_sorted, strict=False):
+            pairs[j] = i
+    return pairs
+
+
+def evaluate(
+    old: Sequence[Observation],
+    new: Sequence[Observation],
+    line_map: dict[int, tuple[int, str]],
+    schemes: Sequence[str],
+    *,
+    keep_records: bool = False,
+) -> FileResult:
+    gt_pairs = ground_truth(old, new, line_map)
+    gt = {
+        "old": len(old),
+        "new": len(new),
+        "known": len(gt_pairs),
+        "known_equal": sum(1 for _, kind in gt_pairs.values() if kind == "equal"),
+        "known_edited": sum(1 for _, kind in gt_pairs.values() if kind == "edited"),
+        "introduced": len(new) - len(gt_pairs),
+        "removed": len(old) - len(gt_pairs),
+    }
+    by_scheme: dict[str, dict[str, int]] = {}
+    statuses: dict[str, list[str]] = {}
+    for scheme in schemes:
+        predicted = scheme_pairs(old, new, scheme)
+        counts = dict.fromkeys(OUTCOMES, 0)
+        counts["false_new_equal"] = counts["false_new_edited"] = 0
+        per_new: list[str] = []
+        for j in range(len(new)):
+            truth = gt_pairs.get(j)
+            guess = predicted.get(j)
+            if truth is None:
+                outcome = "new_ok" if guess is None else "false_known"
+            elif guess is None:
+                outcome = "false_new"
+                counts[f"false_new_{truth[1]}"] += 1
+            else:
+                outcome = "known_ok" if guess == truth[0] else "mispaired"
+            counts[outcome] += 1
+            per_new.append(outcome)
+        by_scheme[scheme] = counts
+        statuses[scheme] = per_new
+
+    records: list[dict[str, object]] = []
+    if keep_records:
+        for j, n in enumerate(new):
+            truth = gt_pairs.get(j)
+            records.append(
+                {
+                    "rule": n.rule,
+                    "line": n.line,
+                    "col": n.col,
+                    "truth": "new" if truth is None else "known",
+                    "truth_kind": None if truth is None else truth[1],
+                    "old_line": None if truth is None else old[truth[0]].line,
+                    "outcomes": {s: statuses[s][j] for s in schemes},
+                }
+            )
+    return FileResult(gt, by_scheme, records)
+
+
+def collisions(observations: Sequence[Observation], schemes: Sequence[str]) -> dict[str, int]:
+    """Per scheme, how many findings share their key with another finding in the same file."""
+    out: dict[str, int] = {}
+    for scheme in schemes:
+        counts: dict[str, int] = defaultdict(int)
+        for o in observations:
+            counts[o.keys[scheme]] += 1
+        out[scheme] = sum(c for c in counts.values() if c > 1)
+    return out
diff --git a/research/finding_identity/replay.py b/research/finding_identity/replay.py
new file mode 100644
index 0000000..3cfe047
--- /dev/null
+++ b/research/finding_identity/replay.py
@@ -0,0 +1,152 @@
+"""Replay a repository's history and score every identity scheme against the ground truth.
+
+For each consecutive pair of commits on the first-parent line, every changed source file
+is scanned before and after the change. Unchanged files are skipped: every scheme
+trivially keeps their findings, so they would only inflate the counts.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import time
+from collections.abc import Sequence
+from datetime import datetime, timezone
+from itertools import pairwise
+from pathlib import Path
+from typing import Any
+
+import codeguard
+
+from .gitrepo import Change, GitRepo
+from .linemap import map_lines
+from .matching import OUTCOMES, evaluate
+from .scan import Scanner, is_candidate
+
+GT_KEYS = ("old", "new", "known", "known_equal", "known_edited", "introduced", "removed")
+SCHEME_KEYS = (*OUTCOMES, "false_new_equal", "false_new_edited")
+
+
+def _empty_totals(schemes: Sequence[str]) -> dict[str, Any]:
+    return {
+        "gt": dict.fromkeys(GT_KEYS, 0),
+        "schemes": {s: dict.fromkeys(SCHEME_KEYS, 0) for s in schemes},
+        "files": 0,
+        "renamed_files": 0,
+        "skipped_files": 0,
+    }
+
+
+def _add(totals: dict[str, Any], other: dict[str, Any]) -> None:
+    for k, v in other["gt"].items():
+        totals["gt"][k] += v
+    for s, counts in other["schemes"].items():
+        for k, v in counts.items():
+            totals["schemes"][s][k] += v
+    for k in ("files", "renamed_files", "skipped_files"):
+        totals[k] += other[k]
+
+
+def _replay_change(
+    repo: GitRepo,
+    scanner: Scanner,
+    ch: Change,
+    schemes: Sequence[str],
+    totals: dict[str, Any],
+    records: list[dict[str, Any]] | None,
+) -> None:
+    old_ok, new_ok = is_candidate(ch.old_path), is_candidate(ch.new_path)
+    if not (old_ok or new_ok):
+        return
+    old_res = new_res = None
+    if old_ok and ch.old_sha and ch.old_path:
+        old_res = scanner.scan(ch.old_sha, ch.old_path, repo.blob(ch.old_sha))
+        if old_res is None:
+            totals["skipped_files"] += 1
+            return
+    if new_ok and ch.new_sha and ch.new_path:
+        new_res = scanner.scan(ch.new_sha, ch.new_path, repo.blob(ch.new_sha))
+        if new_res is None:
+            totals["skipped_files"] += 1
+            return
+    old_obs = old_res.observations if old_res else []
+    new_obs = new_res.observations if new_res else []
+    if not old_obs and not new_obs:
+        return
+    line_map = map_lines(old_res.lines, new_res.lines) if old_res and new_res else {}
+    result = evaluate(old_obs, new_obs, line_map, schemes, keep_records=records is not None)
+    for k, v in result.gt.items():
+        totals["gt"][k] += v
+    for s, counts in result.by_scheme.items():
+        for k, v in counts.items():
+            totals["schemes"][s][k] += v
+    totals["files"] += 1
+    if ch.status == "R":
+        totals["renamed_files"] += 1
+    if records is not None:
+        for rec in result.records:
+            rec.update(path=ch.new_path, old_path=ch.old_path, status=ch.status)
+            records.append(rec)
+
+
+def replay(
+    repo: GitRepo,
+    *,
+    ref: str,
+    commits: int,
+    scanner: Scanner,
+    schemes: Sequence[str],
+    out_dir: Path,
+    label: str,
+    categories: Sequence[str],
+    details: bool = False,
+) -> dict[str, Any]:
+    out_dir.mkdir(parents=True, exist_ok=True)
+    history = repo.first_parent_history(ref, commits + 1)
+    head = repo.resolve(ref)
+    meta = {
+        "label": label,
+        "repo": str(repo.path),
+        "ref": ref,
+        "head": head,
+        "commit_pairs": max(len(history) - 1, 0),
+        "categories": list(categories),
+        "schemes": list(schemes),
+        "codeguard_version": codeguard.__version__,
+        "created": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+    }
+    totals = _empty_totals(schemes)
+    started = time.monotonic()
+    with (
+        (out_dir / "pairs.jsonl").open("w", encoding="utf-8") as pairs_out,
+        (out_dir / "findings.jsonl").open("w", encoding="utf-8") as findings_out,
+    ):
+        for n, (old, new) in enumerate(pairwise(history), start=1):
+            pair_totals = _empty_totals(schemes)
+            records: list[dict[str, Any]] | None = [] if details else None
+            for ch in repo.changes(old, new):
+                _replay_change(repo, scanner, ch, schemes, pair_totals, records)
+            _add(totals, pair_totals)
+            pairs_out.write(json.dumps({"old": old, "new": new, **pair_totals}) + "\n")
+            for rec in records or []:
+                findings_out.write(json.dumps({"commit": new, **rec}) + "\n")
+            if n % 25 == 0 or n == meta["commit_pairs"]:
+                elapsed = time.monotonic() - started
+                print(
+                    f"[{label}] {n}/{meta['commit_pairs']} commit pairs, "
+                    f"{totals['gt']['new']} findings compared, {elapsed:.0f}s",
+                    file=sys.stderr,
+                )
+    meta["elapsed_seconds"] = round(time.monotonic() - started, 1)
+    meta["versions_scanned"] = scanner.versions_scanned
+    meta["parse_failures"] = scanner.parse_failures
+    summary = {
+        **meta,
+        "totals": totals,
+        "collisions": {
+            "observations": scanner.observations,
+            "by_scheme": scanner.collision_totals,
+        },
+    }
+    (out_dir / "summary.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
+    return summary
diff --git a/research/finding_identity/scan.py b/research/finding_identity/scan.py
new file mode 100644
index 0000000..8733cf1
--- /dev/null
+++ b/research/finding_identity/scan.py
@@ -0,0 +1,91 @@
+"""Run CodeGuard on one file version and attach every scheme's key to each finding."""
+
+from __future__ import annotations
+
+import warnings
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass
+
+import codeguard.rules  # noqa: F401  (registers the built-in rules)
+from codeguard.engine.registry import REGISTRY
+from codeguard.engine.runner import AnalysisRunner
+from codeguard.lang.base import Language
+from codeguard.lang.registry import language_for_path
+
+from .matching import Observation, collisions
+from .schemes import FileContext, keys_for
+
+#: Paths that are vendored, generated or minified, and so not worth tracking.
+_SKIP_PARTS = ("node_modules/", "vendor/", "/dist/", "/build/", "site-packages/")
+_SKIP_SUFFIXES = (".min.js", ".d.ts", ".bundle.js")
+MAX_BYTES = 512 * 1024
+
+
+def rule_ids_for(categories: Iterable[str]) -> list[str]:
+    wanted = set(categories)
+    return [r.id for r in REGISTRY.all() if r.category.value in wanted]
+
+
+def is_candidate(path: str | None) -> bool:
+    if path is None or language_for_path(path) is None:
+        return False
+    padded = "/" + path
+    return not any(p in padded for p in _SKIP_PARTS) and not path.endswith(_SKIP_SUFFIXES)
+
+
+@dataclass
+class ScanResult:
+    observations: list[Observation]
+    lines: list[str]
+    collisions: dict[str, int]
+
+
+class Scanner:
+    """Scans file versions, caching by (blob, path) so a version is analysed once."""
+
+    def __init__(self, rule_ids: list[str], schemes: Sequence[str]) -> None:
+        self._runner = AnalysisRunner(rule_ids=rule_ids)
+        self._schemes = tuple(schemes)
+        self._cache: dict[tuple[str, str], ScanResult | None] = {}
+        self.versions_scanned = 0
+        self.parse_failures = 0
+        self.observations = 0
+        self.collision_totals = dict.fromkeys(self._schemes, 0)
+
+    def scan(self, cache_key: str, path: str, data: bytes) -> ScanResult | None:
+        """Findings for *path* with contents *data*; ``None`` if it cannot be analysed."""
+        key = (cache_key, path)
+        if key in self._cache:
+            return self._cache[key]
+        result = self._scan(path, data)
+        self._cache[key] = result
+        if result is not None:
+            self.versions_scanned += 1
+            self.observations += len(result.observations)
+            for scheme, n in result.collisions.items():
+                self.collision_totals[scheme] += n
+        return result
+
+    def _scan(self, path: str, data: bytes) -> ScanResult | None:
+        lang = language_for_path(path)
+        if lang is None or len(data) > MAX_BYTES:
+            return None
+        text = data.decode("utf-8", errors="replace")
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                findings = self._runner.run(text, filename=path, language=lang)
+        except (SyntaxError, ValueError, RecursionError):
+            self.parse_failures += 1
+            return None
+        ctx = FileContext.build(path, text, is_python=lang is Language.PYTHON)
+        observations = [
+            Observation(
+                rule=f.rule_id,
+                line=f.location.line,
+                col=f.location.col,
+                keys=keys_for(f.rule_id, f.location.line, ctx, self._schemes),
+            )
+            for f in findings
+        ]
+        return ScanResult(observations, ctx.lines, collisions(observations, self._schemes))
diff --git a/research/finding_identity/schemes.py b/research/finding_identity/schemes.py
new file mode 100644
index 0000000..006e2db
--- /dev/null
+++ b/research/finding_identity/schemes.py
@@ -0,0 +1,184 @@
+"""Identity schemes: ways of deciding that two findings are "the same finding".
+
+A scheme maps one finding in one file version to a key. Two findings in consecutive
+versions of a file are matched when their keys are equal. Every scheme includes the
+rule ID and the file path, so a finding never matches across rules or across files.
+
+| Scheme        | Key built from                                                     |
+|---------------|--------------------------------------------------------------------|
+| line          | line number                                                        |
+| line_text     | exact text of the flagged line                                     |
+| github        | GitHub code scanning ``primaryLocationLineHash`` (ported below)    |
+| codeguard     | enclosing scope + normalized statement (CodeGuard ``codeguard/v1``)|
+| cg_no_scope   | ablation: codeguard without the enclosing scope                    |
+| cg_no_mask    | ablation: codeguard without masking string and number literals    |
+| cg_raw        | ablation: codeguard without any normalization (scope + raw line)  |
+"""
+
+from __future__ import annotations
+
+import ast
+import hashlib
+import sys
+from array import array
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from codeguard.engine import fingerprint as cg_fp
+
+PRIMARY_SCHEMES = ("line", "line_text", "github", "codeguard")
+ABLATION_SCHEMES = ("cg_no_scope", "cg_no_mask", "cg_raw")
+ALL_SCHEMES = PRIMARY_SCHEMES + ABLATION_SCHEMES
+
+
+# ---------------------------------------------------------------------------
+# GitHub code scanning line hash
+# ---------------------------------------------------------------------------
+# Port of hash() in github/codeql-action src/fingerprints.ts (MIT License,
+# Copyright GitHub, Inc.). For every line it computes a rolling hash over the next
+# 100 characters, ignoring spaces and tabs, starting at the beginning of the line and
+# running into the following lines. CR and CRLF are normalized to LF. Arithmetic is
+# 64-bit wrapping (the original uses the `long` package). The value is
+# "<hex>:<n>", where n counts how many times the same hash occurred earlier in the
+# file. JavaScript strings are UTF-16, so the port walks UTF-16 code units.
+
+_BLOCK_SIZE = 100
+_MOD = 37
+_MASK64 = (1 << 64) - 1
+_FIRST_MOD = pow(_MOD, _BLOCK_SIZE, 1 << 64)
+_TAB, _SPACE, _LF, _CR, _EOF = 9, 32, 10, 13, 65535
+
+
+def _utf16_units(text: str) -> array[int]:
+    units = array("H")
+    units.frombytes(text.encode("utf-16-le", "surrogatepass"))
+    if sys.byteorder == "big":
+        units.byteswap()
+    return units
+
+
+def github_line_hashes(text: str) -> dict[int, str]:
+    """Return ``{line number: primaryLocationLineHash}`` for every line of *text*."""
+    window = [0] * _BLOCK_SIZE
+    line_numbers = [-1] * _BLOCK_SIZE
+    out: dict[int, str] = {}
+    counts: dict[str, int] = {}
+    hash_raw = 0
+    index = 0
+    line_number = 0
+    line_start = True
+    prev_cr = False
+
+    def output_hash() -> None:
+        value = format(hash_raw, "x")
+        counts[value] = counts.get(value, 0) + 1
+        out[line_numbers[index]] = f"{value}:{counts[value]}"
+        line_numbers[index] = -1
+
+    def update_hash(current: int) -> None:
+        nonlocal hash_raw, index
+        begin = window[index]
+        window[index] = current
+        hash_raw = (_MOD * hash_raw + current - _FIRST_MOD * begin) & _MASK64
+        index = (index + 1) % _BLOCK_SIZE
+
+    def process(current: int) -> None:
+        nonlocal prev_cr, line_start, line_number
+        if current in (_SPACE, _TAB) or (prev_cr and current == _LF):
+            prev_cr = False
+            return
+        if current == _CR:
+            current = _LF
+            prev_cr = True
+        else:
+            prev_cr = False
+        if line_numbers[index] != -1:
+            output_hash()
+        if line_start:
+            line_start = False
+            line_number += 1
+            line_numbers[index] = line_number
+        if current == _LF:
+            line_start = True
+        update_hash(current)
+
+    for unit in _utf16_units(text):
+        process(unit)
+    process(_EOF)
+    for _ in range(_BLOCK_SIZE):
+        if line_numbers[index] != -1:
+            output_hash()
+        update_hash(0)
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Per-file context and key computation
+# ---------------------------------------------------------------------------
+
+
+def split_lines(text: str) -> list[str]:
+    """Split on LF, CRLF and CR only: the line breaks parsers count for line numbers."""
+    return text.replace("\r\n", "\n").replace("\r", "\n").split("\n")
+
+
+def _sha(*parts: str) -> str:
+    return hashlib.sha256("\0".join(parts).encode("utf-8")).hexdigest()[:16]
+
+
+def _strip_comments_and_space(text: str) -> str:
+    # CodeGuard's normalization without the literal masking step. The regexes are
+    # CodeGuard's own, so the ablation stays in sync with the tool.
+    return str(cg_fp._WS_RE.sub("", cg_fp._COMMENT_RE.sub("", text)))
+
+
+@dataclass
+class FileContext:
+    """What every scheme may read about one version of one file."""
+
+    path: str
+    text: str
+    lines: list[str]
+    tree: ast.Module | None
+    github: dict[int, str]
+
+    @classmethod
+    def build(cls, path: str, text: str, *, is_python: bool) -> FileContext:
+        tree = None
+        if is_python:
+            try:
+                tree = ast.parse(text)
+            except (SyntaxError, ValueError):
+                tree = None
+        return cls(path, text, split_lines(text), tree, github_line_hashes(text))
+
+    def raw_line(self, line: int) -> str:
+        return self.lines[line - 1] if 1 <= line <= len(self.lines) else ""
+
+    def scope(self, line: int) -> str:
+        return cg_fp.python_scope(self.tree, line) if self.tree is not None else ""
+
+
+def keys_for(rule: str, line: int, ctx: FileContext, schemes: Sequence[str]) -> dict[str, str]:
+    """The key of one finding under each of *schemes*."""
+    raw = ctx.raw_line(line)
+    scope = ctx.scope(line)
+    out: dict[str, str] = {}
+    for name in schemes:
+        if name == "line":
+            out[name] = _sha(rule, ctx.path, str(line))
+        elif name == "line_text":
+            out[name] = _sha(rule, ctx.path, raw)
+        elif name == "github":
+            out[name] = _sha(rule, ctx.path, ctx.github.get(line, ""))
+        elif name == "codeguard":
+            out[name] = cg_fp.compute(rule, ctx.path, ctx.text, line, scope=scope)
+        elif name == "cg_no_scope":
+            out[name] = cg_fp.compute(rule, ctx.path, ctx.text, line, scope="")
+        elif name == "cg_no_mask":
+            out[name] = _sha(rule, ctx.path, scope, _strip_comments_and_space(raw))
+        elif name == "cg_raw":
+            out[name] = _sha(rule, ctx.path, scope, raw)
+        else:
+            raise ValueError(f"unknown scheme {name!r}")
+    return out
diff --git a/research/finding_identity/summarize.py b/research/finding_identity/summarize.py
new file mode 100644
index 0000000..b6e6010
--- /dev/null
+++ b/research/finding_identity/summarize.py
@@ -0,0 +1,113 @@
+"""Turn one or more ``summary.json`` files into the rate table the paper reports.
+
+Rates, per scheme:
+
+- false-new rate    false_new / known        persisting findings reported as new
+- hidden rate       false_known / introduced genuinely new findings the scheme would hide
+- mispair rate      mispaired / known        persisting findings tied to the wrong predecessor
+- collision rate    colliding / observations findings sharing a key within one file version
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any
+
+
+def _rate(num: int, den: int) -> float | None:
+    return num / den if den else None
+
+
+def rates(summary: dict[str, Any]) -> list[dict[str, Any]]:
+    gt = summary["totals"]["gt"]
+    obs = summary["collisions"]["observations"]
+    rows = []
+    for scheme, c in summary["totals"]["schemes"].items():
+        rows.append(
+            {
+                "label": summary.get("label", ""),
+                "scheme": scheme,
+                "known": gt["known"],
+                "introduced": gt["introduced"],
+                "false_new": c["false_new"],
+                "false_new_rate": _rate(c["false_new"], gt["known"]),
+                "false_new_equal_rate": _rate(c["false_new_equal"], gt["known_equal"]),
+                "false_new_edited_rate": _rate(c["false_new_edited"], gt["known_edited"]),
+                "false_known": c["false_known"],
+                "hidden_rate": _rate(c["false_known"], gt["introduced"]),
+                "mispair_rate": _rate(c["mispaired"], gt["known"]),
+                "collision_rate": _rate(summary["collisions"]["by_scheme"][scheme], obs),
+            }
+        )
+    return rows
+
+
+def merge(summaries: Sequence[dict[str, Any]]) -> dict[str, Any]:
+    """Pool several repositories into one summary (sums of counts, not averages of rates)."""
+    first = summaries[0]
+    merged: dict[str, Any] = {
+        "label": "ALL",
+        "totals": json.loads(json.dumps(first["totals"])),
+        "collisions": json.loads(json.dumps(first["collisions"])),
+    }
+    for s in summaries[1:]:
+        for k, v in s["totals"]["gt"].items():
+            merged["totals"]["gt"][k] += v
+        for scheme, counts in s["totals"]["schemes"].items():
+            for k, v in counts.items():
+                merged["totals"]["schemes"][scheme][k] += v
+        merged["collisions"]["observations"] += s["collisions"]["observations"]
+        for scheme, v in s["collisions"]["by_scheme"].items():
+            merged["collisions"]["by_scheme"][scheme] += v
+    return merged
+
+
+def _pct(x: float | None) -> str:
+    return "   n/a" if x is None else f"{100 * x:6.2f}"
+
+
+def print_table(rows: Sequence[dict[str, Any]]) -> None:
+    head = (
+        f"{'scheme':<12} {'false-new %':>11} {'(moved)':>8} {'(edited)':>9} "
+        f"{'hidden %':>9} {'mispair %':>10} {'collide %':>10}"
+    )
+    print(head)
+    print("-" * len(head))
+    for r in rows:
+        print(
+            f"{r['scheme']:<12} {_pct(r['false_new_rate']):>11} "
+            f"{_pct(r['false_new_equal_rate']):>8} {_pct(r['false_new_edited_rate']):>9} "
+            f"{_pct(r['hidden_rate']):>9} {_pct(r['mispair_rate']):>10} "
+            f"{_pct(r['collision_rate']):>10}"
+        )
+
+
+def summarize(paths: Sequence[Path], csv_out: Path | None = None) -> None:
+    summaries = [json.loads((p / "summary.json").read_text(encoding="utf-8")) for p in paths]
+    all_rows: list[dict[str, Any]] = []
+    for s in summaries:
+        gt = s["totals"]["gt"]
+        print(
+            f"\n{s['label']}: {s['commit_pairs']} commit pairs, {s['totals']['files']} file "
+            f"changes with findings, {gt['known']} persisting findings "
+            f"({gt['known_equal']} moved, {gt['known_edited']} edited), "
+            f"{gt['introduced']} introduced"
+        )
+        rows = rates(s)
+        print_table(rows)
+        all_rows += rows
+    if len(summaries) > 1:
+        merged = merge(summaries)
+        gt = merged["totals"]["gt"]
+        print(f"\nALL: {gt['known']} persisting findings, {gt['introduced']} introduced")
+        rows = rates(merged)
+        print_table(rows)
+        all_rows += rows
+    if csv_out is not None:
+        with csv_out.open("w", newline="", encoding="utf-8") as fh:
+            writer = csv.DictWriter(fh, fieldnames=list(all_rows[0]))
+            writer.writeheader()
+            writer.writerows(all_rows)
diff --git a/research/finding_identity/tests/__init__.py b/research/finding_identity/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/research/finding_identity/tests/test_matching.py b/research/finding_identity/tests/test_matching.py
new file mode 100644
index 0000000..1efbefa
--- /dev/null
+++ b/research/finding_identity/tests/test_matching.py
@@ -0,0 +1,40 @@
+from research.finding_identity.linemap import map_lines
+from research.finding_identity.matching import Observation, evaluate, ground_truth
+from research.finding_identity.schemes import ALL_SCHEMES, FileContext, keys_for, split_lines
+
+
+def test_map_lines_insertion_and_edit() -> None:
+    old = ["a = 1", "b = eval(x)", "c = 3"]
+    new = ["import os", "a = 1", "b = eval(y)", "c = 3"]
+    mapping = map_lines(old, new)
+    assert mapping[1] == (2, "equal")
+    assert mapping[2] == (3, "edited")
+    assert mapping[3] == (4, "equal")
+
+
+def test_map_lines_unrelated_replacement_is_unmapped() -> None:
+    mapping = map_lines(["x = eval(data)"], ["print('completely different text here')"])
+    assert 1 not in mapping
+
+
+def _obs(text: str, line: int, rule: str = "R") -> Observation:
+    ctx = FileContext.build("f.py", text, is_python=True)
+    return Observation(rule, line, 5, keys_for(rule, line, ctx, ALL_SCHEMES))
+
+
+def test_line_scheme_hides_a_new_finding_and_loses_the_old_one() -> None:
+    old_text = "def f(x):\n    eval(x)\n"
+    # Two lines inserted above; a *different* new finding lands on the old line 2.
+    new_text = "def f(x):\n    eval(y)\n    pass\n    eval(x)\n"
+    old = [_obs(old_text, 2)]
+    new = [_obs(new_text, 2), _obs(new_text, 4)]
+    line_map = map_lines(split_lines(old_text), split_lines(new_text))
+    truth = ground_truth(old, new, line_map)
+    assert truth == {1: (0, "equal")}  # eval(x) moved from line 2 to line 4
+
+    result = evaluate(old, new, line_map, ["line", "codeguard"])
+    assert result.gt["known"] == 1 and result.gt["introduced"] == 1
+    assert result.by_scheme["line"]["false_known"] == 1  # eval(y) hidden
+    assert result.by_scheme["line"]["false_new"] == 1  # eval(x) reported as new
+    assert result.by_scheme["codeguard"]["known_ok"] == 1
+    assert result.by_scheme["codeguard"]["new_ok"] == 1
diff --git a/research/finding_identity/tests/test_replay.py b/research/finding_identity/tests/test_replay.py
new file mode 100644
index 0000000..12b86cd
--- /dev/null
+++ b/research/finding_identity/tests/test_replay.py
@@ -0,0 +1,71 @@
+"""End to end: build a tiny repository and replay it."""
+
+import os
+import subprocess
+from pathlib import Path
+
+from research.finding_identity.gitrepo import GitRepo
+from research.finding_identity.replay import replay
+from research.finding_identity.scan import Scanner
+from research.finding_identity.schemes import ALL_SCHEMES
+
+ENV = {
+    **os.environ,
+    "GIT_AUTHOR_NAME": "t",
+    "GIT_AUTHOR_EMAIL": "t@example.com",
+    "GIT_COMMITTER_NAME": "t",
+    "GIT_COMMITTER_EMAIL": "t@example.com",
+}
+
+V1 = "def handler(expr):\n    return eval(expr)\n"
+# Lines inserted above: line numbers shift, nothing else changes.
+V2 = "import os\n\n\ndef handler(expr):\n    return eval(expr)\n"
+# Function renamed: the line is untouched but its enclosing scope changes.
+V3 = "import os\n\n\ndef run(expr):\n    return eval(expr)\n"
+
+
+def _git(repo: Path, *args: str) -> None:
+    subprocess.run(["git", "-C", str(repo), *args], check=True, env=ENV, capture_output=True)
+
+
+def _commit(repo: Path, text: str, msg: str) -> None:
+    (repo / "app.py").write_text(text)
+    _git(repo, "add", "app.py")
+    _git(repo, "commit", "-q", "-m", msg)
+
+
+def test_replay_scores_schemes(tmp_path: Path) -> None:
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    _git(repo_dir, "init", "-q")
+    for text, msg in ((V1, "v1"), (V2, "v2"), (V3, "v3")):
+        _commit(repo_dir, text, msg)
+
+    repo = GitRepo(repo_dir)
+    try:
+        summary = replay(
+            repo,
+            ref="HEAD",
+            commits=2,
+            scanner=Scanner(["CG-SEC-003"], ALL_SCHEMES),
+            schemes=ALL_SCHEMES,
+            out_dir=tmp_path / "out",
+            label="tiny",
+            categories=["security"],
+            details=True,
+        )
+    finally:
+        repo.close()
+
+    gt = summary["totals"]["gt"]
+    assert gt["known"] == 2 and gt["introduced"] == 0
+    schemes = summary["totals"]["schemes"]
+    # v1 -> v2 (move): only the line-number scheme loses the finding.
+    # v2 -> v3 (rename): only the scope-based schemes lose it.
+    assert schemes["line"]["false_new"] == 1
+    assert schemes["codeguard"]["false_new"] == 1
+    assert schemes["cg_no_scope"]["false_new"] == 0
+    assert schemes["line_text"]["false_new"] == 0
+    assert (tmp_path / "out" / "summary.json").exists()
+    records = (tmp_path / "out" / "findings.jsonl").read_text().splitlines()
+    assert len(records) == 2
diff --git a/research/finding_identity/tests/test_schemes.py b/research/finding_identity/tests/test_schemes.py
new file mode 100644
index 0000000..c0e72ce
--- /dev/null
+++ b/research/finding_identity/tests/test_schemes.py
@@ -0,0 +1,77 @@
+from codeguard.engine.runner import AnalysisRunner
+from codeguard.lang.base import Language
+from research.finding_identity.schemes import (
+    ALL_SCHEMES,
+    FileContext,
+    github_line_hashes,
+    keys_for,
+)
+
+# Expected values were produced by running GitHub's own fingerprints.ts (codeql-action,
+# with the `long` package) on the same inputs.
+GITHUB_A = "import os\n\ndef f(x):\n\treturn eval(x)   # bad\n\nx = 1\nx = 1\nx = 1\n"
+GITHUB_A_EXPECTED = {
+    1: "420aefba3f9eb175:1",
+    2: "e65a2d6470d3f9a6:1",
+    3: "52e8697e4f828294:1",
+    4: "2abbbe6646aa285e:1",
+    5: "b792415f5b7fbcd:1",
+    6: "b06211268476d237:1",
+    7: "87c133c3324ef59b:1",
+    8: "8e00d41f0ebba6bf:1",
+    9: "c129715d7a2bc9a3:1",
+}
+GITHUB_B = "line one\r\nline two\r\n\r\n  indented\rcr-only\n"
+GITHUB_B_EXPECTED = {
+    1: "4ec050cce04de77c:1",
+    2: "b58a469adea14580:1",
+    3: "70d68996d679060:1",
+    4: "ccff826d0d74b76:1",
+    5: "e0be3c354f7b73a5:1",
+    6: "c129715d7a2bc9a3:1",
+}
+
+
+def test_github_hash_matches_reference() -> None:
+    assert github_line_hashes(GITHUB_A) == GITHUB_A_EXPECTED
+    assert github_line_hashes(GITHUB_B) == GITHUB_B_EXPECTED
+
+
+def test_github_hash_counts_duplicates() -> None:
+    # Short identical lines at the end of a file hash identically and get :1, :2, ...
+    hashes = github_line_hashes("a\n" * 120)
+    assert hashes[1].endswith(":1")
+    assert len({h.split(":")[0] for h in list(hashes.values())[:20]}) == 1
+    assert hashes[2].endswith(":2")
+
+
+SOURCE = "def handler(expr):\n    result = eval(expr)\n    return result\n"
+
+
+def test_codeguard_key_equals_runner_fingerprint() -> None:
+    import codeguard.rules  # noqa: F401
+
+    findings = AnalysisRunner(rule_ids=["CG-SEC-003"]).run(
+        SOURCE, filename="app/views.py", language=Language.PYTHON
+    )
+    assert len(findings) == 1
+    f = findings[0]
+    ctx = FileContext.build("app/views.py", SOURCE, is_python=True)
+    keys = keys_for(f.rule_id, f.location.line, ctx, ALL_SCHEMES)
+    assert keys["codeguard"] == f.fingerprint
+    # With no literals on the line, masking changes nothing: the two keys coincide.
+    assert keys["cg_no_mask"] == keys["codeguard"]
+    others = [keys[s] for s in ALL_SCHEMES if s != "cg_no_mask"]
+    assert len(set(others)) == len(others)
+
+
+def test_codeguard_key_ignores_literals_and_spacing() -> None:
+    a = FileContext.build("m.py", "def f():\n    q = run('a', 1)\n", is_python=True)
+    b = FileContext.build("m.py", 'def f():\n    q =  run("b", 2)  # x\n', is_python=True)
+    ka = keys_for("R", 2, a, ALL_SCHEMES)
+    kb = keys_for("R", 2, b, ALL_SCHEMES)
+    assert ka["codeguard"] == kb["codeguard"]
+    assert ka["cg_no_scope"] == kb["cg_no_scope"]
+    assert ka["cg_no_mask"] != kb["cg_no_mask"]
+    assert ka["line_text"] != kb["line_text"]
+    assert ka["line"] == kb["line"]