Fail CI when hotspots, dead code, or mutation survivors grow

.github/workflows/ci.yml

@@ -20,12 +20,22 @@ jobs:
       - uses: actions/setup-python@v6
         with:
           python-version: "3.12"
-      - run: pip install ruff lizard==1.24.0
+      - run: pip install ruff lizard==1.24.0 vulture==2.16
       - run: ruff check src/ tests/
       - name: Complexity proving test
         run: bash scripts/test-check-complexity.sh
       - name: PR-diff complexity
         run: python3 scripts/check-complexity.py --base origin/main --paths src
+      - name: Hotspot proving test
+        run: bash scripts/test-check-hotspots.sh
+      - name: PR-diff hotspots
+        run: python3 scripts/check-hotspots.py --base origin/main --paths src
+      - name: Vulture proving test
+        run: bash scripts/test-check-vulture.sh
+      - name: Vulture baseline
+        run: python3 scripts/check-vulture.py src tests
+      - name: Mutation score proving test
+        run: bash scripts/test-check-mutation.sh
 
   unit:
     runs-on: ubuntu-latest
@@ -93,3 +103,42 @@ jobs:
       - run: pip install --no-cache-dir "."
       - name: Validate scratch init bundle
         run: bash scripts/ci-validate-scratch-bundle.sh
+
+  # mutmut on src/docgen/path_filters.py only. The job reads the committed
+  # survived ceiling and does not rewrite it.
+  mutation:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+      - name: Install system dependencies
+        run: |
+          set -euo pipefail
+          attempt=1
+          max_attempts=4
+          backoff=4
+          while [ "$attempt" -le "$max_attempts" ]; do
+            echo "apt attempt $attempt/$max_attempts"
+            if timeout 600s sudo apt-get -o Acquire::Retries=3 update \
+              && timeout 600s sudo apt-get -o Acquire::Retries=3 install -y --no-install-recommends ffmpeg tesseract-ocr; then
+              exit 0
+            fi
+            if [ "$attempt" -eq "$max_attempts" ]; then
+              echo "apt failed after $max_attempts attempts"
+              exit 1
+            fi
+            echo "apt failed, retrying in ${backoff}s..."
+            sleep "$backoff"
+            backoff=$((backoff * 2))
+            attempt=$((attempt + 1))
+          done
+      - run: pip install --no-cache-dir ".[dev]" "mutmut==3.8.0"
+      - name: Mutation score for path_filters.py
+        env:
+          PYTHONPATH: ${{ github.workspace }}/src
+        run: |
+          mutmut run --max-children 1
+          mutmut export-cicd-stats
+          python3 scripts/check-mutation-score.py

.gitignore

@@ -32,3 +32,4 @@ tests/.bin-cache/
 htmlcov/
 .mypy_cache/
 .ruff_cache/
+mutants/

config/mutation-path-filters-baseline.json

@@ -0,0 +1,12 @@
+{
+  "module": "src/docgen/path_filters.py",
+  "killed": 11,
+  "survived": 0,
+  "total": 11,
+  "no_tests": 0,
+  "skipped": 0,
+  "suspicious": 0,
+  "timeout": 0,
+  "check_was_interrupted_by_user": 0,
+  "segfault": 0
+}

config/vulture-baseline.txt

@@ -0,0 +1,8 @@
+# High-confidence vulture findings (vulture 2.16, min-confidence 80).
+# Each line is: <count> <path> <message>
+# check-vulture.py fails when a count rises or a new key appears.
+# CI reads this file and must not rewrite it.
+1 src/docgen/cli.py unused variable 'param'
+2 src/docgen/scene_clock_harness.py unused variable 'other'
+1 src/docgen/scene_spec_generate.py unused variable 'source_snippets'
+20 tests/test_ai_client.py unused variable 'clear_ai_env'

pyproject.toml

@@ -66,3 +66,29 @@ target-version = "py310"
 # selection that fails the existing codebase; keep CI on the historical E/F gate.
 [tool.ruff.lint]
 select = ["E4", "E7", "E9", "F"]
