#!/usr/bin/env python3
"""doc_census_check.py -- does a tool's own docstring census match the census
that tool PRINTS?

WHY THIS EXISTS
  This archive's three scanners each carry a census table in their module
  docstring: counts that a reader cites, that no command checks. The docstring
  is prose; prose is not executed. On 2026-10-16 the control plane returned
  exactly this defect against a sibling scanner, in its own words:

    "The scanner's own docstring carries a number that does not match the
     census it prints -- the exact class the tool exists to catch."

  tools/citation_resolve_check.py said `471 UNRESOLVED` on one line and
  `441 UNRESOLVED` nine lines later, over a census the tool prints as 471.
  Nothing was checking the docstring against the output beside it. That is this
  tool's whole job.

WHAT IT MEASURES (read-only: this tool only writes --json)
  Two independent verdicts per tool, and they do not share evidence.

  1. INTERNAL SPLIT -- no oracle, no run. Within one docstring, every numeric
     claim of the form `<value> <unit>` is collected; two claims of ONE unit
     that a single revision scope covers, carrying two different values, is a
     split. This is the verdict that catches the 441/471 instance, and it is
     why the tool is useful with no subprocess at all.

  2. ORACLE -- the tool is RUN, at the revision its own docstring names, and
     each claim is compared against the value that run prints for that unit.
       MATCHES       claim == printed value
       STALE         claim != printed value, and the printed unit is not
                     declared clone-local -- a defect claim
       CLONE_LOCAL   claim != printed value but the tool's own output marks
                     that unit clone-local ("blobs in THIS clone's local object
                     store"), so drift is the documented expectation, not a
                     defect. Reported, never counted as one.
       AMBIGUOUS     the tool prints that unit with 2+ values at that rev
                     (e.g. `citations` prints once per section), so there is no
                     single oracle value; both are printed and no claim is made
       NOT_PRINTED   the unit does not appear in this tool's output at all

  A claim is `<digits> <unitword>` with the unit word ADJACENT, which is the
  archive's own notation. A bare number is not a claim and is not checked.

SCOPE, STATED RATHER THAN IMPLIED
  Only module docstrings of .py tools are scanned: a docstring is where a census
  table lives. Only revisions NAMED IN THE BLOCK are used as oracles -- a block
  naming no revision is checked against the tool's default (HEAD) and tagged, so
  a stale rev is never silently attributed. A tool that cannot be run at a
  revision (queue_claim_check.py has no --rev; the queue is materialised from
  git and handed to it by --queue) is driven through a per-tool recipe declared
  below, with a temp file that is removed.

  WHAT THIS DOES NOT CHECK, and the measured size of the gap. A claim is only
  adjudicated when the unit it names is a unit the tool PRINTS. Over the three
  scanners at d355395 the run read ninety-two claims: twelve were verified
  against the run they point at, eight were in units the tool prints with two
  values (SECTION_SPLIT_BY_TOOL -- no oracle), and seventy-one were in units the
  tool never prints at all. So the oracle's reach is narrow and its precision is
  the reason: it reported exactly one bad claim on that corpus, and that one was
  the returned defect. A wider rule was tried first and falsified by its own
  first run -- num_agreement_check.py flagged a hundred and seventeen
  restatements across the archive, too coarse to be read -- so the narrow rule
  is the one that ships. Two claim forms are read (`<value> <unit>` and
  `<unit>: <value>`); a claim written in neither form is not seen, and that is a
  stated gap, not a silent one.

  NO CENSUS OF THIS TOOL'S OWN VERDICTS APPEARS IN THIS DOCSTRING, deliberately:
  this scanner scans itself, so a count of its own output written here would
  have to agree with the output of the run that the same commit changes. That is
  the limit citation_hash_class.py records in the same words -- a commit cannot
  measure the file it introduces -- and here it is avoided rather than
  discovered.

USAGE
  python3 doc_census_check.py [--repo DIR] [--rev HEAD] [--tools NAME...]
      [--json OUT] [--no-run] [--selftest]
"""
import argparse
import ast
import collections
import json
import os
import re
import subprocess
import sys
import tempfile

DEFAULT_REPO = "/root/troupe-standup"

