troupe-standup/wizard-work/ledger/verify_commits_table.py

#!/usr/bin/env python3
"""Verification for `commits-table.md` (ledger / Extract lane).

The table names its own source on a `Source:` line. This gate re-reads that
line, fetches *that* endpoint, and grades every row against the response body
the endpoint returns now -- so the artifact is tied to its data by a
measurement rather than by the sentence that claims it.

What is graded (rc 1 on any failure):

  TABLE      the document parses: a `Source:` URL, a `Rows: N` claim, a four
             column header over N body rows, 4 cells per row, no unescaped
             pipe (which is how a markdown table silently loses a cell).
  FEED       the response body is a JSON array of commit objects.
  RESOLVE    every row's SHA prefix is >= 7 lowercase hex characters and is a
             prefix of exactly one SHA in the feed (ambiguous -> fault).
  SUBJECT    the row's Subject equals the feed's first message line, byte for
             byte. This is the leg that catches a hand-retyped URL inside a
             merge subject -- one character, invisible to every reader who
             does not re-fetch, indistinguishable from a fabricated path.
  DATE       the row's Date equals the feed's *author* date, day precision.
             The column is day-precision by construction, so a within-day
             author/committer difference is printed as an observation.
  ORDER      rows appear in feed order (row i's feed index strictly increases
             with i), i.e. the table is most-recent-first as its title says.
  PROVENANCE the `Provenance:` line names the same URL as `Source:`, a UTC
             fetch timestamp, the record count the body rows imply, and the
             per_page the URL asked for. Q62 leg (c)'s lesson: an artifact
             that names its source is not the same as one that can be audited.
  RECENCY    OBSERVED by default, not graded: whether these rows are still the
             N most recent commits at the endpoint. A committed file cannot be
             abreast of a moving feed, and a leg that goes red for every reader
             forever is a leg everyone learns to ignore. `--strict-recency`
             promotes it to a graded leg for a periodic sweep.

`--feed PATH` grades against a retained response body instead of the network
(this is how the lane gate runs it: hermetic, no network in CHECK). The
`--record PATH` flag writes whatever body was fetched to PATH, which is how a
retained body is pinned in the first place.

House exit contract (the same one `tools/` uses): 0 = every graded leg passed,
1 = a graded leg failed, 2 = could not grade (missing table, unparsable body,
transport failure). No asserts, so `-O` cannot delete a check.

Every path is resolved from `__file__`; nothing here names the author's
checkout, so the gate grades the commit it sits in and not the tree it was
written in.
"""
import json
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import NoReturn

HERE = Path(__file__).resolve().parent
TABLE = HERE / "commits-table.md"
COLUMNS = ["#", "SHA", "Date", "Subject"]
MIN_PREFIX = 7
# Resolution depth. The artifact cites per_page=5; asking for 100 only widens
# the window rows are resolved in, so a row that has since been pushed down the
# feed is still identifiable instead of looking invented.
RESOLVE_DEPTH = 100
TIMEOUT = 40

GRADED = []


def say(ok, leg, detail=""):
    print("[%s] %-10s %s" % ("PASS" if ok else "FAIL", leg, detail))
    if not ok:
        GRADED.append(leg)


def observe(leg, detail):
    print("[INFO] %-10s %s" % (leg, detail))


def die(code, msg) -> NoReturn:
    print("UNGRADABLE: %s" % msg)
    sys.exit(code)


# ---------------------------------------------------------------- document --

def read_table(path):
    if not path.is_file():
        die(2, "no table at %s" % path)
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError) as exc:
        die(2, "cannot read %s: %s" % (path, exc))
    return text


def parse_table(text):
    """Return dict(url, rows_claim, rows). Rows are 4-cell lists; a row whose
    cell count is not 4 is kept (so it can be named) with `cells` recording it.
    """
    url = None
    for line in text.split("\n"):
        m = re.match(r"^Source:\s*(\S+)\s*$", line)
        if m:
            url = m.group(1)
            break
    m = re.search(r"^Rows:\s*(\d+)\s*$", text, re.MULTILINE)
    rows_claim = int(m.group(1)) if m else None

    rows, header = [], None
    for line in text.split("\n"):
        s = line.strip()
        if not s.startswith("|"):
            continue
        cells = [c.strip() for c in s[1:-1].split("|")] if s.endswith("|") else None
        if cells is None:
            continue
        if header is None:
            header = cells
            continue
        if all(re.fullmatch(r":?-{2,}:?", c) for c in cells):
            continue                      # separator
        if re.fullmatch(r"\d+", cells[0]):
            rows.append(cells)
    return {"url": url, "rows_claim": rows_claim, "header": header, "rows": rows}


