#!/usr/bin/env python3
"""Check the dcvc-drift record against its fixtures, and every figure the text quotes.

    python3 -I verify.py [--selftest] <research/dcvc-drift> [<text.md> ...]

Written from research/dcvc-drift.md and the wire format, sharing no code with
tools/dcvc-drift, which wrote the readings. Standard library only, no ONNX runtime: it reads
committed files and runs nothing else.

For each fixture it reads the envelope and its frame table, and decodes each frame's y stream
with an rANS decoder of its own, under the buckets the encoder coded with: the stream must end
canonically, and the symbols are the encoder's. Then, for each fabric and frame, it decodes
that stream again under the buckets the reading states, wherever those differ, and counts what
the entropy stage did: the buckets and symbols that differ, the first of each, how the stream
ended, and so whether the frame is a desync event. A fabric's buckets come from its scales,
which come from a float network this script does not run, so the buckets are the reading's own
statement; what it can hold them to is the latent. A latent is a symbol plus a float mean, so a
wrong symbol moves it by a whole step and drift by far less, and a reading whose latent says
otherwise than its symbols is refused, an event erased with its bucket among them.

Every drift reading is held to its own histogram and digests, the first bucket change to the
scale table, a run with a forced bucket to the event it repeats and to the digests of the run it
is read against, every summary and every verdict to the readings, and each file to its digest.
Each figure in the text, followed by a `dcvc` marker naming its place, must be the value at that
place, rounded as printed.

With --selftest it then changes copies of the record in memory and requires each change to be
refused, for the reason it is planted for, a changed reading and a miscounted desync among them.

Exit 0 when everything holds, 1 with one line a finding.
"""
import bisect
import copy
import hashlib
import json
import lzma
import math
import pathlib
import re
import struct
import sys

LOW = 1 << 31
MASK = (1 << 16) - 1
BYPASS = 4
GROUPS = 32 // BYPASS
I32_MAX, I32_MIN = 2**31 - 1, -(2**31)
# The line between a latent moved by drift and one moved by a wrong symbol: a symbol's step is 1.
LATENT_STEP_EDGE = 0.5


def sha(b):
    return hashlib.sha256(b).hexdigest()


class Stream:
    """A rans64 stream, precision 16, four-bit bypass groups (CompressAI's coder)."""

    def __init__(self, data):
        if len(data) % 4 or len(data) < 8:
            raise ValueError("a rans64 stream is at least two words of four bytes")
        self.words = struct.unpack(f"<{len(data) // 4}I", data)
        self.state = self.words[0] | (self.words[1] << 32)
        self.at = 2
        self.out = False

    def refill(self):
        if self.state < LOW:
            if self.at < len(self.words):
                self.state = ((self.state << 32) | self.words[self.at]) & (2**64 - 1)
                self.at += 1
            else:
                self.out = True

    def bits(self, n):
        v = self.state & ((1 << n) - 1)
        self.state >>= n
        self.refill()
        return v

    def symbol(self, cdf, length, offset):
        top = length - 2
        cum = self.state & MASK
        s = bisect.bisect_right(cdf, cum, 1, length - 1) - 1
        lo = cdf[s]
        self.state = (cdf[s + 1] - lo) * (self.state >> 16) + cum - lo
        self.refill()
        v = s
        if v == top:
            run = self.bits(BYPASS)
            n = run
            k = 0
            while run == 15 and k < GROUPS and not self.out:
                run = self.bits(BYPASS)
                n += run
                k += 1
            raw = 0
            for j in range(min(n, GROUPS)):
                raw |= self.bits(BYPASS) << (j * BYPASS)
            v = raw >> 1
            v = max(-v - 1, I32_MIN) if raw & 1 else min(v + top, I32_MAX)
        return max(min(v + offset, I32_MAX), I32_MIN)

    def canonical(self):
        return not self.out and self.state == LOW and self.at == len(self.words)


