fix(replay-guard): block targeted base-work reverts and test regressions regardless of size

scripts/ci/pr_head_replay_guard.py

@@ -8,10 +8,21 @@
 
 This guard inspects the PR head history, finds the newest merge commit on the
 first-parent path from the supplied base, and evaluates commits made after that
-merge.  It blocks an exact replay of any pre-merge first-parent tree.  It also
-blocks a conservative bulk-regression signature: at least five tracked files and
-500 lines removed, with deletions at least four times additions.  Every decision
-is printed with the exact SHAs and diff counts so the failure is actionable.
+merge.  It blocks, regardless of diff size:
+
+- an exact replay of any pre-merge first-parent tree;
+- targeted unmerges of base work: any path whose post-merge content reverted
+  exactly to its pre-merge first-parent content, discarding what the base
+  merge brought in (observed in appguardrail#297, where a stale snapshot
+  reverted an accessibility wrapper and deleted its regression tests in a
+  push far below the bulk thresholds);
+- test regression without replacement: post-merge commits deleting or
+  shrinking test files while adding no new test file anywhere in the push;
+- the conservative bulk-regression signature: at least five tracked files and
+  500 lines removed, with deletions at least four times additions.
+
+Every decision is printed with the exact SHAs, diff counts, and offending
+paths so the failure is actionable.
 """
 
 from __future__ import annotations
@@ -27,6 +38,8 @@
 MIN_REMOVED_FILES = 5
 MIN_DELETED_LINES = 500
 MIN_DELETION_RATIO = 4
+MAX_LISTED_PATHS = 10
+TEST_DIR_SEGMENTS = frozenset({"tests", "test", "__tests__", "spec", "specs"})
 
 
 @dataclass(frozen=True)
@@ -41,6 +54,9 @@ class ReplayEvidence:
     removed_files: int = 0
     added_lines: int = 0
     deleted_lines: int = 0
+    unmerged_paths: tuple[str, ...] = ()
+    regressed_test_paths: tuple[str, ...] = ()
+    added_test_files: int = 0
 
     @property
     def suspicious_bulk_regression(self) -> bool:
@@ -51,10 +67,25 @@ def suspicious_bulk_regression(self) -> bool:
             and self.deleted_lines >= MIN_DELETION_RATIO * max(1, self.added_lines)
         )
 
+    @property
+    def unmerges_base_work(self) -> bool:
+        """Return whether any path reverted exactly to its pre-merge content."""
+        return bool(self.unmerged_paths)
+
+    @property
+    def suspicious_test_regression(self) -> bool:
+        """Return whether tests were deleted or shrunk with no replacement test file."""
+        return bool(self.regressed_test_paths) and self.added_test_files == 0
+
     @property
     def blocked(self) -> bool:
-        """Return whether exact-tree or bulk-regression evidence blocks the head."""
-        return self.exact_replay_of is not None or self.suspicious_bulk_regression
+        """Return whether any replay, unmerge, or test-regression evidence blocks the head."""
+        return (
+            self.exact_replay_of is not None
+            or self.unmerges_base_work
+            or self.suspicious_test_regression
+            or self.suspicious_bulk_regression
+        )
 
 
 def git_output(repo_root: Path, args: Sequence[str]) -> str:
@@ -125,6 +156,69 @@ def diff_statistics(repo_root: Path, start: str, end: str) -> tuple[int, int, in
     return removed_files, added_lines, deleted_lines
 
 
+def is_test_path(path: str) -> bool:
+    """Return whether a repository path looks like an automated test file."""
+    parts = path.replace("\\", "/").split("/")
+    if any(part in TEST_DIR_SEGMENTS for part in parts[:-1]):
+        return True
+    name = parts[-1]
+    stem = name.split(".", 1)[0]
+    return (
+        stem.startswith("test_")
+        or stem.endswith("_test")
+        or ".spec." in name
+        or ".test." in name
+    )
+
+
+def changed_paths(repo_root: Path, start: str, end: str) -> set[str]:
+    """Return the set of paths whose content differs between two commits."""
+    output = git_output(repo_root, ["diff", "--name-only", start, end])
+    return {line for line in output.splitlines() if line}
+
+
+def unmerged_base_paths(repo_root: Path, merge_anchor: str, head_sha: str) -> tuple[str, ...]:
+    """Return post-merge paths reverted exactly to their pre-merge content.
+
+    A path that changed after the merge anchor yet is byte-identical to the
+    pre-merge first parent means the push discarded exactly what the base
+    merge brought in for that path — the targeted-revert signature of a stale
+    agent workspace snapshot, however small the diff.
+    """
+    since_merge = changed_paths(repo_root, merge_anchor, head_sha)
+    since_pre_merge = changed_paths(repo_root, f"{merge_anchor}^1", head_sha)
+    return tuple(sorted(since_merge - since_pre_merge))
+
+
+def test_file_changes(repo_root: Path, start: str, end: str) -> tuple[tuple[str, ...], int]:
+    """Return regressed (deleted or net-shrunk) test paths and the added-test count."""
+    regressed: set[str] = set()
+    added = 0
+    for line in git_output(repo_root, ["diff", "--name-status", start, end]).splitlines():
+        fields = line.split("\t")
+        if len(fields) < 2 or not is_test_path(fields[-1]):
+            continue
+        status = fields[0][:1]
+        if status == "D":
+            regressed.add(fields[-1])
+        elif status == "A":
+            added += 1
+    for line in git_output(repo_root, ["diff", "--numstat", start, end]).splitlines():
+        fields = line.split("\t", 2)
+        if len(fields) < 3 or not fields[0].isdigit() or not fields[1].isdigit():
+            continue
+        if is_test_path(fields[2]) and int(fields[1]) > int(fields[0]):
+            regressed.add(fields[2])
+    return tuple(sorted(regressed)), added
+
+
+def summarize_paths(paths: Sequence[str]) -> str:
+    """Render a bounded, comma-separated path list for the report."""
+    listed = ", ".join(paths[:MAX_LISTED_PATHS])
+    extra = len(paths) - MAX_LISTED_PATHS
+    return f"{listed} (+{extra} more)" if extra > 0 else listed
+
+
 def collect_evidence(repo_root: Path, base_sha: str, head_sha: str) -> ReplayEvidence:
     """Collect exact-tree and bulk-diff evidence for the supplied PR head."""
     git_output(repo_root, ["rev-parse", "--verify", f"{base_sha}^{{commit}}"])
@@ -149,6 +243,8 @@ def collect_evidence(repo_root: Path, base_sha: str, head_sha: str) -> ReplayEvi
         merge_anchor,
         head_sha,
     )
+    unmerged = unmerged_base_paths(repo_root, merge_anchor, head_sha)
+    regressed_tests, added_tests = test_file_changes(repo_root, merge_anchor, head_sha)
     return ReplayEvidence(
         base_sha=base_sha,
         head_sha=head_sha,
@@ -158,6 +254,9 @@ def collect_evidence(repo_root: Path, base_sha: str, head_sha: str) -> ReplayEvi
         removed_files=removed_files,
         added_lines=added_lines,
         deleted_lines=deleted_lines,
+        unmerged_paths=unmerged,
+        regressed_test_paths=regressed_tests,
+        added_test_files=added_tests,
     )
 
 
@@ -171,24 +270,35 @@ def format_report(evidence: ReplayEvidence) -> str:
         f"- Post-merge commits: {evidence.post_merge_commits}",
         f"- Post-merge removed files: {evidence.removed_files}",
         f"- Post-merge added/deleted lines: {evidence.added_lines}/{evidence.deleted_lines}",
+        f"- Paths reverted to pre-merge content: {summarize_paths(evidence.unmerged_paths) or 'none'}",
+        f"- Regressed test files: {summarize_paths(evidence.regressed_test_paths) or 'none'}",
+        f"- Added test files: {evidence.added_test_files}",
     ]
+    reasons = []
     if evidence.exact_replay_of is not None:
-        lines.extend(
-            [
-                "- Result: FAIL",
-                "- Reason: current HEAD exactly replays the tree of pre-merge ancestor "
-                f"{evidence.exact_replay_of}; a stale agent workspace discarded the base merge.",
-            ]
+        reasons.append(
+            "current HEAD exactly replays the tree of pre-merge ancestor "
+            f"{evidence.exact_replay_of}; a stale agent workspace discarded the base merge."
         )
-    elif evidence.suspicious_bulk_regression:
-        lines.extend(
-            [
-                "- Result: FAIL",
-                "- Reason: post-merge changes match the stale bulk-replay signature "
-                f"(removed files >= {MIN_REMOVED_FILES}, deleted lines >= {MIN_DELETED_LINES}, "
-                f"deletion/addition ratio >= {MIN_DELETION_RATIO}:1).",
-            ]
+    if evidence.unmerges_base_work:
+        reasons.append(
+            "post-merge commits reverted base-merged work back to its exact pre-merge "
+            f"content (unmerged base work): {summarize_paths(evidence.unmerged_paths)}."
+        )
+    if evidence.suspicious_test_regression:
+        reasons.append(
+            "post-merge commits deleted or shrank test files without adding any "
+            f"replacement test file: {summarize_paths(evidence.regressed_test_paths)}."
+        )
+    if evidence.suspicious_bulk_regression:
+        reasons.append(
+            "post-merge changes match the stale bulk-replay signature "
+            f"(removed files >= {MIN_REMOVED_FILES}, deleted lines >= {MIN_DELETED_LINES}, "
+            f"deletion/addition ratio >= {MIN_DELETION_RATIO}:1)."
         )
+    if reasons:
+        lines.append("- Result: FAIL")
+        lines.extend(f"- Reason: {reason}" for reason in reasons)
     elif evidence.merge_anchor is None:
         lines.extend(
             [
@@ -208,7 +318,8 @@ def format_report(evidence: ReplayEvidence) -> str:
         lines.extend(
             [
                 "- Result: PASS",
-                "- Reason: post-merge changes neither match a pre-merge tree nor exceed the conservative bulk-regression thresholds.",
+                "- Reason: post-merge changes neither match a pre-merge tree, revert base-merged work, "
+                "regress tests without replacement, nor exceed the conservative bulk-regression thresholds.",
             ]
         )
     return "\n".join(lines)

tests/test_pr_head_replay_guard.py

@@ -154,6 +154,152 @@ def test_diff_statistics_skips_non_numeric_numstat(monkeypatch, tmp_path):
     assert guard.diff_statistics(tmp_path, "a", "b") == (2, 3, 4)
 
 
+def fixture_repo_with_base_tests(tmp_path: Path) -> tuple[Path, str, str]:
+    """Create a merged PR whose base merge brought a wrapper and its regression test."""
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    git(repo, "init", "-b", "main")
+    git(repo, "config", "user.name", "Test")
+    git(repo, "config", "user.email", "test@example.com")
+    write(repo, "feature.txt", "base\n")
+    commit(repo, "base")
+
+    git(repo, "checkout", "-b", "feature")
+    write(repo, "feature.txt", "feature\n")
+    write(repo, "tests/test_feature.py", "def test_feature():\n    assert True\n\n\ndef test_edge():\n    assert True\n")
+    commit(repo, "feature with tests")
+
+    git(repo, "checkout", "main")
+    write(repo, "src/wrapper.py", "def wrap():\n    return 'accessible'\n")
+    write(repo, "tests/test_wrapper.py", "def test_wrap():\n    assert True\n")
+    current_base = commit(repo, "base adds wrapper and regression test")
+
+    git(repo, "checkout", "feature")
+    git(repo, "merge", "--no-ff", "--no-edit", "main")
+    merge_anchor = git(repo, "rev-parse", "HEAD")
+    return repo, current_base, merge_anchor
+
+
+def test_targeted_unmerge_of_base_work_fails_below_bulk_thresholds(tmp_path, capsys):
+    """A small stale revert that drops base-merged files is blocked and named."""
+    repo, current_base, merge_anchor = fixture_repo_with_base_tests(tmp_path)
+    (repo / "src/wrapper.py").unlink()
+    (repo / "tests/test_wrapper.py").unlink()
+    git(repo, "add", "-A")
+    git(repo, "commit", "-m", "stale targeted revert")
+    head = git(repo, "rev-parse", "HEAD")
+
+    evidence = guard.collect_evidence(repo, current_base, head)
+
+    assert evidence.merge_anchor == merge_anchor
+    assert evidence.unmerged_paths == ("src/wrapper.py", "tests/test_wrapper.py")
+    assert evidence.unmerges_base_work
+    assert evidence.regressed_test_paths == ("tests/test_wrapper.py",)
+    assert evidence.suspicious_test_regression
+    assert not evidence.suspicious_bulk_regression
+    assert evidence.blocked
+    assert guard.main(["--repo-root", str(repo), "--base-sha", current_base, "--head-sha", head]) == 1
+    report = capsys.readouterr().out
+    assert "Result: FAIL" in report
+    assert "unmerged base work" in report
+    assert "src/wrapper.py" in report
+    assert "tests/test_wrapper.py" in report
+
+
+def test_shrunk_test_without_replacement_fails(tmp_path):
+    """Weakening an existing test file with no new test file is blocked."""
+    repo, current_base, _ = fixture_repo_with_base_tests(tmp_path)
+    write(repo, "tests/test_feature.py", "def test_feature():\n    assert True\n")
+    head = commit(repo, "swap regression test for weaker one")
+
+    evidence = guard.collect_evidence(repo, current_base, head)
+
+    assert evidence.unmerged_paths == ()
+    assert evidence.regressed_test_paths == ("tests/test_feature.py",)
+    assert evidence.suspicious_test_regression
+    assert evidence.blocked
+    assert "deleted or shrank test files" in guard.format_report(evidence)
+
+
+def test_test_refactor_with_replacement_passes(tmp_path):
+    """Deleting a test while adding a replacement test file is not blocked."""
+    repo, current_base, _ = fixture_repo_with_base_tests(tmp_path)
+    (repo / "tests/test_feature.py").unlink()
+    write(repo, "tests/test_feature_v2.py", "def test_feature_v2():\n    assert True\n\n\ndef test_edge_v2():\n    assert True\n")
+    git(repo, "add", "-A")
+    git(repo, "commit", "-m", "rename test module")
+    head = git(repo, "rev-parse", "HEAD")
+
+    evidence = guard.collect_evidence(repo, current_base, head)
+
+    assert evidence.regressed_test_paths == ("tests/test_feature.py",)
+    assert evidence.added_test_files == 1
+    assert not evidence.suspicious_test_regression
+    assert evidence.unmerged_paths == ()
+    assert not evidence.blocked
+
+
+def test_is_test_path_covers_common_layouts():
+    """Test-path detection recognizes directories, prefixes, suffixes, and spec names."""
+    assert guard.is_test_path("tests/test_guard.py")
+    assert guard.is_test_path("pkg/__tests__/button.js")
+    assert guard.is_test_path("test_guard.py")
+    assert guard.is_test_path("pkg/guard_test.go")
+    assert guard.is_test_path("app/button.spec.ts")
+    assert guard.is_test_path("app/button.test.tsx")
+    assert guard.is_test_path("tests\\test_windows.py")
+    assert not guard.is_test_path("scripts/ci/guard.py")
+    assert not guard.is_test_path("docs/testing.md")
+
+
+def test_signal_properties_require_their_evidence():
+    """Unmerge and test-regression signals fire only on their exact evidence."""
+    common = {"base_sha": "base", "head_sha": "head", "merge_anchor": "merge", "post_merge_commits": 1}
+    assert not guard.ReplayEvidence(**common).blocked
+    assert guard.ReplayEvidence(**common, unmerged_paths=("a.py",)).blocked
+    assert guard.ReplayEvidence(**common, regressed_test_paths=("tests/test_a.py",)).blocked
+    assert not guard.ReplayEvidence(
+        **common, regressed_test_paths=("tests/test_a.py",), added_test_files=1
+    ).blocked
+
+
+def test_summarize_paths_bounds_long_lists():
+    """Path lists in reports are capped with an explicit overflow count."""
+    paths = [f"tests/test_{index}.py" for index in range(12)]
+    summary = guard.summarize_paths(paths)
+    assert summary.endswith("(+2 more)")
+    assert "tests/test_9.py" in summary
+    assert guard.summarize_paths(["one.py"]) == "one.py"
+
+
+def test_test_file_changes_parses_status_and_numstat(monkeypatch, tmp_path):
+    """Deleted, shrunk, added, malformed, and non-test records are classified correctly."""
+    name_status = "\n".join(
+        [
+            "D\ttests/test_gone.py",
+            "A\ttests/test_new.py",
+            "M\ttests/test_kept.py",
+            "D\tsrc/app_old.py",
+            "badline",
+        ]
+    )
+    numstat = "\n".join(
+        [
+            "1\t5\ttests/test_shrunk.py",
+            "5\t1\ttests/test_grown.py",
+            "-\t-\ttests/blob.bin",
+            "2\t9\tsrc/big.py",
+        ]
+    )
+    outputs = iter([name_status, numstat])
+    monkeypatch.setattr(guard, "git_output", lambda _root, _args: next(outputs))
+
+    regressed, added = guard.test_file_changes(tmp_path, "a", "b")
+
+    assert regressed == ("tests/test_gone.py", "tests/test_shrunk.py")
+    assert added == 1
+
+
 def test_invalid_commit_fails_closed_with_reason(tmp_path, capsys):
     """Missing git evidence fails closed and exposes the exact git reason."""
     repo, _, _, _ = fixture_repo(tmp_path)