# A claim, in either of the two forms this archive writes:
#   `<value> <unit>`      -- "471 UNRESOLVED", "1,612 citations"
#   `<unit> [: =] <value>` -- "ACS = 72", "overlap = 20" (markdown bold allowed)
# A bare number is not a claim. Deliberately not "any word": `471 cannot be
# resolved` is prose, and a checker that flags prose is a checker nobody reads.
CLAIM = re.compile(
    r"(?<![0-9A-Za-z_.])(?P<num>[0-9][0-9,]*)(?P<sep>[^A-Za-z0-9\n]{0,2})"
    r"(?P<unit>[A-Za-z_][A-Za-z0-9_]*|[A-Z][A-Z_]{2,})\b")
CLAIM_EQ = re.compile(
    r"(?P<unit>[A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*\**\s*(?P<num>[0-9][0-9,]*)")
REV_LONG = re.compile(r"\b[0-9a-f]{7,40}\b")

# Function words and units this tool does not adjudicate. `at` in "2026-10-15 at
# rev 9188dd3" is not a census quantity; neither is `B` (bytes differ by
# revision by design, and "1,201,550 B" is a size, not a count). Stated so the
# exclusion is a rule rather than a habit.
STOPWORDS = {
    "at", "the", "of", "in", "is", "are", "was", "and", "or", "to", "for", "on",
    "by", "as", "it", "its", "that", "this", "these", "those", "with", "from",
    "no", "not", "one", "two", "three", "all", "any", "be", "been", "has",
    "have", "had", "his", "her", "our", "your", "their", "but", "if", "then",
    "than", "so", "up", "out", "off", "over", "into", "run", "runs", "may",
    "can", "same", "each", "both", "per", "via", "b", "kb", "mb", "gb",
}

# How to run each tool at a revision. `queue_claim_check.py` takes no --rev: its
# subject is a FILE, so the file is materialised at the revision and handed to
# it. Any tool not listed here is run as `<script> --repo R --rev REV`.
def _recipe_queue(repo, rev, tmpdir):
    dst = os.path.join(tmpdir, f"WORK-QUEUE.{rev}.md")
    blob = subprocess.run(["git", "-C", repo, "show", f"{rev}:WORK-QUEUE.md"],
                          capture_output=True).stdout
    with open(dst, "wb") as fh:
        fh.write(blob)
    return ["--queue", dst, "--census"]


RECIPES = {"queue_claim_check.py": _recipe_queue}


def git(repo, *a):
    return subprocess.run(["git", "-C", repo, *a], capture_output=True).stdout


def docstring_span(path, text):
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return None
    if not tree.body or not isinstance(tree.body[0], ast.Expr):
        return None
    node = tree.body[0]
    if not isinstance(node.value, ast.Constant) or not isinstance(node.value.value, str):
        return None
    return node.lineno, node.end_lineno


def blocks(lines, lo, hi):
    """Blank-line-delimited paragraphs -- the same scope the sibling scanners
    use (citation_resolve_check.py for inheritance, num_agreement_check.py for
    quantity agreement), so one paragraph is one scope in all three."""
    start, buf = lo, []
    for ln in range(lo, hi + 1):
        raw = lines[ln - 1]
        if raw.strip() == "":
            if buf:
                yield start, buf
            buf, start = [], ln + 1
        else:
            buf.append((ln, raw))
    if buf:
        yield start, buf


