#!/usr/bin/env python3
"""selftest_sweep.py -- every instrument proves itself, in one command, and the
number it advertises is the number it prints.

WHY THIS EXISTS
  `tools/append_only_audit.py` carried, on its own usage line, the words
  "synthetic repo, N cells" beside a selftest that printed a different N.
  The control plane found it and flagged it twice; it stood until 580a41c because
  nothing in this archive ever compared a tool's self-advertised count against the
  count the tool prints. `doc_census_check.py` now checks that class -- but its
  subject is the census blocks (AMBIGUOUS / MATCHES / NOT_PRINTED / ...) and it
  drives every tool against a revision, so a full run is minutes, not seconds.

  The selftest cell count is a cheaper and more basic claim: it is the number a
  reader uses to decide whether a tool is worth trusting, and the number a
  check-in quotes as proof. Eight tools carry one. One was wrong for days. This
  is the gate for that sentence, and it runs in seconds because a selftest is a
  unit test, not a census.

WHAT IT MEASURES -- one row per tracked tool, one verdict per row
  NO_SELFTEST_FLAG   the source never mentions "--selftest"; the tool is not run
                     (a tool without the flag may do real work on `--selftest`).
  TIMEOUT            the selftest did not finish inside --timeout seconds.
  RUN_FAILED         non-zero exit and no count line: the run died, or failed
                     before it could report.
  NO_COUNT_LINE      exit 0 but nothing matching
                     `selftest: <n> cells|fixtures, <n> passed, <n> failed`.
  SELFTEST_FAILED    a count line exists and reports failed > 0 or passed < total.
  SPLIT              two different values of one unit on the tool's own
                     `--selftest` usage line(s) -- the four-hundred-and-seventy-one
                     over-count defect, at
                     selftest scale.
  COUNT_MISMATCH     exactly one advertised value, and it is not the printed one
                     (value or unit). This is the append_only_audit defect.
  NO_CLAIM           prints a count, advertises none. ADVISORY, not a fault: an
                     absent claim cannot be false.
  OK                 printed a count, every advertised value agrees with it.

  FATAL = every class above except NO_CLAIM. Exit 1 if any row is fatal, else 0.

WHAT COUNTS AS A CLAIM, PRECISELY
  Only the module docstring (read through `ast.get_docstring`, so a fixture
  string in the BODY is not a claim -- `doc_census_check.py` contains the literal
  "selftest: N cells, N passed, N failed" inside a test, and reading the file as
  text would call that a claim), and only on a line that also contains the token
  `--selftest`. Prose elsewhere in the docstring is not a claim: the same
  doc_census_check docstring quotes a thirteen-cell claim as the defect the tool
  itself catches, and a scanner that flagged its own bug report would be wrong.

DISCOVERY, AND WHY UNTRACKED FILES ARE PRINTED
  Tracked paths only, from `git ls-files` in --repo -- the lesson doc_census_check
  paid for: an untracked tool is invisible to a revision-scoped census, and this
  gate inherits that blindness rather than pretending to fix it. Every `tools/*.py`
  that IS present but untracked is therefore listed as SKIPPED_UNTRACKED with its
  name, never silently dropped. With --tools-dir, or with no git, discovery is a
  plain glob of that directory and the row says so.

READ-ONLY: the only write path is `--json PATH`. It does not edit a tool, and it
does not decide that a claim SHOULD exist -- NO_CLAIM is advisory by design.

USAGE:
  python3 selftest_sweep.py --repo /root/troupe-standup
  python3 selftest_sweep.py --tools-dir /tmp/fake --json /tmp/sweep.json
  python3 selftest_sweep.py --selftest          # 18 cells, parser and sweep
"""
from __future__ import annotations

import argparse
import ast
import glob
import json
import os
import re
import subprocess
import sys
import tempfile

UNITS = ("cells", "fixtures")

