"""R49 Stage D3 gate — do any of the author's own proper nouns survive the rename? The rename exists so a voice adapter fits *prose style* and not the author's characters and worldbuilding. That only holds if the renamed copies are actually clean, and "actually clean" is a measurement, not a property of having run the script. Brontë's run reached 0 of 203; BabyYarros opened at 86 of 232. The gate is a whole-corpus scan, not a per-work one, and that distinction is load-bearing. A name detected in `iron-flame` but below threshold in `fourth-wing` is renamed in one copy and printed verbatim in the other, and a per-work gate reports that as clean. CONTROLS. A detector that only ever sees the renamed text cannot tell "absent" from "blind", so this instrument runs both directions every time: * POSITIVE -- the same scan over the UNRENAMED source. Every entity must be found there. A miss means the matcher is broken and its zeroes are worthless. * NEGATIVE -- a nonce string that appears in neither tree. A hit means the matcher is manufacturing signal. Exit code is the gate: 0 iff the controls pass AND no source entity survives. """ from __future__ import annotations import argparse, json, re, sys from collections import Counter, defaultdict from pathlib import Path NONCE = "Qxzvwolfram" # negative control: appears in no corpus def load_works(corpus: Path) -> dict[str, str]: man = json.loads((corpus / "manifest.json").read_text()) out = {} for w in man["works"]: rows = [json.loads(l) for l in (corpus / w["path"]).read_text(encoding="utf-8").splitlines() if l.strip()] out[w["slug"]] = "\n\n".join(r["text"] for r in rows) return out def load_copies(renamed: Path) -> dict[str, str]: out = {} for p in sorted((renamed / "copies").glob("*.jsonl")): rows = [json.loads(l) for l in p.read_text(encoding="utf-8").splitlines() if l.strip()] out[p.name] = "\n\n".join(r["text"] for r in rows) return out def scan(texts: dict[str, str], surfaces: list[str]) -> dict[str, dict[str, int]]: """surface -> {text_name: hits}. One alternation pass per text, not one per name. ⚠ Longest-first alternation, so `Xaden Riorson` is consumed before `Xaden` and a two-part name is not counted twice. """ if not surfaces: return {} pat = re.compile(r"\b(" + "|".join(re.escape(s) for s in sorted(surfaces, key=len, reverse=True)) + r")\b") hits: dict[str, dict[str, int]] = defaultdict(dict) for name, text in texts.items(): local: dict[str, int] = defaultdict(int) for m in pat.finditer(text): local[m.group(1)] += 1 for s, n in local.items(): hits[s][name] = n return hits def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("corpus", help="source corpus dir (manifest.json + works/)") ap.add_argument("--entities", required=True) ap.add_argument("--renamed", required=True, help="rename.py --out dir") ap.add_argument("--min-cap", type=int, default=8, help="rename.py's renameable threshold; entities below it are " "reported separately because rename never touched them") ap.add_argument("--phrase-map", default=None, help="the JSON rename.py used; its `allow` list names the phrases judged " "real-world or generic. Without it the phrase audit does not run.") ap.add_argument("--phrase-min", type=int, default=5, help="a capitalised 2-3gram must recur this often in the source to be audited") ap.add_argument("--report", default=None, help="write the full JSON breakdown here") a = ap.parse_args() corpus, renamed = Path(a.corpus), Path(a.renamed) ents_all = json.loads(Path(a.entities).read_text()) source = load_works(corpus) copies = load_copies(renamed) if not copies: print("== no copy files found -- nothing to gate"); return 1 # Mirror rename.py's own renameable predicate so the two cannot drift apart. renameable, sub_threshold = {}, {} for slug, w in ents_all.items(): for key, e in w["entities"].items(): surf = e.get("surface") or key if "’" in key or "'" in key: continue (renameable if e["cap"] >= a.min_cap else sub_threshold).setdefault(surf, set()).add(slug) surfaces = sorted(set(renameable) | set(sub_threshold)) print(f" {len(renameable)} renameable surfaces (cap >= {a.min_cap}) · " f"{len(sub_threshold)} sub-threshold · {len(copies)} copy files") # ---- controls -------------------------------------------------------- src_hits = scan(source, surfaces + [NONCE]) missing = [s for s in surfaces if s not in src_hits] pos_ok = not missing neg_ok = NONCE not in src_hits print(f" [{'PASS' if pos_ok else 'FAIL'}] positive control: every surface found in the " f"unrenamed source ({len(surfaces) - len(missing)}/{len(surfaces)})" + ("" if pos_ok else f" -- MISSING {missing[:10]}")) print(f" [{'PASS' if neg_ok else 'FAIL'}] negative control: nonce `{NONCE}` absent from source") # ---- the measurement ------------------------------------------------- copy_hits = scan(copies, surfaces + [NONCE]) neg_ok = neg_ok and NONCE not in copy_hits surv_renameable = {s: copy_hits[s] for s in renameable if s in copy_hits} surv_sub = {s: copy_hits[s] for s in sub_threshold if s in copy_hits} print(f"\n SURVIVING renameable: {len(surv_renameable)} of {len(renameable)}") for s, where in sorted(surv_renameable.items(), key=lambda kv: -sum(kv[1].values()))[:40]: tot = sum(where.values()) print(f" {s:<18} {tot:>6} hits across {len(where)} copies " f"(detected in: {','.join(sorted(renameable[s]))})") if len(surv_renameable) > 40: print(f" ... and {len(surv_renameable) - 40} more") print(f"\n SURVIVING sub-threshold (cap < {a.min_cap}, rename never saw them): " f"{len(surv_sub)} of {len(sub_threshold)}") for s, where in sorted(surv_sub.items(), key=lambda kv: -sum(kv[1].values()))[:15]: print(f" {s:<18} {sum(where.values()):>6} hits") # ---- phrase audit ---------------------------------------------------- # ⚠ The unigram scan above cannot see `Riders Quadrant` or `Fourth Wing`: # every component is an ordinary word the detector correctly refuses. This # pass is what caught them AFTER the unigram gate read 0 of 314. surviving_phrases = {} if a.phrase_map: pm = json.loads(Path(a.phrase_map).read_text()) allow = set(pm.get("allow", [])) PH = re.compile(r"\b([A-Z][a-z]{2,}(?: [A-Z][a-z]{2,}){1,2})\b") src_ph = Counter() for t in source.values(): src_ph.update(PH.findall(t)) cop_ph = Counter() for t in copies.values(): cop_ph.update(PH.findall(t)) # A heading word cannot start a leak: `Chapter Twenty` is the book's own # scaffolding, not the author's invention. STRUCT = ("Chapter", "Prologue", "Epilogue", "Part", "Appendix", "Volume", "Book") audited = {p for p, n in src_ph.items() if n >= a.phrase_min and not p.startswith(STRUCT)} - allow surviving_phrases = {p: {"source": src_ph[p], "copies": cop_ph[p]} for p in audited if cop_ph[p] > 0} print(f"\n PHRASE AUDIT: {len(audited)} capitalised 2-3grams recur >= {a.phrase_min} " f"times in the source ({len(allow)} allow-listed as real-world/generic)") print(f" SURVIVING phrases: {len(surviving_phrases)}") for ph, w in sorted(surviving_phrases.items(), key=lambda kv: -kv[1]["source"])[:30]: print(f" {ph:<34} source {w['source']:>4} copies {w['copies']:>5}") if a.report: Path(a.report).write_text(json.dumps({ "renameable_total": len(renameable), "sub_threshold_total": len(sub_threshold), "controls": {"positive_pass": pos_ok, "negative_pass": neg_ok, "missing": missing}, "surviving_renameable": {s: {"hits": sum(w.values()), "copies": len(w), "detected_in": sorted(renameable[s])} for s, w in surv_renameable.items()}, "surviving_sub_threshold": {s: {"hits": sum(w.values()), "copies": len(w)} for s, w in surv_sub.items()}, "surviving_phrases": surviving_phrases, }, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\n wrote {a.report}") if not (pos_ok and neg_ok): print("\n== CONTROLS FAILED -- this gate's verdict is not trustworthy"); return 2 if surv_renameable or surv_sub or surviving_phrases: print(f"\n== GATE FAILED: {len(surv_renameable) + len(surv_sub)} source entities and " f"{len(surviving_phrases)} phrases survive"); return 1 print("\n== GATE PASSED: 0 source entities and 0 audited phrases survive in any copy") print(f" ⚠ sensitivity floor: a name appearing fewer than {a.min_cap} times per work is " f"never detected, and a phrase recurring fewer than {a.phrase_min} times is never " f"audited. Neither is renamed, and neither is reported here.") return 0 if __name__ == "__main__": sys.exit(main())