def bucket_of(sigma, table):
    """`GaussianConditional.build_indexes`, in the table's own single precision."""
    sigma = max(sigma, table[0])
    return len(table) - 1 - sum(1 for s in table[:-1] if sigma <= s)


class Fixture:
    def __init__(self, directory):
        self.dir = directory
        self.meta = json.loads((directory / "fixture.json").read_text())
        self.raw = (directory / "test.dcvc.wai").read_bytes()
        if self.raw[:4] != b"WAI1":
            raise ValueError("not a WAI1 envelope")
        (mlen,) = struct.unpack_from("<I", self.raw, 4)
        self.manifest = json.loads(self.raw[8:8 + mlen])
        (plen,) = struct.unpack_from("<I", self.raw, 8 + mlen)
        p = self.raw[12 + mlen:12 + mlen + plen]
        if len(p) != plen or 12 + mlen + plen != len(self.raw):
            raise ValueError("the envelope's lengths do not fill it")
        self.h, self.w, self.n, _ = struct.unpack_from("<IIII", p, 0)
        self.cy, self.cz = struct.unpack_from("<HH", p, 16)
        self.sy = 1 << p[20]
        self.q = struct.unpack_from("<H", p, 22)[0]
        self.elements = self.cy * self.sy * self.sy
        at = 26 + 8 * self.n
        self.y = []
        for i in range(self.n):
            lz, ly = struct.unpack_from("<II", p, 26 + 8 * i)
            self.y.append(p[at + lz:at + lz + ly])
            at += lz + ly
        if at != plen:
            raise ValueError("the frame table does not fill the payload")
        g = json.loads((directory / "set" / f"gauss_cdfs_q{self.q}.json").read_text())
        self.table = [struct.unpack("<f", struct.pack("<f", s))[0] for s in g["scale_table"]]
        self.cdf, self.len, self.off = g["cdf"], g["cdf_length"], g["offset"]
        self.enc_buckets = lzma.decompress((directory / "encoder-buckets.bin.xz").read_bytes())
        self.cache = {}

    def frame_buckets(self, blob, i):
        return blob[i * self.elements:(i + 1) * self.elements]

    def decode(self, i, buckets):
        key = (i, sha(buckets))
        if key not in self.cache:
            s = Stream(self.y[i])
            out = [s.symbol(self.cdf[b], self.len[b], self.off[b]) for b in buckets]
            self.cache[key] = (out, s.canonical())
        return self.cache[key]


def digest_symbols(syms):
    return sha(struct.pack(f"<{len(syms)}i", *syms))


def span(rs, samples):
    worst = max((r["rgb_sse"] for r in rs), default=0)
    return {
        "frames": len(rs),
        "frames_changed": sum(1 for r in rs if r["rgb_changed"] > 0),
        "rgb_max": max((r["rgb_max"] for r in rs), default=0),
        "rgb_changed_max": max((r["rgb_changed"] for r in rs), default=0),
        "rgb_sse_max": worst,
        "psnr_min_db": None if worst == 0 else 10 * math.log10(65025.0 * samples / worst),
        "feature_max": max((r["feature_max"] for r in rs), default=0.0),
    }


def summary_of(readings, samples):
    events = [r["frame"] for r in readings if r["desync"]]
    first = events[0] if events else None
    return {
        "frames_exact": sum(1 for r in readings if r["rgb_changed"] == 0),
        "desync_events": len(events),
        "first_desync_frame": first,
        "frames_scale_changed": sum(1 for r in readings if r["scale_changed"] > 0),
        "bucket_changes": sum(r["bucket_changed"] for r in readings),
        "symbol_changes": sum(r["symbol_changed"] for r in readings),
        "noncanonical_ends": sum(1 for r in readings if r["canonical_end"] is False),
        "before_first_desync": span([r for r in readings if first is None or r["frame"] < first], samples),
        "from_first_desync": span([r for r in readings if first is not None and r["frame"] >= first], samples),
    }


