#!/usr/bin/env python3 """Freeze the instrument: the three task sets, verbatim, from their sources. Run ON THE LAPTOP (this laptop), where the docent repo and the hub docroot live; the output is a self-contained kit the harness carries to the bench box, so the box that runs the arms needs neither repo. Nothing here calls a model. SOURCES, each named so a reader can go back to it: (a)+(b) `~/projects/s2s-docent/bench/bank.json` (63 questions, frozen) and the hub docroot the docent's own `corpus.load` reads. The STATE is `prompt.fenced_article(article.prompt_body)` — byte-for-byte the span the deployed seat reads, fence and all. (c) the long table's public wall JSON. The SET IS HELD OUT of every prompt: a line's own speaker is never in its state, and any line whose text names its own speaker is dropped (see NAME_TOKENS). WHY A KIT AND NOT A LIVE CALL. An arm re-run a week later must read the same bytes, and the bench box has no docent checkout. Every state is written here once, hashed, and the hash travels in every row. """ from __future__ import annotations import argparse import hashlib import json import os import random import re import sys SEED = 20260921 #: A line that names its own speaker leaks the label into the state, so it is #: dropped from task (c) rather than scored. Surnames and given names both: #: "as Darwin observed" is as much a give-away as "I, Charles Darwin". NAME_TOKENS = { "darwin": ["darwin", "charles"], "hypatia": ["hypatia"], "ibn_sina": ["ibn sina", "avicenna"], "sagan": ["sagan", "carl"], "socrates": ["socrates", "socratic"], "einstein": ["einstein", "albert"], } def sha(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def build_articles(docroot: str, docent_repo: str, slugs: set[str]) -> dict: sys.path.insert(0, docent_repo) from docent import corpus, prompt # noqa: E402 c = corpus.load(docroot) out = {} for slug in sorted(slugs): a = c.get(slug) if a is None: raise SystemExit("kit: slug %r is not in the corpus" % slug) state = prompt.fenced_article(a.prompt_body) sections = [{"id": s.id, "heading": s.heading} for s in a.sections] known = {s["id"] for s in sections} # the cite enum is what the gate admits, which is one more than the # file carries (the synthetic short-version section); an id with no # heading in `sections` is carried with its id as its own label. for aid in sorted(a.anchor_ids): if aid not in known: sections.append({"id": aid, "heading": aid.replace("-", " ")}) out[slug] = { "slug": slug, "title": a.title, "state": state, "state_sha256": sha(state), "state_bytes": len(state.encode("utf-8")), "est_tokens": a.est_tokens, "sections": sorted(sections, key=lambda s: s["id"]), "body_sha": a.body_sha, } return out def build_ab(bank: dict, articles: dict) -> tuple[list, list]: task_a, task_b = [], [] for q in bank["questions"]: on = q["kind"] == "on-page" task_a.append({ "id": q["id"], "slug": q["slug"], "question": q["q"], "kind": q["kind"], "label": "yes" if on else "no", }) if on: item = {"id": q["id"], "slug": q["slug"], "question": q["q"], "options": [s["id"] for s in articles[q["slug"]]["sections"]], "label": None} if q.get("expects"): item["label"] = q["expects"] item["label_source"] = "bank.expects" task_b.append(item) return task_a, task_b def build_c(wall: dict, per_guest: int | None) -> tuple[list, list]: roster: dict[str, str] = {} pool: dict[str, list] = {} for seat, e in enumerate(wall["entries"]): if e.get("taken_down"): continue for g in e["guests"]: roster.setdefault(g["id"], g["name"]) for course in e["courses"]: for i, ln in enumerate(course["lines"]): sp = ln["speaker"] if sp == "host" or ln.get("withheld"): continue text = (ln.get("text") or "").strip() if len(text) < 80: continue low = text.lower() if any(tok in low for tok in NAME_TOKENS.get(sp, [sp])): continue # the line names its own speaker: it would leak pool.setdefault(sp, []).append({ # THE ENTRY IS PART OF A LINE'S IDENTITY. Every wall entry is # edition 1, so (guest, edition, course, line_index) is not # unique across dinners: the 2026-09-21 kit put 108 distinct # lines under 50 distinct ids. `seat` is the entry's position # in the wall JSON, which is what makes the id an address. "seat": seat, "edition": e["edition"], "course": course["key"], "line_index": i, "speaker": sp, "text": text, }) options = sorted(roster) # the SAME option order for every item and arm n = min(len(v) for v in pool.values()) if per_guest: n = min(n, per_guest) rng = random.Random(SEED) items = [] for gid in options: picks = sorted(pool[gid], key=lambda r: (r["seat"], r["course"], r["line_index"])) for row in rng.sample(picks, n): items.append({ "id": "c-%s-e%02d-s%02d-%s-%02d" % (gid, row["edition"], row["seat"], row["course"], row["line_index"]), "text": row["text"], "label": gid, "options": options, "source": {k: row[k] for k in ("edition", "seat", "course", "line_index")}, }) if len({it["id"] for it in items}) != len(items): raise SystemExit("kit: task (c) ids are not unique - %d ids for %d items" % (len({it["id"] for it in items}), len(items))) items.sort(key=lambda r: r["id"]) return items, [{"id": k, "name": roster[k]} for k in options] def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--docroot", default=os.path.expanduser("~/s2s-research-hub/site")) ap.add_argument("--docent", default=os.path.expanduser("~/projects/s2s-docent")) ap.add_argument("--wall", required=True, help="a saved copy of the wall JSON") ap.add_argument("--out", default=os.path.dirname(os.path.abspath(__file__))) ap.add_argument("--per-guest", type=int, default=0) a = ap.parse_args() bank_path = os.path.join(a.docent, "bench", "bank.json") bank = json.load(open(bank_path, encoding="utf-8")) wall = json.load(open(a.wall, encoding="utf-8")) articles = build_articles(a.docroot, a.docent, {q["slug"] for q in bank["questions"]}) task_a, task_b = build_ab(bank, articles) task_c, roster = build_c(wall, a.per_guest or None) manifest = { "built": None, # stamped by the caller from the clock "seed": SEED, "bank_sha256": sha(open(bank_path, encoding="utf-8").read()), "wall_sha256": sha(open(a.wall, encoding="utf-8").read()), "wall_url": "https://longtable.strata2signal.com/api/wall", "docroot": a.docroot, "counts": {"articles": len(articles), "a": len(task_a), "b": len(task_b), "c": len(task_c)}, "a_labels": {"yes": sum(1 for r in task_a if r["label"] == "yes"), "no": sum(1 for r in task_a if r["label"] == "no")}, "b_labelled": sum(1 for r in task_b if r["label"]), "c_roster": roster, "c_per_guest": len(task_c) // max(1, len(roster)), } for name, obj in (("articles", articles), ("task_a", task_a), ("task_b", task_b), ("task_c", task_c), ("manifest", manifest)): p = os.path.join(a.out, name + ".json") with open(p, "w", encoding="utf-8") as fh: json.dump(obj, fh, ensure_ascii=False, indent=1, sort_keys=True) print("%-12s %8d bytes %s" % (name, os.path.getsize(p), p)) print(json.dumps(manifest["counts"]), json.dumps(manifest["a_labels"]), "b labelled:", manifest["b_labelled"], "c/guest:", manifest["c_per_guest"]) return 0 if __name__ == "__main__": raise SystemExit(main())