troupe-standup/wizard-work/sphinx/arxiv-digest-verified-2026-10-25/CHECK.py

#!/usr/bin/env python3
"""sphinx archive-lane CHECK harness.

One bare command -- `python3 CHECK.py` -- whose exit code IS the verdict.

It re-reads the committed artifacts and independently re-derives every headline
number from the frozen HTTP responses captured in this directory, so the check
cannot pass on stale or hand-edited JSON:

  1. digest_sha256 in VERIFICATION.json matches the digest on disk
  2. window_size == the count of corpus records with published >= 2026-10-19
  3. claims == window_size, and verified_exact_lookup == claims
  4. unverified == [] and title_mismatch == [] and errors == []
  5. every row's id is a real corpus record (no invented claims)
  6. every row's title_sha256 == sha256(title) recomputed from the row itself
  7. the 9 search-residual IDs are exactly the ones whose primary category is
     outside cs.CL/CV/LG (the measured query artifact, not a corpus error)
  8. S3 listing sha256 values match the captured listing-*.html files on disk
  9. ADVERSARIAL_TEST.json records that the fabricated ID was rejected
 10. ABS_PROBE.json records that all 9 residual IDs returned HTTP 200

Exit 0 only if all ten hold.
"""
import hashlib
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
CORPUS = "/root/.hermes/wizard-outputs/sphinx/arxiv_recent_harvest/merged_all.jsonl"
DIGEST = "/root/.hermes/wizard-outputs/sphinx/arxiv_recent_harvest/arxiv-recent-digest-2026-10-25.md"
CUTOFF = "2026-10-19"

checks = []


def check(name, ok, detail=""):
    checks.append((name, bool(ok), detail))
    print(f"[{'PASS' if ok else 'FAIL'}] {name}" + (f"  -- {detail}" if detail else ""))
    return ok


def main():
    ver = json.load(open(os.path.join(HERE, "VERIFICATION.json"), encoding="utf-8"))

    # 1 digest binding
    digest_sha = hashlib.sha256(open(DIGEST, "rb").read()).hexdigest()
    check("digest sha256 matches VERIFICATION.json",
          ver["digest_sha256"] == digest_sha,
          f"{digest_sha[:16]}...")

    # 2 window recomputed from the corpus file
    corpus = {}
    for line in open(CORPUS, encoding="utf-8"):
        line = line.strip()
        if not line:
            continue
        rec = json.loads(line)
        m = re.search(r"/abs/([0-9]{4}\.[0-9]{4,5})", rec.get("id", ""))
        if m:
            corpus[m.group(1)] = rec
    window = [b for b, r in corpus.items()
              if (r.get("published") or "")[:10] >= CUTOFF]
    check("window_size matches corpus records on/after cutoff",
          ver["window_size"] == len(window),
          f"reported={ver['window_size']} recomputed={len(window)}")

    # 3 all claims verified
    c = ver["counts"]
    check("claims == window_size", c["claims"] == ver["window_size"],
          f"claims={c['claims']}")
    check("verified_exact_lookup == claims",
          c["verified_exact_lookup"] == c["claims"],
          f"{c['verified_exact_lookup']}/{c['claims']}")

    # 4 nothing left over
    check("no unverified ids", ver["unverified_ids"] == [], str(ver["unverified_ids"]))
    check("no title mismatches", ver["title_mismatch_ids"] == [],
          str(ver["title_mismatch_ids"]))
    check("no fetch errors", ver["errors"] == [], str(ver["errors"]))

    # 5 no invented claims: every verified row is a real corpus record
    rows = ver["rows"]
    check("every row id exists in corpus",
          all(r["id"] in corpus for r in rows),
          f"{len(rows)} rows")

    # 6 title hashes recompute
    bad_hash = [r["id"] for r in rows
                if r["title_sha256"] != hashlib.sha256(r["title"].encode()).hexdigest()]
    check("every title_sha256 recomputes from its own title", not bad_hash,
          str(bad_hash[:5]))

    # 7 the measured query artifact: search-residual == primary category outside
    s2_missed = sorted(r["id"] for r in rows if not r["s2_windowed_search"])
    prim = json.load(open(os.path.join(HERE, "RESIDUAL_CATEGORIES.json"),
                          encoding="utf-8"))
    outside = sorted(b for b, v in prim.items()
                     if not v["in_sc_CL_CV_LG_primary"])
    check("search-residual ids == ids with primary outside cs.CL/CV/LG",
          s2_missed == outside,
          f"residual={len(s2_missed)} outside={len(outside)}")

    # 8 listing hashes match the frozen HTML on disk
    lh = ver["listing_hashes"]
    ok8 = True
    for cat, meta in lh.items():
        path = os.path.join(HERE, f"listing-{cat}.html")
        if not os.path.exists(path):
            ok8 = False
            continue
        actual = hashlib.sha256(open(path, "rb").read()).hexdigest()
        if actual != meta["sha256"]:
            ok8 = False
    check("listing page sha256 values match captured HTML files", ok8)

    # 9 adversarial rejection
    adv = json.load(open(os.path.join(HERE, "ADVERSARIAL_TEST.json"), encoding="utf-8"))
    check("fabricated ID was rejected by the verifier",
          adv["fabricated_accepted_by_verifier"] is False
          and adv["all_genuine_ids_still_verified"] is True,
          adv["verdict"])

    # 10 abs pages really live
    ap = json.load(open(os.path.join(HERE, "ABS_PROBE.json"), encoding="utf-8"))
    check("all residual IDs returned HTTP 200 abs pages",
          ap["exists"] == ap["tested"] and ap["not_found"] == 0,
          f"{ap['exists']}/{ap['tested']}")

    failed = [n for n, ok, _ in checks if not ok]
    print(f"\n{len(checks) - len(failed)}/{len(checks)} checks passed")
    if failed:
        print("FAILED:", failed)
        return 1
    print("ALL CHECKS PASSED")
    return 0


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