"""RUN-021 - Round 18: the production shard emitter, Rice-coded overrides and replay economics.

Round 18 turns Round 17's stack into a checkpoint format. The D3 root, split officially to
depth 3, gives eight shards; each is emitted to global depth 16 as a preorder topology stream
(1 bit a node), a leaf-tag stream (2 bits: SYM, CERT, PENDING) and, for the tail window,
Rice-coded override events. An independent verifier replays them, and a lifetime cost model
prices the two schedulers. Here both certificates are decoded from the text's grammar and
the format spec alone, every leaf is re-proved (the pilot's bound and exact cores), both
forests are re-emitted byte for byte, the sizes and the cost arithmetic are recomputed, and
the shipped emitter and verifier are read, after that, for what they do.

Checks
  A0 anchors      the numbers and formulas below occur in Round 18 as written
  W1 decode       sec. 2-10, 12: both certificates decode from the text's grammar; counts,
                  bits, digests and the pending files agree; the conventions the spec leaves
                  open are fixed by the data, the others failing
  W2 leaves       sec. 4-6, 20, 27: SYM by geometry, PENDING at depth 16, every CERT by the
                  pilot's bound (the recorded worst slacks) and by exact cores
  W3 emit         sec. 13-14: the official forest re-emitted byte for byte from the text; the
                  tail forest only with its emitter's lookahead, which does not count a SYM
                  child as certified - so it is not Round 17's tail window
  W4 numbers      sec. 14-24: the reductions, the axis metadata, the stream, zlib and zip
                  sizes, the evaluation counts and the cost model's arithmetic
  W5 verifier     sec. 19, 29, 31: what the shipped verifier checks, read after the rebuild
  P1 package      benchmark, manifests, summaries, cost policy, spec and crystal agree
  P2 package      VALIDATION against the archive; its eight checks from the data
  R1 registry     Round 18's judgments reviewed, first-used statuses defined

Usage: python code/run021_round18_shard_emitter.py [--drill] [--blind CID]
"""
import argparse
import copy
import hashlib
import inspect
import io
import json
import math
import re
import zipfile
import zlib
from multiprocessing import Pool

import numpy as np

import gate
import registry
import run008_round05_saturation_gate as r8
import run009_round06_witness_exchange as r9
import run011_round08_conditional_lift as r11
import run012_round09_adaptive_atlas as r12
import run013_round10_reference_dry_run as r13
import run019_round16_global_pilot as r19
import run020_round17_d3_cpg as r20
from corpus import SOURCES, TREE, decode_name, source_bytes, walk_sources

PKG = "main.zip!/AMRAL_LUC_FC_Round_18_Package_v0.1.zip!/"
DOC = PKG + "AMRAL_Lebesgue_Finite_Closure_Round_18_Production_Shard_Emitter_v0.1.md"
CRYSTAL = PKG + "AMRAL_LUC_FC_State_Crystal_Round_18_v0.1.md"
FORMAT = PKG + "AMRAL_LUC_FC_Round_18_FORMAT_SPEC.json"
BENCH = PKG + "AMRAL_LUC_FC_Round_18_BENCHMARK_SUMMARY.json"
COST = PKG + "AMRAL_LUC_FC_Round_18_COST_POLICY.json"
VALIDATION = PKG + "AMRAL_LUC_FC_Round_18_VALIDATION.json"
REF = PKG + "AMRAL_LUC_FC_Round_18_reference/"
EMITTER = REF + "production_shard_emitter.py"
VERIFIER = REF + "production_shard_verifier.py"
BASEPY = REF + "round16_base_pilot.py"
R16CODE = "main.zip!/AMRAL_LUC_FC_Round_16_Package_v0.1.zip!/AMRAL_LUC_FC_Round_16_global_pilot.py"
SUMS = {k: REF + k for k in ("emitter_summary.json", "emitter_stdout.json", "verifier_summary.json", "verifier_stdout.json")}
CERTS = ("certificate_official", "certificate_tail_window")
LOG = TREE / "data" / "gate-logs" / "run021-round18-shard-emitter.json"
CACHE = TREE / "data" / "cache" / "run021-certificates.json"
MANIFEST = TREE / "data" / "SOURCE_MANIFEST.v1.json"

T, TS = 0.835, 0.8350002
GATE = T - 0.025

ANCHORS = {
    "W1": [r"\boxed{\text{seeddepth}=3.}", r"\boxed{S=2^3=8}", r"\boxed{1\text{bit}}", r"\boxed{2\text{bits}}", r"00=SYM",
           r"01=CERT", r"10=PENDING", r"11=RESERVED", r"g_j=i_j-i_{j-1}-1,", r"i_0=-1.", r"k\in\{0,1,\ldots,6\}.",
           r"g_j=2^kq_j+r_j,", r"N_s=2L_s-1.", r"\boxed{N=2L-S,\qquadS=8.}", r"\boxed{N_O=12350}", r"L_O=6179.",
           r"\texttt{SYM}=109,", r"\texttt{CERT}=3304,", r"\boxed{\texttt{PENDING}=2766.}", r"\boxed{N_T=10086}",
           r"L_T=5047.", r"\texttt{SYM}=106,", r"\texttt{CERT}=2893,", r"\boxed{\texttt{PENDING}=2048.}", r"\boxed{573.}"],
    "W2": [r"y_{3,\min}>\sqrt3\,x_{3,\max}.", r"T_{\rmsearch}=0.8350002,", r"6.15\times10^{-6}.", r"1.14\times10^{-5}.",
           r"\texttt{FP\_PAD}=10^{-9}."],
    "W4": [r"4983.", r"\boxed{11.37\%.}", r"\boxed{18.33\%.}", r"\boxed{25.96\%.}", r"\boxed{718}", r"15117\text{bits}.",
           r"\boxed{3823\text{bits}.}", r"\boxed{25.29\%.}", r"3094\text{bytes}.", r"\boxed{3011\text{bytes}.}",
           r"\boxed{2.68\%.}", r"1660\text{bytes}.", r"1782\text{bytes}.", r"7.35\%", r"25045\text{bytes}.",
           r"\boxed{22263\text{bytes}.}", r"\boxed{11.11\%.}", r"\boxed{2.3140\text{s}}", r"\boxed{1.9887\text{s}}",
           r"\boxed{14.06\%\text{faster}.}", r"E_O\approx6.98\text{s}.", r"E_T\approx17.32\text{s}.", r"\boxed{2.48\times}",
           r"\boxed{C=E+RV+\kappaP+\muB.}", r"R>\frac{E_T-E_O}{V_O-V_T}.", r"\boxed{R_\star\approx31.79.}",
           r"\DeltaP=2766-2048=718", r"\kappa>\frac{E_T-E_O}{718}.", r"\boxed{\kappa>0.0144\text{s/pendingleaf}.}"],
}