# `selftest: 18 cells, 18 passed, 0 failed`
# `self-test: 19 fixtures, 19 passed, 0 failed`
# `selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed`
PRINTED_RE = re.compile(
    r"(?:selftest|self-test)\s*:\s*([\d,]+)\s+(cells|fixtures)\b"
    r"[^\n]*?\b([\d,]+)\s+passed\b[^\n]*?\b([\d,]+)\s+failed\b"
)
CLAIM_RE = re.compile(r"(?<![\w.])([\d,]+)\s+(cells|fixtures)\b")
SELFTEST_FLAG_RE = re.compile(r"""["']--selftest["']""")

FATAL = (
    "SELFTEST_FAILED",
    "RUN_FAILED",
    "TIMEOUT",
    "NO_COUNT_LINE",
    "NO_SELFTEST_FLAG",
    "COUNT_MISMATCH",
    "SPLIT",
    "UNREADABLE",
)


def _int(text: str) -> int:
    return int(text.replace(",", ""))


def parse_printed(text: str):
    """(count, unit, passed, failed) from the first count line, or None."""
    m = PRINTED_RE.search(text or "")
    if not m:
        return None
    return _int(m.group(1)), m.group(2), _int(m.group(3)), _int(m.group(4))


def usage_claims(doc: str):
    """[(value, unit, line)] -- claims on lines that name --selftest."""
    out = []
    for line in (doc or "").splitlines():
        if "--selftest" not in line:
            continue
        for m in CLAIM_RE.finditer(line):
            out.append((_int(m.group(1)), m.group(2), line.strip()))
    return out


def module_docstring(path: str):
    """(docstring, source). The docstring is AST-read so body fixtures are not claims."""
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            src = fh.read()
    except OSError as exc:
        return None, f"<unreadable: {exc}>"
    try:
        tree = ast.parse(src)
    except SyntaxError:
        return None, src
    return (ast.get_docstring(tree) or ""), src


# ---------------------------------------------------------------- discovery

def discover(repo: str, tools_dir: str | None):
    """(rows, source, untracked).

    rows: [(display_name, path)] -- path is absolute, display_name is tools/x.py.
    """
    if tools_dir:
        d = os.path.abspath(tools_dir)
        paths = sorted(p for p in glob.glob(os.path.join(d, "*.py"))
                       if os.path.isfile(p))
        return ([(os.path.basename(p), p) for p in paths],
                f"glob of --tools-dir {d} (no git consulted)", [])

    d = os.path.join(os.path.abspath(repo), "tools")
    have_git = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse",
                               "--git-dir"], capture_output=True).returncode == 0
    if not have_git:
        paths = sorted(p for p in glob.glob(os.path.join(d, "*.py"))
                       if os.path.isfile(p))
        return ([(os.path.basename(p), p) for p in paths],
                f"glob of {d} (not a git tree)", [])

    tracked = subprocess.run(
        ["git", "-C", os.path.abspath(repo), "ls-files", "--", "tools"],
        capture_output=True, text=True).stdout.splitlines()
    named = [t for t in tracked
             if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]]
    rows = [(os.path.basename(t), os.path.join(os.path.abspath(repo), t))
            for t in sorted(named)]

    untracked = subprocess.run(
        ["git", "-C", os.path.abspath(repo), "ls-files", "--others",
         "--exclude-standard", "--", "tools"],
        capture_output=True, text=True).stdout.splitlines()
    stray = [t for t in untracked
             if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]]

    rev = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse",
                          "--short", "HEAD"], capture_output=True, text=True).stdout.strip()
    return rows, f"git ls-files at {rev or 'HEAD'} ({len(rows)} tracked in tools/)", sorted(stray)


# ------------------------------------------------------------------- judging

