fix(review): distinguish test cleanup from stale replay

scripts/ci/pr_head_replay_guard.py

@@ -16,8 +16,8 @@
   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;
+- test regression without replacement: post-merge commits deleting test files
+  or reducing declared test cases while adding no replacement test file;
 - the conservative bulk-regression signature: at least five tracked files and
   500 lines removed, with deletions at least four times additions.
 
@@ -28,6 +28,8 @@
 from __future__ import annotations
 
 import argparse
+import ast
+import re
 import subprocess
 from dataclasses import dataclass
 from pathlib import Path
@@ -40,6 +42,21 @@
 MIN_DELETION_RATIO = 4
 MAX_LISTED_PATHS = 10
 TEST_DIR_SEGMENTS = frozenset({"tests", "test", "__tests__", "spec", "specs"})
+TEST_CASE_PATTERNS = {
+    ".bats": re.compile(r"(?m)^\s*@test\b"),
+    ".go": re.compile(
+        r"(?m)^\s*func\s+(?:Test|Benchmark|Fuzz)[A-Z0-9_][A-Za-z0-9_]*\s*\("
+    ),
+    ".js": re.compile(r"\b(?:it|test)(?:\.(?:concurrent|each|only|skip|todo))*\s*\("),
+    ".jsx": re.compile(r"\b(?:it|test)(?:\.(?:concurrent|each|only|skip|todo))*\s*\("),
+    ".r": re.compile(r"\b(?:testthat::)?test_that\s*\("),
+    ".rs": re.compile(
+        r"#\s*\[\s*(?:[A-Za-z_][A-Za-z0-9_:]*::)?test"
+        r"(?:\s*\([^]]*\))?\s*\]"
+    ),
+    ".ts": re.compile(r"\b(?:it|test)(?:\.(?:concurrent|each|only|skip|todo))*\s*\("),
+    ".tsx": re.compile(r"\b(?:it|test)(?:\.(?:concurrent|each|only|skip|todo))*\s*\("),
+}
 
 
 @dataclass(frozen=True)
@@ -74,7 +91,7 @@ def unmerges_base_work(self) -> bool:
 
     @property
     def suspicious_test_regression(self) -> bool:
-        """Return whether tests were deleted or shrunk with no replacement test file."""
+        """Return whether test cases were lost with no replacement test file."""
         return bool(self.regressed_test_paths) and self.added_test_files == 0
 
     @property
@@ -190,8 +207,35 @@ def unmerged_base_paths(repo_root: Path, merge_anchor: str, head_sha: str) -> tu
     return tuple(sorted(since_merge - since_pre_merge))
 
 
+def test_case_count(
+    repo_root: Path,
+    revision: str,
+    path: str,
+) -> int | None:
+    """Return a supported test file's declared test-case count at one revision."""
+    try:
+        source = git_output(repo_root, ["show", f"{revision}:{path}"])
+    except RuntimeError:
+        return None
+
+    suffix = Path(path).suffix.lower()
+    if suffix == ".py":
+        try:
+            tree = ast.parse(source)
+        except SyntaxError:
+            return None
+        return sum(
+            isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
+            and node.name.startswith("test")
+            for node in ast.walk(tree)
+        )
+
+    pattern = TEST_CASE_PATTERNS.get(suffix)
+    return len(pattern.findall(source)) if pattern is not None else None
+
+
 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."""
+    """Return deleted or test-case-reducing 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():
@@ -207,8 +251,13 @@ def test_file_changes(repo_root: Path, start: str, end: str) -> tuple[tuple[str,
         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])
+        path = fields[2]
+        if not is_test_path(path) or int(fields[1]) <= int(fields[0]):
+            continue
+        before_count = test_case_count(repo_root, start, path)
+        after_count = test_case_count(repo_root, end, path)
+        if before_count is None or after_count is None or after_count < before_count:
+            regressed.add(path)
     return tuple(sorted(regressed)), added
 
 
@@ -287,8 +336,9 @@ def format_report(evidence: ReplayEvidence) -> str:
         )
     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)}."
