#!/usr/bin/env python3 """Exercise the page-building code paths that the completeness guard otherwise hides. build_page.py refuses to render until every scan is finished, which means its newest and most error-prone blocks — the adversarial-minus-control difference table, and the pairing of books that landed in both corpora — would run for the first time at the moment of publication. This runs them against whatever real scores exist right now. It fabricates nothing; on a partial scan it prints partial numbers, which are a smoke test and not a result. python3 selftest.py """ import ast, hashlib, json, os, re, sys HERE = os.path.dirname(os.path.abspath(__file__)) # Do not run this beside a live scan. # # score.py holds a roberta checkpoint and a batch of activations: about 1.0 GB on a box with 1.9 GB # and no swap. This script reads passages.jsonl into a list of dicts, which is another few hundred # megabytes, and the kernel resolves the disagreement by killing the largest process — which is # always the scan, never this. run_scan.sh makes that survivable rather than fatal, but every kill # costs a model reload and whatever was in the current batch. # # I have written "never analyse beside a live scan" down twice and then done it six times in one # sitting, because each individual run feels free. So it stops being a resolution and becomes a # check. Override deliberately with --anyway when the scan is finished or when you mean it. # # The check itself now lives in build_page.py, which reads more than this script does and had no # guard at all until I ran it twice against a live scan while *testing an unrelated refusal*. Two # hand-written copies of one rule is the arrangement that put a `scripts/|data/` filter in the # orphan check twice; importing it costs one line and cannot drift. sys.path.insert(0, HERE) import build_page as B def static_checks(): """Everything checkable without opening a data file — so it runs while the scan is running. These read source, not scores, which makes them the only half of this script that can run during the eleven hours when the answer is being computed. That is not a small window: it is when I am editing the build, and when a break is cheapest to fix. The arity check earned its place immediately. Adding a fifth return value to `headline()` broke this script's four-value unpack, and the call sits inside a `try` that prints "headline() FAILED" — so the break would have surfaced, hours later, looking like a data problem rather than a signature I had changed myself. """ src = open(os.path.join(HERE, "build_page.py")).read() tree = ast.parse(src) arity = {} for fn in [n for n in tree.body if isinstance(n, ast.FunctionDef)]: rets = {len(r.value.elts) for r in ast.walk(fn) if isinstance(r, ast.Return) and isinstance(r.value, ast.Tuple)} if rets: arity[fn.name] = rets bad = [] for path in ("build_page.py", "selftest.py"): for n in ast.walk(ast.parse(open(os.path.join(HERE, path)).read())): if isinstance(n, ast.Assign) and isinstance(n.targets[0], ast.Tuple) \ and isinstance(n.value, ast.Call): f = n.value.func name = f.attr if isinstance(f, ast.Attribute) else getattr(f, "id", None) if name in arity and len(n.targets[0].elts) not in arity[name]: bad.append("%s:%d unpacks %d from %s(), which returns %s" % (path, n.lineno, len(n.targets[0].elts), name, "/".join(str(x) for x in sorted(arity[name])))) # Every {{KEY}} in the template comes from the build or from copy.json. Assignment, not # mention: `V["HEADLINE_INTRO"]` appears in build_page.py as a *read*, and a substring search # counts that as produced -- so this walks the AST for real writes to V. written = set() for n in ast.walk(tree): if isinstance(n, ast.Assign): for t in n.targets: if isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name) \ and t.value.id == "V" and isinstance(t.slice, ast.Constant): written.add(t.slice.value) if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "update" \ and isinstance(n.func.value, ast.Name) and n.func.value.id == "V": for a in n.args: if isinstance(a, ast.Dict): written |= {k.value for k in a.keys if isinstance(k, ast.Constant) and isinstance(k.value, str)} keys = set(re.findall(r"\{\{([A-Z0-9_]+)\}\}", open(os.path.join(HERE, "page.tmpl.html")).read())) copy = set(json.load(open(os.path.join(HERE, "copy.json")))) # The outcome-dependent strings were absent on purpose until the numbers existed; naming them # kept "not written yet" and "quietly dropped" as different states. They are written now, so the # exemption is computed from copy.json rather than typed: a hardcoded list of keys-to-forgive # goes on forgiving them after they arrive, and would have let TITLE be deleted from copy.json # without this file noticing. Subtracting a *computed* pending set from the orphan test would be # worse still — it would empty that test by construction — and there is no third state left to # model anyway: a template key that nothing writes and copy.json does not carry is an orphan, # and it fails here. orphan_keys = sorted(keys - written - copy) print("static: %d template keys, %d written by the build, %d from copy.json, %d pending copy" % (len(keys), len(keys & written), len(keys & copy), len(orphan_keys))) for b in bad: print(" ARITY MISMATCH: " + b) if orphan_keys: print(" PLACEHOLDER PRODUCED BY NOTHING: " + ", ".join(orphan_keys)) if bad or orphan_keys: sys.exit(1) static_checks() B.refuse_if_scan_live() S = {k: B.jl(os.path.join(HERE, "scores_%s.jsonl" % k)) for k, _ in B.MODELS} C = {k: B.jl(os.path.join(HERE, "scores_control_%s.jsonl" % k)) for k, _ in B.MODELS} corpus = [json.loads(l) for l in open(os.path.join(HERE, "passages.jsonl"))] control = [] for line in open(os.path.join(HERE, "control.jsonl")): d = json.loads(line) d["n"] = "%s#%d" % (d["id"], d["seq"]) # control rows carry no corpus-wide id control.append(d) print("corpus %d passages, control %d passages" % (len(corpus), len(control))) for k, _ in B.MODELS: print(" %-7s %d/%d corpus, %d/%d control scored" % (k, len(S[k]), len(corpus), len(C[k]), len(control))) shared = {r["item"] for r in corpus} & {r["id"] for r in control} ERA = 1990 failures = 0 # --- the difference table ------------------------------------------------------------------- for k, _ in B.MODELS: a = [r for r in corpus if r["n"] in S[k]] b = [r for r in control if r["n"] in C[k]] cases = [("as drawn", a, b), ("excluding shared", [r for r in a if r["item"] not in shared], [r for r in b if r["id"] not in shared]), ("%d onward" % ERA, [r for r in a if (r.get("year") or 0) >= ERA], [r for r in b if (r.get("year") or 0) >= ERA])] for label, ar, br in cases: if not ar or not br: print(" %-7s %-18s skipped, a side is empty (%d/%d)" % (k, label, len(ar), len(br))) continue try: row = B.diff_row(label, ar, S[k], br, C[k]) assert len(row) == 5 and all(isinstance(x, str) for x in row), row print(" %-7s %-18s %s" % (k, label, " | ".join(row[1:]))) except Exception as e: failures += 1 print(" %-7s %-18s FAILED: %r" % (k, label, e)) # --- the headline table and the two rules that govern what it may claim ---------------------- # On a partial scan these rates will move, which is the point of watching them early rather than # meeting them at publication. A refusal here is not a failure of the test — it is the rule firing # on incomplete data, and it is reported as such. try: hrows, hrefusals, hrates, hwins, hbounds = B.headline(S, C, corpus, control) for row in hrows: print(" headline %-46s n=%-7s books=%-6s flagged=%-6s %s | %s | %s" % (re.sub(r"<[^>]+>", "", row[0]), row[1], row[2], row[3], row[4], row[5], row[6])) print(" wider-interval rule: by book wins %d rows, by passage %d" % (hwins["by book"], hwins["by passage"])) print(" adversarial rates: %s (null rule fires under %.0f%%)" % (", ".join("%s %.2f%%" % (k, 100 * v) for k, v in sorted(hrates.items())), 100 * B.NULL_RATE)) for r in hrefusals: print(" WOULD REFUSE: %s" % r) except Exception as e: failures += 1 print(" headline() FAILED: %r" % e) # --- the shared-book pairing ---------------------------------------------------------------- # Free padding experiment: the same bytes sit in two corpora under two different shuffles, so in # two different batches, padded to two different widths. Any drift shows up here. adup = [r for r in corpus if r["item"] in shared] cdup = [r for r in control if r["id"] in shared] ah = {hashlib.sha1(r["text"].encode()).hexdigest(): r["n"] for r in adup} if len(ah) != len(adup): print("NOTE: %d of %d shared-book passages are not distinct by hash" % (len(adup) - len(ah), len(adup))) pairs = [] for k, _ in B.MODELS: for r in cdup: m = ah.get(hashlib.sha1(r["text"].encode()).hexdigest()) if m is not None and m in S[k] and r["n"] in C[k]: pairs.append((S[k][m]["p_ai"], C[k][r["n"]]["p_ai"])) worst = max((abs(x - y) for x, y in pairs), default=0.0) flips = sum((x > B.THRESH) != (y > B.THRESH) for x, y in pairs) print("shared items %d; %d paired scores so far, worst |delta| %.3g, flips %d" % (len(shared), len(pairs), worst, flips)) # --- template placeholders vs keys the builder produces -------------------------------------- # A {{NAME}} typo'd on one side is invisible until the build runs, and the build cannot run until # the last scan lands. Seven names used to be exempted here, absent from copy.json on purpose # because their wording depended on how the numbers came out. They are written now, and the # exemption came out with them: `built` below already counts every copy.json key, so keeping the # list would only have meant that deleting one of the seven from copy.json — or leaving it in the # file after dropping its {{NAME}} from the template — passed this check in silence. An exemption # outlives the condition that justified it unless removing it is part of satisfying it. tmpl = open(os.path.join(HERE, "page.tmpl.html")).read() src = open(os.path.join(HERE, "build_page.py")).read() copy = json.load(open(os.path.join(HERE, "copy.json"))) used = set(re.findall(r"\{\{([A-Z0-9_]+)\}\}", tmpl)) built = (set(re.findall(r'V\[\s*"([A-Z0-9_]+)"\s*\]', src)) | set(re.findall(r'^\s*"([A-Z0-9_]+)":', src, re.M)) | {k for k in copy if not k.startswith("_")}) consumed = set(re.findall(r'V\["([A-Z0-9_]+)"\]', src)) # folded into a block, not substituted unbuilt = sorted(used - built) unused = sorted(built - used - consumed) if unbuilt: failures += 1 print("PLACEHOLDER in template but never built: %s" % ", ".join(unbuilt)) if unused: failures += 1 print("KEY built but never reaches the page: %s" % ", ".join(unused)) print("placeholders: %d in template, %d supplied by copy.json, %d unmatched" % (len(used), len(used & {k for k in copy if not k.startswith("_")}), len(unbuilt) + len(unused))) # --- invariants the remaining blocks depend on ---------------------------------------------- # EXAMPLES_BLOCK, GENRE_BLOCK and OOV_BLOCK are inline string formatting over dict lookups. Their # realistic failure is not bad arithmetic, it is a key that isn't there — by_n[s["n"]] on a scored # id with no corpus row, a None title, a missing oov entry. Assert the lookups instead of # refactoring four blocks apart while a scan is running. import registers by_n = {r["n"]: r for r in corpus} oov = B.jl(os.path.join(HERE, "oov_passages.jsonl")) for k, _ in B.MODELS: orphan = [n for n in S[k] if n not in by_n] if orphan: failures += 1 print("SCORED id absent from the corpus (%s): %d, e.g. %r" % (k, len(orphan), orphan[:2])) missing_field = [r["n"] for r in corpus if r.get("year") is None or "item" not in r] if missing_field: failures += 1 print("CORPUS rows missing year/item: %d, e.g. %r" % (len(missing_field), missing_field[:2])) # The examples block slices r["title"] — None would raise, and it falls back to item, so only a # missing *key* is fatal. Check the fallback is reachable for every row it could pick. untitled = [r["n"] for r in corpus if "title" not in r and "item" not in r] if untitled: failures += 1 print("CORPUS rows with neither title nor item: %d" % len(untitled)) scored_any = set(S["hello"]) | set(S["openai"]) no_oov = [n for n in scored_any if n not in oov] fam = sum(1 for r in corpus if registers.family(r) is not None) print("invariants: %d scored ids, %d without an oov entry, %d/%d corpus rows have a register" % (len(scored_any), len(no_oov), fam, len(corpus))) # --- the pre-registered tier rule ------------------------------------------------------------ # PREREG: if the book-date-verified and catalogue-year-only subsets diverge by more than # B.TIER_LIMIT points, the strict subset becomes the headline. build_page refuses to render when # that fires. The threshold and the split are both imported, not restated: this file's job is to # warn me the rule is about to fire, which it cannot do against its own stale copy of either. # Report the current gap here so it is a number I watch rather than a surprise at publication — # on a partial scan it will move, and that is the point of looking at it early. for k, _ in B.MODELS: g = B.grouped(corpus, S[k], B.tier) if g["strict"][1] and g["loose"][1]: gap = 100 * (g["strict"][0] / g["strict"][1] - g["loose"][0] / g["loose"][1]) print("tier gap %-7s strict %d/%d, loose %d/%d, gap %+.2f pp%s" % (k, g["strict"][0], g["strict"][1], g["loose"][0], g["loose"][1], gap, " <-- WOULD FIRE" if abs(gap) > B.TIER_LIMIT else "")) else: print("tier gap %-7s not evaluable yet (strict %d, loose %d passages scored)" % (k, g["strict"][1], g["loose"][1])) # --- every href on the page vs every file the build actually writes ------------------------- # build_page has a link checker, but it only runs against a finished build — which cannot happen # until the last scan lands. This is the same check done statically, so a promise pointing at a # file nothing produces is caught now rather than in eleven hours. That failure has already # happened twice on this page: a "float-level noise measured below" that was measured nowhere, # and an "analysis output" no script wrote. WEBROOT = "/var/www/aaw" writes = set(B.ROOT_FILES) | {"scripts/" + f for f in B.SCRIPT_FILES} \ | {"data/" + f for f in B.DATA_FILES} \ | {"scripts/upstream/" + f for f in B.UPSTREAM_SCRIPTS} hrefs = set(re.findall(r'href="([^"#:]+)"', tmpl)) for v in copy.values(): if isinstance(v, str): hrefs |= set(re.findall(r'href="([^"#:]+)"', v)) for href in sorted(hrefs): if href.startswith("/"): where = os.path.join(WEBROOT, href.lstrip("/")) if os.path.isdir(WEBROOT) and not os.path.exists(where): failures += 1 print("DEAD site-absolute link: %s" % href) elif href not in writes: failures += 1 print("LINK to a file the build never writes: %s" % href) # And the other direction, which is the one that actually failed. The check above walks the links # that exist; it is silent about a file that ships with nothing pointing at it, and eighteen of # these did. build_page now refuses on that, but its version runs against the rendered page — i.e. # only after the last scan lands. This is the same rule read off the manifests, so a file added to # DATA_FILES and linked from nowhere is caught the moment it is added. orphan = B.orphans(writes, hrefs) if orphan: failures += 1 print("SHIPPED but linked from nowhere (%d): %s" % (len(orphan), ", ".join(orphan))) print("links: %d checked against %d published paths, %d orphaned" % (len(hrefs), len(writes), len(orphan))) # --- the copy gate, exercised --------------------------------------------------------------- # build_page refuses if a rate in the hand-written copy is not one the build computed. That guard # cannot run until the scan lands, and a guard nothing has ever seen fire is a guard I am trusting # on its source code. So run it here on synthetic inputs: it has to pass a percentage written to # fewer decimals than the computed one, and catch both a rate that is merely close and a "k of n" # that appears nowhere. Then run it on the real copy.json against an empty build, where every rate # is by definition unbacked — which is how the current copy proves it states no rates at all. FAKE = {"HEADLINE_BLOCK": "1.70%0.24%3 of 212 books"} cases = [({"A": "flagged 1.7% of them"}, 0, "1.7% vs computed 1.70%"), ({"A": "flagged 1.8% of them"}, 1, "1.8% is not 1.70%"), ({"A": "3 of 212 books"}, 0, "verbatim k-of-n"), ({"A": "4 of 212 books"}, 1, "k-of-n nobody computed"), ({"A": "python 3.11 on 2026-08-28"}, 0, "versions and dates are not rates")] for authored, want, why in cases: got = len(B.unbacked_rates(authored, FAKE)) if got != want: failures += 1 print("COPY GATE wrong on %s: %d unbacked, expected %d" % (why, got, want)) live = B.unbacked_rates(copy, {}) print("copy gate: %d synthetic cases, %d rate(s) in the current copy.json" % (len(cases), len(live))) if live: print(" (these will need a computed key before the page can build: %s)" % "; ".join(live)) # --- the batching check's sample, across process boundaries ---------------------------------- # padding_check.py picks which passages to re-score singly; build_page.py imports the same function # and refuses to render if the published file does not hold exactly the ids it returns. Those two # calls happen in two different interpreters, hours apart, and the rule shuffles a set difference — # so if the ordering depended on hash randomisation, the build would refuse at the end of an # eleven-hour chain with nothing wrong. `sorted(..., key=str)` before the seeded shuffle is what # makes that safe; this is the check that says so rather than the comment that claims it. import subprocess from padding_check import select_sample SYN = {"syn#%d" % i: round(((i * 37) % 1000) / 1000.0, 6) for i in range(400)} mine = select_sample(SYN)[0] prog = ("import json,sys; sys.path.insert(0,%r); from padding_check import select_sample; " "print(json.dumps(select_sample(json.load(sys.stdin))[0]))" % HERE) for hs in ("0", "1", "424242"): r = subprocess.run([sys.executable, "-c", prog], input=json.dumps(SYN), capture_output=True, text=True, env=dict(os.environ, PYTHONHASHSEED=hs)) if r.returncode != 0 or json.loads(r.stdout) != mine: failures += 1 print("SAMPLE RULE not reproducible under PYTHONHASHSEED=%s: rc=%d %s" % (hs, r.returncode, r.stderr.strip()[-200:])) print("sample rule: %d of %d synthetic passages selected, stable across 3 hash seeds" % (len(mine), len(SYN))) # --- the cluster interval, under a permutation of its input ---------------------------------- # Same family as the sample rule above, one layer down. cluster_bootstrap has a fixed seed, which # I had been calling reproducible; it is not, on its own. It resamples cluster *positions*, so the # same books in a different order are a different sequence of draws and a different interval. The # two programs that call it walk the corpus in different orders by construction — build_page.py # reads passages.jsonl in file order, analyse.py iterates the score file, which is in the scan's # shuffle order — so the page and analysis.txt would have published different bounds from one # corpus. by_cluster sorts by book id to make the interval a function of the data alone. A seed is # not what makes that true, so a test that permutes the input is what has to say so. import clusterci rows_a = [{"n": "p%d" % i, "item": "bk%d" % (i % 80)} for i in range(600)] scr = {"p%d" % i: {"p_ai": 0.9 if i % 3 == 0 else 0.1} for i in range(600)} rows_b = sorted(rows_a, key=lambda r: (int(r["n"][1:]) * 37) % 601) # 601 prime: a real permutation ca, cb = (clusterci.by_cluster(r, scr, "item") for r in (rows_a, rows_b)) ia, ib = (clusterci.cluster_bootstrap(c) for c in (ca, cb)) if ca != cb or ia != ib: failures += 1 print("CLUSTER INTERVAL depends on row order: %s vs %s" % (ia, ib)) print("cluster interval: %d books, 95%% CI %.2f-%.2f%%, unchanged under a permutation of the rows" % (len(ca), 100 * ia[0], 100 * ia[1])) # --- the median bound, which the build refuses on -------------------------------------------- # Same reasoning as the tier gap: report it now so the number is watched rather than met at # publication. Both medians are properties of the corpora, not of the scan, so this is final. # The build's function and the build's bound, imported rather than restated — this line exists to # warn me early, and a warning computed from its own copy of the rule can go stale quietly. ma, mc = B.median_words(corpus), B.median_words(control) print("median length: adversarial %d words, control %d, gap %d (build refuses above %d)%s" % (ma, mc, abs(ma - mc), B.MED_GAP_LIMIT, " <-- WOULD FIRE" if abs(ma - mc) > B.MED_GAP_LIMIT else "")) print("SELFTEST %s" % ("FAILED (%d)" % failures if failures else "OK")) sys.exit(1 if failures else 0)