def judge(doc, out: str, rc: int, timed_out: bool):
    """One verdict dict from a completed run. Pure: no I/O, no clock."""
    printed = parse_printed(out)
    claims = usage_claims(doc or "")
    rec = {
        "printed": ({"count": printed[0], "unit": printed[1],
                     "passed": printed[2], "failed": printed[3]} if printed else None),
        "claims": [{"value": v, "unit": u, "line": ln} for v, u, ln in claims],
        "verdict": None,
        "detail": "",
    }

    if timed_out:
        rec["verdict"] = "TIMEOUT"
        rec["detail"] = "selftest did not finish"
        return rec

    if printed is None:
        if rc != 0:
            rec["verdict"] = "RUN_FAILED"
            tail = [l for l in (out or "").splitlines() if l.strip()][-1:] or [""]
            rec["detail"] = f"rc={rc}; last line: {tail[0][:160]}"
        else:
            rec["verdict"] = "NO_COUNT_LINE"
            rec["detail"] = "rc=0 but no '<n> cells|fixtures, <n> passed, <n> failed' line"
        return rec

    count, unit, passed, failed = printed
    if failed > 0 or passed < count:
        rec["verdict"] = "SELFTEST_FAILED"
        rec["detail"] = f"{passed}/{count} passed, {failed} failed"
        return rec

    same_unit = [(v, ln) for v, u, ln in claims if u == unit]
    values = sorted({v for v, _ in same_unit})
    if len(values) > 1:
        rec["verdict"] = "SPLIT"
        rec["detail"] = ("advertises " + " and ".join(f"{v} {unit}" for v in values)
                         + f"; prints {count}")
        return rec
    if not claims:
        rec["verdict"] = "NO_CLAIM"
        rec["detail"] = f"prints {count} {unit}; advertises none (advisory)"
        return rec
    if values and values[0] == count:
        rec["verdict"] = "OK"
        rec["detail"] = f"{count} {unit}"
        return rec

    rec["verdict"] = "COUNT_MISMATCH"
    adv = ", ".join(f"{v} {u}" for v, u, _ in claims)
    rec["detail"] = f"advertises {adv}; prints {count} {unit}"
    return rec


def sweep(rows, repo: str, timeout: int, verbose: bool):
    results = []
    for name, path in rows:
        doc, src = module_docstring(path)
        rec = {"tool": name, "path": path}
        if doc is None and not SELFTEST_FLAG_RE.search(src or ""):
            rec.update({"verdict": "UNREADABLE", "detail": "source unreadable",
                        "printed": None, "claims": []})
            results.append(rec)
            continue
        if not SELFTEST_FLAG_RE.search(src or ""):
            rec.update({"verdict": "NO_SELFTEST_FLAG", "printed": None,
                        "claims": [{"value": v, "unit": u, "line": ln}
                                   for v, u, ln in usage_claims(doc or "")],
                        "detail": "no \"--selftest\" in source; not run"})
            results.append(rec)
            continue

        timed_out = False
        rc, out = 0, ""
        try:
            proc = subprocess.run([sys.executable, path, "--selftest"],
                                  cwd=repo, capture_output=True, text=True,
                                  errors="replace", timeout=timeout)
            rc, out = proc.returncode, (proc.stdout or "") + (proc.stderr or "")
        except subprocess.TimeoutExpired:
            timed_out = True
        except OSError as exc:
            rc, out = 127, f"<exec failed: {exc}>"

        verdict = judge(doc, out, rc, timed_out)
        rec.update(verdict)
        rec["rc"] = rc
        results.append(rec)
        if verbose:
            print(f"  . {name}: {rec['verdict']} rc={rc}")
    return results


# --------------------------------------------------------------------- report