def grade_document(doc):
    say(doc["url"] is not None, "TABLE", "Source: %s" % doc["url"] if doc["url"] else "no `Source:` line")
    if doc["url"]:
        parsed = urllib.parse.urlsplit(doc["url"])
        say(parsed.scheme == "https" and bool(parsed.netloc), "TABLE", "source is an https URL")
    say(doc["header"] == COLUMNS, "TABLE", "header %r" % (doc["header"],))
    say(doc["rows_claim"] is not None, "TABLE", "Rows: %s" % doc["rows_claim"])
    say(doc["rows_claim"] == len(doc["rows"]), "TABLE",
        "Rows: %s vs %d body rows" % (doc["rows_claim"], len(doc["rows"])))
    bad = [(r[0], len(r)) for r in doc["rows"] if len(r) != len(COLUMNS)]
    say(not bad, "TABLE", "cell counts %s" % (bad or "all %d" % len(COLUMNS)))
    say(len(doc["rows"]) > 0, "TABLE", "%d data row(s)" % len(doc["rows"]))


# ------------------------------------------------------------------- feed ---

def fetch(url):
    sep = "&" if "?" in url else "?"
    deep = "%s%sper_page=%d" % (url, sep, RESOLVE_DEPTH) if "per_page=" not in url else re.sub(
        r"per_page=\d+", "per_page=%d" % RESOLVE_DEPTH, url)
    req = urllib.request.Request(deep, headers={
        "User-Agent": "ledger-commits-table-gate/1.0",
        "Accept": "application/vnd.github+json",
    })
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
            body = r.read()
            status = r.status
    except urllib.error.HTTPError as exc:
        die(2, "HTTP %s from %s" % (exc.code, deep))
    except (urllib.error.URLError, OSError) as exc:
        die(2, "transport failure for %s: %s" % (deep, exc))
    if status != 200:
        die(2, "HTTP %s from %s" % (status, deep))
    observe("FEED", "fetched %s (%d bytes, per_page=%d)" % (deep, len(body), RESOLVE_DEPTH))
    return body, deep


def load_feed(body, origin):
    try:
        feed = json.loads(body)
    except (ValueError, UnicodeDecodeError) as exc:
        die(2, "response body is not JSON (%s): %s" % (origin, exc))
    if not isinstance(feed, list) or not feed:
        die(2, "response body is not a non-empty JSON array (%s)" % origin)
    for i, rec in enumerate(feed):
        try:
            rec["sha"], rec["commit"]["message"], rec["commit"]["author"]["date"]
        except (KeyError, TypeError):
            die(2, "record %d lacks sha/commit.message/commit.author.date (%s)" % (i, origin))
    return feed


# ------------------------------------------------------------------ grading --

def record_view(rec):
    msg = rec["commit"]["message"]
    subject = msg.split("\n", 1)[0]
    author = rec["commit"]["author"]["date"]
    committer = rec["commit"].get("committer", {}).get("date", author)
    return {"sha": rec["sha"], "subject": subject, "author": author, "committer": committer}


def grade_rows(doc, feed):
    views = [record_view(r) for r in feed]
    index_of = {}
    for i, v in enumerate(views):
        index_of.setdefault(v["sha"], i)

    resolved = []
    for row in doc["rows"]:
        num, prefix = row[0], row[1]
        if not re.fullmatch(r"[0-9a-f]{%d,}" % MIN_PREFIX, prefix):
            say(False, "RESOLVE", "row %s: %r is not a >=%d-char lowercase hex prefix" % (num, prefix, MIN_PREFIX))
            resolved.append(None)
            continue
        hits = [i for i, v in enumerate(views) if v["sha"].startswith(prefix)]
        if len(hits) != 1:
            say(False, "RESOLVE", "row %s: %r matches %d feed SHAs (need exactly 1)" % (num, prefix, len(hits)))
            resolved.append(None)
            continue
        resolved.append(hits[0])
    ok = [r for r in resolved if r is not None]
    say(len(ok) == len(resolved), "RESOLVE",
        "%d/%d rows resolve to exactly one feed SHA" % (len(ok), len(resolved)))

    for row, idx in zip(doc["rows"], resolved):
        if idx is None:
            continue
        v = views[idx]
        num, _, date, subject = row[0], row[1], row[2], row[3]
        if subject != v["subject"]:
            say(False, "SUBJECT", "row %s differs from feed %s" % (num, v["sha"][:10]))
            if len(subject) == len(v["subject"]):
                k = next((j for j in range(len(subject)) if subject[j] != v["subject"][j]), None)
                if k is not None:
                    print("        first difference at character %d" % k)
                    print("        table: ...%s..." % subject[max(0, k - 24):k + 25])
                    print("        feed : ...%s..." % v["subject"][max(0, k - 24):k + 25])
            else:
                print("        table: %r" % subject)
                print("        feed : %r" % v["subject"])
            continue
        say(True, "SUBJECT", "row %s == %s" % (num, v["sha"][:10]))
        day = v["author"][:10]
        say(date == day, "DATE", "row %s: table %s vs feed author %s" % (num, date, v["author"]))
        if v["committer"][:10] != v["author"][:10]:
            observe("DATE", "row %s author %s and committer %s fall on different days" % (num, v["author"], v["committer"]))

    seq = [r for r in resolved if r is not None]
    say(seq == sorted(set(seq)) and len(seq) == len(set(seq)), "ORDER",
        "row order vs feed index %s" % (seq,))
    return resolved, views


