warm_ue_cache.py

"""Build the FULL UE 5.8 cart (Aaron 2026-07-01: whole Runtime tree, not a sample).

Three stages, ordered so the cheap gates run before the expensive one:
  1. COUNT the full filtered corpus (files, chunks, index size) — the real cost numbers.
  2. DETERMINISTIC EVAL GATE on the full corpus (parse-only, no embeddings) — must hold ~100%
     on structural types before we spend hours embedding.
  3. PRE-WARM the embed cache (the long part) — so the registered server NEVER cold-embeds
     at startup. Larger batches to keep the per-batch cache rewrite from dominating.

Run:  .venv/Scripts/python.exe warm_ue_cache.py
"""

from __future__ import annotations

import os

os.environ["CARTRIDGE_DOMAIN"] = "ue"
os.environ["CARTRIDGE_FILTER"] = "UFUNCTION"
os.environ.pop("CARTRIDGE_MAXFILES", None)   # FULL tree — no cap

import time

from corpus import CPP_EXTS, domain_roots, file_chunk_lists
from generate_claims import generate
from checker import DeterministicChecker


def main() -> int:
    t0 = time.time()
    print("=== 1. FULL-CORPUS COUNTS ===", flush=True)
    fcl = file_chunk_lists(domain_roots(), CPP_EXTS)
    chunks = [c for f in fcl for c in f]
    print(f"  filtered files: {len(fcl)}   chunks: {len(chunks)}   scan: {time.time()-t0:.0f}s", flush=True)

    t1 = time.time()
    print("=== 2. DETERMINISTIC EVAL GATE (full corpus, parse-only) ===", flush=True)
    det = DeterministicChecker()
    print(f"  index: {len(det.idx)} classes, {sum(len(e['methods']) for e in det.idx.values())} methods, "
          f"build {time.time()-t1:.0f}s", flush=True)
    claims = generate()
    flagged = [det.check(c["text"]).verdict == "flag" for c in claims]
    real = [c["label"] != "fake" for c in claims]
    structural = [i for i, c in enumerate(claims) if c["type"] != "fires_event"]
    rp = sum(not flagged[i] for i in structural if real[i]) / max(sum(real[i] for i in structural), 1)
    fc = sum(flagged[i] for i in structural if not real[i]) / max(sum(not real[i] for i in structural), 1)
    print(f"  structural gate: real-pass={rp:.0%} fake-catch={fc:.0%} (n={len(structural)})", flush=True)
    if rp < 0.99 or fc < 0.99:
        print("  GATE FAILED — do not proceed to embed; investigate parser on full-tree headers", flush=True)
        return 1

    t2 = time.time()
    print("=== 3. PRE-WARM EMBED CACHE (the long part) ===", flush=True)
    from embed import Embedder
    Embedder().embed(chunks, batch_size=256)
    print(f"  embed warm: {time.time()-t2:.0f}s   TOTAL: {time.time()-t0:.0f}s", flush=True)
    print("DONE — cache warm; safe to start/register ue-verify", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())