def near(a, b):
    if a is None or b is None:
        return a is b
    return a == b or abs(a - b) <= 1e-9 * max(abs(a), abs(b))


def same(a, b):
    if isinstance(a, dict) and isinstance(b, dict):
        return a.keys() == b.keys() and all(same(a[k], b[k]) for k in a)
    if isinstance(a, float) or isinstance(b, float):
        return near(a, b)
    return a == b


def check_reading(name, r, fx, base, bad, forced=False):
    i = r["frame"]
    where = f"{name} frame {i}"
    ours = fx.frame_buckets(fx.enc_buckets, i)
    theirs = bytearray(ours)
    for element, bucket in r["bucket_changes"]:
        theirs[element] = bucket
    theirs = bytes(theirs)
    changed = [k for k in range(fx.elements) if theirs[k] != ours[k]]
    if sha(theirs) != r["buckets_sha256"]:
        bad(f"{where}: buckets_sha256 is not the digest of its buckets")
    if r["bucket_changed"] != len(changed):
        bad(f"{where}: bucket_changed is {r['bucket_changed']}, its buckets differ from the encoder's in {len(changed)}")
    first = r["first_bucket_change"]
    if (first is None) != (not changed):
        bad(f"{where}: first_bucket_change {first} against {len(changed)} changed buckets")
    elif changed:
        k = changed[0]
        if first["element"] != k or first["encoder_bucket"] != ours[k] or first["decoder_bucket"] != theirs[k]:
            bad(f"{where}: first_bucket_change is {first}, the buckets give element {k}: {ours[k]} to {theirs[k]}")
        if first["edge"] != fx.table[min(ours[k], theirs[k])] or first["edges_crossed"] != abs(ours[k] - theirs[k]):
            bad(f"{where}: first_bucket_change's edge is not the table's")
        if bucket_of(first["scale"], fx.table) != (ours[k] if forced else theirs[k]):
            bad(f"{where}: scale {first['scale']} selects bucket {bucket_of(first['scale'], fx.table)}, not {theirs[k]}")
    syms, canonical = fx.decode(i, theirs)
    want, _ = fx.decode(i, ours)
    wrong = [k for k in range(fx.elements) if syms[k] != want[k]]
    if digest_symbols(syms) != r["symbols_sha256"]:
        bad(f"{where}: symbols_sha256 is not the digest of the stream decoded under its buckets")
    if r["symbol_changed"] != len(wrong):
        bad(f"{where}: symbol_changed is {r['symbol_changed']}, the stream gives {len(wrong)}")
    if r["first_symbol_changed"] != (wrong[0] if wrong else None):
        bad(f"{where}: first_symbol_changed is {r['first_symbol_changed']}, the stream gives {wrong[0] if wrong else None}")
    if r["canonical_end"] != canonical:
        bad(f"{where}: canonical_end is {r['canonical_end']}, the stream gives {canonical}")
    if r["desync"] != bool(wrong):
        bad(f"{where}: desync is {r['desync']}, the stream gives {bool(wrong)}")
    if (r["symbol_changed"] == 0) != (r["latent_max"] < LATENT_STEP_EDGE):
        bad(f"{where}: latent_max is {r['latent_max']} with {r['symbol_changed']} symbols changed, and a wrong symbol is a whole step")

    hist = {int(k): v for k, v in r["rgb_delta_hist"].items()}
    if any(k <= 0 or k > 255 or v <= 0 for k, v in hist.items()):
        bad(f"{where}: a histogram bin outside 1 to 255, or empty")
    if r["rgb_changed"] != sum(hist.values()):
        bad(f"{where}: rgb_changed is {r['rgb_changed']}, its histogram gives {sum(hist.values())}")
    if r["rgb_max"] != max(hist, default=0):
        bad(f"{where}: rgb_max is {r['rgb_max']}, its histogram gives {max(hist, default=0)}")
    if r["rgb_sse"] != sum(k * k * v for k, v in hist.items()):
        bad(f"{where}: rgb_sse is {r['rgb_sse']}, its histogram gives {sum(k * k * v for k, v in hist.items())}")
    if (r["rgb_sha256"] == base["rgb_sha256"]) != (r["rgb_changed"] == 0):
        bad(f"{where}: rgb_sha256 and the histogram disagree on whether the frame is the baseline's")
    for field in ("feature", "latent"):
        if (r[f"{field}_sha256"] == base[f"{field}_sha256"]) != (r[f"{field}_changed"] == 0):
            bad(f"{where}: {field}_sha256 and {field}_changed disagree")
        if (r[f"{field}_changed"] == 0) != (r[f"{field}_max"] == 0):
            bad(f"{where}: {field}_changed and {field}_max disagree")
    if (r["scale_changed"] == 0) != (r["scale_max"] == 0):
        bad(f"{where}: scale_changed and scale_max disagree")
    if r["scale_changed"] == 0 and r["buckets_sha256"] != base["buckets_sha256"] and not forced:
        bad(f"{where}: its buckets are not the baseline's, with no scale changed")
    if r["desync"] and r["bucket_changed"] == 0:
        bad(f"{where}: a desync with every bucket the encoder's")


