Coverage Analysis skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Coverage Analysis"
  short_description: "Gate the diff, ratchet the total, read the gaps"
  default_prompt: "$coverage-analysis Add coverage to our CI and fail PRs that add untested code"

#!/usr/bin/env python3
"""Report coverage of the lines changed on a branch (patch / diff coverage).

Reads one coverage report (LCOV, Cobertura XML, or a Go coverprofile), asks git which
lines changed against a base ref, and prints covered/uncovered changed lines per file.

Usage:
    python3 diff_coverage.py coverage/lcov.info --base origin/main
    python3 diff_coverage.py coverage.xml --base origin/main --fail-under 80
    python3 diff_coverage.py cover.out --base main --include-worktree

Exit codes: 0 ok, 1 below --fail-under, 2 usage or parse error.
"""
from __future__ import annotations

import argparse
import re
import subprocess
import sys
import xml.etree.ElementTree as ET
from collections import defaultdict
from pathlib import Path

Coverage = dict[str, dict[int, int]]  # path -> {line: hits}


def parse_lcov(text: str) -> Coverage:
    cov: Coverage = defaultdict(dict)
    current = None
    for line in text.splitlines():
        if line.startswith("SF:"):
            current = line[3:].strip()
        elif line.startswith("DA:") and current:
            num, hits = line[3:].split(",")[:2]
            n, h = int(num), int(float(hits))
            cov[current][n] = max(cov[current].get(n, 0), h)
        elif line == "end_of_record":
            current = None
    return cov


def parse_cobertura(text: str) -> Coverage:
    root = ET.fromstring(text)
    sources = [s.text.strip() for s in root.iter("source") if s.text and s.text.strip()]
    cov: Coverage = defaultdict(dict)
    for cls in root.iter("class"):
        fname = cls.get("filename", "")
        # Prefer a path relative to the repo; keep the raw name and let suffix matching resolve it.
        path = fname
        for src in sources:
            candidate = Path(src) / fname
            if candidate.exists():
                path = str(candidate)
                break
        for ln in cls.iter("line"):
            n, h = int(ln.get("number", "0")), int(ln.get("hits", "0"))
            cov[path][n] = max(cov[path].get(n, 0), h)
    return cov


def parse_go(text: str) -> Coverage:
    cov: Coverage = defaultdict(dict)
    pat = re.compile(r"^(.+):(\d+)\.\d+,(\d+)\.\d+ \d+ (\d+)$")
    for line in text.splitlines()[1:]:
        m = pat.match(line.strip())
        if not m:
            continue
        path, start, end, count = m.group(1), int(m.group(2)), int(m.group(3)), int(m.group(4))
        for n in range(start, end + 1):
            cov[path][n] = max(cov[path].get(n, 0), count)
    return cov


def load(path: Path) -> Coverage:
    text = path.read_text(encoding="utf-8", errors="replace")
    head = text.lstrip()[:200]
    if head.startswith("<?xml") or head.startswith("<coverage"):
        return parse_cobertura(text)
    if head.startswith("mode:"):
        return parse_go(text)
    if "SF:" in text:
        return parse_lcov(text)
    raise ValueError(f"unrecognised coverage format: {path}")


def changed_lines(base: str, include_worktree: bool) -> dict[str, set[int]]:
    args = ["git", "diff", "--unified=0", "--no-color", "--no-ext-diff"]
    args += [base] if include_worktree else [f"{base}...HEAD"]
    out = subprocess.run(args, check=True, capture_output=True, text=True).stdout
    changes: dict[str, set[int]] = defaultdict(set)
    current = None
    for line in out.splitlines():
        if line.startswith("+++ "):
            target = line[4:]
            current = None if target == "/dev/null" else re.sub(r"^b/", "", target)
        elif line.startswith("@@") and current:
            m = re.search(r"\+(\d+)(?:,(\d+))?", line)
            if m:
                start, count = int(m.group(1)), int(m.group(2) or 1)
                changes[current].update(range(start, start + count))
    return changes


def match(cov: Coverage, repo_path: str) -> dict[int, int] | None:
    if repo_path in cov:
        return cov[repo_path]
    best = None
    for p in cov:
        norm = p.replace("\\", "/")
        if norm.endswith("/" + repo_path) or repo_path.endswith("/" + norm.lstrip("./")):
            if best is None or len(p) < len(best):
                best = p
    return cov[best] if best else None


def ranges(nums: list[int]) -> str:
    out, start, prev = [], None, None
    for n in sorted(nums):
        if start is None:
            start = prev = n
        elif n == prev + 1:
            prev = n
        else:
            out.append(f"{start}-{prev}" if start != prev else str(start))
            start = prev = n
    if start is not None:
        out.append(f"{start}-{prev}" if start != prev else str(start))
    return ", ".join(out)


def main() -> int:
    ap = argparse.ArgumentParser(description="Coverage of changed lines against a git base ref.")
    ap.add_argument("report", type=Path, help="lcov.info, cobertura coverage.xml or Go cover.out")
    ap.add_argument("--base", default="origin/main", help="base ref (default origin/main)")
    ap.add_argument("--include-worktree", action="store_true", help="diff base against the working tree, not HEAD")
    ap.add_argument("--fail-under", type=float, default=None, help="exit 1 if diff coverage %% is below this")
    a = ap.parse_args()

    try:
        cov = load(a.report)
        changes = changed_lines(a.base, a.include_worktree)
    except (OSError, ValueError, ET.ParseError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except subprocess.CalledProcessError as exc:
        print(f"error: git diff failed: {exc.stderr.strip()}", file=sys.stderr)
        return 2

    total_cov = total_exec = 0
    rows, missing_files = [], []
    for path in sorted(changes):
        file_cov = match(cov, path)
        if file_cov is None:
            missing_files.append(path)
            continue
        executable = [n for n in changes[path] if n in file_cov]
        if not executable:
            continue
        covered = [n for n in executable if file_cov[n] > 0]
        uncovered = [n for n in executable if file_cov[n] == 0]
        total_cov += len(covered)
        total_exec += len(executable)
        rows.append((path, len(covered), len(executable), uncovered))

    print(f"diff coverage vs {a.base}\n")
    for path, c, e, unc in rows:
        pct = 100 * c / e
        print(f"  {pct:5.1f}%  {c:>4}/{e:<4} {path}")
        if unc:
            print(f"          uncovered lines: {ranges(unc)}")
    if missing_files:
        print("\n  changed files absent from the report (not measured, or not source):")
        for p in missing_files:
            print(f"    {p}")
    if total_exec == 0:
        print("\nno changed executable lines found in the report")
        return 0
    pct = 100 * total_cov / total_exec
    print(f"\ntotal: {total_cov}/{total_exec} changed executable lines covered = {pct:.1f}%")
    if a.fail_under is not None and pct < a.fail_under:
        print(f"FAIL: below --fail-under {a.fail_under}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())