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

#!/usr/bin/env python3
"""check_queue_extract.py -- gate for the queue census extraction artifact.

Verdict (exit 0 == correct, nonzero == wrong) is the exit code of this script.
Checks, in order, with no fallbacks:
  1. Round-trip determinism: re-extracting WORK-QUEUE.md with queue-extract.py
     reproduces the committed artifact TSV byte for byte.
  2. Structural agreement: every item in the lane's pinned parse (docs/queue-index.md)
     matches the corresponding TSV row on (id, line) for end_line, lines, bytes, role.
  3. Byte accounting: the sum of the TSV's byte column equals the queue's own
     sha256-derived byte count (2,149,154) and the file's line count (13,555).
  4. No gap: items tile WORK-QUEUE.md with no overlap and no gap.
"""
import json
import re
import subprocess
import sys

ROOT = "/root/troupe-standup"
EXTRACT = ROOT + "/wizard-work/ledger/queue-extract.py"
TSV = ROOT + "/wizard-work/ledger/queue-extracted.tsv"
QUEUE = ROOT + "/WORK-QUEUE.md"
DOC = ROOT + "/docs/queue-index.md"

BEGIN = "<!-- queue-index:begin -->"
END = "<!-- queue-index:end -->"

FAILS = []


def check(name, ok, detail=""):
    if ok:
        print("PASS %s" % name)
    else:
        print("FAIL %s %s" % (name, detail))
        FAILS.append(name)


def read_pinned():
    doc = open(DOC, encoding="utf-8").read()
    body = doc.split(BEGIN, 1)[1].split(END, 1)[0]
    m = re.search(r"```json\n(.*?)\n```", body, re.S)
    if not m:
        check("pinned block present", False, "no json fence")
        sys.exit(2)
    return json.loads(m.group(1))


def read_tsv():
    rows = []
    for ln in open(TSV, encoding="utf-8"):
        ln = ln.rstrip("\n")
        if ln.startswith("id\t"):
            continue
        if ln == "---\n":
            continue
        parts = ln.split("\t")
        if len(parts) < 11:
            check("tsv row width", False, ln[:80])
            continue
        rows.append({
            "id": parts[0], "kind": parts[1], "line": int(parts[2]),
            "end_line": int(parts[3]), "lines": int(parts[4]),
            "bytes": int(parts[5]), "role": parts[6], "role_detail": parts[7],
            "status": parts[8], "status_line": int(parts[9]) if parts[9] not in ("None", "") else None,
            "title": parts[10],
        })
    return rows


def main():
    # 1. round-trip determinism
    p1 = subprocess.run([sys.executable, EXTRACT], capture_output=True, text=True)
    first = p1.stdout
    p2 = subprocess.run([sys.executable, EXTRACT], capture_output=True, text=True)
    check("re-extract rc=0", p1.returncode == 0 and p2.returncode == 0,
          "rc=%s,%s" % (p1.returncode, p2.returncode))
    check("re-extract deterministic", first == p2.stdout)

    tsv = read_tsv()
    pinned = read_pinned()

    # 2. structural agreement against the pinned parse
    by_key = {}
    for it in pinned["items"]:
        by_key[(it["id"], it["line"])] = it
    check("pinned item count", len(pinned["items"]) == len(tsv),
          "pinned=%d tsv=%d" % (len(pinned["items"]), len(tsv)))
    matched = 0
    for t in tsv:
        k = (t["id"], t["line"])
        p = by_key.get(k)
        if p is None:
            continue
        matched += 1
        same_end = t["end_line"] == p["end_line"]
        same_lines = t["lines"] == p["lines"]
        same_bytes = t["bytes"] == p["bytes"]
        same_role = t["role"] == p["role"]
        if not (same_end and same_lines and same_bytes and same_role):
            check("item %s@%d" % k,
                  False,
                  "tsv(end=%d,li=%d,by=%d,ro=%s) pin(end=%d,li=%d,by=%d,ro=%s)"
                  % (t["end_line"], t["lines"], t["bytes"], t["role"],
                     p["end_line"], p["lines"], p["bytes"], p["role"]))
    check("all pinned items matched in TSV", matched == len(pinned["items"]),
          "matched=%d" % matched)

    # 3. byte accounting
    total_bytes = sum(t["bytes"] for t in tsv)
    content = open(QUEUE, encoding="utf-8").read()
    total_lines = len(content.split("\n")) - (1 if content.endswith("\n") else 0)
    check("tsv byte sum", total_bytes == 2149154, "sum=%d" % total_bytes)
    check("queue line count", total_lines == pinned["queue_lines"],
          "lines=%d pinned=%d" % (total_lines, pinned["queue_lines"]))

    # 4. items tile WORK-QUEUE.md with no gap/overlap
    tsv_sorted = sorted(tsv, key=lambda t: (t["line"], t["id"]))
    ok_tile = True
    prev_end = 0
    for t in tsv_sorted:
        if t["line"] != prev_end + 1:
            ok_tile = False
            break
        prev_end = t["end_line"]
    check("items tile the queue", ok_tile and prev_end == total_lines - 1,
          "first_gap_after=%d last_end=%d" % (prev_end, prev_end))

    # top-level summary numbers echoed into the TSV
    summ = {}
    for ln in first.split("---\n"):
        if ln.startswith("items="):
            summ["items"] = ln.split("=", 1)[1].split("\n")[0]
    check("artifact header count", summ.get("items") == str(len(tsv)),
          "header=%s tsv=%d" % (summ.get("items"), len(tsv)))

    if FAILS:
        print("VERDICT: %d FAIL(S)" % len(FAILS))
        sys.exit(1)
    print("VERDICT: correct")
    sys.exit(0)


if __name__ == "__main__":
    main()