def check_events(name, doc, fx, bad):
    """Runs with one bucket forced, each repeating one event a fabric read, alone: nothing but
    that bucket and what follows from it may differ from the run it is read against."""
    ev = doc.get("events")
    if ev is None:
        return
    if ev["format"] != "wai.research.dcvc-drift.events/1":
        bad(f"{name}: not a dcvc-drift events file")
    plain = next((f for f in doc["fabrics"] if f["fabric"] == ev["against"]["fabric"]), None)
    if plain is None:
        bad(f"{name}: events are read against {ev['against']['fabric']}, which is no fabric of the set")
        return
    if not plain["sink_route"] or any(ev["against"][k] != plain[k] for k in ("arch", "os", "provider", "threads")):
        bad(f"{name}: events are not read against a sink's route as the set read it")
    for run in ev["runs"]:
        fab = f"{name} {run['fabric']}"
        if len(run["forced"]) != 1 or len(run["readings"]) != fx.n:
            bad(f"{fab}: {len(run['forced'])} buckets forced and {len(run['readings'])} readings, not one and {fx.n}")
            continue
        frame, element, bucket = run["forced"][0]
        if not any(f["readings"][frame]["bucket_changes"] == [[element, bucket]] for f in doc["fabrics"]):
            bad(f"{fab}: no fabric read this bucket change at frame {frame}")
        if run["repeat_equal"] is not True:
            bad(f"{fab}: its two passes differ")
        for r, b in zip(run["readings"], plain["readings"]):
            check_reading(fab, r, fx, b, bad, forced=True)
            where = f"{fab} frame {r['frame']}"
            at = r["frame"] == frame
            if r["bucket_changes"] != ([[element, bucket]] if at else []):
                bad(f"{where}: bucket_changes are {r['bucket_changes']}, and only the forced bucket moves")
            if r["scale_changed"] or (r["latent_changed"] and not at):
                bad(f"{where}: a scale or a latent moved beyond the forced frame's entropy decode")
            if r["frame"] < frame and (r["rgb_changed"] or r["feature_changed"]):
                bad(f"{where}: a frame before the forced one moved")
        want = summary_of(run["readings"], doc["samples_per_frame"])
        if not same(run["summary"], want):
            bad(f"{fab}: summary is {run['summary']}, the readings give {want}")