def report(results, source, untracked, quiet=False):
    fatal = [r for r in results if r["verdict"] in FATAL]
    if not quiet:
        print(f"discovery: {source}")
        for name in untracked:
            print(f"  SKIPPED_UNTRACKED {name}  (present in tools/, not in the tree)")
        width = max([len(r["tool"]) for r in results] + [4])
        print()
        print(f"{'tool':<{width}}  {'verdict':<16} {'claim':>12} {'printed':>12}  detail")
        for r in results:
            claims = r.get("claims") or []
            claim_s = (", ".join(f"{c['value']} {c['unit']}" for c in claims)
                       if claims else "-")
            p = r.get("printed")
            printed_s = f"{p['count']} {p['unit']}" if p else "-"
            print(f"{r['tool']:<{width}}  {r['verdict']:<16} {claim_s:>12} "
                  f"{printed_s:>12}  {r['detail']}")
        print()
    ok = sum(1 for r in results if r["verdict"] == "OK")
    print(f"sweep: {len(results)} tools, {ok} OK, {len(fatal)} fatal"
          + (f" ({', '.join(sorted({r['verdict'] for r in fatal}))})" if fatal else ""))
    return 1 if fatal else 0


# ------------------------------------------------------------------- selftest

def selftest():
    bad = []

    def cell(name, ok, detail=""):
        print(f"{'PASS' if ok else 'FAIL'} {name}"
              + (f"  [{detail}]" if (detail and not ok) else ""))
        if not ok:
            bad.append(name)

    cell("c1 parse plain cells line",
         parse_printed("selftest: 18 cells, 18 passed, 0 failed") == (18, "cells", 18, 0))
    cell("c2 parse fixtures with parenthetical",
         parse_printed("selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed")
         == (14, "fixtures", 14, 0))
    cell("c3 parse hyphenated self-test",
         parse_printed("self-test: 19 fixtures, 19 passed, 0 failed")
         == (19, "fixtures", 19, 0))
    cell("c4 parse thousands separator",
         parse_printed("selftest: 1,204 cells, 1,204 passed, 0 failed")
         == (1204, "cells", 1204, 0))
    cell("c5 no count line -> None",
         parse_printed("all good, everything fine") is None)

    doc = ("USAGE:\n  python3 t.py --selftest          # synthetic repo, 18 cells\n")
    cell("c6 claim on the --selftest usage line",
         usage_claims(doc) and usage_claims(doc)[0][:2] == (18, "cells"))
    prose = ("  returned a docstring claim of `13 cells` against a selftest that\n"
             "  prints `18`\n")
    cell("c7 prose beside the word selftest is not a claim",
         usage_claims(prose) == [])
    cell("c8 claim with a unit the tool never prints is still a claim",
         usage_claims("  --selftest   # 7 fixtures\n")[0][:2] == (7, "fixtures"))

    j = judge(doc, "selftest: 13 cells, 13 passed, 0 failed", 0, False)
    cell("c9 advertised 18 vs printed 13 -> COUNT_MISMATCH",
         j["verdict"] == "COUNT_MISMATCH", j["verdict"])
    j = judge(doc, "selftest: 18 cells, 18 passed, 0 failed", 0, False)
    cell("c10 advertised 18 vs printed 18 -> OK", j["verdict"] == "OK", j["verdict"])
    j = judge("  --selftest   # 7 fixtures\n", "selftest: 7 cells, 7 passed, 0 failed", 0, False)
    cell("c11 unit mismatch is a mismatch, not a pass",
         j["verdict"] == "COUNT_MISMATCH", j["verdict"])
    j = judge("  --selftest   # 18 cells\n  --selftest   # 13 cells\n",
              "selftest: 18 cells, 18 passed, 0 failed", 0, False)
    cell("c12 two advertised values -> SPLIT", j["verdict"] == "SPLIT", j["verdict"])
    j = judge("", "selftest: 8 cells, 7 passed, 1 failed", 0, False)
    cell("c13 failed>0 is fatal even at rc=0",
         j["verdict"] == "SELFTEST_FAILED", j["verdict"])
    j = judge("", "Traceback (most recent call last):", 1, False)
    cell("c14 rc!=0 with no count line -> RUN_FAILED", j["verdict"] == "RUN_FAILED", j["verdict"])
    j = judge("", "", 0, False)
    cell("c15 rc=0 with no count line -> NO_COUNT_LINE", j["verdict"] == "NO_COUNT_LINE", j["verdict"])

    # c16/c17: end to end through main(), on a directory of fake instruments.
    good = ('#!/usr/bin/env python3\n"""fake_good.py -- a fake.\n\nUSAGE:\n'
            '  python3 fake_good.py --selftest     # 7 cells\n"""\n'
            'import sys\nif "--selftest" in sys.argv:\n'
            '    print("selftest: 7 cells, 7 passed, 0 failed")\n    sys.exit(0)\n')
    bad_tool = ('#!/usr/bin/env python3\n"""fake_bad.py -- a fake.\n\nUSAGE:\n'
                '  python3 fake_bad.py --selftest      # 13 cells\n"""\n'
                'import sys\nif "--selftest" in sys.argv:\n'
                '    print("selftest: 18 cells, 18 passed, 0 failed")\n    sys.exit(0)\n')
    flagless = ('#!/usr/bin/env python3\n"""fake_flagless.py -- no selftest."""\n'
                'print("did some work")\n')
    with tempfile.TemporaryDirectory() as td:
        for nm, body in (("fake_good.py", good), ("fake_bad.py", bad_tool),
                         ("fake_flagless.py", flagless)):
            with open(os.path.join(td, nm), "w") as fh:
                fh.write(body)
        rc_bad = main(["--tools-dir", td, "--quiet"])
        rows, source, untracked = discover(td, td)
        res = sweep(rows, td, 30, False)
        verdicts = {r["tool"]: r["verdict"] for r in res}
        cell("c16 end to end: a wrong claim is fatal",
             rc_bad == 1 and verdicts.get("fake_bad.py") == "COUNT_MISMATCH",
             f"rc={rc_bad} {verdicts}")
        cell("c17 end to end: correct claim OK, flagless caught, exit 1",
             verdicts.get("fake_good.py") == "OK"
             and verdicts.get("fake_flagless.py") == "NO_SELFTEST_FLAG",
             str(verdicts))
        jd = os.path.join(td, "sweep.json")
        main(["--tools-dir", td, "--quiet", "--json", jd])
        try:
            with open(jd) as fh:
                blob = json.load(fh)
            ok_json = (isinstance(blob, dict) and len(blob.get("tools", [])) == 3
                       and blob.get("fatal") == 2)
        except (OSError, ValueError):
            ok_json = False
        cell("c18 --json carries every row and the fatal count", ok_json)

    print(f"selftest: 18 cells, {18 - len(bad)} passed, {len(bad)} failed")
    return 1 if bad else 0