def grade_provenance(text, doc):
    m = re.search(r"^Provenance:\s+(\S+)\s+fetched\s+(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z)"
                  r"\s+\((\d+)\s+records?,\s*per_page=(\d+)\)\s*$", text, re.MULTILINE)
    if not m:
        say(False, "PROVENANCE", "no `Provenance: <url> fetched <UTC> (<n> records, per_page=<k>)` line")
        return
    url, stamp, records, per_page = m.group(1), m.group(2), int(m.group(3)), int(m.group(4))
    say(url == doc["url"], "PROVENANCE", "names the same endpoint as Source: (%s)" % url)
    try:
        datetime.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
        parsed = True
    except ValueError:
        parsed = False
    say(parsed, "PROVENANCE", "fetch timestamp %s parses as UTC" % stamp)
    say(records == len(doc["rows"]), "PROVENANCE",
        "%d records claimed for %d body rows" % (records, len(doc["rows"])))
    m2 = re.search(r"per_page=(\d+)", doc["url"] or "")
    say(bool(m2) and per_page == int(m2.group(1)), "PROVENANCE",
        "per_page=%d matches the Source URL's per_page=%s" % (per_page, m2.group(1) if m2 else "absent"))
    observe("PROVENANCE", "captured %s" % stamp if parsed else "captured (unparsed)")


def grade_recency(doc, resolved, strict):
    n = len(doc["rows"])
    top = list(range(n))
    seq = [r for r in resolved if r is not None]
    if seq == top:
        observe("RECENCY", "rows are the %d most recent records at the endpoint" % n)
        if strict:
            say(True, "RECENCY", "rows are still the top %d" % n)
        return
    if strict:
        say(False, "RECENCY", "rows are not the top %d (%s)" % (n, seq))
    else:
        newest = seq[0] if seq else None
        observe("RECENCY", "rows are not the top %d at this reading (row 1 sits at feed index %s, "
                           "so %s commit(s) have landed since capture); not graded, "
                           "pass --strict-recency to grade it"
                           % (n, newest, newest))


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

def main(argv):
    table_path, feed_path, record_path = TABLE, None, None
    strict = False
    i = 0
    while i < len(argv):
        a = argv[i]
        if a == "--table" and i + 1 < len(argv):
            table_path = Path(argv[i + 1]).expanduser()
            i += 2
        elif a == "--feed" and i + 1 < len(argv):
            feed_path = Path(argv[i + 1]).expanduser()
            i += 2
        elif a == "--record" and i + 1 < len(argv):
            record_path = Path(argv[i + 1]).expanduser()
            i += 2
        elif a == "--strict-recency":
            strict = True
            i += 1
        elif a in ("-h", "--help"):
            print(__doc__)
            return 0
        else:
            print("unknown argument: %r" % a, file=sys.stderr)
            return 2

    text = read_table(table_path)
    doc = parse_table(text)
    grade_document(doc)
    if not doc["url"]:
        die(2, "the document names no source, so there is nothing to fetch")

    if feed_path is not None:
        if not feed_path.is_file():
            die(2, "no retained feed body at %s" % feed_path)
        try:
            body = feed_path.read_bytes()
        except OSError as exc:
            die(2, "cannot read %s: %s" % (feed_path, exc))
        origin = "retained body %s" % feed_path.name
        observe("FEED", "%s (%d bytes)" % (origin, len(body)))
    else:
        body, origin = fetch(doc["url"])
        if record_path is not None:
            try:
                record_path.write_bytes(body)
                observe("FEED", "wrote %d bytes to %s" % (len(body), record_path))
            except OSError as exc:
                die(2, "cannot write %s: %s" % (record_path, exc))

    feed = load_feed(body, origin)
    observe("FEED", "%d record(s) in %s" % (len(feed), origin))
    resolved, views = grade_rows(doc, feed)
    grade_provenance(text, doc)
    grade_recency(doc, resolved, strict)

    if GRADED:
        print("TABLE FAIL: %s" % ", ".join(GRADED))
        return 1
    print("TABLE PASS: %d row(s) reproduce from %s byte for byte."
          % (len(doc["rows"]), "the retained body" if feed_path else "a live fetch"))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))