def verdicts_of(fabrics):
    sink = [f for f in fabrics if f["sink_route"]]

    def holds(group, key):
        return all(f["summary"][key] == 0 for f in group)

    exact = lambda group: all(f["summary"]["frames_exact"] == len(f["readings"]) for f in group)

    def frames_of(f):
        return [(r["rgb_sha256"], r["feature_sha256"]) for r in f["readings"]]

    equal = [[a["fabric"], b["fabric"]] for i, a in enumerate(fabrics) for b in fabrics[i + 1:] if frames_of(a) == frames_of(b)]
    return {
        "sink_route_fabrics": [f["fabric"] for f in sink],
        "decode_equivalence_sink_route": exact(sink),
        "entropy_consistency_sink_route": holds(sink, "desync_events"),
        "decode_equivalence_all": exact(fabrics),
        "entropy_consistency_all": holds(fabrics, "desync_events"),
        "repeats_on_one_fabric": all(f["repeat_equal"] for f in fabrics),
        "fabrics_with_equal_frames": equal,
    }


def check_set(name, doc, fx, bad):
    if doc["format"] != "wai.research.dcvc-drift/1" or doc["capability"] != "wai.neural.dcvc_rt":
        bad(f"{name}: not a dcvc-drift record")
    if fx.manifest["model_requirement"]["capability"] != doc["capability"]:
        bad(f"{name}: the envelope is not a {doc['capability']} envelope")
    if doc["frames"] != fx.n or doc["elements_per_frame"] != fx.elements or doc["samples_per_frame"] != fx.h * fx.w * 3:
        bad(f"{name}: frame counts disagree with the envelope")
    if len(doc["encoder"]) != fx.n:
        bad(f"{name}: {len(doc['encoder'])} encoder readings for {fx.n} frames")
    for i, e in enumerate(doc["encoder"]):
        syms, canonical = fx.decode(i, fx.frame_buckets(fx.enc_buckets, i))
        if not canonical:
            bad(f"{name} frame {i}: the encoder's buckets do not end the y stream canonically")
        if digest_symbols(syms) != e["symbols_sha256"]:
            bad(f"{name} frame {i}: symbols_sha256 is not the digest of the encoder's symbols")
        if fx.meta["frames"][i]["rgb_sha256"] != e["rgb_sha256"]:
            bad(f"{name} frame {i}: the reference digest is not the fixture's")
        # The baseline holds the encoder's float image: rounded the encoder's way it is the
        # reference byte for byte, and only ties can separate its frames from it.
        if e["baseline_rgb_even_sha256"] != e["rgb_sha256"]:
            bad(f"{name} frame {i}: the baseline, rounded half to even, is not the encoder's reconstruction")
    fabrics = doc["fabrics"]
    if not fabrics:
        bad(f"{name}: no fabric")
        return
    base = fabrics[0]
    if not base["sink_route"]:
        bad(f"{name}: the baseline is not the sink's route")
    for r, e in zip(base["readings"], doc["encoder"]):
        if (r["rgb_sha256"] == e["rgb_sha256"]) != (e["baseline_tie_samples"] == 0):
            bad(f"{name} frame {r['frame']}: baseline_tie_samples disagrees with the frame digests")
    for f in fabrics:
        fab = f"{name} {f['fabric']}"
        if len(f["readings"]) != fx.n or [r["frame"] for r in f["readings"]] != list(range(fx.n)):
            bad(f"{fab}: its readings are not one a frame")
            continue
        if f["sink_route"] != (f["provider"] == "cpu" and f["threads"] == 1):
            bad(f"{fab}: sink_route does not follow from its provider and threads")
        if f["sink_route"] and f["sink_frames_equal"] is not True:
            bad(f"{fab}: the traced frames were not held to the sink's")
        for r, b in zip(f["readings"], base["readings"]):
            check_reading(fab, r, fx, b, bad)
        want = summary_of(f["readings"], doc["samples_per_frame"])
        if not same(f["summary"], want):
            for k in want:
                if not same(f["summary"].get(k), want[k]):
                    bad(f"{fab}: summary {k} is {f['summary'].get(k)}, the readings give {want[k]}")