# ----------------------------------------------------------------------- main

def main(argv=None):
    ap = argparse.ArgumentParser(
        description="Run every tool's --selftest and check its advertised cell count.")
    ap.add_argument("--repo", default="/root/troupe-standup")
    ap.add_argument("--tools-dir", default=None,
                    help="sweep this directory instead of the repo's tracked tools/")
    ap.add_argument("--timeout", type=int, default=120, metavar="S")
    ap.add_argument("--json", default=None, metavar="PATH")
    ap.add_argument("--quiet", action="store_true")
    ap.add_argument("--verbose", action="store_true")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args(argv)

    if a.selftest:
        return selftest()

    rows, source, untracked = discover(a.repo, a.tools_dir)
    if not rows:
        print("sweep: 0 tools, nothing discovered")
        return 1
    results = sweep(rows, a.repo if not a.tools_dir else os.path.abspath(a.tools_dir),
                    a.timeout, a.verbose)
    rc = report(results, source, untracked, a.quiet)

    if a.json:
        fatal = [r for r in results if r["verdict"] in FATAL]
        with open(a.json, "w") as fh:
            json.dump({"repo": os.path.abspath(a.repo),
                       "tools_dir": a.tools_dir,
                       "discovery": source,
                       "untracked": untracked,
                       "tools": results,
                       "fatal": len(fatal),
                       "ok": sum(1 for r in results if r["verdict"] == "OK")}, fh, indent=1)
    return rc


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