# ------------------------------------------------- printed census -> oracle
def parse_printed(stdout):
    """stdout -> {unit: {value: [line, ...]}} plus the set of clone-local units.

    Four line shapes are read, all of them the scanners' own:
      `  RESOLVES        113`      indented label then count (family rows)
      `  citations: 410`           label then count
      `corpus: 307 files at ...`   count then unit
      `overlap A&B = 20 ; ...`     inline assignments, split on the separators
    A unit whose printed line (or the line above it, for a wrapped phrase) says
    "clone" is clone-local: its count is expected to move with the clone.
    """
    units = collections.defaultdict(lambda: collections.defaultdict(list))
    clone = set()
    lines = stdout.split("\n")

    def add(unit, value, ln, cloneline):
        u = unit.strip().lower()
        if len(u) < 3 or not re.fullmatch(r"[a-z_][a-z0-9_]*", u) or u in STOPWORDS:
            return
        units[u][value].append(ln)
        if "clone" in cloneline.lower():
            clone.add(u)

    for i, raw in enumerate(lines, 1):
        ctx = raw + " " + (lines[i - 2] if i >= 2 else "")
        m = re.match(r"^\s{2,}([A-Za-z_][A-Za-z0-9_]*)\s+([0-9][0-9,]*)\s*$", raw)
        if m:                                        # family row: label  count
            add(m.group(1), int(m.group(2).replace(",", "")), i, ctx)
            continue
        m = re.match(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*([0-9][0-9,]*)\b", raw)
        if m:                                        # label: count
            add(m.group(1), int(m.group(2).replace(",", "")), i, ctx)
        for m in re.finditer(r"([0-9][0-9,]*)\s+(?:[a-z]+ )?([A-Za-z_][A-Za-z0-9_]*)\b", raw):
            add(m.group(2), int(m.group(1).replace(",", "")), i, ctx)
        for seg in re.split(r"[;]", raw):            # inline `X = N`
            m = re.match(r"^\s*(.*?)\s*[:=]\s*([0-9][0-9,]*)\s*$", seg)
            if not m or not m.group(1):
                continue
            words = re.findall(r"[A-Za-z_][A-Za-z0-9_]*", m.group(1))
            if not words:
                continue
            val = int(m.group(2).replace(",", ""))
            add("_".join(w.lower() for w in words), val, i, ctx)   # whole phrase
            add(words[-1], val, i, ctx)      # the nearest label: `overlap A&B`
        # the scanners' own unit declaration: `... 6961 (unit: occurrences)`.
        # The unit is named AFTER its number, which no other rule reads.
        for m in re.finditer(r"\(\s*unit\s*:\s*([A-Za-z_][A-Za-z0-9_]*)\s*\)", raw):
            left = raw[:m.start()]
            nums = re.findall(r"([0-9][0-9,]*)\s*$", left.rstrip())
            if nums:
                add(m.group(1), int(nums[-1].replace(",", "")), i, ctx)
    return {k: dict(v) for k, v in units.items()}, clone


def claims_in(blk):
    """[(line, value, unit, token, offset)] for one block.

    Both claim forms are read and a claim found by both is reported once. The
    offset is measured on the block text so a RESTATEMENT (far apart) can be
    told from an ENUMERATION of neighbouring sub-counts, the same
    discrimination num_agreement_check.py applies.
    """
    out, off = [], 0
    for ln, raw in blk:
        found = {}
        for m in CLAIM.finditer(raw):
            u = m.group("unit")
            if len(u) < 3 or u.lower() in STOPWORDS or re.fullmatch(r"[qQ]\d+", u):
                continue
            found[m.start()] = (int(m.group("num").replace(",", "")), u, m.group(0))
        for m in CLAIM_EQ.finditer(raw):
            u = m.group("unit")
            if (len(u) < 3 or u.lower() in STOPWORDS
                    or re.fullmatch(r"[qQ]\d+", u) or m.start() in found):
                continue
            if m.start() and re.match(r"[0-9A-Za-z_.]", raw[m.start() - 1]):
                continue                      # `WORK-QUEUE.md:8586` is a citation
            found[m.start()] = (int(m.group("num").replace(",", "")), u, m.group(0))
        for pos in sorted(found):
            val, u, tok = found[pos]
            out.append((ln, val, u, tok, off + pos))
        off += len(raw) + 1
    return out


def is_label(unit):
    """An uppercase census label -- the vocabulary a split is checked in when
    the tool has not been run (no oracle available)."""
    return len(unit) >= 3 and re.fullmatch(r"[A-Z][A-Z_]*", unit) is not None


def keys_for(unit, printed):
    """Candidate oracle keys for a claimed unit word."""
    cands = [unit, unit.lower(), unit.lower().rstrip("s"), unit.upper()]
    seen, out = set(), []
    for c in cands:
        if c.lower() in printed and c.lower() not in seen:
            seen.add(c.lower())
            out.append(c.lower())
    return out


def oracle(printed, clone, unit, value):
    """(verdict, detail) for one claim; `printed` is one revision's map."""
    keys = keys_for(unit, printed)
    vals = set()
    for k in keys:
        vals |= set(printed[k])
    if not vals:
        return "NOT_PRINTED", ""
    if len(vals) > 1:
        return "AMBIGUOUS", "printed: " + ", ".join(str(v) for v in sorted(vals))
    printed_val = vals.pop()
    if printed_val == value:
        return "MATCHES", ""
    if any(k in clone for k in keys):
        return "CLONE_LOCAL", f"printed {printed_val}"
    return "STALE", f"printed {printed_val}"


# --------------------------------------------------------------- self-test
# Synthetic stdout, so the parsers and the verdicts are executed, not asserted.
FAKE_STDOUT = """\
corpus: 297 markdown files at babdadc in /root/troupe-standup
--- EXPLICIT `<path>:<line>` -- a CLAIM (unit: citations) ---
  citations: 410
  RESOLVES        113
  STALE_LINE      0
--- BARE `:<line>` -- a CANDIDATE, never a claim (unit: citations) ---
  citations: 1612
  UNRESOLVED      471
file mentions with no line: 6961 (unit: occurrences) -- these seed the scope
index: 871 blobs (unit: blobs in THIS clone's local object store), 680 commits
overlap A&B = 20 ; ACS-only = 52
"""
FAKE_DOC = """\
THE CENSUS AT HEAD, WITH ITS UNITS (measured 2026-10-16, rev babdadc, 297 .md)
    BARE     1612 citations: 255 EXTERNAL_NAMED / 471 UNRESOLVED.
  Read every figure as a citation count, never as a defect count: 441 UNRESOLVED
  is not 441 errors.

A SECOND BLOCK, no revision named, quoting 471 UNRESOLVED and nothing else.
"""


def _fixture_checks():
    printed, clone = parse_printed(FAKE_STDOUT)
    ok = True

    def check(name, got, want):
        nonlocal ok
        good = got == want
        ok = ok and good
        print(f"{'PASS' if good else 'FAIL'} {name}")
        if not good:
            print(f"      got  {got}")
            print(f"      want {want}")

    check("parse: UNRESOLVED single", sorted(printed.get("unresolved", {})), [471])
    check("parse: citations ambiguous", sorted(printed.get("citations", {})), [410, 1612])
    check("parse: count-then-unit", sorted(printed.get("occurrences", {})), [6961])
    check("parse: one word skipped", sorted(printed.get("files", {})), [297])
    check("parse: clone-local marked", "blobs" in clone, True)
    # the phrase rule: `ACS-only = 52` must NOT register as the unit `acs`
    check("parse: inline phrase key", sorted(printed.get("acs_only", {})), [52])

    lines = FAKE_DOC.split("\n")
    got, first_block, revs_seen = [], [], set()
    for first, blk in blocks(lines, 1, len(lines)):
        revs = sorted(set(REV_LONG.findall("\n".join(r for _, r in blk))))
        revs_seen |= set(revs)
        for ln, val, unit, _tok, _pos in claims_in(blk):
            got.append((unit, val))
            if first == 1:
                first_block.append((unit, val))
    # the fixture that matters: one unit, one rev scope, two values -> a split
    check("claims: both UNRESOLVED values in one block",
          [uv for uv in first_block if uv[0] == "UNRESOLVED"],
          [("UNRESOLVED", 471), ("UNRESOLVED", 441)])
    check("claims: `471 cannot be resolved` is not a claim",
          any(u in ("cannot", "resolved") for u, _ in got), False)
    check("claims: the block's rev is read", "babdadc" in revs_seen, True)
    check("oracle: 471 matches", oracle(printed, clone, "UNRESOLVED", 471),
          ("MATCHES", ""))
    check("oracle: 441 is stale", oracle(printed, clone, "UNRESOLVED", 441),
          ("STALE", "printed 471"))
    check("oracle: ambiguous unit makes no claim",
          oracle(printed, clone, "citations", 1612),
          ("AMBIGUOUS", "printed: 410, 1612"))
    check("oracle: clone-local is not a defect",
          oracle(printed, clone, "blobs", 1234), ("CLONE_LOCAL", "printed 871"))
    check("oracle: unprinted unit", oracle(printed, clone, "wombats", 3),
          ("NOT_PRINTED", ""))
    return 0 if ok else 1


# ------------------------------------------------------------------- driver
def run_tool(repo, script, rev, tmpdir, timeout=600):
    args = RECIPES.get(os.path.basename(script), None)
    if args is None:
        argv = [sys.executable, script, "--repo", repo, "--rev", rev]
    else:
        argv = [sys.executable, script] + args(repo, rev, tmpdir)
    p = subprocess.run(argv, capture_output=True, timeout=timeout)
    return p.stdout.decode("utf-8", "replace"), p.returncode


def unit_scope(unit, printed, clone, no_run):
    """Is a split in this unit checkable, and why not if it is not?

    With the oracle (printed map) in hand a unit is checkable when the tool
    prints it with exactly ONE value at the block's revision: the docstring and
    the output then talk about one number. A unit the tool prints with 2+ values
    is SECTION-SPLIT BY THE TOOL'S OWN DESIGN (`citations` prints once under
    EXPLICIT and once under BARE), so a docstring that mirrors both is correct
    and is reported as such, not flagged. A unit the tool never prints is not a
    census unit here. Without a run, only uppercase census labels are checked.
    """
    if no_run or printed is None:
        return ("LABEL_ONLY" if is_label(unit) else "UNCHECKED_NO_ORACLE")
    keys = keys_for(unit, printed)
    vals = set()
    for k in keys:
        vals |= set(printed[k])
    if not vals:
        return "NOT_A_CENSUS_UNIT"
    if len(vals) > 1:
        return "SECTION_SPLIT_BY_TOOL"
    return "CHECKABLE"


def check_tool(repo, script, name, cache, run=True, tmpdir=".", isolation=240):
    text = open(script, encoding="utf-8", errors="replace").read()
    span = docstring_span(script, text)
    if span is None:
        return None
    lines = text.split("\n")
    lo, hi = span
    blk_claims = []
    for first, blk in blocks(lines, lo, hi):
        btext = "\n".join(raw for _, raw in blk)
        revs = sorted(set(REV_LONG.findall(btext)))
        cl = claims_in(blk)
        if cl:
            blk_claims.append((first, revs, cl, btext))

    # ---- the oracle first: every revision the docstring names is run once.
    printed_by_rev, verdicts, failures = {}, [], []
    if run:
        for first, revs, cl, _b in blk_claims:
            for rev in (revs or ["HEAD"]):
                key = (name, rev)
                if key not in cache:
                    try:
                        out, rc = run_tool(repo, script, rev, tmpdir)
                        cache[key] = (parse_printed(out), rc, out)
                    except Exception as exc:                    # noqa: BLE001
                        cache[key] = (None, None, f"<run failed: {exc}>")
                printed_by_rev.setdefault((first, rev), cache[key])

    # 1. INTERNAL SPLIT: one unit, one revision scope, two values, far enough
    #    apart to be a restatement rather than a neighbouring sub-count.
    splits, unchecked = [], collections.Counter()
    by_unit = collections.defaultdict(list)
    for first, revs, cl, _b in blk_claims:
        scope = tuple(revs)
        for ln, val, unit, _t, pos in cl:
            by_unit[(unit.upper() if unit.isupper() else unit.lower(), scope)].append(
                (first, ln, val, pos))
    for (unit, scope), items in sorted(by_unit.items()):
        vals = {v for _, _, v, _ in items}
        if len(vals) < 2:
            continue
        # a restatement is a pair far apart; an enumeration is neighbours
        far = any(abs(b[3] - a[3]) >= isolation
                  for a, b in [(x, y) for x in items for y in items]
                  if a[1] != b[1] and a[2] != b[2])
        if not far:
            continue
        rev_for_oracle = (items[0][0], scope[0] if len(scope) == 1 else "HEAD")
        entry = printed_by_rev.get(rev_for_oracle)
        pr, pr_clone = (entry[0] if entry else None) or (None, None)
        why = unit_scope(unit, pr, pr_clone, not run)
        if why != "CHECKABLE" and why != "LABEL_ONLY":
            unchecked[why] += 1
            continue
        splits.append({"unit": unit, "revs": list(scope),
                       "values": sorted(vals), "how": why,
                       "rows": [{"line": ln, "value": v, "block": b}
                                for b, ln, v, _ in items]})

    # 2. ORACLE: every claim compared against what the tool prints.
    if run:
        for first, revs, cl, _b in blk_claims:
            for rev in (revs or ["HEAD"]):
                pr, rc, out = printed_by_rev[(first, rev)]
                if pr is None:
                    failures.append({"file": name, "block": first, "rev": rev,
                                     "unit": "-", "claim": None,
                                     "verdict": "RUN_FAILED", "detail": str(out)[:90]})
                    continue
                printed, clone = pr
                for ln, val, unit, tok, _pos in cl:
                    v, detail = oracle(printed, clone, unit, val)
                    verdicts.append({"file": name, "block": first, "line": ln,
                                     "rev": rev, "unit": unit, "claim": val,
                                     "token": tok, "verdict": v, "detail": detail})
    return {"tool": name, "splits": splits, "unchecked": dict(unchecked),
            "failures": failures, "verdicts": verdicts,
            "blocks_with_claims": len(blk_claims)}


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--repo", default=DEFAULT_REPO)
    ap.add_argument("--rev", default="HEAD")
    ap.add_argument("--tools", nargs="*")
    ap.add_argument("--json")
    ap.add_argument("--no-run", action="store_true",
                    help="internal-split verdicts only; no subprocess")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()

    if a.selftest:
        return _fixture_checks()

    store = git(a.repo, "ls-tree", "-r", "--name-only", a.rev)
    tree = {p for p in store.decode("utf-8", "replace").split("\n") if p}
    tools = a.tools or sorted(p for p in tree
                              if p.startswith("tools/") and p.endswith(".py"))
    if not tools:
        print(f"no tools found at {a.rev}", file=sys.stderr)
        return 2

    cache, reports = {}, []
    with tempfile.TemporaryDirectory(prefix="doccensus-") as tmpdir:
        for t in tools:
            src = subprocess.run(["git", "-C", a.repo, "show", f"{a.rev}:{t}"],
                                 capture_output=True).stdout
            local = os.path.join(tmpdir, os.path.basename(t))
            with open(local, "wb") as fh:
                fh.write(src)
            try:
                rep = check_tool(a.repo, local, t, cache, run=not a.no_run,
                                 tmpdir=tmpdir)
            except subprocess.TimeoutExpired:
                rep = {"tool": t, "splits": [], "blocks_with_claims": 0,
                       "verdicts": [{"file": t, "verdict": "TIMEOUT", "rev": "-",
                                     "unit": "-", "claim": None, "block": 0}]}
            if rep:
                reports.append(rep)

    print(f"tools scanned at {a.rev}: {len(reports)} "
          f"({'internal-split only, label vocabulary' if a.no_run else 'docstring vs printed census'})")
    tot = collections.Counter()
    for rep in reports:
        bad = rep["failures"] + [v for v in rep["verdicts"] if v["verdict"] == "STALE"]
        print(f"\n{rep['tool']}: {len(rep['splits'])} INTERNAL SPLIT, "
              f"{len(bad)} STALE/failed, {len(rep['verdicts'])} claims checked")
        for s in rep["splits"]:
            vals = ", ".join(f"{r['value']}@{r['line']}" for r in s["rows"])
            scope = ",".join(s["revs"]) or "no rev named"
            print(f"  SPLIT  {s['unit']} [{scope}] ({s['how']}): {vals}")
        for k, c in sorted(rep["unchecked"].items()):
            print(f"  (not flagged: {c} split(s) in units {k})")
        for v in bad:
            print(f"  {v['verdict']:10s} {v['file']}:{v.get('line', '-')} "
                  f"{v['unit']} claimed {v['claim']} at {v['rev']} -- {v['detail']}")
        for k, c in collections.Counter(v["verdict"] for v in rep["verdicts"]).items():
            tot[k] += c
    print("\n--- verdict totals (unit: claims) ---")
    for k in sorted(tot):
        print(f"  {k:12s} {tot[k]}")
    if a.json:
        with open(a.json, "w") as fh:
            json.dump({"rev": a.rev, "repo": a.repo, "reports": reports}, fh, indent=1)
        print(f"wrote {a.json}")
    return 0


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