troupe-standup/wizard-work/ledger/2026-10-31-workqueue-census/CHECK

#!/usr/bin/env python3
"""Independent verification for the 2026-10-31 workqueue-census artifact.

RE-APPROACH: this checker does NOT import the extractor. It re-derives the
expected rows from WORK-QUEUE.md with plain string operations (split on the
first " — ", check the leading "Q<n>", strip trailing notes); the committed TSV
is the artifact of record. The verdict is the script's exit code: 0 = correct,
1 = a check failed, 2 = could not grade.

Run:  cd wizard-work/ledger/2026-10-31-workqueue-census && python3 CHECK
"""
import csv
import json
import os
import re
import sys
from datetime import datetime, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE)))
SRC = os.path.join(REPO_ROOT, "WORK-QUEUE.md")
TSV_NAME = "workqueue-census.tsv"
MANIFEST_NAME = "extract_manifest.json"
STALE = "2026-10-28T00:00:00Z"
TS_FMT = "%Y-%m-%dT%H:%M:%SZ"

# Independent derivation: split on the FIRST " — ", keep the token that is
# "Q<n>" followed by digits. This is a different code path from the extractor's
# HEADING_RE regex, so agreement is a real cross-check, not a shared import.
SEP = "\u2014"


def rederive(src_path):
    golden = []
    qid, title = None, ""
    with open(src_path, encoding="utf-8") as fh:
        lines = fh.read().splitlines()
    for line in lines:
        if not line.startswith("## "):
            continue
        rest = line[3:].strip()
        if " \u2014 " not in rest:
            continue
        pid, _, t = rest.partition(" \u2014 ")
        if not (pid.startswith("Q") and pid[1:].isdigit()):
            continue
        # strip trailing "*(notes)*" exactly as the extractor does
        t = re.sub(r"\s*\*.*$", "", t).strip()
        if qid is not None:
            golden.append((qid, title))
        qid, title = pid[1:], t
    if qid is not None:
        golden.append((qid, title))
    return golden


def _read_tsv(path):
    with open(path, encoding="utf-8", newline="") as fh:
        return list(csv.reader(fh, delimiter="\t"))


def main():
    fails = []

    def check(name, cond, detail=""):
        print(("PASS " if cond else "FAIL ") + name + ("" if cond else " " + repr(detail)))
        if not cond:
            fails.append(name)

    # --- source exists and is readable -------------------------------------
    try:
        src_text = open(SRC, encoding="utf-8").read()
    except OSError as exc:
        check("source WORK-QUEUE.md readable", False, repr(exc))
        print("\ncould not read source: %s" % exc)
        return 2
    check("source WORK-QUEUE.md readable", True)

    # --- golden re-derivation ----------------------------------------------
    golden = rederive(SRC)

    # --- committed TSV ------------------------------------------------------
    tsv_path = os.path.join(HERE, TSV_NAME)
    try:
        rows = _read_tsv(tsv_path)
    except OSError as exc:
        check("committed TSV readable", False, repr(exc))
        print("\nmissing committed TSV: %s" % exc)
        return 2
    header, committed = rows[0], rows[1:]

    check("tsv header", header == ["Q", "Title", "Status", "Target"], repr(header))
    check("row count",
          len(committed) == len(golden),
          "committed=%d golden=%d" % (len(committed), len(golden)))
    mism = [k for k, (c, g) in enumerate(zip(committed, golden))
            if len(c) < 2 or c[0] != g[0] or c[1] != g[1]]
    check("q+title identical to independent re-derive", not mism,
          "first mismatch row %r" % (mism[0] if mism else None))

    # --- cross-count: '## Q<n> —' headings on disk vs committed rows --------
    headings = [ln for ln in src_text.splitlines()
                if ln.startswith("## ") and (" \u2014 " in ln[3:])]
    q_headings = [ln for ln in headings
                  if ln[3:].strip().split(" \u2014 ", 1)[0].startswith("Q")
                  and ln[3:].strip().split(" \u2014 ", 1)[0][1:].isdigit()]
    check("'## Q<n> —' headings == committed rows",
          len(q_headings) == len(committed),
          "headings=%d rows=%d" % (len(q_headings), len(committed)))

    # --- manifest -----------------------------------------------------------
    manifest_path = os.path.join(HERE, MANIFEST_NAME)
    try:
        man = json.load(open(manifest_path, encoding="utf-8"))
    except (OSError, ValueError) as exc:
        check("manifest readable JSON", False, repr(exc))
        man = {}
    check("manifest.rows == rows", man.get("rows") == len(golden),
          "manifest=%r actual=%d" % (man.get("rows"), len(golden)))
    auto = man.get("extracted_at_utc", "")
    check("manifest.extracted_at_utc present",
          isinstance(auto, str) and auto, repr(auto))
    parseable = False
    if isinstance(auto, str) and auto:
        try:
            datetime.strptime(auto, TS_FMT).replace(tzinfo=timezone.utc)
            parseable = True
        except ValueError:
            pass
    check("manifest.extracted_at_utc parseable UTC", parseable, repr(auto))
    check("manifest.extracted_at_utc not the stale constant", auto != STALE, repr(auto))
    check("manifest.generated_by present", isinstance(man.get("generated_by"), str)
          and man["generated_by"], repr(man.get("generated_by")))

    # --- corruption ----------------------------------------------------------
    empty = [r for r in committed if len(r) < 2 or not r[0] or not r[1]]
    check("no empty q/title", not empty, repr(empty[:1]))
    bad_tab = [r for r in committed if len(r) >= 3 and "\t" in r[2]]
    check("status contains no literal tab", not bad_tab, repr(bad_tab[:1]))
    bad_tab2 = [r for r in committed if len(r) >= 4 and "\t" in r[3]]
    check("target contains no literal tab", not bad_tab2, repr(bad_tab2[:1]))

    if fails:
        print("\n%d check(s) FAILED" % len(fails))
        return 1
    print("\nALL CHECKS PASSED (%d rows, %d q-numbers, %d distinct)"
          % (len(committed), len(set(q_headings)), len(set(q_headings))))
    return 0


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