vh
707fae2b2c
Per-surface scanning is O(surfaces x copies x corpus). lv-mccarthy (108 surfaces, 36 copies) finished in 8 s; lv-hemingway (881 surfaces, 10 copies) was still running at 5 minutes and had to be killed. A gate too slow to run is not a gate. Same trick scan() already uses: build one alternation, map the matched string back to its surface by stripping separators. Regression: identical verdict and identical per-surface hit counts on the pre-fix lv-mccarthy tree (5 surfaces, 78 hits) and on the fixed one (0). Re-derived on the two shipped corpora with the committed instrument rather than a scratch probe: lv-hemingway GATE FAILED Pasionaria, Primitivo, Chicote -- 6 hits each, all 6 copies lv-bronte GATE PASSED 0
304 lines
16 KiB
Python
304 lines
16 KiB
Python
"""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
|
||
|
||
# ⚠⭐ THE SEPARATOR CLASS — a leak the `\b(Surface)\b` scan above structurally CANNOT see.
|
||
# Any character inserted inside a name defeats a word-boundary pattern outright, so a mangled
|
||
# occurrence is not merely unrepaired: it is unrenameable AND unreportable, and the gate prints
|
||
# a clean zero over it. Two mechanisms are already measured in this line:
|
||
#
|
||
# lv-bronte 2026-09-16 `_Antigua_` `_` is a word character, so \bAntigua\b cannot match
|
||
# inside it. Found by hand, not by this gate.
|
||
# lv-mccarthy 2026-09-17 `B ell` a small-caps drop cap survived as its own token
|
||
# `Toad-vine` a print line-break hyphen survived the extraction
|
||
#
|
||
# McCarthy's tree had already PASSED this gate at "0 of 75 renameable and 0 of 37
|
||
# sub-threshold" while carrying 13 occurrences of five protagonist names — Bell, Chigurh,
|
||
# Moss, Toadvine, Glanton — in all six copies. Every VISIBLE occurrence had been renamed
|
||
# correctly, which is exactly what makes the residue invisible to a spot-read.
|
||
#
|
||
# ⚠ A NAIVE separator-tolerant scan is dominated by FALSE POSITIVES, because it matches across
|
||
# genuine word boundaries. Measured on lv-hemingway: 21 raw hits, of which 18 are ordinary text
|
||
# (`God damn` for the surface `Goddamn` ×14, plus `I run`/`On an`/`Do me`/`Si le`) and 3 are
|
||
# real (`Primi tivo`, `Pasionar ia`, `Chi cote`). The discriminator that separates them without
|
||
# a dictionary: in a genuine split, at least one FRAGMENT is not a word — it occurs as a
|
||
# standalone token almost nowhere in the source. `God` and `damn` occur constantly; `Primi`,
|
||
# `tivo`, `ell`, `higurh` and `Toad` do not. That single test cleared all 18 and kept all 3.
|
||
SPLIT_SEP = r"[ _\-\u00ad\u2010\u2011]"
|
||
TOKEN = re.compile(r"[A-Za-z\u00c0-\u017f']+")
|
||
|
||
|
||
def split_scan(source: dict[str, str], copies: dict[str, str], surfaces: list[str],
|
||
frag_max: int, allow: set[str]) -> dict[str, dict]:
|
||
"""Surfaces surviving in the copies with ONE separator hiding them from the unigram scan.
|
||
|
||
Returns surface -> {"forms": {matched_string: hits}, "copies": n}. Exact matches are the
|
||
unigram pass's business and are excluded here so the two cannot double-report.
|
||
"""
|
||
src_tokens = Counter()
|
||
for t in source.values():
|
||
src_tokens.update(TOKEN.findall(t))
|
||
# ⚠ ONE ALTERNATION PASS PER TEXT, not one per name -- the same reason `scan()` does it.
|
||
# Per-surface scanning is O(surfaces x copies x corpus) and it is not a theoretical cost:
|
||
# lv-mccarthy (108 surfaces, 36 copies) finished in 8 s and lv-hemingway (881 surfaces,
|
||
# 10 copies) was still running at 5 minutes and had to be killed. A gate too slow to run
|
||
# is not a gate.
|
||
cands = [s for s in surfaces if len(s) >= 4 and TOKEN.fullmatch(s)]
|
||
if not cands:
|
||
return {}
|
||
cands.sort(key=len, reverse=True)
|
||
by_stripped = {}
|
||
for s in cands:
|
||
by_stripped.setdefault(s, s) # exact form maps to itself
|
||
pat = re.compile(r"(?<![A-Za-z])(" +
|
||
"|".join((SPLIT_SEP + "?").join(re.escape(c) for c in s) for s in cands) +
|
||
r")(?![A-Za-z])")
|
||
strip = re.compile(SPLIT_SEP)
|
||
out: dict[str, dict] = {}
|
||
forms: dict[str, Counter] = defaultdict(Counter)
|
||
seen: dict[str, set] = defaultdict(set)
|
||
for name, text in copies.items():
|
||
for m in pat.finditer(text):
|
||
form = m.group(1)
|
||
surf = by_stripped.get(strip.sub("", form))
|
||
if surf is None or form == surf or form in allow:
|
||
continue
|
||
frags = TOKEN.findall(form)
|
||
# A genuine split leaves a fragment that is not a word of this corpus.
|
||
if len(frags) < 2 or all(src_tokens[f] > frag_max for f in frags):
|
||
continue
|
||
forms[surf][form] += 1
|
||
seen[surf].add(name)
|
||
for surf, f in forms.items():
|
||
out[surf] = {"forms": dict(f), "copies": len(seen[surf])}
|
||
return out
|
||
|
||
|
||
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("--split-frag-max", type=int, default=3,
|
||
help="a separator-split candidate is reported only when one of its "
|
||
"fragments occurs as a standalone token no more than this often in "
|
||
"the source. Raising it reports more and flags more ordinary text; "
|
||
"0 turns the fragment filter off entirely.")
|
||
ap.add_argument("--no-split-scan", action="store_true",
|
||
help="skip the separator-split pass. It is ON by default because the pass "
|
||
"exists to catch a class that made this gate print a false zero.")
|
||
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}")
|
||
|
||
# ---- separator-split audit --------------------------------------------
|
||
# ⚠ Its own controls, because a scan that only ever sees clean text cannot tell `absent`
|
||
# from `blind` -- the same argument that put the positive control on the unigram pass.
|
||
split_surv: dict[str, dict] = {}
|
||
split_ok = True
|
||
if not a.no_split_scan:
|
||
allow = set()
|
||
if a.phrase_map:
|
||
allow = set(json.loads(Path(a.phrase_map).read_text()).get("split_allow", []))
|
||
probe = next((s for s in sorted(surfaces, key=len, reverse=True)
|
||
if len(s) >= 4 and TOKEN.fullmatch(s)), None)
|
||
if probe:
|
||
# POSITIVE: the same surface with one separator inserted MUST be detected.
|
||
planted = {"__control__": f"the {probe[0]} {probe[1:]} rode on"}
|
||
pos = split_scan(source, planted, [probe], a.split_frag_max, set())
|
||
# NEGATIVE: the same split nonce, hunted in the REAL copies, must NOT be found.
|
||
# ⚠ The first version of this control planted the nonce in its own probe text and
|
||
# then asserted it was absent, so it failed by construction on every run. A control
|
||
# that cannot pass is not a control; it is an alarm wired to itself.
|
||
neg = split_scan(source, copies, [NONCE], a.split_frag_max, set())
|
||
split_ok = bool(pos) and not neg
|
||
print(f"\n [{'PASS' if split_ok else 'FAIL'}] split-scan controls: planted "
|
||
f"`{probe[0]} {probe[1:]}` {'detected' if pos else 'MISSED'}, split nonce "
|
||
f"{'absent' if not neg else 'FALSELY DETECTED'}")
|
||
split_surv = split_scan(source, copies, surfaces, a.split_frag_max, allow)
|
||
print(f" SEPARATOR-SPLIT survivors: {len(split_surv)} surfaces hidden from the "
|
||
f"unigram scan by a space, hyphen or underscore inside the name")
|
||
for s_, w in sorted(split_surv.items(), key=lambda kv: -sum(kv[1]["forms"].values())):
|
||
print(f" {s_:<18} {sum(w['forms'].values()):>5} hits in {w['copies']} copies "
|
||
f"as {', '.join(repr(k) for k in sorted(w['forms']))}")
|
||
|
||
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,
|
||
"split_scan_ran": not a.no_split_scan,
|
||
"split_scan_controls_pass": split_ok,
|
||
"split_frag_max": a.split_frag_max,
|
||
"surviving_separator_split": split_surv,
|
||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"\n wrote {a.report}")
|
||
|
||
if not (pos_ok and neg_ok and split_ok):
|
||
print("\n== CONTROLS FAILED -- this gate's verdict is not trustworthy"); return 2
|
||
if surv_renameable or surv_sub or surviving_phrases or split_surv:
|
||
print(f"\n== GATE FAILED: {len(surv_renameable) + len(surv_sub)} source entities, "
|
||
f"{len(surviving_phrases)} phrases and {len(split_surv)} separator-split "
|
||
f"surfaces survive"); return 1
|
||
print("\n== GATE PASSED: 0 source entities, 0 audited phrases and 0 separator-split "
|
||
"surfaces 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())
|