esh-pfi-infrastructure/scripts/r49-corpus/leak_gate.py

"""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))
    out: dict[str, dict] = {}
    for s in surfaces:
        if len(s) < 4 or not TOKEN.fullmatch(s):
            continue
        body = (SPLIT_SEP + "?").join(re.escape(c) for c in s)
        pat = re.compile(r"(?<![A-Za-z])(" + body + r")(?![A-Za-z])")
        forms: Counter = Counter()
        seen_copies = set()
        for name, text in copies.items():
            for m in pat.finditer(text):
                form = m.group(1)
                if form == s 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[form] += 1
                seen_copies.add(name)
        if forms:
            out[s] = {"forms": dict(forms), "copies": len(seen_copies)}
    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())