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


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())