+            "post-merge commits deleted test files or reduced declared test cases "
+            "without adding any replacement test file: "
+            f"{summarize_paths(evidence.regressed_test_paths)}."
         )
     if evidence.suspicious_bulk_regression:
         reasons.append(

tests/test_pr_head_replay_guard.py

@@ -166,7 +166,7 @@ def fixture_repo_with_base_tests(tmp_path: Path) -> tuple[Path, str, str]:
 
     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")
+    write(repo, "tests/test_feature.py", "def test_feature():\n    assert True\n    assert True\n\n\ndef test_edge():\n    assert True\n")
     commit(repo, "feature with tests")
 
     git(repo, "checkout", "main")
@@ -218,7 +218,24 @@ def test_shrunk_test_without_replacement_fails(tmp_path):
     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)
+    assert "reduced declared test cases" in guard.format_report(evidence)
+
+
+def test_duplicate_assertion_cleanup_without_test_case_loss_passes(tmp_path):
+    """Removing duplicate assertions while preserving test cases is not stale replay."""
+    repo, current_base, _ = fixture_repo_with_base_tests(tmp_path)
+    write(
+        repo,
+        "tests/test_feature.py",
+        "def test_feature():\n    assert True\n\n\ndef test_edge():\n    assert True\n",
+    )
+    head = commit(repo, "remove duplicate assertion")
+
+    evidence = guard.collect_evidence(repo, current_base, head)
+
+    assert evidence.regressed_test_paths == ()
+    assert evidence.unmerged_paths == ()
+    assert not evidence.blocked
 
 
 def test_test_refactor_with_replacement_passes(tmp_path):
@@ -252,6 +269,53 @@ def test_is_test_path_covers_common_layouts():
     assert not guard.is_test_path("docs/testing.md")
 
 
+def test_test_case_count_fails_closed_and_supports_known_formats(
+    monkeypatch,
+    tmp_path,
+):
+    """Missing, invalid, supported, and unsupported test sources are classified."""
+
+    def missing_source(_root, _args):
+        raise RuntimeError("missing revision")
+
+    monkeypatch.setattr(guard, "git_output", missing_source)
+    assert guard.test_case_count(tmp_path, "base", "tests/test_missing.py") is None
+
+    monkeypatch.setattr(
+        guard,
+        "git_output",
+        lambda _root, _args: "def test_broken(",
+    )
+    assert guard.test_case_count(tmp_path, "base", "tests/test_broken.py") is None
+
+    monkeypatch.setattr(
+        guard,
+        "git_output",
+        lambda _root, _args: "def test_ok():\n    assert True\n",
+    )
+    assert guard.test_case_count(tmp_path, "base", "tests/test_ok.py") == 1
+
+    supported_sources = (
+        ("tests/guard.bats", '@test "works" {\n  true\n}\n'),
+        ("tests/guard.go", "func TestGuard(t *testing.T) {}\n"),
+        ("tests/guard.test.js", "test.concurrent.each(cases)('works', () => {});\n"),
+        ("tests/guard.test.jsx", "it.only('works', () => {});\n"),
+        ("tests/test_guard.R", "testthat::test_that('works', { expect_true(TRUE) })\n"),
+        ("tests/guard_test.rs", "#[tokio::test]\nasync fn works() {}\n"),
+        ("tests/guard.test.ts", "test.skip('works', () => {});\n"),
+        ("tests/guard.test.tsx", "it.todo('works');\n"),
+    )
+    for path, source in supported_sources:
+        monkeypatch.setattr(
+            guard,
+            "git_output",
+            lambda _root, _args, source=source: source,
+        )
+        assert guard.test_case_count(tmp_path, "base", path) == 1
+
+    assert guard.test_case_count(tmp_path, "base", "tests/README.md") is None
+
+
 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}
@@ -273,7 +337,7 @@ def test_summarize_paths_bounds_long_lists():
 
 
 def test_test_file_changes_parses_status_and_numstat(monkeypatch, tmp_path):
-    """Deleted, shrunk, added, malformed, and non-test records are classified correctly."""
+    """Deleted, weakened, added, malformed, and non-test records are classified."""
     name_status = "\n".join(
         [
             "D\ttests/test_gone.py",
@@ -293,6 +357,11 @@ def test_test_file_changes_parses_status_and_numstat(monkeypatch, tmp_path):
     )
     outputs = iter([name_status, numstat])
     monkeypatch.setattr(guard, "git_output", lambda _root, _args: next(outputs))
+    monkeypatch.setattr(
+        guard,
+        "test_case_count",
+        lambda _root, revision, _path: 2 if revision == "a" else 1,
+    )
 
     regressed, added = guard.test_file_changes(tmp_path, "a", "b")