diff --git a/scripts/audit_consolidation_orphans.py b/scripts/audit_consolidation_orphans.py new file mode 100644 index 0000000000..4d7750df9b --- /dev/null +++ b/scripts/audit_consolidation_orphans.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +"""Consolidation-orphans audit -- the ORGAN for issue #16473 (acceptance 1). + +User mandate (2026-09-17, verbatim excerpt from the Epic body): the open-issue +pool grows, consolidation debt is paid poorly, and part of the stock ADDS +entropy instead of absorbing it. The Epic therefore asks for an audit that +LISTS the orphan issues (not attached to any consolidation chaperone) BY AXIS +(A-F), each with the PROOF of what was searched -- "un tri, pas un verdict". + +Attachment semantics (a TRIAGE, deliberately generous -- a human arbitrates): + + - ``body`` the issue's own body cites ``#``; + - ``chaperone_body`` a chaperone's body cites ``#``; + - ``chaperone_comment`` a comment on a chaperone cites ``#``; + - ``issue_comment`` (``--deep`` only) a comment on the issue cites + ``#``. + +A bare ``#N`` mention counts as attachment here even when the prose around it +is negative ("ne pas dupliquer #5081") -- the report surfaces the line, the +reader decides. This is the opposite failure mode of a verdict tool: a false +"attached" costs a glance, a false "orphan" costs a duplicate grain. + +Chaperones (from the Epic body): #5081 (notebook canon), #4362 (Lean lakes), +#13737 (structural, 30 perimeters), #9535 (cleanup), #16473 (this umbrella). + +ADVISORY by default (same posture as orphan_branch_scan.py): exit 0 whatever +the orphan count; the payload is the report. ``--fail-on-orphans`` exists for +a future ratchet wire, NOT for this delivery. + +Usage:: + + python scripts/audit_consolidation_orphans.py --fetch # gh live + python scripts/audit_consolidation_orphans.py --fetch --deep # + per-issue comments + python scripts/audit_consolidation_orphans.py --input state.json # offline + ... --format md|json [--out report.md] [--fail-on-orphans] + +``--input`` expects ``{"issues": [{"number", "title", "body"}, ...], +"chaperones": {"5081": {"body": "...", "comments": [{"body": "..."}]}}}`` +(comments optional). The pure core (attachment proofs, axis classification, +report rendering) is network-free and unit-tested in +scripts/tests/test_audit_consolidation_orphans.py; only ``--fetch`` talks to +GitHub via ``gh``. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from typing import Any + +CHAPERONES = (5081, 4362, 13737, 9535, 16473) + +AXIS_LABELS = { + "A": "Renumeration / parcours notebooks", + "B": "Doublons / twins / consolidation notebooks", + "C": "Lakes Lean", + "D": "Scripts sprawl & doublons CLI", + "E": "Doc sprawl & catalogue", + "F": "Fraicheur des bodies d'Epic", +} + +# Deliberately narrow patterns: a wrong axis is noise, an unclassified orphan +# is still listed. Multi-word anchors beat bare keywords ("lean" alone would +# swallow every notebook issue; "cli" alone every tooling one). +AXIS_PATTERNS: dict[str, list[re.Pattern[str]]] = { + "A": [ + re.compile(r"renum", re.IGNORECASE), + re.compile(r"renommage|renommer|rename", re.IGNORECASE), + re.compile(r"num[ée]rotation|numeration|renumber", re.IGNORECASE), + re.compile(r"accr[ée]tion|accretion", re.IGNORECASE), + re.compile(r"suffixe (noyau|kernel)|noyau canonique", re.IGNORECASE), + re.compile(r"nomenclature", re.IGNORECASE), + re.compile(r"casse .*(lien|catalogue|catalog)", re.IGNORECASE), + ], + "B": [ + re.compile(r"doublon|doublons", re.IGNORECASE), + re.compile(r"jumeau|jumelle|jumeaux", re.IGNORECASE), + re.compile(r"\btwins?\b|\btwin_", re.IGNORECASE), + re.compile(r"duplicat|duplicate", re.IGNORECASE), + re.compile(r"solution_leak|check_twin|duplicate_notebook|link_label", re.IGNORECASE), + ], + "C": [ + re.compile(r"\blakes?\b", re.IGNORECASE), + re.compile(r"mathlib", re.IGNORECASE), + re.compile(r"junction", re.IGNORECASE), + re.compile(r"find_lake_root|lake build|lakefile", re.IGNORECASE), + ], + "D": [ + re.compile(r"scripts? .*(sprawl|consolidat|obsol[èe]te|doublon)", re.IGNORECASE), + re.compile(r"one[- ]shots?", re.IGNORECASE), + re.compile(r"infralink", re.IGNORECASE), + re.compile(r"kill[-_]mcp", re.IGNORECASE), + re.compile(r"command sprawl", re.IGNORECASE), + ], + "E": [ + re.compile(r"\bdocs?\b .*(sprawl|consolidat|hygiene|hygi[èe]ne)", re.IGNORECASE), + re.compile(r"readme", re.IGNORECASE), + re.compile(r"catalogue|catalog\b", re.IGNORECASE), + re.compile(r"changelog", re.IGNORECASE), + re.compile(r"rebrand", re.IGNORECASE), + re.compile(r"\.planning\b", re.IGNORECASE), + re.compile(r"markdown files|fichiers markdown", re.IGNORECASE), + ], + "F": [ + re.compile(r"epic", re.IGNORECASE), + re.compile(r"staleness|stale body|body p[ée]rim[ée]", re.IGNORECASE), + ], +} + + +def mention_lines(text: str, number: int) -> list[str]: + """Lines of ``text`` citing ``#`` (word-bounded: #1647 != #16473).""" + if not text: + return [] + pattern = re.compile(rf"#{number}\b") + return [line.strip() for line in text.splitlines() if pattern.search(line)] + + +def attachment_proofs( + issue: dict[str, Any], + chaperones: dict[int, dict[str, Any]], + issue_comments: list[dict[str, Any]] | None = None, +) -> list[dict[str, Any]]: + """All attachment proofs found for one issue against the chaperones.""" + proofs: list[dict[str, Any]] = [] + number = issue["number"] + for chap in CHAPERONES: + for line in mention_lines(issue.get("body") or "", chap): + proofs.append({"type": "body", "chaperone": chap, "lines": [line]}) + detail = chaperones.get(chap) + if not detail: + continue + for line in mention_lines(detail.get("body") or "", number): + proofs.append({"type": "chaperone_body", "chaperone": chap, "lines": [line]}) + for comment in detail.get("comments") or []: + for line in mention_lines(comment.get("body") or "", number): + proofs.append({"type": "chaperone_comment", "chaperone": chap, "lines": [line]}) + if issue_comments is not None: + for comment in issue_comments: + for chap in CHAPERONES: + for line in mention_lines(comment.get("body") or "", chap): + proofs.append({"type": "issue_comment", "chaperone": chap, "lines": [line]}) + return proofs + + +def classify_axes(title: str, body: str) -> list[str]: + """Axes A-F matched by title+body, in A-F order. Empty list = UNCLASSIFIED.""" + haystack = f"{title}\n{body or ''}" + return [ + axis + for axis in sorted(AXIS_PATTERNS) + if any(pat.search(haystack) for pat in AXIS_PATTERNS[axis]) + ] + + +def audit( + issues: list[dict[str, Any]], + chaperones: dict[int, dict[str, Any]], + issues_comments: dict[int, list[dict[str, Any]]] | None = None, +) -> dict[str, Any]: + """Pure core: tri of open issues into attached / orphaned-by-axis.""" + chaperone_numbers = {i["number"] for i in issues} & set(CHAPERONES) + attached: list[dict[str, Any]] = [] + orphans: list[dict[str, Any]] = [] + for issue in issues: + number = issue["number"] + if number in CHAPERONES: + continue + comments = (issues_comments or {}).get(number) + proofs = attachment_proofs(issue, chaperones, comments) + if proofs: + attached.append({ + "number": number, + "title": issue.get("title") or "", + "proofs": [ + {"type": p["type"], "chaperone": p["chaperone"], "line": p["lines"][0]} + for p in proofs + ], + }) + continue + absence = ["body: 0 citation des chaperons (#%s)" % " #".join(map(str, CHAPERONES))] + if issues_comments is not None: + absence.append("comments: 0 citation des chaperons (--deep)") + orphans.append({ + "number": number, + "title": issue.get("title") or "", + "axes": classify_axes(issue.get("title") or "", issue.get("body") or ""), + "absence": absence, + }) + by_axis: dict[str, list[dict[str, Any]]] = {axis: [] for axis in AXIS_LABELS} + by_axis["UNCLASSIFIED"] = [] + for orphan in orphans: + if orphan["axes"]: + for axis in orphan["axes"]: + by_axis[axis].append(orphan) + else: + by_axis["UNCLASSIFIED"].append(orphan) + return { + "chaperones": list(CHAPERONES), + "scanned": len(issues) - len(chaperone_numbers), + "attached": len(attached), + "orphans": len(orphans), + "deep": issues_comments is not None, + "attached_details": attached, + "orphans_by_axis": by_axis, + } + + +def render_markdown(report: dict[str, Any]) -> str: + lines = [ + "# Audit consolidation-orphans (#16473, acceptance 1)", + "", + "Chaperons : %s" % " ".join(f"#{c}" for c in report["chaperones"]), + f"Issues ouvertes scannées : {report['scanned']} ; rattachées : {report['attached']} ; orphelines : {report['orphans']}" + + (" (scan --deep : commentaires inclus)" if report["deep"] else " (bodies seulement, lancer --deep pour les commentaires)"), + "", + "Un tri, pas un verdict : chaque ligne orpheline liste la preuve de ce qui a été cherché. L'arbitrage de rattachement (acceptance 2) reste humain.", + "", + ] + for axis in [*AXIS_LABELS, "UNCLASSIFIED"]: + entries = report["orphans_by_axis"][axis] + if not entries: + continue + label = AXIS_LABELS[axis] if axis in AXIS_LABELS else "Hors périmètre consolidation apparent" + lines.append(f"## Axe {axis} — {label} ({len(entries)})") + lines.append("") + lines.append("| Issue | Titre | Preuve d'absence |") + lines.append("|---|---|---|") + for orphan in sorted(entries, key=lambda o: o["number"]): + absence = " ; ".join(orphan["absence"]) + lines.append(f"| #{orphan['number']} | {orphan['title'][:90]} | {absence} |") + lines.append("") + return "\n".join(lines) + + +def _gh_json(repo: str, args: list[str]) -> Any: + result = subprocess.run( + ["gh", *args, "--repo", repo], + capture_output=True, + text=True, + encoding="utf-8", + ) + if result.returncode != 0: + raise RuntimeError(f"gh {' '.join(args)} failed (rc={result.returncode}): {result.stderr.strip()}") + return json.loads(result.stdout) if result.stdout.strip() else None + + +def fetch_state(repo: str, limit: int, deep: bool) -> dict[str, Any]: + """Collect live state: open issues + chaperone bodies/comments (+ per-issue comments with --deep).""" + issues = _gh_json(repo, ["issue", "list", "--state", "open", "--limit", str(limit), + "--json", "number,title,body"]) or [] + chaperones: dict[int, dict[str, Any]] = {} + for chap in CHAPERONES: + detail = _gh_json(repo, ["issue", "view", str(chap), "--json", "body,comments"]) or {} + chaperones[chap] = detail + issues_comments: dict[int, list[dict[str, Any]]] | None = None + if deep: + issues_comments = {} + for issue in issues: + number = issue["number"] + detail = _gh_json(repo, ["issue", "view", str(number), "--json", "comments"]) or {} + issues_comments[number] = detail.get("comments") or [] + return {"issues": issues, "chaperones": {str(k): v for k, v in chaperones.items()}, + "issues_comments": {str(k): v for k, v in (issues_comments or {}).items()}, + "deep": deep} + + +def load_input(path: str) -> tuple[list[dict[str, Any]], dict[int, dict[str, Any]], dict[int, list[dict[str, Any]]] | None]: + with open(path, encoding="utf-8") as handle: + state = json.load(handle) + issues = state.get("issues") or [] + raw_chap = state.get("chaperones") or {} + chaperones = {int(k): v for k, v in raw_chap.items()} + raw_comments = state.get("issues_comments") + issues_comments = None + if raw_comments: + issues_comments = {int(k): v for k, v in raw_comments.items()} + return issues, chaperones, issues_comments + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--fetch", action="store_true", help="collect live state via gh") + source.add_argument("--input", metavar="FILE", help="offline JSON state (see module docstring)") + parser.add_argument("--repo", default="jsboige/CoursIA") + parser.add_argument("--limit", type=int, default=400) + parser.add_argument("--deep", action="store_true", help="also scan comments on open issues (N+5 gh calls)") + parser.add_argument("--format", choices=("md", "json"), default="md") + parser.add_argument("--out", help="write the report to FILE instead of stdout") + parser.add_argument("--fail-on-orphans", action="store_true", + help="exit 1 when orphans > 0 (future ratchet wire; NOT the delivery posture)") + args = parser.parse_args(argv) + + if args.fetch: + state = fetch_state(args.repo, args.limit, args.deep) + issues = state["issues"] + chaperones = {int(k): v for k, v in state["chaperones"].items()} + issues_comments = None + if state["issues_comments"]: + issues_comments = {int(k): v for k, v in state["issues_comments"].items()} + else: + issues, chaperones, issues_comments = load_input(args.input) + + report = audit(issues, chaperones, issues_comments) + rendered = json.dumps(report, ensure_ascii=False, indent=2) if args.format == "json" else render_markdown(report) + if args.out: + with open(args.out, "w", encoding="utf-8", newline="\n") as handle: + handle.write(rendered) + print(f"report written to {args.out}", file=sys.stderr) + else: + print(rendered) + if args.fail_on_orphans and report["orphans"] > 0: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_audit_consolidation_orphans.py b/scripts/tests/test_audit_consolidation_orphans.py new file mode 100644 index 0000000000..94fd0c7daf --- /dev/null +++ b/scripts/tests/test_audit_consolidation_orphans.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""Unit tests for the pure core of audit_consolidation_orphans.py (#16473). + +The attachment/axis/rendering functions are network-free; only --fetch talks +to gh and is exercised end-to-end at delivery time, not here. Fixtures encode +the semantics fixed by the Epic body: a bare #N mention counts as attachment +(triage, not verdict), word-bounded numbers (#1647 != #16473), and an orphan +with no axis match stays UNCLASSIFIED rather than being force-classified. +""" + +import json +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from audit_consolidation_orphans import ( # noqa: E402 + CHAPERONES, + attachment_proofs, + audit, + classify_axes, + load_input, + main, + mention_lines, + render_markdown, +) + + +CHAP_body = "#5081 est le chaperon canon." +CHAPERONES_STATE = { + 5081: {"body": "Programme canonique. Sous-grains : #1001.", "comments": []}, + 4362: {"body": "Lakes.", "comments": [{"body": "rattache aussi #1002"}]}, + 13737: {"body": "Structure.", "comments": []}, + 9535: {"body": "Nettoyage.", "comments": []}, + 16473: {"body": "Parapluie.", "comments": []}, +} + + +def _issue(number, title="Une issue", body=""): + return {"number": number, "title": title, "body": body} + + +def test_mention_lines_word_boundary(): + text = "cites #16473 ici\nmais pas #1647 la\n#16473 encore" + assert mention_lines(text, 16473) == ["cites #16473 ici", "#16473 encore"] + assert mention_lines(text, 1647) == ["mais pas #1647 la"] + + +def test_attachment_via_issue_body_see(): + issue = _issue(2001, body="Tranche renumerotation. See #5081") + proofs = attachment_proofs(issue, CHAPERONES_STATE) + assert any(p["type"] == "body" and p["chaperone"] == 5081 for p in proofs) + + +def test_attachment_via_chaperone_body(): + issue = _issue(1001, body="rien") + proofs = attachment_proofs(issue, CHAPERONES_STATE) + assert any(p["type"] == "chaperone_body" and p["chaperone"] == 5081 for p in proofs) + + +def test_attachment_via_chaperone_comment(): + issue = _issue(1002, body="rien") + proofs = attachment_proofs(issue, CHAPERONES_STATE) + assert any(p["type"] == "chaperone_comment" and p["chaperone"] == 4362 for p in proofs) + + +def test_attachment_via_issue_comment_deep_only(): + issue = _issue(3001, body="rien") + assert attachment_proofs(issue, CHAPERONES_STATE) == [] + proofs = attachment_proofs(issue, CHAPERONES_STATE, + issue_comments=[{"body": "Part of #9535"}]) + assert any(p["type"] == "issue_comment" and p["chaperone"] == 9535 for p in proofs) + + +def test_bare_mention_counts_as_attachment(): + # Triage posture: even a negative-prose mention surfaces, human arbitrates. + issue = _issue(4001, body="ne pas dupliquer la doctrine de #5081") + assert attachment_proofs(issue, CHAPERONES_STATE) != [] + + +def test_classify_axes_matches(): + assert classify_axes("Renum Search Part3", "") == ["A"] + assert "C" in classify_axes("Mutualiser les lakes Mathlib", "junctions NTFS") + assert "E" in classify_axes("Doc sprawl", "consolider 150+ markdown files") + assert classify_axes("Backtest strategie X", "Sharpe, CAGR") == [] + + +def test_audit_tri_and_chaperone_exclusion(tmp_path): + issues = [ + _issue(5081, title="chaperon lui-meme", body=CHAP_body), + _issue(5001, title="Renum Infer", body="voir serie"), + _issue(5002, title="Backtest QC", body="Sharpe"), + ] + report = audit(issues, CHAPERONES_STATE) + assert report["scanned"] == 2 # 5081 excluded as a chaperone + assert report["orphans"] == 2 + assert report["attached"] == 0 + axes_5001 = next(o for o in report["orphans_by_axis"]["A"] if o["number"] == 5001) + assert axes_5001["absence"] == ["body: 0 citation des chaperons (#%s)" % " #".join(map(str, CHAPERONES))] + assert any(o["number"] == 5002 for o in report["orphans_by_axis"]["UNCLASSIFIED"]) + + +def test_audit_deep_records_comment_absence(): + issues = [_issue(6001, title="Doublons twins", body="")] + report = audit(issues, CHAPERONES_STATE, issues_comments={6001: [{"body": "rien"}]}) + orphan = next(o for o in report["orphans_by_axis"]["B"] if o["number"] == 6001) + assert any("comments: 0 citation" in a for a in orphan["absence"]) + + +def test_render_markdown_contains_axes_and_absence_proof(): + issues = [_issue(7001, title="Renum quoi", body="")] + report = audit(issues, CHAPERONES_STATE) + md = render_markdown(report) + assert "## Axe A" in md + assert "Preuve d'absence" in md + assert "0 citation des chaperons" in md + assert "pas un verdict" in md + + +def test_main_offline_json_and_exit_codes(tmp_path, capsys): + state = { + "issues": [_issue(8001, title="Renum test", body="See #5081")], + "chaperones": {"5081": {"body": "", "comments": []}}, + } + input_path = tmp_path / "state.json" + input_path.write_text(json.dumps(state), encoding="utf-8") + out_path = tmp_path / "report.json" + + rc = main(["--input", str(input_path), "--format", "json", "--out", str(out_path)]) + assert rc == 0 + data = json.loads(out_path.read_text(encoding="utf-8")) + assert data["attached"] == 1 and data["orphans"] == 0 + + state["issues"][0]["body"] = "aucun lien" + input_path.write_text(json.dumps(state), encoding="utf-8") + rc_ok = main(["--input", str(input_path), "--format", "json", "--out", str(out_path)]) + rc_fail = main(["--input", str(input_path), "--format", "md", + "--fail-on-orphans", "--out", str(tmp_path / "r.md")]) + assert rc_ok == 0 # advisory posture by default + assert rc_fail == 1 # opt-in ratchet wire + + +def test_load_input_roundtrip(tmp_path): + state = { + "issues": [_issue(9001, title="t", body="")], + "chaperones": {"5081": {"body": "b", "comments": [{"body": "c"}]}}, + "issues_comments": {"9001": [{"body": "voir #4362"}]}, + } + path = tmp_path / "s.json" + path.write_text(json.dumps(state), encoding="utf-8") + issues, chaperones, comments = load_input(str(path)) + assert issues[0]["number"] == 9001 + assert 5081 in chaperones + assert comments is not None and 9001 in comments