+
+# One module. validate.py's generated mutant module is about 4.6MB and the
+# runner was OOM-killed above 12G, so that file is not the CI target.
+# path_filters.py is the module that can finish. source_paths copies the
+# package so imports resolve; only_mutate keeps the run off the rest of src.
+[tool.mutmut]
+source_paths = ["src"]
+only_mutate = ["src/docgen/path_filters.py"]
+pytest_add_cli_args_test_selection = ["tests/test_path_filters.py"]
+also_copy = [
+    ".github",
+    ".cursor",
+    "scripts",
+    "AGENTS.md",
+    "README.md",
+    "NOTICE",
+    "SESSION_NOTES.md",
+    "packaging",
+    "docs",
+    "milestones",
+    "third_party",
+    "issues",
+    "specs-on-hold",
+    "session-notes",
+    "config",
+]

scripts/check-hotspots.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Hotspot gate.
+
+Fails when a changed Python file is both complex and in the top change-frequency
+set. A change to a quiet file passes, including when that file is complex. A
+change to a frequent file that is not complex passes.
+
+Complex means any function with CCN > 10 or NLOC > 80 (same limits as
+check-complexity.py). Frequency is the number of commits on --base that touch
+the file, among Python files present at that revision. The top set is the
+--top files by that count; ties at the cutoff are included.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+DEFAULT_CCN = 10
+DEFAULT_NLOC = 80
+DEFAULT_TOP = 10
+
+
+def git(*args: str, cwd: Path) -> str:
+    return subprocess.check_output(["git", *args], cwd=cwd, text=True).rstrip("\n")
+
+
+def changed_python_files(repo: Path, base: str, paths: list[str]) -> list[str]:
+    rels = git(
+        "diff",
+        "--name-only",
+        "--diff-filter=ACMR",
+        f"{base}...HEAD",
+        "--",
+        *paths,
+        cwd=repo,
+    )
+    files = []
+    for line in rels.splitlines():
+        line = line.strip()
+        if line.endswith(".py"):
+            files.append(line)
+    return files
+
+
+def python_files_at(repo: Path, rev: str, paths: list[str]) -> set[str]:
+    listed = git("ls-tree", "-r", "--name-only", rev, "--", *paths, cwd=repo)
+    return {line.strip() for line in listed.splitlines() if line.strip().endswith(".py")}
+
+
+def commit_counts(repo: Path, rev: str, paths: list[str]) -> dict[str, int]:
+    log = git("log", rev, "--pretty=format:", "--name-only", "--", *paths, cwd=repo)
+    counts: dict[str, int] = {}
+    for line in log.splitlines():
+        line = line.strip()
+        if not line.endswith(".py"):
+            continue
+        counts[line] = counts.get(line, 0) + 1
+    return counts
+
+
+def top_frequency(counts: dict[str, int], top_n: int) -> set[str]:
+    ranked = sorted(counts.items(), key=lambda item: (-item[1], item[0]))
+    if not ranked or top_n < 1:
+        return set()
+    cutoff = ranked[min(top_n, len(ranked)) - 1][1]
+    return {path for path, count in ranked if count >= cutoff and count > 0}
+
+
+def lizard_rows(source: str, filename: str) -> list[dict[str, str]]:
+    if not source.strip():
+        return []
+    with tempfile.TemporaryDirectory() as tmp:
+        dest = Path(tmp) / Path(filename).name
+        dest.write_text(source, encoding="utf-8")
+        csv_path = Path(tmp) / "out.csv"
+        subprocess.run(
+            ["lizard", "-l", "python", "-C", "999", "-L", "999999", "-o", str(csv_path), str(dest)],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        text = csv_path.read_text(encoding="utf-8")
+    rows = []
+    for rec in csv.reader(io.StringIO(text)):
+        if len(rec) < 9:
+            continue
+        rows.append({"nloc": rec[0], "ccn": rec[1], "name": rec[7]})
+    return rows
+
+
+def function_peaks(repo: Path, rel: str) -> tuple[int, int] | None:
+    try:
+        source = git("show", f"HEAD:{rel}", cwd=repo)
+    except subprocess.CalledProcessError:
+        return None
+    rows = lizard_rows(source, rel)
+    if not rows:
+        return (0, 0)
+    max_ccn = max(int(row["ccn"]) for row in rows)
+    max_nloc = max(int(row["nloc"]) for row in rows)
+    return max_ccn, max_nloc
+
+
+def hotspot_failures(
+    repo: Path,
+    changed: list[str],
+    hot: set[str],
+    counts: dict[str, int],
+    ccn_limit: int,
+    nloc_limit: int,
+) -> list[str]:
+    failures = []
+    for rel in changed:
+        if rel not in hot:
+            continue
+        peaks = function_peaks(repo, rel)
+        if peaks is None:
+            continue
+        ccn, nloc = peaks
+        if ccn > ccn_limit or nloc > nloc_limit:
+            failures.append(
+                f"HOTSPOT {rel} commits={counts.get(rel, 0)} "
+                f"maxCCN={ccn} maxNLOC={nloc} (limits {ccn_limit}/{nloc_limit})"
+            )
+    return failures
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", default="origin/main")
+    parser.add_argument("--repo", default=".")
+    parser.add_argument("--ccn", type=int, default=DEFAULT_CCN)
+    parser.add_argument("--nloc", type=int, default=DEFAULT_NLOC)
+    parser.add_argument("--top", type=int, default=DEFAULT_TOP)
+    parser.add_argument("--paths", nargs="*", default=["."])
+    args = parser.parse_args()
+    repo = Path(args.repo).resolve()
+    try:
+        changed = changed_python_files(repo, args.base, args.paths)
+        present = python_files_at(repo, args.base, args.paths)
+        counts = commit_counts(repo, args.base, args.paths)
+    except subprocess.CalledProcessError as exc:
+        print(f"git failed: {exc}", file=sys.stderr)
+        return 2
+    if not changed:
+        print("check-hotspots: no changed Python files")
+        return 0
+    ranked = {path: count for path, count in counts.items() if path in present}
+    hot = top_frequency(ranked, args.top)
+    try:
+        failures = hotspot_failures(repo, changed, hot, counts, args.ccn, args.nloc)
+    except subprocess.CalledProcessError as exc:
+        print(f"lizard failed: {exc}", file=sys.stderr)
+        return 2
+    if failures:
+        print("check-hotspots: FAIL", file=sys.stderr)
+        for line in failures:
+            print(line, file=sys.stderr)
+        return 1
+    print(f"check-hotspots: PASS ({len(changed)} file(s), {len(hot)} hotspot path(s))")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/check-mutation-score.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""Compare a mutmut export to the committed survived ceiling.
+
+Reads mutants/mutmut-cicd-stats.json and config/mutation-path-filters-baseline.json.
+Fails when survived mutants increase. Does not write the baseline.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_BASELINE = ROOT / "config" / "mutation-path-filters-baseline.json"
+DEFAULT_STATS = ROOT / "mutants" / "mutmut-cicd-stats.json"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE)
+    parser.add_argument("--stats", type=Path, default=DEFAULT_STATS)
+    args = parser.parse_args()
+    if not args.baseline.is_file():
+        print(f"missing mutation baseline: {args.baseline}", file=sys.stderr)
+        return 2
+    if not args.stats.is_file():
+        print(f"missing mutmut stats: {args.stats}", file=sys.stderr)
+        return 2
+    baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
+    stats = json.loads(args.stats.read_text(encoding="utf-8"))
+    if int(stats.get("check_was_interrupted_by_user") or 0) > 0:
+        print("check-mutation-score: FAIL interrupted run", file=sys.stderr)
+        return 1
+    if int(stats.get("total") or 0) <= 0:
+        print("check-mutation-score: FAIL empty mutmut export", file=sys.stderr)
+        return 1
+    allowed = int(baseline["survived"])
+    survived = int(stats["survived"])
+    if survived > allowed:
+        print(
+            f"check-mutation-score: FAIL survived {allowed} -> {survived}",
+            file=sys.stderr,
+        )
+        return 1
+    print(f"check-mutation-score: PASS survived {survived} (ceiling {allowed})")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())