def check_verdicts(name, doc, fx, stated, entry, bad):
    want = verdicts_of(doc["fabrics"])
    if not same(stated, want):
        bad(f"{name}: verdicts are {stated}, the readings give {want}")
    ties = [e["baseline_tie_samples"] for e in doc["encoder"]]
    derived = {"frames": doc["frames"], "ref_gain": fx.meta["ref_gain"], "image_gain": fx.meta["image_gain"],
               "tie_samples": sum(ties), "tie_frames": sum(1 for t in ties if t),
               "feature_rms_last": fx.meta["frames"][-1]["feature_rms"]}
    if not same(entry, derived):
        bad(f"{name}: record.json's set entry is {entry}, the set gives {derived}")


MARK = re.compile(r"([-+]?[0-9][0-9,]*(?:\.[0-9]+)?(?:[eE][-+]?[0-9]+)?)(?:\s*(?:%|dB|frames|bytes))?<!--dcvc:([^>]+)-->")


def value_at(record, docs, fixtures, path):
    """`<set>.fabric:<name>.<member>...`, `<set>.events.frame:<n>.<member>...` for the run that
    forced frame n, `fixture.<set>.<member>...` for a fixture.json, or a member path of
    record.json."""
    parts = path.split(".")
    if parts[0] in docs:
        node = docs[parts[0]]
        parts = parts[1:]
    elif parts[0] == "fixture":
        node = fixtures[parts[1]].meta
        parts = parts[2:]
    else:
        node = record
    for p in parts:
        if p.startswith("fabric:"):
            node = next(f for f in node["fabrics"] if f["fabric"] == p[7:])
        elif p.startswith("frame:"):
            node = next(run for run in node["runs"] if run["forced"][0][0] == int(p[6:]))
        elif isinstance(node, list):
            node = node[int(p)]
        else:
            node = node[p]
    return node


def check_text(root, record, docs, fixtures, name, text, bad):
    for m in re.finditer(r"`([0-9a-f]{64})`<!--dcvc-file:([^>]+)-->", text):
        if sha((root / m.group(2)).read_bytes()) != m.group(1):
            bad(f"{name}: {m.group(2)} is not the file whose digest the text states")
    count = 0
    for m in MARK.finditer(text):
        count += 1
        printed, path = m.group(1), m.group(2)
        try:
            want = value_at(record, docs, fixtures, path)
        except (KeyError, IndexError, StopIteration, ValueError) as e:
            bad(f"{name}: marker {path}: no such place ({e!r})")
            continue
        got = float(printed.replace(",", ""))
        if "e" in printed.lower():
            mant = printed.lower().split("e")[0]
            places = len(mant.split(".")[1]) if "." in mant else 0
            if float(f"{float(want):.{places}e}") != got:
                bad(f"{name}: {printed} at {path}: the record gives {want}")
        else:
            places = len(printed.split(".")[1]) if "." in printed else 0
            if round(float(want), places) != got:
                bad(f"{name}: {printed} at {path}: the record gives {want}")
    if text.count("<!--dcvc:") != count:
        bad(f"{name}: {text.count('<!--dcvc:')} markers, {count} follow a figure")


def verify(root, record, docs, fixtures, texts):
    findings = []
    bad = findings.append
    for name, doc in docs.items():
        check_set(name, doc, fixtures[name], bad)
        check_events(name, doc, fixtures[name], bad)
        check_verdicts(name, doc, fixtures[name], record["verdicts"][name], record["sets"][name], bad)
    for name, text in texts.items():
        check_text(root, record, docs, fixtures, name, text, bad)
    return findings