# ---------------------------------------------------------------- the grammar (sec. 3, 9-10; the format spec)

class Bits:
    """a bit reader; msb=False reads each byte from its low bit"""

    def __init__(self, data, msb=True):
        self.data, self.msb, self.pos = data, msb, 0

    def read(self, n=1):
        v = 0
        for _ in range(n):
            byte = self.data[self.pos // 8]
            b = (byte >> (7 - self.pos % 8)) & 1 if self.msb else (byte >> (self.pos % 8)) & 1
            v, self.pos = (v << 1) | b, self.pos + 1
        return v


def rice_events(data, k, count, unary_one=True, msb=True):
    """(gap, rank): q in unary, k bits of r, a 2-bit rank; gap = 2^k q + r (sec. 10)"""
    b, out = Bits(data, msb), []
    for _ in range(count):
        q = 0
        while b.read() == (1 if unary_one else 0):
            q += 1
        r = b.read(k) if k else 0
        out.append(((q << k) + r, b.read(2)))
    return out, b.pos


def seed_boxes():
    """sec. 2: the D3 root split by the official rule to depth 3; the seed path's bits are the
    halves taken, 0 the lower"""
    c, h = r20.d3_root()
    seq = r19.split_sequence(h, r19.official_axis, 3)
    return {format(i, "03b"): r19.decode(format(i, "03b"), seq, c, h) for i in range(8)}


def decode_shard(seed, c0, h0, topo, tags, events, msb=True, low_first=True):
    """preorder; an override event names the internal node by its preorder index among
    internal nodes, and the axis by its rank among the four others in increasing order"""
    tb, lb = Bits(topo, msb), Bits(tags, msb)
    over, i = {}, -1
    for gap, rank in events:
        i += gap + 1
        over[i] = rank
    leaves, internal = [], [0]

    def walk(path, c, h):
        if tb.read():
            idx = internal[0]
            internal[0] += 1
            a = r19.official_axis(h)
            if idx in over:
                a = sorted(set(range(5)) - {a})[over[idx]]
            kids = r20.children(c, h, a)
            for k in ((0, 1) if low_first else (1, 0)):
                walk(path + str(k), *kids[k])
        else:
            leaves.append((seed + path, lb.read(2), c, h))

    walk("", c0, h0)
    return leaves, internal[0], tb.pos, lb.pos


def decode_certificate(files, msb=True, low_first=True, unary_one=True):
    """files: {relative path: bytes} of one certificate directory"""
    roots, out = seed_boxes(), {"shards": [], "leaves": [], "problems": []}
    for s in range(8):
        sd = f"shards/shard-{s:02d}/"
        man = json.loads(files[sd + "manifest.json"])
        topo, tags, ev = (files[sd + f] for f in ("topology.bin", "leaf_tags.bin", "override_events.bin"))
        for name, data in (("topology", topo), ("leaf_tags", tags), ("override_events", ev)):
            rec = man["streams"][name]
            if hashlib.sha256(data).hexdigest() != rec["sha256"] or len(data) != rec["bytes"] or rec["bytes"] != (rec["bits"] + 7) // 8:
                out["problems"].append(f"shard {s} {name}: digest, bytes or bit count")
        try:
            events, used = rice_events(ev, man["rice_k"], man["overrides"], unary_one, msb) if man["overrides"] else ([], 0)
            leaves, internal, tbits, lbits = decode_shard(man["seed_path"], *roots[man["seed_path"]], topo, tags, events,
                                                          msb, low_first)
        except IndexError:
            out["problems"].append(f"shard {s}: a stream runs out")
            continue
        got = {"nodes": internal + len(leaves), "internal": internal, "leaves": len(leaves),
               "counts": {name: sum(1 for x in leaves if x[1] == t) for t, name in ((0, "SYM"), (1, "CERT"), (2, "PENDING"))},
               "overrides": len(events), "bits": [tbits, lbits, used]}
        want = {k: man[k] for k in ("nodes", "internal", "leaves", "counts", "overrides")}
        want["bits"] = [man["streams"][n]["bits"] for n in ("topology", "leaf_tags", "override_events")]
        if got != want:
            out["problems"].append(f"shard {s}: decoded {got}, manifest {want}")
        if any(x[1] == 3 for x in leaves):
            out["problems"].append(f"shard {s}: a reserved tag")
        pend = [p for p, t, c, h in leaves if t == 2]
        if "\n".join(pend) + ("\n" if pend else "") != files[sd + "pending_paths.txt"].decode("utf-8"):
            out["problems"].append(f"shard {s}: pending_paths.txt is not the decoded PENDING set")
        if any(x[1] == 0 and not r20.sym(x[2], x[3], 1e-15) for x in leaves):
            out["problems"].append(f"shard {s}: a SYM leaf inside the wedge")
        if any(x[1] == 2 and len(x[0]) != 16 for x in leaves):
            out["problems"].append(f"shard {s}: a PENDING leaf above depth 16")
        out["shards"].append(got)
        out["leaves"] += leaves
    allpend = sorted(p for p, t, c, h in out["leaves"] if t == 2)
    if "\n".join(allpend) + ("\n" if allpend else "") != files["pending_frontier.txt"].decode("utf-8"):
        out["problems"].append("pending_frontier.txt is not the decoded PENDING set")
    return out


# ---------------------------------------------------------------- re-emission (sec. 2-3, 13-14)

def modes_at(c, h, target):
    """Round 16's routes against a closing threshold, with the pilot's sampling"""
    c, h = np.asarray(c, dtype=float), np.asarray(h, dtype=float)
    apr = max(r19.f_apr(r19.box_distance(c[0:2], h[0:2])), r19.f_apr(r19.box_distance(c[3:5], h[3:5])))
    tau = r19.tau_f3(h)
    rep = r19.sampled_hull([r9.placed_reuleaux(3, 0.0, c[:2]), r9.placed_reuleaux(5, c[2], c[3:5])]) \
        - (4 * T * tau + math.pi * tau * tau)
    k3, k5 = r19.cores(c, h)
    core = r19.sampled_hull([k3, k5]) if k3 is not None and k5 is not None else -1.0
    best = max(apr, rep, core)
    return best >= target, best


def _modes(args):
    return modes_at(args[0], args[1], TS)


class Writer:
    def __init__(self):
        self.bits = []

    def put(self, v, n):
        self.bits += [(v >> (n - 1 - i)) & 1 for i in range(n)]

    def bytes(self):
        b = self.bits + [0] * (-len(self.bits) % 8)
        return bytes(int("".join(map(str, b[i:i + 8])), 2) for i in range(0, len(b), 8))


def rice_encode(events):
    """the shortest k in 0..6 (sec. 10): q ones and a zero, k bits of r, the 2-bit rank"""
    best = None
    for k in range(7):
        w, last = Writer(), -1
        for i, rank in events:
            g, last = i - last - 1, i
            w.put((1 << ((g >> k) + 1)) - 2, (g >> k) + 1)
            if k:
                w.put(g & ((1 << k) - 1), k)
            w.put(rank, 2)
        if best is None or len(w.bits) < len(best[1].bits):
            best = (k, w)
    return best


def emit_shard(seed, c0, h0, policy, pool, sym_certifies=False):
    """level by level to global depth 16, then written in preorder. policy 'tail': from
    global depth 10, a parent whose official children are both uncertified and whose best
    bound is at least T - 0.025 tries the other axes in increasing order, keeps the largest
    (certified, min L, L- + L+, -tau, -axis) among those certifying a child, and stops at one
    certifying both. sym_certifies=True counts a SYM child as certified with L = 1e9, as
    Round 17's tail window does; Round 18's emitter evaluates it like any cell."""
    look = lambda cc, hh: (True, 1e9) if sym_certifies and r20.sym(cc, hh, 1e-15) else modes_at(cc, hh, TS)
    nodes, level, alternates = {}, [("", c0, h0, None)], 0
    for d in range(len(seed), 17):
        todo = [i for i, x in enumerate(level) if x[3] is None and not r20.sym(x[1], x[2], 1e-15)]
        got = dict(zip(todo, pool.map(_modes, [(level[i][1].tolist(), level[i][2].tolist()) for i in todo], chunksize=16)))
        nxt = []
        for i, (p, c, h, pre) in enumerate(level):
            if r20.sym(c, h, 1e-15):
                nodes[p] = (0,)
                continue
            ok, best = pre if pre is not None else got[i]
            if ok:
                nodes[p] = (1,)
                continue
            if d == 16:
                nodes[p] = (2,)
                continue
            ao = r19.official_axis(h)
            axis, kids = ao, r20.children(c, h, ao)
            vals = [look(cc, hh) for cc, hh in kids] if policy == "tail" else [None, None]
            if policy == "tail" and d >= 10 and not any(v[0] for v in vals) and best >= GATE:
                choice = None
                for a in sorted(set(range(5)) - {ao}):
                    alternates += 1
                    kk = r20.children(c, h, a)
                    vv = [look(cc, hh) for cc, hh in kk]
                    cert = sum(v[0] for v in vv)
                    if cert > 0:
                        key = (cert, min(v[1] for v in vv), sum(v[1] for v in vv), -r19.tau_f3(kk[0][1]), -a)
                        if choice is None or key > choice[0]:
                            choice = (key, a, kk, vv)
                        if cert == 2:
                            break
                if choice is not None:
                    axis, kids, vals = choice[1], choice[2], choice[3]
            nodes[p] = ("split", ao, axis)
            for k, (cc, hh) in enumerate(kids):
                keep = vals[k] if policy == "tail" and not r20.sym(cc, hh, 1e-15) else None
                nxt.append((p + str(k), cc, hh, keep))
        level = nxt
    topo, tags, events, idx = Writer(), Writer(), [], [0]

    def walk(p):
        v = nodes[p]
        if v[0] != "split":
            topo.put(0, 1)
            tags.put(v[0], 2)
            return
        topo.put(1, 1)
        if v[2] != v[1]:
            events.append((idx[0], sorted(set(range(5)) - {v[1]}).index(v[2])))
        idx[0] += 1
        walk(p + "0")
        walk(p + "1")

    walk("")
    k, w = rice_encode(events)
    internal = sum(1 for v in nodes.values() if v[0] == "split")
    return {"topology": topo.bytes().hex(), "leaf_tags": tags.bytes().hex(), "override_events": w.bytes().hex() if events else "",
            "rice_k": k if events else 0, "overrides": len(events), "internal": internal, "nodes": len(nodes),
            "alternates": alternates}


def compute(files):
    with Pool(14) as pool:
        out = {"emit": {}}
        boxes = seed_boxes()
        for name, policy, sc in (("official", "official", False), ("tail", "tail", False), ("tail_r17", "tail", True)):
            out["emit"][name] = [emit_shard(format(s, "03b"), *boxes[format(s, "03b")], policy, pool, sc) for s in range(8)]
        out["reproof"] = {}
        for cert in CERTS:
            leaves = [(c.tolist(), h.tolist()) for p, t, c, h in decode_certificate(files[cert])["leaves"] if t == 1]
            sampled = pool.map(_reproof_sampled, leaves, chunksize=32)
            exact = pool.map(_reproof_exact, leaves, chunksize=16)
            out["reproof"][cert] = {"cert_leaves": len(leaves), "sampled": sampled, "exact": exact}
    return out


def _reproof_sampled(args):
    m = r19.modes(args[0], args[1], False)
    return [m[0], m[1]]


def _reproof_exact(args):
    m = r19.modes(args[0], args[1], True)
    return [m[0], m[1]]


def _code_key(files):
    digest = hashlib.sha256()
    for fn in (Bits, rice_events, seed_boxes, decode_shard, decode_certificate, modes_at, _modes, Writer, rice_encode,
               emit_shard, compute, _reproof_sampled, _reproof_exact):
        digest.update(inspect.getsource(fn).encode())
    for mod in (r8, r9, r11, r12, r13, r19, r20):
        digest.update(open(mod.__file__, "rb").read())
    for cert in CERTS:
        for k in sorted(files[cert]):
            digest.update(k.encode() + files[cert][k])
    return digest.hexdigest()


_COMPUTED = {}


def computed(files):
    key = _code_key(files)
    if key not in _COMPUTED:
        if CACHE.exists():
            cached = json.loads(CACHE.read_text(encoding="utf-8"))
            if cached.get("key") == key:
                _COMPUTED[key] = cached["results"]
        if key not in _COMPUTED:
            _COMPUTED[key] = json.loads(json.dumps(compute(files)))
            CACHE.parent.mkdir(parents=True, exist_ok=True)
            CACHE.write_text(json.dumps({"key": key, "results": _COMPUTED[key]}) + "\n", encoding="utf-8")
    return _COMPUTED[key]


# ---------------------------------------------------------------- checks

def totals(dec):
    t = {k: sum(s[k] for s in dec["shards"]) for k in ("nodes", "internal", "leaves", "overrides")}
    t.update({k: sum(s["counts"][k] for s in dec["shards"]) for k in ("SYM", "CERT", "PENDING")})
    t["rice_bits"] = sum(s["bits"][2] for s in dec["shards"])
    return t


def w1_decode(w):
    out, got = [], {}
    for cert in CERTS:
        files = w["certs"][cert]
        gm = json.loads(files["global_manifest.json"])
        for s, rec in enumerate(gm["shards"]):
            if hashlib.sha256(files[f"shards/shard-{s:02d}/manifest.json"]).hexdigest() != rec["manifest_sha256"]:
                out.append(f"{cert} shard {s}: its manifest is not the one the global manifest records")
        dec = decode_certificate(files)
        out += [f"{cert}: {p}" for p in dec["problems"]]
        t = totals(dec)
        got[cert] = t
        if len(dec["shards"]) == 8 and (t["nodes"] != 2 * t["leaves"] - 8 or t["internal"] != t["leaves"] - 8):
            out.append(f"{cert}: N = 2L - S fails")
        want = {k: gm["totals"][k] for k in ("nodes", "internal", "leaves", "SYM", "CERT", "PENDING", "overrides")}
        if {k: t[k] for k in want} != want or {k: t[k] for k in w["text_totals"][cert]} != w["text_totals"][cert]:
            out.append(f"{cert}: decoded {t}, manifest {want}, text {w['text_totals'][cert]}")
    tail = w["certs"]["certificate_tail_window"]
    others = {"LSB first": dict(msb=False), "higher child first": dict(low_first=False), "zeros then a one": dict(unary_one=False)}
    fails = {name: bool(decode_certificate(tail, **kw)["problems"]) for name, kw in others.items()}
    if all(fails.values()) != w["other_conventions_fail"]:
        out.append(f"the other conventions fail to decode: {fails}")
    w["w1"] = {"decoded": got, "other_conventions_fail": fails}
    return out


def w2_leaves(w):
    out, rp = [], w["computed"]["reproof"]
    rec = {"certificate_official": w["bench"]["official"]["verifier"]["worst_cert_slack"],
           "certificate_tail_window": w["bench"]["tail_window"]["verifier"]["worst_cert_slack"]}
    got = {}
    for cert in CERTS:
        s, e = rp[cert]["sampled"], rp[cert]["exact"]
        if any(m != "CORE" or v < T for m, v in s):
            out.append(f"{cert}: a CERT leaf the pilot's bound does not close, or not by CORE")
        if any(v < T for m, v in e):
            out.append(f"{cert}: a CERT leaf exact cores do not close")
        ws, we = min(v for m, v in s) - T, min(v for m, v in e) - T
        if abs(ws - rec[cert]) > 1e-12:
            out.append(f"{cert}: worst sampled slack {ws!r}, verifier {rec[cert]!r}")
        got[cert] = {"cert_leaves": rp[cert]["cert_leaves"], "worst_sampled_slack": ws, "worst_exact_slack": we}
    if round(min(g["worst_exact_slack"] for g in got.values()), 4) != w["exact_worst"]:
        out.append(f"exact cores clear 0.835 by {min(g['worst_exact_slack'] for g in got.values()):.6f}, recorded {w['exact_worst']}")
    w["w2"] = got
    return out


def w3_emit(w):
    out, em = [], w["computed"]["emit"]
    for name, cert in (("official", "certificate_official"), ("tail", "certificate_tail_window")):
        files = w["certs"][cert]
        for s, r in enumerate(em[name]):
            sd = f"shards/shard-{s:02d}/"
            man = json.loads(files[sd + "manifest.json"])
            for stream in ("topology", "leaf_tags", "override_events"):
                if bytes.fromhex(r[stream]) != files[sd + stream + ".bin"]:
                    out.append(f"{name} shard {s}: {stream} re-emitted differs")
            if r["overrides"] and r["rice_k"] != man["rice_k"]:
                out.append(f"{name} shard {s}: Rice k {r['rice_k']}, recorded {man['rice_k']}")
    acc = []
    for name, cert, extra in (("official", "certificate_official", False), ("tail", "certificate_tail_window", True)):
        files = w["certs"][cert]
        for s, r in enumerate(em[name]):
            man = json.loads(files[f"shards/shard-{s:02d}/manifest.json"])
            e = 1 + 2 * r["internal"] + (r["internal"] + 2 * r["alternates"] if extra else 0)
            acc.append(e == man["lower_evaluations"])
    if all(acc) != w["evaluation_accounting"]:
        out.append("the emitter's evaluations are 1 + 2I (official) and 1 + 3I + 2A (tail) per shard: "
                   f"{sum(acc)} of {len(acc)} shards")
    r17 = em["tail_r17"]
    same = all(bytes.fromhex(r["topology"]) == w["certs"]["certificate_tail_window"][f"shards/shard-{s:02d}/topology.bin"]
               for s, r in enumerate(r17))
    n17 = sum(r["overrides"] for r in r17)
    if same != w["r17_rule_reproduces"] or n17 != w["r17_tail_overrides"]:
        out.append(f"with Round 17's lookahead the tail forest {'is' if same else 'is not'} the certificate's; "
                   f"{n17} overrides")
    w["w3"] = {"r17_rule_overrides": n17, "r17_rule_nodes": sum(r["nodes"] for r in r17),
               "tail_alternates": sum(r["alternates"] for r in em["tail"])}
    return out


def w4_numbers(w):
    out, b, cost = [], w["bench"], w["cost"]
    dec = {cert: totals(decode_certificate(w["certs"][cert])) for cert in CERTS}
    o, t = dec["certificate_official"], dec["certificate_tail_window"]
    streams = {cert: sum(len(v) for k, v in w["certs"][cert].items() if k.endswith(".bin")) for cert in CERTS}
    zl = {cert: len(zlib.compress(b"".join(v for k, v in sorted(w["certs"][cert].items()) if k.endswith(".bin")), 9))
          for cert in CERTS}
    zp = w["zip_sizes"]
    Eo, Et = cost["measured_reference"]["official"]["E"], cost["measured_reference"]["tail_window"]["E"]
    Vo, Vt = cost["measured_reference"]["official"]["V"], cost["measured_reference"]["tail_window"]["V"]
    got = {"node_reduction_fraction": (o["nodes"] - t["nodes"]) / o["nodes"],
           "pending_reduction_fraction": (o["PENDING"] - t["PENDING"]) / o["PENDING"],
           "raw_binary_stream_reduction_fraction": (streams["certificate_official"] - streams["certificate_tail_window"])
                                                   / streams["certificate_official"],
           "full_zip_reduction_fraction": (zp["certificate_official"] - zp["certificate_tail_window"]) / zp["certificate_official"],
           "zlib_concatenated_binary_change_fraction": zl["certificate_tail_window"] / zl["certificate_official"] - 1,
           "emitter_slowdown_factor": Et / Eo, "replay_speedup_fraction": (Vo - Vt) / Vo,
           "replay_only_break_even_count": (Et - Eo) / (Vo - Vt),
           "pending_continuation_break_even_seconds_per_saved_pending_leaf_at_R0": (Et - Eo) / (o["PENDING"] - t["PENDING"]),
           "override_fraction_of_tail_internal_nodes": t["overrides"] / t["internal"],
           "rice_sparse_axis_bits_fraction_of_explicit_3bit_axes": t["rice_bits"] / (3 * t["internal"])}
    for k, v in got.items():
        if abs(v - b["comparison"][k]) > 1e-12:
            out.append(f"{k}: recomputed {v!r}, recorded {b['comparison'][k]!r}")
    rec = {"certificate_official": b["official"]["emitter"], "certificate_tail_window": b["tail_window"]["emitter"]}
    for cert in CERTS:
        e = rec[cert]
        if (streams[cert], zl[cert], zp[cert]) != (e["binary_stream_bytes"], e["zlib9_stream_bytes"],
                                                  b["official" if "official" in cert else "tail_window"]["full_zip_bytes"]):
            out.append(f"{cert}: streams {streams[cert]}, zlib {zl[cert]}, zip {zp[cert]} differ from the benchmark")
        if 3 * dec[cert]["internal"] != e["axis_metadata"]["explicit_3bit_axis_all_internal_bits"] \
                or dec[cert]["rice_bits"] != e["axis_metadata"]["sparse_override_event_bits"]:
            out.append(f"{cert}: axis metadata differs")
    if (t["internal"] != w["text_internal"]) != w["text_internal_wrong"]:
        out.append(f"the text's 'internal nodes {w['text_internal']}' against the decoded {t['internal']}")
    Eot, Ett = b["official"]["emitter"]["totals"]["lower_evaluations"], b["tail_window"]["emitter"]["totals"]["lower_evaluations"]
    redundant = t["internal"]
    ratios = [round(Ett / Eot, 4), round((Ett - redundant) / Eot, 4)]
    if [redundant, ratios] != w["redundant"]:
        out.append(f"the tail emitter re-evaluates {redundant} parents; evaluation ratio {ratios}, recorded {w['redundant']}")
    w["w4"] = {"recomputed": got, "streams": streams, "zlib9": zl, "zip": zp, "evaluations": [Eot, Ett],
               "redundant_parent_evaluations": redundant, "evaluation_ratio_with_and_without": ratios}
    return out


def function_body(code, name):
    m = re.search(rf"^def {name}\(.*?(?=^def |\Z)", code, re.S | re.M)
    return m.group(0) if m else ""


def w5_verifier(w):
    out, v, e, base = [], w["verifier_code"], w["emitter_code"], w["base_code"]
    marker = "P[j]=p1 if np.linalg.norm(p1-old)<np.linalg.norm(p2-old) else p2"
    facts = {"reads_pending_files": bool(re.search(r"pending_(paths|frontier)", v)),
             "compares_manifest_counts": bool(re.search(r"M\[['\"]counts['\"]\]", v)),
             "geometry_transcribed_from_pilot": marker in v and marker in base,
             "own_hull": "def hull(" in v and "ConvexHull" not in v,
             "lookahead_ignores_sym": "sym_out" not in function_body(e, "choose") and "sym_out" in e,
             "gate_re_evaluates_parent": "parent=evaluate(c,h,threshold=TARGET)" in e}
    if facts != w["code_facts"]:
        out.append(f"the shipped code: {facts}, recorded {w['code_facts']}")
    w["w5"] = facts
    return out


def p1_package(w):
    out, b = [], w["bench"]
    gms = {cert: json.loads(w["certs"][cert]["global_manifest.json"]) for cert in CERTS}
    if b["official"]["emitter"] != gms["certificate_official"] or b["tail_window"]["emitter"] != gms["certificate_tail_window"]:
        out.append("the benchmark's emitter blocks are not the global manifests")
    es = json.loads(w["sums"]["emitter_summary.json"])
    if es != {"OFFICIAL": gms["certificate_official"], "TAIL-WINDOW-CPG": gms["certificate_tail_window"]} \
            or json.loads(w["sums"]["emitter_stdout.json"]) != es:
        out.append("the emitter's summary or stdout is not the two global manifests")
    vs = json.loads(w["sums"]["verifier_summary.json"])
    if vs != {"OFFICIAL": b["official"]["verifier"], "TAIL-WINDOW-CPG": b["tail_window"]["verifier"]} \
            or json.loads(w["sums"]["verifier_stdout.json"]) != vs:
        out.append("the verifier's summary or stdout is not the benchmark's verifier blocks")
    m = w["cost"]["measured_reference"]
    for key, name in (("official", "official"), ("tail_window", "tail_window")):
        x = b[name]
        if (m[key]["E"], m[key]["V"], m[key]["P"], m[key]["B_zip"]) != (x["emitter"]["emitter_seconds"], x["verifier"]["replay_seconds"],
                                                                      x["verifier"]["counts"]["PENDING"], x["full_zip_bytes"]):
            out.append(f"the cost policy's {key} figures differ from the benchmark")
    f = w["format"]["streams"]
    if (f["topology"]["bits_per_node"], f["topology"]["internal"], f["topology"]["leaf"], f["leaf_tags"]["bits_per_leaf"],
            f["leaf_tags"]["SYM"], f["leaf_tags"]["CERT"], f["leaf_tags"]["PENDING"]) != (1, 1, 0, 2, 0, 1, 2):
        out.append("the format spec's grammar is not the text's")
    cr = gate.squash(w["crystal"])
    frags = ["N=12350,", "\\texttt{PENDING}=2766,", "3094\\text{bytes}.", "2.3140\\text{s}.", "N=10086,",
             "\\texttt{PENDING}=2048,", "573\\text{overrides}.", "3011\\text{bytes}.", "1.9887\\text{s}.", "$-18.33\\%$",
             "$-25.96\\%$", "$-2.68\\%$", "$-11.11\\%$", "$-14.06\\%$", "25.29\\%", "6.98\\text{s}.", "17.32\\text{s}.",
             "$2.48\\times$", "R_\\star\\approx31.79.", "14.4\\text{mspersavedpendingleaf}."]
    out += [f"the crystal lacks {x}" for x in frags if x not in cr]
    comp = b["comparison"]
    for x, fmt in ((comp["node_reduction_fraction"], "$-{:.2f}\\%$"), (comp["pending_reduction_fraction"], "$-{:.2f}\\%$"),
                   (comp["replay_only_break_even_count"], "R_\\star\\approx{:.2f}.")):
        if fmt.format(100 * x if "%" in fmt else x) not in cr:
            out.append(f"the crystal does not round {x}")
    return out


def p2_package(w):
    out, leaves = [], {row[0]: row for row in w["manifest"]["leaves"]}
    v = w["validation"]
    for rec, leaf_id, text in ((v["round18"], DOC, w["text"]), (v["state_crystal"], CRYSTAL, w["crystal"])):
        _, size, digest = leaves[leaf_id]
        if rec["sha256"] != digest or rec["bytes"] != size:
            out.append(f"{rec['file']}: recorded sha256/bytes differ from the archive")
        n = text.count("$$")
        if rec["double_dollar_count"] != n or rec["double_dollar_even"] != (n % 2 == 0):
            out.append(f"{rec['file']}: recorded {rec['double_dollar_count']} $$, document has {n}")
        if rec["contains_unicode_escape_sequence"] != bool(re.search(r"\\u[0-9a-fA-F]{4}", text)):
            out.append(f"{rec['file']}: unicode-escape flag differs")
        if rec["contains_bracket_math_delimiter"] != bool(re.search(r"\\[\[\]()]", text)):
            out.append(f"{rec['file']}: bracket-delimiter flag differs")
    for key, leaf_id in (("format_spec_sha256", FORMAT), ("benchmark_summary_sha256", BENCH), ("cost_policy_sha256", COST),
                         ("emitter_sha256", EMITTER), ("verifier_sha256", VERIFIER)):
        if v[key] != leaves[leaf_id][2]:
            out.append(f"{key}: recorded digest differs from the archive")
    b = w["bench"]
    o, t = b["official"], b["tail_window"]
    checks = {"official_verified": o["verifier"]["status"] == "VERIFIED-CHECKPOINT",
              "tail_verified": t["verifier"]["status"] == "VERIFIED-CHECKPOINT",
              "both_not_theorem_ready": not o["verifier"]["theorem_ready"] and not t["verifier"]["theorem_ready"],
              "pending_reduced": t["verifier"]["counts"]["PENDING"] < o["verifier"]["counts"]["PENDING"],
              "raw_streams_reduced": t["emitter"]["binary_stream_bytes"] < o["emitter"]["binary_stream_bytes"],
              "rice_metadata_less_than_explicit": t["emitter"]["axis_metadata"]["sparse_override_event_bits"]
                                                  < t["emitter"]["axis_metadata"]["explicit_3bit_axis_all_internal_bits"],
              "tail_replay_faster": t["verifier"]["replay_seconds"] < o["verifier"]["replay_seconds"],
              "global_claim_false": not o["emitter"]["global_lebesgue_claim"] and not t["emitter"]["global_lebesgue_claim"]}
    if checks != v["checks"]:
        out.append(f"VALIDATION's checks {v['checks']}, the data gives {checks}")
    if leaves[BASEPY][2] != leaves[R16CODE][2]:
        out.append("the reference's round16_base_pilot.py is not Round 16's code")
    return out


# ---------------------------------------------------------------- registry (Round 18)

R18_DEFINITIONS = {
    "FORMALIZED + EMPIRICALLY CALIBRATED": ("in-line", "sec. 31: stated as a formula and calibrated on the reference "
                                                       "run's timings"),
    "IMPLEMENTED + VERIFIED": ("in-line", "sec. 31: implemented, and checked by the independent replay"),
    "Tail-Window preferred": ("in-line", "sec. 25: the engineering preference for a long-running checkpoint"),
}
R18_REVIEWS = [
    ("PASS", "Official checkpoint replay", "computation", "PASS",
     "decoded from the text's grammar and re-proved leaf by leaf - worst slack 6.15e-6 with the pilot's bound, 2.4e-3 "
     "with exact cores (W1, W2) - and re-emitted byte for byte (W3)"),
    ("PASS", "Tail-Window checkpoint replay", "computation", "PASS",
     "W1, W2; re-emitted byte for byte only with its emitter's lookahead, which does not count a SYM child as "
     "certified - not Round 17's tail window, which gives 536 overrides here, not 573 (W3)"),
    ("PASS", "Pending frontier preservation", "computation", "PASS",
     "every pending file is the decoded PENDING set, checked here (W1); the shipped verifier never reads them (W5)"),
    ("CLOSED", "Production shard grammar", "reading", "PARTIAL",
     "it decodes as written (W1), but three conventions are stated nowhere - bits most significant first, the lower "
     "child first, unary as ones ended by a zero - and only trial against the data fixes them"),
    ("COMPUTE-DEFERRED", "Final $0.8350$ theorem", "n/a", "n/a", ""),
    ("VERIFIED", "PRODUCTION CHECKPOINT FORMAT", "computation", "PASS",
     "W1-W3: decoded, re-proved and re-emitted independently of the shipped verifier, whose own checks omit the "
     "pending files and the manifests' counts (W5)"),
    ("VERIFIED", "D3 root replay", "computation", "PASS",
     "the seed boxes are the D3 root split officially to depth 3, and every SYM leaf lies outside the wedge (W1)"),
    ("COMPLETE", "28. Shard status", "n/a", "n/a", "the status sec. 28 says no shard has; defined in R11"),
    ("EXPANDED", "28. Shard status", "computation", "PASS", "every shard holds PENDING leaves (W1)"),
    ("MISHRA-FIRST-ORDER", "7. Default split is implicit", "n/a", "n/a",
     "the default axis, defined in R17 (RUN-020); the decoder rebuilds it (W1)"),
    ("Tail-Window preferred", "long-running checkpoint", "computation", "PASS",
     "the arithmetic holds (W4); the break-even thresholds come from single timings of an emitter that spends 17% of "
     "its evaluations re-evaluating parents it has already evaluated, which only strengthens the preference"),
    ("IMPLEMENTED + VERIFIED", "Rice-coded sparse override", "computation", "PASS",
     "the events decode to 3823 bits, each shard's k is the shortest, and they re-encode byte for byte (W1, W3)"),
    ("FORMALIZED + EMPIRICALLY CALIBRATED", "Lifetime cost model", "reading", "PARTIAL",
     "the formula and its break-evens are arithmetic of four single, unrepeated timings (W4); the tail emitter's includes "
     "an avoidable re-evaluation of every parent (5039 of its 30229 evaluations)"),
]


def r1_registry(w):
    return registry.completeness(w["registry"], "R18", w["definitions"], w["reviews"])


# ---------------------------------------------------------------- world, checks, drill

CHECKS = {"A0": gate.anchors_check(ANCHORS), "W1": w1_decode, "W2": w2_leaves, "W3": w3_emit, "W4": w4_numbers,
          "W5": w5_verifier, "P1": p1_package, "P2": p2_package, "R1": r1_registry}
DEPENDS = {"A0": {"text"}, "W1": {"certs", "text_totals", "other_conventions_fail"},
           "W2": {"computed", "bench", "exact_worst"},
           "W3": {"computed", "certs", "evaluation_accounting", "r17_rule_reproduces", "r17_tail_overrides"},
           "W4": {"certs", "bench", "cost", "zip_sizes", "text_internal", "text_internal_wrong", "redundant"},
           "W5": {"verifier_code", "emitter_code", "base_code", "code_facts"},
           "P1": {"bench", "certs", "sums", "cost", "format", "crystal"},
           "P2": {"validation", "manifest", "text", "crystal", "bench"},
           "R1": {"registry", "definitions", "reviews"}}
CLAIMS = {
    "text_totals": {"certificate_official": {"nodes": 12350, "leaves": 6179, "SYM": 109, "CERT": 3304, "PENDING": 2766},
                    "certificate_tail_window": {"nodes": 10086, "leaves": 5047, "SYM": 106, "CERT": 2893, "PENDING": 2048,
                                                "overrides": 573}},
    "other_conventions_fail": True, "exact_worst": 0.0024,
    "evaluation_accounting": True, "r17_rule_reproduces": False, "r17_tail_overrides": 536,
    "text_internal": 4983, "text_internal_wrong": True, "redundant": [5039, [2.4477, 2.0397]],
    "code_facts": {"reads_pending_files": False, "compares_manifest_counts": False, "geometry_transcribed_from_pilot": True,
                   "own_hull": True, "lookahead_ignores_sym": True, "gate_re_evaluates_parent": True},
}


def load_inputs():
    blobs = {n: source_bytes(n) for n in SOURCES}
    want = {DOC, CRYSTAL, FORMAT, BENCH, COST, VALIDATION, EMITTER, VERIFIER, BASEPY, *SUMS.values()}
    certs = {c: {} for c in CERTS}
    got = {}
    for leaf, d in walk_sources(blobs, keep=lambda l: l.id in want or any(l.id.startswith(REF + c + "/") for c in CERTS)):
        if d is None:
            continue
        for c in CERTS:
            if leaf.id.startswith(REF + c + "/"):
                certs[c][leaf.id[len(REF + c + "/"):]] = d
        got[leaf.id] = d
    main = zipfile.ZipFile(io.BytesIO(blobs["main.zip"]))
    pkg = zipfile.ZipFile(io.BytesIO(main.read(next(i for i in main.infolist() if "Round_18_Package" in decode_name(i)))))
    sizes = {decode_name(i): i.file_size for i in pkg.infolist()}
    return {"text": got[DOC].decode("utf-8"), "crystal": got[CRYSTAL].decode("utf-8"), "format": json.loads(got[FORMAT]),
            "bench": json.loads(got[BENCH]), "cost": json.loads(got[COST]), "validation": json.loads(got[VALIDATION]),
            "emitter_code": got[EMITTER].decode("utf-8"), "verifier_code": got[VERIFIER].decode("utf-8"),
            "base_code": got[BASEPY].decode("utf-8"), "sums": {k: got[v] for k, v in SUMS.items()}, "certs": certs,
            "zip_sizes": {c: sizes[f"AMRAL_LUC_FC_Round_18_reference/{c}.zip"] for c in CERTS},
            "manifest": json.loads(MANIFEST.read_text(encoding="utf-8"))}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--drill", action="store_true")
    ap.add_argument("--blind", metavar="CID")
    args = ap.parse_args()
    inputs = load_inputs()
    results_computed = computed(inputs["certs"])

    def build():
        return {**copy.deepcopy(inputs), "computed": results_computed, "registry": registry.load(),
                "definitions": dict(R18_DEFINITIONS), "reviews": list(R18_REVIEWS), **copy.deepcopy(CLAIMS)}

    if args.blind:
        CHECKS[args.blind] = lambda w: []
        args.drill = True
    w = build()
    results = {c: fn(w) for c, fn in CHECKS.items()}
    for c, f in results.items():
        print(f"{c} {'PASS' if not f else 'FAIL'}", *(f"\n     {x}" for x in f[:6]))
    for key in ("w1", "w2", "w3", "w4", "w5"):
        print(key, json.dumps(w.get(key), default=float)[:900])
    if not args.drill:
        return
    certs = inputs["certs"]
    pend = copy.deepcopy(certs)
    txt = pend["certificate_tail_window"]["pending_frontier.txt"].decode("utf-8").split("\n")
    txt[5] = txt[5][:-1] + ("1" if txt[5][-1] == "0" else "0")
    pend["certificate_tail_window"]["pending_frontier.txt"] = "\n".join(txt).encode("utf-8")
    count = copy.deepcopy(certs)
    man = json.loads(count["certificate_tail_window"]["shards/shard-02/manifest.json"])
    man["counts"]["CERT"] += 1
    count["certificate_tail_window"]["shards/shard-02/manifest.json"] = json.dumps(man, indent=2).encode("utf-8")
    bench = copy.deepcopy(inputs["bench"])
    bench["official"]["verifier"]["worst_cert_slack"] += 1e-9
    cost = copy.deepcopy(inputs["cost"])
    cost["measured_reference"]["official"]["E"] = 6.9
    fmt = json.loads(json.dumps(inputs["format"], sort_keys=True))
    cases = [
        ("K0-unmodified", "control", {}, set()),
        ("K1-doc-crlf", "control", {"text": inputs["text"].replace("\n", "\r\n")}, set()),
        ("K2-certificate-files-reordered", "control",
         {"certs": {c: dict(sorted(certs[c].items(), reverse=True)) for c in CERTS}}, set()),
        ("K3-format-spec-keys-sorted", "control", {"format": fmt}, set()),
        ("M1-doc-official-nodes", "mutation", {"text": inputs["text"].replace("N_O=12350", "N_O=12351")}, {"A0"}),
        ("M2-pending-path", "mutation", {"certs": pend}, {"W1"}),
        ("M3-manifest-count", "mutation", {"certs": count}, {"W1"}),
        ("M4-other-conventions-decode", "mutation", {"other_conventions_fail": False}, {"W1"}),
        ("M5-exact-slack", "mutation", {"exact_worst": 0.0034}, {"W2"}),
        ("M6-verifier-slack", "mutation", {"bench": bench}, {"W2", "P1"}),
        ("M7-r17-rule-reproduces", "mutation", {"r17_rule_reproduces": True}, {"W3"}),
        ("M8-evaluation-accounting", "mutation", {"evaluation_accounting": False}, {"W3"}),
        ("M9-text-internal-right", "mutation", {"text_internal_wrong": False}, {"W4"}),
        ("M10-cost-timing", "mutation", {"cost": cost}, {"W4", "P1"}),
        ("M11-zip-size", "mutation", {"zip_sizes": {**inputs["zip_sizes"], "certificate_official": 25046}}, {"W4"}),
        ("M12-redundant-evaluations", "mutation", {"redundant": [0, [2.4477, 2.4477]]}, {"W4"}),
        ("M13-verifier-reads-pending", "mutation",
         {"code_facts": {**CLAIMS["code_facts"], "reads_pending_files": True}}, {"W5"}),
        ("M14-lookahead-counts-sym", "mutation",
         {"code_facts": {**CLAIMS["code_facts"], "lookahead_ignores_sym": False}}, {"W5"}),
        ("M15-emitter-digest", "mutation", {"validation": {**inputs["validation"], "emitter_sha256": "0" * 64}}, {"P2"}),
        ("M16-validation-check", "mutation",
         {"validation": {**inputs["validation"], "checks": {**inputs["validation"]["checks"], "pending_reduced": False}}}, {"P2"}),
        ("M17-review-dropped", "mutation", {"reviews": list(R18_REVIEWS[1:])}, {"R1"}),
    ]
    if args.blind:
        cases = [c for c in cases if args.blind in c[3]]
    d = gate.run_drill(build, CHECKS, DEPENDS, cases)
    summary = gate.summarize(d, CHECKS)
    print(gate.report_drill(summary))
    for x in d:
        if not x["ok"]:
            print("   DRILL FAIL", x)
    if args.blind:
        n = sum(1 for x in d if args.blind in x["expected"] and args.blind not in x["tripped"])
        print(f"blind {args.blind}: drill reported it blind in {n} cases (nothing written)")
        return
    log = {"run": "RUN-021", "subject": "Round 18: the production shard emitter, Rice-coded overrides and replay economics",
           "document": DOC, "checks": {c: {"pass": not f, "failures": f} for c, f in results.items()},
           **{k: w.get(k) for k in ("w1", "w2", "w3", "w4", "w5")}, "drill": summary}
    LOG.write_text(json.dumps(log, ensure_ascii=False, indent=1, default=float) + "\n", encoding="utf-8")
    if not results["R1"]:
        registry.record("R18", "RUN-021", R18_DEFINITIONS, R18_REVIEWS)
        print(f"registry: {len(R18_DEFINITIONS)} definitions, {len(R18_REVIEWS)} reviews recorded for R18")


if __name__ == "__main__":
    main()