Jev maintenance tools: shared client, spec coverage audit, voice check, nightly triage

scripts/jev/spec_coverage.py

@@ -0,0 +1,238 @@
+"""Which functional spec items have a test? Gaps between SIMURGARRAY_FUNCTIONAL_SPEC.md and tests/.
+
+1. Code parses the spec items (``- **CLI-021** ...`` up to the next item or heading) and every test function.
+2. An item cited by ID in a test file counts as covered by citation.
+3. For the other items, code ranks test functions by word overlap (BM25) and keeps the best three; Jev answers one
+   yes/no question per candidate: does this test check what the item requires?
+4. The report lists, per prefix, the items covered by citation, covered by judgment, maybe covered and uncovered.
+
+Open questions (OQ) and scope statements (SCOPE) are not behaviours and are skipped.
+
+    python scripts/jev/spec_coverage.py                  # writes spec_coverage_report.md
+    python scripts/jev/spec_coverage.py --dry-run        # parsing and candidates only, no API calls
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import math
+import re
+import sys
+from collections import Counter
+from dataclasses import dataclass
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from client import ROOT, noul, system_one
+
+SPEC = ROOT / "SIMURGARRAY_FUNCTIONAL_SPEC.md"
+TESTS = ROOT / "tests"
+SKIP_PREFIXES = {"OQ", "SCOPE"}
+COVERED, MAYBE = 0.8, 0.5
+_ITEM = re.compile(r"^- \*\*([A-Z]+)-(\d+)\*\*\s*(`[^`]*`)?\s*(.*)")
+_ID = re.compile(r"\b([A-Z]{2,5})-(\d{3})\b")
+_WORD = re.compile(r"[a-z][a-z0-9_]{2,}")
+STOP = frozenset(
+    {
+        "the",
+        "and",
+        "for",
+        "with",
+        "from",
+        "that",
+        "this",
+        "are",
+        "not",
+        "any",
+        "all",
+        "its",
+        "into",
+        "each",
+        "when",
+        "then",
+        "than",
+        "per",
+        "but",
+        "has",
+        "have",
+        "was",
+        "which",
+        "their",
+        "them",
+        "they",
+        "one",
+        "two",
+        "set",
+        "get",
+        "use",
+        "used",
+        "uses",
+        "test",
+        "tests",
+        "should",
+        "must",
+        "may",
+        "can",
+    }
+)
+
+
+@dataclass
+class Item:
+    id: str
+    prefix: str
+    tag: str
+    text: str
+
+
+@dataclass
+class TestFn:
+    path: str
+    line: int
+    name: str
+    text: str
+
+
+def parse_spec(text: str) -> list[Item]:
+    items: list[Item] = []
+    cur: Item | None = None
+    for ln in text.splitlines():
+        m = _ITEM.match(ln)
+        if m or ln.startswith("#"):
+            if cur:
+                items.append(cur)
+            cur = None
+            if m:
+                cur = Item(f"{m.group(1)}-{m.group(2)}", m.group(1), (m.group(3) or "").strip("`"), m.group(4))
+            continue
+        if cur and len(cur.text) < 1800:
+            cur.text += "\n" + ln
+    if cur:
+        items.append(cur)
+    return [i for i in items if i.prefix not in SKIP_PREFIXES]
+
+
+def parse_tests(root: Path) -> list[TestFn]:
+    """Every ``test_*`` function: name, docstring and source (truncated), with the module docstring as context."""
+    out: list[TestFn] = []
+    for f in sorted(root.rglob("test_*.py")):
+        src = f.read_text(encoding="utf-8")
+        tree = ast.parse(src)
+        mod_doc = (ast.get_docstring(tree) or "").split("\n")[0]
+        for node in ast.walk(tree):
+            if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) and node.name.startswith("test_"):
+                body = ast.get_source_segment(src, node) or ""
+                text = f"# {f.relative_to(ROOT)} ({mod_doc})\n{body}"[:1600]
+                out.append(TestFn(str(f.relative_to(ROOT)), node.lineno, node.name, text))
+    return out
+
+
+def cited_ids(root: Path) -> dict[str, set[str]]:
+    """Spec IDs mentioned anywhere in each test file."""
+    out: dict[str, set[str]] = {}
+    for f in sorted(root.rglob("*.py")):
+        for m in _ID.finditer(f.read_text(encoding="utf-8")):
+            out.setdefault(f"{m.group(1)}-{m.group(2)}", set()).add(str(f.relative_to(ROOT)))
+    return out
+
+
+def _tokens(s: str) -> list[str]:
+    return [w for w in _WORD.findall(s.lower().replace("-", "_")) if w not in STOP]
+
+
+class Bm25:
+    def __init__(self, docs: list[str], k1: float = 1.4, b: float = 0.75) -> None:
+        self.toks = [Counter(_tokens(d)) for d in docs]
+        self.len = [sum(t.values()) for t in self.toks]
+        self.avg = sum(self.len) / max(len(self.len), 1)
+        df = Counter(w for t in self.toks for w in t)
+        n = len(docs)
+        self.idf = {w: math.log(1 + (n - c + 0.5) / (c + 0.5)) for w, c in df.items()}
+        self.k1, self.b = k1, b
+
+    def top(self, query: str, k: int) -> list[int]:
+        q = set(_tokens(query))
+        scores = []
+        for i, t in enumerate(self.toks):
+            s = 0.0
+            for w in q & t.keys():
+                tf = t[w]
+                s += self.idf[w] * tf * (self.k1 + 1) / (tf + self.k1 * (1 - self.b + self.b * self.len[i] / self.avg))
+            scores.append((s, i))
+        scores.sort(reverse=True)
+        return [i for s, i in scores[:k] if s > 0]
+
+
+QUESTION = (
+    "Does the test function in `tests[{i}]` check the behaviour that `requirement` specifies? Read the test's "
+    "assertions: a test that only touches the same area without asserting this behaviour does not count."
+)
+
+
+def judge(item: Item, cands: list[TestFn]) -> list[float]:
+    state = {"requirement": f"{item.id} ({item.tag}): {item.text.strip()}", "tests": [c.text for c in cands]}
+    qs = {
+        f"t{i}": noul(
+            QUESTION.format(i=i),
+            true="The test exercises this behaviour and asserts on its outcome",
+            false="The test checks something else, or only touches the area",
+        )
+        for i in range(len(cands))
+    }
+    ans = system_one(state, qs)
+    return [float(ans[f"t{i}"]["noul"]) for i in range(len(cands))]
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    ap.add_argument("--out", type=Path, default=ROOT / "spec_coverage_report.md")
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--prefix", nargs="*", help="only these prefixes, for example CLI OUT")
+    a = ap.parse_args(argv)
+    items = parse_spec(SPEC.read_text(encoding="utf-8"))
+    if a.prefix:
+        items = [i for i in items if i.prefix in a.prefix]
+    tests = parse_tests(TESTS)
+    cited = cited_ids(TESTS)
+    bm = Bm25([t.text for t in tests])
+    rows: list[tuple[Item, str, float, TestFn | None, str]] = []
+    for it in items:
+        if it.id in cited:
+            rows.append((it, "cited", 1.0, None, ", ".join(sorted(cited[it.id]))))
+            continue
+        cands = [tests[i] for i in bm.top(f"{it.id} {it.text}", 5)]
+        if not cands:
+            rows.append((it, "uncovered", 0.0, None, "no candidate test"))
+            continue
+        if a.dry_run:
+            rows.append((it, "planned", 0.0, cands[0], f"{len(cands)} candidates"))
+            continue
+        ps = judge(it, cands)
+        best = max(range(len(ps)), key=lambda k: ps[k])
+        status = "covered" if ps[best] >= COVERED else "maybe" if ps[best] >= MAYBE else "uncovered"
+        rows.append((it, status, ps[best], cands[best], ""))
+    order = ["uncovered", "maybe", "covered", "cited", "planned"]
+    lines = ["# Spec coverage by tests", ""]
+    tot = Counter(r[1] for r in rows)
+    lines.append(f"{len(rows)} items: " + " · ".join(f"{k} {tot[k]}" for k in order if tot[k]))
+    lines += ["", "| Prefix | Items | cited | covered | maybe | uncovered |", "|---|---|---|---|---|---|"]
+    for pfx in sorted({r[0].prefix for r in rows}):
+        c = Counter(r[1] for r in rows if r[0].prefix == pfx)
+        lines.append(f"| {pfx} | {sum(c.values())} | {c['cited']} | {c['covered']} | {c['maybe']} | {c['uncovered']} |")
+    for st in order:
+        sel = [r for r in rows if r[1] == st and st != "cited"]
+        if not sel:
+            continue
+        lines += ["", f"## {st} ({len(sel)})", ""]
+        for it, _, p, t, extra in sel:
+            first = it.text.strip().split("\n")[0][:140]
+            where = f" · best test `{t.path}:{t.line}` {t.name} (p {p:.2f})" if t else ""
+            lines.append(f"* **{it.id}** `{it.tag}` {first}{where}{(' · ' + extra) if extra else ''}")
+    a.out.write_text("\n".join(lines) + "\n")
+    print(f"{len(rows)} items: " + ", ".join(f"{k} {tot[k]}" for k in order if tot[k]) + f" → {a.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())