def check_files(root, record, fixtures, bad, read=None):
    read = read or (lambda p: p.read_bytes())
    for rel, digest in record["files"].items():
        p = root / rel
        if not p.exists():
            bad(f"{rel}: named by record.json and absent")
        elif sha(read(p)) != digest:
            bad(f"{rel}: not the file record.json names")
    for name, fx in fixtures.items():
        for rel, digest in fx.meta["files"].items():
            if rel in ("ref.rgb",):
                continue
            if rel == "encoder-buckets.bin":
                if sha(fx.enc_buckets) != digest:
                    bad(f"{name}: encoder-buckets.bin.xz is not the buckets fixture.json names")
                continue
            p = fx.dir / rel
            if not p.exists():
                bad(f"{name}: {rel} is named by fixture.json and absent")
            elif sha(p.read_bytes()) != digest:
                bad(f"{name}: {rel} is not the file fixture.json names")


def selftest(root, record, docs, fixtures):
    """Each change must be refused, by the readings and not by a page that quotes them, and for
    the reason it is planted for."""
    fails = []
    set_name = next(iter(docs))

    def expect(what, change, target=None, says=""):
        r = copy.deepcopy(record)
        d = copy.deepcopy(docs)
        change(r, d[target or set_name])
        found = verify(root, r, d, fixtures, {})
        if not any(says in f for f in found):
            fails.append(f"selftest: {what} was not refused" + (f" ({says})" if found else ""))

    def reading(d, fabric=0, frame=0):
        return d["fabrics"][fabric]["readings"][frame]

    last = len(docs[set_name]["fabrics"]) - 1
    expect("a changed sample delta", lambda r, d: reading(d, last).__setitem__("rgb_max", reading(d, last)["rgb_max"] + 1))
    expect("a changed frame digest", lambda r, d: reading(d, last).__setitem__("symbols_sha256", "0" * 64))
    expect("a changed histogram", lambda r, d: reading(d, last)["rgb_delta_hist"].__setitem__("9", 3))
    expect("a feature called unchanged", lambda r, d: reading(d, last, 1).__setitem__("feature_changed", 0))
    expect("a frame called a desync with no wrong symbol",
           lambda r, d: reading(d, 0).__setitem__("desync", True))
    expect("a summary with one desync too many",
           lambda r, d: d["fabrics"][last]["summary"].__setitem__("desync_events", d["fabrics"][last]["summary"]["desync_events"] + 1))
    expect("a summary with one desync too few",
           lambda r, d: d["fabrics"][last]["summary"].__setitem__("desync_events", d["fabrics"][last]["summary"]["desync_events"] - 1))
    expect("a bucket count changed", lambda r, d: reading(d, last).__setitem__("bucket_changed", reading(d, last)["bucket_changed"] + 1))
    expect("a bucket changed", lambda r, d: reading(d, last)["bucket_changes"].append([0, 1]))
    expect("a tie count changed", lambda r, d: r["sets"][set_name].__setitem__("tie_samples", r["sets"][set_name]["tie_samples"] + 1))
    expect("a verdict the readings do not give",
           lambda r, d: r["verdicts"][set_name].__setitem__("repeats_on_one_fabric", not r["verdicts"][set_name]["repeats_on_one_fabric"]))

    # A reading the record cannot recompute, a feature's largest difference, say, is held by
    # the digest of its file: one changed byte is a finding.
    def edited(p):
        raw = p.read_bytes()
        return raw.replace(b"feature_max", b"feature_mix", 1) if p.name == "readings.json" else raw

    found = []
    check_files(root, record, fixtures, found.append, edited)
    if not found:
        fails.append("selftest: a changed byte of a readings file was not refused")

    # A real event hidden: every reading of one, and its summary, written as a clean frame.
    for name, doc in docs.items():
        for fi, f in enumerate(doc["fabrics"]):
            for ri, r0 in enumerate(f["readings"]):
                if r0["desync"]:
                    def hide(r, d, fi=fi, ri=ri):
                        x = d["fabrics"][fi]["readings"][ri]
                        x["desync"] = False
                        x["symbol_changed"] = 0
                        x["first_symbol_changed"] = None
                        d["fabrics"][fi]["summary"] = summary_of(d["fabrics"][fi]["readings"], d["samples_per_frame"])
                    expect(f"a hidden desync in {name} {f['fabric']} frame {ri}", hide, name)
                    break
            else:
                continue
            break

    # An event erased together with its bucket, on every fabric that read it: buckets, digests,
    # symbols and summaries all agree, so the stream alone has nothing to refuse. The latent
    # still moved by a whole step. It takes the event with the fewest wrong symbols.
    def erased(frame):
        def change(r, d):
            enc = d["encoder"][frame]
            encoder_buckets = fixtures["gop"].frame_buckets(fixtures["gop"].enc_buckets, frame)
            for f in d["fabrics"]:
                x = f["readings"][frame]
                if not x["desync"] or x["symbol_changed"] != 1:
                    continue
                x.update(bucket_changes=[], bucket_changed=0, first_bucket_change=None, buckets_sha256=sha(encoder_buckets),
                         symbols_sha256=enc["symbols_sha256"], symbol_changed=0, first_symbol_changed=None,
                         canonical_end=True, desync=False)
                f["summary"] = summary_of(f["readings"], d["samples_per_frame"])
        return change

    one = next((r["frame"] for f in docs["gop"]["fabrics"] for r in f["readings"]
                if r["desync"] and r["symbol_changed"] == 1), None)
    if one is None:
        fails.append("selftest: no one-symbol event in gop to erase")
    else:
        expect("an event erased with its bucket", erased(one), "gop", "latent_max")

    # Runs with a forced bucket.
    runs = docs["gop"].get("events", {}).get("runs")
    if not runs:
        fails.append("selftest: no run with a forced bucket in gop")
    else:
        frame = runs[0]["forced"][0][0]

        def run0(d):
            return d["events"]["runs"][0]

        def other_bucket(r, d):
            run0(d)["forced"][0][2] += 1

        def moved_before(r, d):
            run0(d)["readings"][frame - 1]["feature_changed"] = 1

        def moved_latent(r, d):
            run0(d)["readings"][frame + 1]["latent_changed"] = 1

        def changed_summary(r, d):
            run0(d)["summary"]["from_first_desync"]["rgb_max"] += 1

        def another_fabric(r, d):
            d["events"]["against"]["threads"] = 8

        expect("a forced bucket no fabric read", other_bucket, "gop", "no fabric read")
        expect("a forced run that moved before its frame", moved_before, "gop", "before the forced one")
        expect("a forced run whose later latent moved", moved_latent, "gop", "beyond the forced frame")
        expect("a forced run's summary changed", changed_summary, "gop", "the readings give")
        expect("events read against another fabric", another_fabric, "gop", "sink's route")
    return fails


def main():
    args = sys.argv[1:]
    want_self = "--selftest" in args
    args = [a for a in args if a != "--selftest"]
    if not args:
        raise SystemExit(__doc__)
    root = pathlib.Path(args[0])
    record = json.loads((root / "record.json").read_text())
    docs, fixtures = {}, {}
    for name in record["sets"]:
        docs[name] = json.loads((root / "readings" / name / "readings.json").read_text())
        events = root / "readings" / name / "events.json"
        if f"readings/{name}/events.json" in record["files"] and events.exists():
            docs[name]["events"] = json.loads(events.read_text())
        fixtures[name] = Fixture(root / "fixture" / name)
    texts = {a: pathlib.Path(a).read_text() for a in args[1:]}
    findings = verify(root, record, docs, fixtures, texts)
    check_files(root, record, fixtures, findings.append)
    if want_self and not findings:
        findings += selftest(root, record, docs, fixtures)
    for f in findings:
        print(f)
    if not findings:
        n = sum(len(d["fabrics"]) * d["frames"] for d in docs.values())
        print(f"dcvc-drift: {n} readings hold" + (", and every planted change is refused" if want_self else ""))
    return 1 if findings else 0


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