open-standards/wai/live-conformance/rung_price_verify.py

#!/usr/bin/env python3
"""Independent verifier of the rung price corpus (extensions/rung-price.md).

Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. It takes from the second implementations of the texts this extension builds on
what they define: documents, their canonical form and digest, and a request's quantities from
route-record's (`route_record_verify.py`, route-record §2 and §3.1), per-hop records from
live-metering's (`live_metering_verify.py`, live-metering §3.6), room descriptors from fleet's
(`fleet_verify.py`, fleet §4.4), and Ed25519 verification as SPEC §7 *Signatures* states it from
the routine every Python verifier of WAI's corpora shares (`../claim-conformance/_ed25519.py`).
Everything this extension adds is written here: §2's region code, its country and *holds* (WAI's
one definition, which other verifiers import from here as `region_code`, `country` and `holds`);
the list's members, rules, digest and signature (§4); the evidence an energy figure names, the
texts that give it and the rules that refute it (§4.6); its verdict at an instant (§5); a request,
the records and rooms the evidence texts give, its candidates and removals, cost and energy per
receiver, each candidate's evidence and energy rank, ranking, decision and break-even audience
(§6); and §7.2's pass rate with the tier that prices it (`pass_rate`, `tier`).

For every case it requires the outcome the corpus states, member for member; that every reason of
§8's table appears in the corpus and that the corpus states no other; that every limit of §6.2,
every decision of §6.5, every evidence of §6.4 and every energy rank occurs; and that no outcome
depends on the order the lists or the evidence are given in, beyond the positions of the lists'
verdicts, nor on the order of a request's members, nor on a list or a text being given twice. It
checks what no vector carries: the bounds of a text, a list, a request and an evidence text, the
ceiling the `tier` limit reads when two lower ones cover a rung, and §7.2's pass rate. Then it
mutates copies of the corpus, one defect each, and requires every copy to fail; and it drops each
of its own checks in turn and requires the corpus to catch the weakened verifier.

    python3 -I rung_price_verify.py <live-conformance/prices>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier is
caught. Python 3.8 or later and the standard library.
"""

import copy
import functools
import hashlib
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import _ed25519 as ED  # noqa: E402
import route_record_verify as RR  # noqa: E402
import live_metering_verify as LMV  # noqa: E402
import fleet_verify as FV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "rung-price.md")
KIND = "wai-rung-prices"                         # §4.1
REQUEST_KIND = "wai-rung-request"                # §6.1
DOMAIN = b"wai:rung-prices\x01"                  # §4.4
MAX = (1 << 53) - 1
MAX_TEXT = 1 << 20                               # §4.5: 1,048,576 bytes
MAX_REQUEST = 1 << 16                            # §6.1: 65,536 bytes
MAX_ENTRIES = 1024                               # §4.1
MAX_FOOTPRINT = 8192                             # §4.2
U32 = (1 << 32) - 1
DIRECTIONS = ("downlink", "uplink")              # §2
BEARERS = ("bidirectional", "multicast", "unidirectional")   # bearer-receipts §2
CLASSES = ("CalibratedInstrument", "OnChipCounter", "ModelBased", "Estimator")   # route-record §2's classes: rule 1's form
SHARED = ("multicast", "unidirectional")
# §4.3: unit -> (bearers, directions, divisor, the most a price may be).
UNITS = {
    "bytes": (("bidirectional",), ("downlink",), 10 ** 9, MAX),
    "subscriber_us": (("bidirectional",), ("downlink",), 10 ** 6, U32),
    "publisher_us": (("bidirectional",), ("uplink",), 10 ** 6, U32),
    "bytes_sent": (SHARED, DIRECTIONS, 10 ** 9, MAX),
    "pass_us": (SHARED, DIRECTIONS, 10 ** 6, MAX),
}
LIST = ["kind", "version", "seller", "currency", "issued", "expires", "entries", "sig"]                 # §4.1
ENTRY = ["direction", "bearer", "footprint", "max_bps", "unit", "divisor", "price", "energy_uj", "energy_class",
         "energy_ppm", "energy_evidence"]                                                               # §4.2
REQUEST = ["kind", "direction", "to", "rate_bps", "pass_rate_bps", "bearers", "currency", "quantities", "max_price",
           "audience", "sort", "confirmed", "evidence_age_us"]                                          # §6.1
DECISIONS = ["none", "only_survivor", "price", "energy", "seller", "entry"]                             # §6.5
MAX_READINGS = 64                                                               # §4.2
MAX_CONFIRMED = 256                                                             # §6.1
READ_TAGS = {"CalibratedInstrument": 5, "OnChipCounter": 4, "ModelBased": 2}    # §4.6 rule 3, in §6.4's rank order
HARDWARE = ("CalibratedInstrument", "OnChipCounter")                            # §4.2: the classes that declare an uncertainty
READ_UNITS = ("bytes", "subscriber_us")                                         # §4.5 rule 5
EVIDENCE = ["checked", "absent", "outside-window", "unconfirmed"]               # §6.4, besides `refuted reason=…`
CURRENCY = re.compile(r"\A[A-Z]{3}\Z")

# A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything.
DROPPED = None


def drops(name):
    return DROPPED == name


def section(heading):
    text = open(EXTENSION, encoding="utf-8").read()
    at = text.index("\n" + heading)
    end = text.find("\n#", at + 1)
    return text[at:end if end >= 0 else len(text)]


def reasons_table():
    """§8's codes, in order."""
    return [r.split("`")[1] for r in section("## 8. Reasons").splitlines() if r.startswith("| `")]


def limits_table():
    """§6.2's limits, in its table's order."""
    return [r.split("`")[1] for r in section("### 6.2 Candidates and removals").splitlines() if r.startswith("| `")]


class Refused(Exception):
    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason


# ------------------------------------------------------------------ §2: the region code

REGION = re.compile(r"\A[A-Z]{2}(?:-[A-Z0-9]{1,3})?\Z")
LENIENT = re.compile(r"\A[A-Za-z]{2,3}(?:-[A-Za-z0-9]{1,3})?\Z")


def region_code(v):
    """§2 *Region code*: an ISO 3166-1 alpha-2 code (two capitals), or an ISO 3166-2 code (one, `-`,
    then 1 to 3 capitals or digits). Whether ISO assigns it is not checked."""
    rx = LENIENT if drops("region_grammar") else REGION
    return isinstance(v, str) and rx.match(v) is not None


def country(r):
    """§2: a code's country is its first two characters."""
    return r[:2]


def holds(codes, r):
    """§2 *Holds*: `codes` lists `r`, or `r` is an ISO 3166-2 code whose country `codes` lists. Never a
    country because a subdivision of it is listed. Codes are compared byte for byte."""
    if r in codes:
        return True
    if len(r) > 2 and country(r) in codes and not drops("footprint_exact"):
        return True
    return drops("footprint_upward") and any(len(c) > 2 and country(c) == r for c in codes)


# ------------------------------------------------------------------ §4: the list

class PriceList:
    def __init__(self, doc, digest):
        self.doc, self.digest = doc, digest
        self.seller, self.currency = doc["seller"], doc["currency"]
        self.issued, self.expires, self.entries = doc["issued"], doc["expires"], doc["entries"]


def members(v, names):
    """Exactly `names`; a verifier without its `members` check admits more."""
    if drops("members"):
        return isinstance(v, dict) and set(names) <= set(v)
    return RR.exactly(v, names)


def entry_form(e):
    """§4.2, each member of its form (rule 1). The footprint's order is rule 3's."""
    if not members(e, ENTRY):
        return False
    if not (isinstance(e["direction"], str) and e["direction"] in DIRECTIONS) and not drops("direction_form"):
        return False
    if not (isinstance(e["bearer"], str) and e["bearer"] in BEARERS) and not drops("bearer_form"):
        return False
    fp = e["footprint"]
    if not isinstance(fp, list) or not all(region_code(c) for c in fp):
        return False
    if not fp and not drops("footprint_empty"):
        return False
    if len(fp) > MAX_FOOTPRINT and not drops("footprint_bound"):
        return False
    if not RR.is_int(e["max_bps"], 0 if drops("max_bps_form") else 1, MAX):
        return False
    if not (RR.is_unit(e["unit"]) or (drops("unit_form") and isinstance(e["unit"], str))):
        return False
    if not (RR.is_int(e["divisor"], 0 if drops("divisor_form") else 1) and RR.is_int(e["price"])):
        return False
    if not RR.null_or(e["energy_uj"], lambda x: RR.is_int(x, 0 if drops("energy_zero") else 1)):
        return False
    if not RR.null_or(e["energy_class"], lambda c: isinstance(c, str) and (c in CLASSES or drops("energy_class"))):
        return False
    if not RR.null_or(e["energy_ppm"], lambda p: RR.is_int(p, 0, MAX) if drops("ppm_bound") else RR.is_int(p, 1, U32)):
        return False
    if not RR.null_or(e["energy_evidence"], readings_form):
        return False
    if (e["energy_ppm"] is not None) != (e["energy_class"] in HARDWARE) and not drops("ppm_pair"):
        return False
    if (e["energy_evidence"] is not None) != (e["energy_class"] in READ_TAGS) and not drops("readings_pair"):
        return False
    return (e["energy_uj"] is None) == (e["energy_class"] is None) or drops("energy_pair")


def readings_form(v):
    """§4.2 `energy_evidence`: 1 to 64 digests, each 64 lowercase hexadecimal digits, strictly
    ascending."""
    if not isinstance(v, list):
        return False
    if (not v or len(v) > MAX_READINGS) and not drops("readings_bound"):
        return False
    if not all(RR.is_hex(d, 64) or (drops("readings_digest") and isinstance(d, str) and len(d) == 64) for d in v):
        return False
    return ascending(v) or drops("readings_order")


def list_form(v):
    """§4.1 and §4.2: rule 1."""
    if not members(v, LIST):
        return False
    if not (v["kind"] == KIND or drops("kind")):
        return False
    if not (RR.is_int(v["version"]) and (v["version"] == 1 or drops("version"))):
        return False
    if not (RR.is_hex(v["seller"], 64) or (drops("seller_form") and isinstance(v["seller"], str) and len(v["seller"]) == 64)):
        return False
    if not (isinstance(v["currency"], str) and (CURRENCY.match(v["currency"]) or drops("currency_form"))):
        return False
    if not (RR.is_int(v["issued"]) and RR.is_int(v["expires"])):
        return False
    if not (RR.is_hex(v["sig"], 128) or (drops("sig_form") and isinstance(v["sig"], str) and len(v["sig"]) == 128)):
        return False
    es = v["entries"]
    if not isinstance(es, list) or len(es) > MAX_ENTRIES or (not es and not drops("entries_empty")):
        return False
    return all(entry_form(e) for e in es)


def tier_key(e):
    return e["direction"], e["bearer"], e["unit"], e["max_bps"]


def order_key(e):
    """Rule 4's order: strings by their bytes, integers by value."""
    return (e["direction"].encode(), e["bearer"].encode(), e["unit"].encode(), e["max_bps"],
            e["footprint"][0].encode() if e["footprint"] else b"")


def ascending(xs):
    return all(xs[i].encode() < xs[i + 1].encode() for i in range(len(xs) - 1))


_VERIFIED = {}


def signature_ok(seller, body, sig):
    k = (seller, hashlib.sha256(body).digest(), sig)
    if k not in _VERIFIED:
        try:
            _VERIFIED[k] = ED.verify(bytes.fromhex(seller), DOMAIN + body, bytes.fromhex(sig)) is None
        except ValueError:
            _VERIFIED[k] = False
    return _VERIFIED[k]


def read_list(text):
    """§4.5: rules 1 to 4 (`rung_prices_malformed`), then rule 5, rule 6 and the signature. A
    PriceList, or Refused."""
    bad = Refused("rung_prices_malformed")
    if not isinstance(text, str) or len(text.encode("utf-8", "surrogatepass")) > MAX_TEXT:
        raise bad
    v = RR.read_document(text)
    # Rule 1.
    if v is None or not list_form(v):
        raise bad
    es = v["entries"]
    # Rule 2.
    if v["issued"] >= v["expires"] and not drops("window"):
        raise bad
    # Rule 3.
    for e in es:
        fp = e["footprint"]
        if not ascending(fp) and not drops("footprint_order"):
            raise bad
        listed = set(fp)
        if any(len(c) > 2 and country(c) in listed for c in fp) and not drops("footprint_redundant"):
            raise bad
    # Rule 4.
    keys = [order_key(e) for e in es]
    if any(keys[i] >= keys[i + 1] for i in range(len(keys) - 1)) and not drops("order"):
        raise bad
    seen = set()
    for e in es:
        k = tier_key(e) + (e["price"], e["energy_uj"], e["energy_class"]) \
            + (() if drops("mergeable_ppm") else (e["energy_ppm"],)) \
            + (() if drops("mergeable_readings") else
               (None if e["energy_evidence"] is None else tuple(e["energy_evidence"]),))
        if k in seen and not drops("mergeable"):
            raise bad
        seen.add(k)
    # Rule 5.
    for e in es:
        if e["unit"] not in UNITS:
            if drops("unit_known"):
                continue
            raise Refused("rung_prices_unit_invalid")
        bearers, directions, divisor, bound = UNITS[e["unit"]]
        if (e["bearer"] not in bearers or e["direction"] not in directions) and not drops("unit_basis"):
            raise Refused("rung_prices_unit_invalid")
        if e["divisor"] != divisor and not drops("fixed_divisor"):
            raise Refused("rung_prices_unit_invalid")
        if e["price"] > bound and not drops("room_price_bound"):
            raise Refused("rung_prices_unit_invalid")
        if e["energy_evidence"] is not None and e["unit"] not in READ_UNITS and not drops("readings_unit"):
            raise Refused("rung_prices_unit_invalid")
    # Rule 6.
    groups = {}
    for i, e in enumerate(es):
        groups.setdefault(tier_key(e), []).append(i)
    for idx in groups.values():
        owner = {}
        for i in idx:
            for c in es[i]["footprint"]:
                if c in owner and owner[c] != i and not drops("ambiguous"):
                    raise Refused("rung_prices_ambiguous")
                owner[c] = i
        for i in idx:
            for c in es[i]["footprint"]:
                o = owner.get(country(c)) if len(c) > 2 else None
                if o is not None and o != i and not drops("ambiguous_containment"):
                    raise Refused("rung_prices_ambiguous")
    # §4.4: the body is the list without `sig`, in its canonical form.
    body = {k: x for k, x in v.items() if k != "sig"}
    canon = RR.canonical(body).encode("utf-8")
    signed = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8") if drops("canonical") else canon
    if not drops("signature") and not signature_ok(v["seller"], signed, v["sig"]):
        raise Refused("rung_prices_signature_invalid")
    named = RR.canonical(v).encode("utf-8") if drops("body_digest") else canon
    return PriceList(v, hashlib.sha256(named).hexdigest())


def list_outcome(text):
    try:
        p = read_list(text)
    except Refused as r:
        return {"outcome": "refuse", "reason": r.reason}
    return {"outcome": "read", "digest": p.digest, "seller": p.seller, "entries": len(p.entries)}


# ------------------------------------------------------------------ §5: verdicts

def verdicts(texts, at_us):
    """§5: for each text, in order, [label, PriceList or None, digest or None]. The digest is None
    for a list refused at reading (§4.5)."""
    out = []
    for t in texts:
        try:
            p = read_list(t)
        except Refused as r:
            out.append(["reject reason=" + r.reason, None, None])
            continue
        # §5.1.
        if at_us < p.issued and not drops("not_yet"):
            out.append(["reject reason=rung_prices_not_yet", None, p.digest])
            continue
        out.append([None, p, p.digest])
    # §5.2: of each seller's lists not refused, the greatest `issued` is the latest.
    by_seller = {}
    for k, (_, p, _) in enumerate(out):
        if p is not None:
            by_seller.setdefault(p.seller, []).append(k)
    for ks in by_seller.values():
        lapsed = []
        while True:
            live = [k for k in ks if k not in lapsed]
            latest = max(out[k][1].issued for k in live)
            at_latest = [k for k in live if out[k][1].issued == latest]
            contradicted = len({out[k][2] for k in at_latest}) > 1 and not drops("contradiction")
            gone = [k for k in at_latest if out[k][1].expires <= at_us]
            # A weakened verifier lets an earlier list stand when the latest lapsed; the text
            # never revives one.
            if drops("no_revive") and gone and not contradicted and len(live) > len(at_latest):
                lapsed += gone
                continue
            break
        for k in ks:
            p = out[k][1]
            if k in lapsed:
                out[k][0] = "reject reason=rung_prices_expired"
            elif p.issued < latest and not drops("supersede"):
                out[k][0] = "superseded"
            elif p.issued == latest and contradicted:
                out[k][0] = "set-aside reason=rung_prices_contradicted"
            elif p.expires <= at_us and not drops("expired"):
                out[k][0] = "reject reason=rung_prices_expired"
            else:
                out[k][0] = "stands"
    return out


# ------------------------------------------------------------------ §6: comparing

def max_price_ok(v):
    """route-record §3.1's `max_price`: 0 to 16 objects of exactly `unit`, `price` and `divisor` (at
    least 1), in strictly ascending order of `unit`."""
    return (isinstance(v, list) and len(v) <= 16
            and all(RR.exactly(m, ["unit", "price", "divisor"]) and RR.is_unit(m["unit"]) and RR.is_int(m["price"])
                    and RR.is_int(m["divisor"], 1) for m in v)
            and ascending([m["unit"] for m in v]))


def confirmed_ok(v):
    """§6.1 `confirmed`: 0 to 256 objects of exactly `key` and `class`, strictly ascending by
    (`key`, `class`), each by its bytes."""
    if not isinstance(v, list) or len(v) > MAX_CONFIRMED:
        return False
    if not all(RR.exactly(c, ["key", "class"]) and RR.is_hex(c["key"], 64) and isinstance(c["class"], str)
               and c["class"] in READ_TAGS for c in v):
        return False
    ks = [(c["key"].encode(), c["class"].encode()) for c in v]
    return all(ks[i] < ks[i + 1] for i in range(len(ks) - 1))


def request_form(r):
    """§6.1's table, each member of its form."""
    if not (RR.exactly(r, REQUEST) or (drops("request_members") and isinstance(r, dict) and set(REQUEST) <= set(r))):
        return False
    rate = lambda x: RR.is_int(x, 0 if drops("request_rate") else 1, MAX)          # noqa: E731
    pass_rate = lambda x: RR.is_int(x, 0 if drops("request_pass_rate") else 1, MAX)  # noqa: E731
    bs = r["bearers"]
    bearers_ok = bs is None or (isinstance(bs, list) and 1 <= len(bs) <= 3
                                and all(isinstance(b, str) and (b in BEARERS or drops("request_bearer_class")) for b in bs)
                                and (ascending(bs) or drops("request_bearers")))
    return ((r["kind"] == REQUEST_KIND or drops("request_kind"))
            and (isinstance(r["direction"], str) and r["direction"] in DIRECTIONS or drops("request_direction"))
            and (region_code(r["to"]) or (drops("request_to") and isinstance(r["to"], str)))
            and RR.null_or(r["rate_bps"], rate) and RR.null_or(r["pass_rate_bps"], pass_rate) and bearers_ok
            and isinstance(r["currency"], str) and (CURRENCY.match(r["currency"]) is not None or drops("request_currency"))
            and (RR.quantities_ok(r["quantities"]) or drops("request_quantities"))
            and (max_price_ok(r["max_price"]) or drops("request_max_price"))
            and RR.null_or(r["audience"], lambda n: RR.is_int(n, 0 if drops("request_audience") else 1))
            and (r["sort"] in ("price", "energy") or (drops("request_sort") and isinstance(r["sort"], str)))
            and (confirmed_ok(r["confirmed"]) or drops("request_confirmed"))
            and (RR.is_int(r["evidence_age_us"], 1, MAX) or drops("request_evidence_age")))


def read_request(text):
    """§6.1: the request, or None (`rung_request_invalid`)."""
    if not isinstance(text, str) or len(text.encode("utf-8", "surrogatepass")) > MAX_REQUEST:
        return None
    v = RR.read_document(text)
    return v if v is not None and request_form(v) else None


def quantity(req, unit):
    for q in req["quantities"]:
        if q["unit"] == unit:
            return q["quantity"]
    if drops("units_never_converted") and unit == "bytes_sent":
        return quantity(req, "bytes")
    return None


def basis(e):
    """§4.3, §6.3: a `bidirectional` bearer reaches each receiver; the others transmit once."""
    return "receiver" if e["bearer"] == "bidirectional" else "transmission"


def rate_of(req, e):
    """§6.2: the rung's rate for a receiver-basis entry, the pass rate for a transmission-basis one;
    `null` stands for 2^53 - 1."""
    r = req["rate_bps"] if basis(e) == "receiver" or drops("pass_rate") else req["pass_rate_bps"]
    return MAX if r is None else r


def tier(entries, direction, bearer, unit, to, rate):
    """§6.2 `tier` (and §7.2's *a pass rate above every ceiling has no price*): the position, in a
    list's `entries`, of the entry of (`direction`, `bearer`, `unit`) whose footprint holds `to` with
    the least ceiling at least `rate` (None: 2^53 - 1); None when no ceiling covers it."""
    rate = MAX if rate is None else rate
    best = None
    for i, e in enumerate(entries):
        if (e["direction"], e["bearer"], e["unit"]) == (direction, bearer, unit) and holds(e["footprint"], to) \
                and e["max_bps"] >= rate and (best is None or e["max_bps"] < entries[best]["max_bps"]):
            best = i
    return best


def pass_rate(bytes_sent, count, duration_ms):
    """§7.2: P = `bytes_sent` / `count` octets a pass, a division that is exact, and the pass rate
    ⌈8000 × P / `duration_ms`⌉ bits per second."""
    if count < 1 or duration_ms < 1 or bytes_sent % count:
        raise ValueError("not a schedule entry's figures")
    return -(-8000 * (bytes_sent // count) // duration_ms)


def figure(q, x, d, unbounded=False):
    """§6.3: ⌊q × x / d⌋, None past 2^53 - 1 or when either is None."""
    if q is None or x is None:
        return None
    v = q * x // d
    return v if v <= MAX or unbounded else None


def per_receiver(t, n):
    if t is None:
        return None
    return t // n if drops("ceiling") else -(-t // n)


def removals(req, p, e):
    """§6.2, in its order: (limit, value)."""
    out = []
    if p.currency != req["currency"] and not drops("currency"):
        out.append(("currency", p.currency))
    if req["bearers"] is not None and e["bearer"] not in req["bearers"] and not drops("bearer"):
        out.append(("bearer", e["bearer"]))
    if quantity(req, e["unit"]) is None and not drops("unit"):
        out.append(("unit", e["unit"]))
    if not holds(e["footprint"], req["to"]) and not drops("footprint"):
        out.append(("footprint", list(e["footprint"])))
    rate = rate_of(req, e)
    if e["max_bps"] < rate and not drops("rate"):
        out.append(("rate", e["max_bps"]))
    lower = [o["max_bps"] for o in p.entries
             if o is not e and (o["direction"] == e["direction"] or drops("tier_direction"))
             and o["bearer"] == e["bearer"] and o["unit"] == e["unit"]
             and (holds(o["footprint"], req["to"]) or drops("tier_holds")) and rate <= o["max_bps"] < e["max_bps"]]
    if lower and not drops("tier"):
        out.append(("tier", min(lower)))
    for m in req["max_price"]:
        if m["unit"] == e["unit"] and e["price"] * m["divisor"] > m["price"] * e["divisor"] and not drops("unit_price"):
            out.append(("unit_price", e["price"]))
    return out


_HOPS, _ROOMS = {}, {}


def hop_of(b):
    """live-metering §3.5 and §3.6: the per-hop record a text holds, of the room its own `room`
    names, or None; remembered by its bytes. The run that drops `evidence_signature` leaves its
    signature unchecked."""
    k = (b, drops("evidence_signature"))
    if k not in _HOPS:
        check = LMV.check_sig
        if drops("evidence_signature"):
            LMV.check_sig = lambda *a: None
        try:
            _HOPS[k] = LMV.read_hop(b, b[5:37])
        except LMV.Refuse:
            _HOPS[k] = None
        finally:
            LMV.check_sig = check
    return _HOPS[k]


def room_of(b):
    """fleet §4.4: the `t0` of the room descriptor a text holds, or None. The run that drops
    `evidence_room_signature` leaves its signature unchecked."""
    k = (b, drops("evidence_room_signature"))
    if k not in _ROOMS:
        if drops("evidence_room_signature"):
            FV.SKIP.add("descriptor-signature")
        try:
            _ROOMS[k] = FV.read_descriptor(b)["t0"]
        except FV.Refuse:
            _ROOMS[k] = None
        finally:
            FV.SKIP.discard("descriptor-signature")
    return _ROOMS[k]


def named(hex_texts):
    """§4.6: the texts given, by the digest each names: SHA-256 over all but the last 64 bytes of a
    text of 64 to 1,048,576 bytes."""
    out = {}
    for h in hex_texts:
        b = bytes.fromhex(h)
        if 64 <= len(b) <= MAX_TEXT:
            out.setdefault(hashlib.sha256(b[:-64]).hexdigest(), []).append(b)
    return out


def given(hex_texts):
    """§4.6: the records, and the rooms' `t0`, that the texts give, by digest: each by a text naming
    it that reads and verifies."""
    records, rooms = {}, {}
    for d, bs in named(hex_texts).items():
        hs = [hop_of(b) for b in bs]
        ok = [x for x in hs if x is not None]
        if ok and (len(ok) == len(hs) or not drops("evidence_any_copy")):
            records[d] = ok[0]
        ts = [t for t in map(room_of, bs) if t is not None]
        if ts:
            rooms[d] = ts[0]
    return records, rooms


def quantity_of(x, unit):
    """§4.6 rule 5: media bytes out, or subscriber time of layer classes 0 to 7."""
    if unit == "bytes":
        return sum(t["bytes"] for t in x["tracks"] if (t["kind"] == 1 or drops("evidence_media_only"))
                   and (t["dir"] == 2 or drops("evidence_media_out")))
    return sum(x["sub"] if drops("evidence_layer_8") else x["sub"][:8])


def breaks(x, rule, seller, cls):
    """§4.6 rules 1 to 3: whether the given record `x` breaks `rule` for an entry of `cls`."""
    if rule == "role":
        return x["role"] != 1 and not drops("evidence_role") and not (drops("evidence_role_compositor") and x["role"] == 2)
    if rule == "signer":
        return x["hop"].hex() != seller and not drops("evidence_signer")
    tag = 0 if x["class"] is None else x["class"][0]
    return tag != READ_TAGS[cls] and not (drops("evidence_class") and tag != 0) \
        and not (drops("evidence_unmetered") and tag == 0)


def rules_1_to_3(xs, seller, cls):
    """§6.4 item 1: the reason of the first of rules 1 to 3 that a given record breaks, its records
    taken in the order named (None where a record is not given); None when none breaks one."""
    rules = ("role", "signer", "class")
    if drops("evidence_rules_reversed"):
        rules = ("class", "signer", "role")
    elif drops("evidence_class_before_signer"):
        rules = ("role", "class", "signer")
    if drops("evidence_rule_major"):
        pairs = [(x, r) for r in rules for x in xs]
    else:
        pairs = [(x, r) for x in xs for r in rules]
    for x, r in pairs:
        if x is None:
            if drops("evidence_first_missing"):
                return "absent"
            continue
        if breaks(x, r, seller, cls):
            return "refuted reason=rung_prices_evidence_" + r
    return None


def rules_4_and_5(e, read):
    """§4.6 rules 4 and 5 over the records together, in order: the reason of the first that fails,
    or None."""
    def rule_4():
        # The largest declared uncertainty.
        ppms = [x["class"][1] for x, _ in read if x["class"] is not None and x["class"][0] in (4, 5)]
        if not ppms:
            want = None
        elif drops("evidence_ppm_last"):
            want = ppms[-1]
        else:
            want = (min if drops("evidence_ppm_max") else max)(ppms)
        if e["energy_ppm"] is None or drops("evidence_ppm"):
            return None
        wrong = e["energy_ppm"] < want if drops("evidence_ppm_over") and want is not None else e["energy_ppm"] != want
        return "refuted reason=rung_prices_evidence_class" if wrong else None

    def rule_5():
        # The records' rate per `divisor` units, rounded up.
        if drops("evidence_figure"):
            return None
        j = sum(x["joules"] for x, _ in read)
        q = sum(quantity_of(x, e["unit"]) for x, _ in read)
        rate = None if q == 0 else (j * e["divisor"] // q if drops("evidence_round") else -(-j * e["divisor"] // q))
        wrong = rate is None or (e["energy_uj"] < rate if drops("evidence_figure_over") else rate != e["energy_uj"])
        return "refuted reason=rung_prices_evidence_figure" if wrong else None
    for rule in ((rule_5, rule_4) if drops("evidence_figure_before_ppm") else (rule_4, rule_5)):
        r = rule()
        if r:
            return r
    return None


def window(read, req, at_us):
    """§6.4 item 4: `outside-window` when a record's instant is below `at_us` - `evidence_age_us`, or at
    or after `at_us`; None otherwise."""
    age = req["evidence_age_us"]
    held = read[:1] if drops("evidence_window_first") else read[-1:] if drops("evidence_window_last") else read
    for x, t0 in held:
        if t0 is None:
            continue                                    # only in the run that drops evidence_room_given
        i = t0 + x["end"] - 1
        low = (i > at_us - age) if drops("evidence_window_edge") else (i >= at_us - age)
        if not (low and (i < at_us or drops("evidence_window_upper"))) and not drops("evidence_window"):
            return "outside-window"
    return None


def evidence_of(p, e, g, req, at_us):
    """§4.6 and §6.4: a candidate's evidence, or None when its entry names no records."""
    names = e["energy_evidence"]
    if names is None:
        return None
    records, rooms = g
    if drops("evidence_precedence") and any(d not in records for d in names):
        return "absent"
    cls = e["energy_class"]
    xs = [records.get(d) for d in names]
    refuted = rules_1_to_3(xs, p.seller, cls)
    # The records given, each with its room's `t0`; absent when a record, or the room of one, is not.
    read, absent = [], any(x is None for x in xs)
    for x in xs:
        if x is None:
            continue
        t0 = rooms.get(x["room"].hex())
        if t0 is None and not drops("evidence_room_given"):
            absent = True
            continue
        read.append((x, t0))
    absent = absent and not drops("evidence_given")

    def unconfirmed():
        # §6.1: `confirmed` holds the seller's key with the entry's class; a weakened run matches
        # the key alone.
        held = any(c["key"] == p.seller and (c["class"] == cls or drops("evidence_confirmed_class"))
                   for c in req["confirmed"])
        return None if held or drops("evidence_confirmed") else "unconfirmed"
    steps = {
        "rules 1 to 3": lambda: refuted,
        "absent": lambda: "absent" if absent else None,
        "rules 4 and 5": lambda: rules_4_and_5(e, read),
        "window": lambda: window(read, req, at_us),
        "confirmed": unconfirmed,
    }
    # §6.4's order, or the order of a weakened run that moves one step ahead of another.
    order = ("rules 1 to 3", "absent", "rules 4 and 5", "window", "confirmed")
    if drops("evidence_window_before_figure"):
        order = ("rules 1 to 3", "absent", "window", "rules 4 and 5", "confirmed")
    elif drops("evidence_confirmed_before_window"):
        order = ("rules 1 to 3", "absent", "rules 4 and 5", "confirmed", "window")
    elif drops("evidence_window_before_absent"):
        order = ("rules 1 to 3", "window", "absent", "rules 4 and 5", "confirmed")
    for step in order:
        r = steps[step]()
        if r:
            return r
    return "checked"


def energy_rank(c):
    """§6.4: 4 when the energy is `null`; 0, 1 or 2 by class when the evidence is checked; 3
    otherwise."""
    if c["energy"] is None and not drops("energy_null_rank"):
        return 4
    if (c["evidence"] == "checked" or drops("label_decides")) and c["energy_class"] in READ_TAGS:
        return 0 if drops("class_first") else list(READ_TAGS).index(c["energy_class"])
    return 3 if c["energy_class"] is not None else 4


def nulls_last(a, b):
    if a is None and b is None:
        return 0
    if a is None or b is None:
        return 1 if a is None else -1
    return (a > b) - (a < b)


def compare_by(key, a, b):
    """-1 when `a` comes first under `key`, 1 when `b` does, 0 when they do not differ."""
    if key == "price":
        zero = (lambda x: 0 if x is None else x) if drops("null_last") else (lambda x: x)
        return nulls_last(zero(a["cost"]), zero(b["cost"]))
    if key == "energy":
        ra, rb = a["energy_rank"], b["energy_rank"]
        if ra != rb:
            return -1 if ra < rb else 1
        if ra == 4 or drops("energy_figure"):
            return 0
        if ra <= 1 and not drops("uncertainty") and not (ra == 1 and drops("uncertainty_counter")):
            # One declared standard uncertainty above each figure: a comparison, never stated.
            def raised(c):
                if c["energy"] is None:
                    return None
                if drops("uncertainty_product"):
                    return c["energy"] * (c["energy_ppm"] or 0)
                if drops("uncertainty_ppm_first"):
                    return (c["energy_ppm"] or 0, c["energy"])
                return c["energy"] * (10 ** 6 + (c["energy_ppm"] or 0))
            return nulls_last(raised(a), raised(b))
        return nulls_last(a["energy"], b["energy"])
    if key == "seller":
        sa, sb = bytes.fromhex(a["seller"]), bytes.fromhex(b["seller"])
        return (sa > sb) - (sa < sb)
    return (a["entry"] > b["entry"]) - (a["entry"] < b["entry"])


def break_even(t, c_star):
    """§6.5: the least audience at which a transmission's cost per receiver is no more than c*."""
    if t is None or c_star is None:
        return None
    if c_star == 0:
        return 1 if t == 0 or drops("break_even_zero") else None
    return max(1, -(-t // c_star))


def compare(texts, evidence, at_us, request_text):
    """§6: the refusal, or the plan of §6.6, with the texts `evidence` given as lowercase
    hexadecimal (§4.6). A list refused at reading (§4.5) is stated in `lists` with its verdict and a
    `null` digest."""
    req = read_request(request_text)
    if req is None:
        return {"outcome": "refuse", "reason": "rung_request_invalid"}
    vs = verdicts(texts, at_us)
    g = given(evidence)
    standing = {}
    for label, p, _ in vs:
        # Two latest lists with one digest are one list (§5.2).
        if label == "stands" and p.seller not in standing:
            standing[p.seller] = p
    n = req["audience"] if req["audience"] is not None else (1000 if drops("audience_default") else 1)
    cands = []
    for seller in (list(standing) if drops("candidate_order") else sorted(standing, key=bytes.fromhex)):
        p = standing[seller]
        for i, e in enumerate(p.entries):
            if e["direction"] != req["direction"] and not drops("direction"):
                continue
            q = quantity(req, e["unit"])
            t = figure(q, e["price"], e["divisor"], drops("cost_bound"))
            en = figure(q, e["energy_uj"], e["divisor"], drops("energy_bound"))
            shared = basis(e) == "transmission"
            removed = removals(req, p, e)
            # §6.6: every candidate's evidence, a removed one's too.
            ev = None if removed and drops("evidence_survivors_only") else evidence_of(p, e, g, req, at_us)
            refuted = ev is not None and ev.startswith("refuted") and not drops("refuted_kept")
            cands.append({
                "seller": seller, "entry": i, "bearer": e["bearer"], "unit": e["unit"], "max_bps": e["max_bps"],
                "basis": basis(e),
                "removed": [{"limit": lim, "value": val} for lim, val in removed],
                "cost": per_receiver(t, n) if shared else t,
                "energy": None if refuted else (per_receiver(en, n) if shared and not drops("energy_share") else en),
                "energy_class": e["energy_class"], "energy_ppm": e["energy_ppm"], "evidence": ev, "energy_rank": None,
                "transmission_cost": None, "break_even": None, "rank": None,
                "_t": t,
            })
    for c in cands:
        c["energy_rank"] = energy_rank(c)
    survivors = [c for c in cands if not c["removed"]]
    keys = [req["sort"], "energy" if req["sort"] == "price" else "price", "seller", "entry"]
    keys = [k for k in keys if not drops(k + "_key")]

    def first(a, b):
        for k in keys:
            r = compare_by(k, a, b)
            if r:
                return r, k
        return 0, None
    order = sorted(survivors, key=functools.cmp_to_key(lambda a, b: first(a, b)[0]))
    for r, c in enumerate(order):
        c["rank"] = r + 1
    pool = cands if drops("c_star_survivors") else survivors
    stated = [c["cost"] for c in pool if c["basis"] == "receiver" and c["cost"] is not None]
    c_star = min(stated) if stated else None
    for c in (cands if drops("break_even_survivors") else survivors):
        if c["basis"] == "transmission":
            c["transmission_cost"] = c["_t"]
            c["break_even"] = break_even(c["_t"], c_star)
    for c in cands:
        del c["_t"]
    if not order:
        decision = {"decided_by": "none", "runner_up": None}
    elif len(order) == 1:
        decision = {"decided_by": "only_survivor", "runner_up": None}
    else:
        decision = {"decided_by": first(order[0], order[1])[1],
                    "runner_up": {"seller": order[1]["seller"], "entry": order[1]["entry"]}}
    return {
        "outcome": "plan",
        "request": RR.digest(req),
        "at_us": at_us,
        "lists": [{"verdict": label, "digest": dg} for label, _, dg in vs],
        # §6.6: the digests of the records and rooms the evidence texts gave, ascending.
        "evidence": sorted(set(g[0]) | set(g[1])),
        "audience": n,
        "candidates": cands,
        "order": [{"seller": c["seller"], "entry": c["entry"]} for c in order],
        "decision": decision,
    }


# ------------------------------------------------------------------ the corpus

KINDS = ("lists", "verdicts", "comparisons")
# A verifier that crashes on a case has not given it its outcome.
ERRORS = (Refused, KeyError, TypeError, ValueError, IndexError, AttributeError, ArithmeticError)


def run(kind, c):
    if kind == "lists":
        return list_outcome(c["text"])
    if kind == "verdicts":
        return {"lists": [v[0] for v in verdicts(c["texts"], c["at_us"])]}
    return compare(c["texts"], c["evidence"], c["at_us"], c["request"])


def same(a, b):
    """One JSON value, member for member: `1` is not `true`."""
    return RR.canonical(a) == RR.canonical(b)


def strings(v, out):
    if isinstance(v, str):
        out.append(v)
    elif isinstance(v, list):
        for x in v:
            strings(x, out)
    elif isinstance(v, dict):
        for x in v.values():
            strings(x, out)


def members_reversed(text):
    """The same document with every object's members in the reverse order, indented."""
    def rev(v):
        if isinstance(v, dict):
            return {k: rev(v[k]) for k in reversed(list(v))}
        if isinstance(v, list):
            return [rev(x) for x in v]
        return v
    return json.dumps(rev(json.loads(text)), ensure_ascii=False, indent=1)


def check(doc, quiet=False, full=True):
    """The failures of a corpus document: each case's outcome, §8's codes, §6.2's limits and §6.5's
    decisions; and in a full run, the invariances of §6 and `tier` against every survivor, which
    are properties of this verifier rather than of the corpus, so a weakened verifier is held to
    the corpus alone."""
    fails = []

    def fail(what):
        fails.append(what)
        if not quiet:
            print("FAIL", what)
    got_of = {}
    for kind in KINDS:
        for c in doc[kind]:
            try:
                got = run(kind, c)
            except ERRORS as e:
                got = {"error": repr(e)}
            got_of[c["name"]] = got
            if not same(got, c["outcome"]):
                fails.append(c["name"])
                if not quiet:
                    print(f"FAIL {c['name']}\n  stated {json.dumps(c['outcome'])}\n  got    {json.dumps(got)}")
    # §8: every code appears, and the corpus states no other.
    stated = []
    for kind in KINDS:
        for c in doc[kind]:
            strings(c["outcome"], stated)
    codes = reasons_table()
    for code in codes:
        if not any(s == code or s.endswith("reason=" + code) for s in stated):
            fail("no case gives " + code)
    raised = set()
    for kind in KINDS:
        for c in doc[kind]:
            o = c["outcome"]
            if isinstance(o, dict) and "reason" in o:
                raised.add(o["reason"])
            labels = o.get("lists", []) if isinstance(o, dict) else []
            for x in labels:
                lab = x if isinstance(x, str) else (x or {}).get("verdict", "")
                if isinstance(lab, str) and "reason=" in lab:
                    raised.add(lab.split("reason=", 1)[1])
            for cd in (o.get("candidates", []) if isinstance(o, dict) else []):
                ev = cd.get("evidence") if isinstance(cd, dict) else None
                if isinstance(ev, str) and "reason=" in ev:
                    raised.add(ev.split("reason=", 1)[1])
    for r in sorted(raised - set(codes)):
        fail(f"§8 does not list {r}, which the corpus states")
    # §6.2 and §6.5: every limit and every decision occurs.
    plans = [c["outcome"] for c in doc["comparisons"] if c["outcome"].get("outcome") == "plan"]
    limits = {x["limit"] for o in plans for cd in o["candidates"] for x in cd["removed"]}
    for lim in limits_table():
        if lim not in limits:
            fail("no candidate is removed by " + lim)
    decided = {o["decision"]["decided_by"] for o in plans}
    for d in DECISIONS:
        if d not in decided:
            fail("no plan is decided by " + d)
    # §6.4: every evidence and every energy rank occurs.
    seen = {cd.get("evidence") for o in plans for cd in o["candidates"] if cd.get("evidence")}
    for ev in EVIDENCE:
        if ev not in seen:
            fail("no candidate's evidence is " + ev)
    ranks = {cd.get("energy_rank") for o in plans for cd in o["candidates"]}
    for r in range(5):
        if r not in ranks:
            fail(f"no candidate has energy rank {r}")
    if full:
        invariances(doc, got_of, fail)
    return fails


def invariances(doc, got_of, fail):
    """§6: the order the lists and the evidence are given in decides nothing but the positions of
    the lists' verdicts, and neither the order of a request's members nor a list or a text given
    twice (§5.2, §4.6) decides anything; and each survivor is the entry `tier` names for its rung."""
    for c in doc["verdicts"]:
        got = got_of[c["name"]]
        try:
            back = run("verdicts", dict(c, texts=list(reversed(c["texts"]))))
        except ERRORS as e:
            back = {"error": repr(e)}
        if "lists" in got and not same(back, {"lists": list(reversed(got["lists"]))}):
            fail(f"{c['name']}: the verdicts depend on the order of the lists")
    for c in doc["comparisons"]:
        got = got_of[c["name"]]
        try:
            back = run("comparisons", dict(c, texts=list(reversed(c["texts"])), evidence=list(reversed(c["evidence"]))))
            moved = run("comparisons", dict(c, request=members_reversed(c["request"]))) \
                if RR.read_document(c["request"]) is not None else got
        except ERRORS as e:
            back = moved = {"error": repr(e)}
        if "lists" in got:
            back = dict(back, lists=list(reversed(back.get("lists", []))))
        if not same(back, got):
            fail(f"{c['name']}: the plan depends on the order of the lists")
        if not same(moved, got):
            fail(f"{c['name']}: the plan depends on the order of the request's members")
        # §5.2: two latest lists with one digest are one list, so each list given twice offers
        # its entries once; and a text given twice gives what it gives once (§4.6).
        try:
            twice = run("comparisons", dict(c, texts=c["texts"] + c["texts"], evidence=c["evidence"] + c["evidence"]))
        except ERRORS as e:
            twice = {"error": repr(e)}
        if "lists" in got:
            twice = dict(twice, lists=twice.get("lists", [])[:len(c["texts"])])
        if not same(twice, got):
            fail(f"{c['name']}: the plan changes when each list is given twice")
        if got.get("outcome") != "plan":
            continue
        try:
            req = RR.read_document(c["request"])
            by_digest = {}
            for t in c["texts"]:
                try:
                    p = read_list(t)
                    by_digest[p.digest] = p
                except Refused:
                    pass
            standing = {by_digest[x["digest"]].seller: by_digest[x["digest"]] for x in got["lists"] if x["verdict"] == "stands"}
            for cd in got["candidates"]:
                if cd["rank"] is None:
                    continue
                es = standing[cd["seller"]].entries
                e = es[cd["entry"]]
                if tier(es, e["direction"], e["bearer"], e["unit"], req["to"], rate_of(req, e)) != cd["entry"]:
                    fail(f"{c['name']}: the survivor {cd['seller']} {cd['entry']} is not the entry its tier names")
        except ERRORS as e:
            fail(f"{c['name']}: {e!r}")


def mutations(doc):
    """Copies of the corpus, one defect each, every one of which must fail."""
    def case(d, name):
        return next(c for k in KINDS for c in d[k] if c["name"] == name)

    def m_price_unsigned(d):
        c = case(d, "unicast_tiers")
        c["text"] = c["text"].replace('"price":10000000', '"price":10000001', 1)

    def m_digest(d):
        case(d, "unicast_tiers")["outcome"]["digest"] = "00" * 32

    def m_twin_digest(d):
        c = case(d, "same_body_two_signatures")
        c["outcome"]["digest"] = hashlib.sha256(RR.canonical(json.loads(c["text"])).encode()).hexdigest()

    def m_entries(d):
        case(d, "broadcast_capacity")["outcome"]["entries"] -= 1

    def m_instant(d):
        case(d, "last_instant")["at_us"] += 1

    def m_revive(d):
        case(d, "expired_latest_revives_nothing")["outcome"]["lists"][0] = "stands"

    def m_texts_order(d):
        c = case(d, "later_list_stands")
        c["texts"] = list(reversed(c["texts"]))

    def m_audience(d):
        c = case(d, "audience_below_break_even")
        c["request"] = c["request"].replace('"audience":4999', '"audience":5000', 1)

    def m_rate(d):
        c = case(d, "rate_at_ceiling")
        c["request"] = c["request"].replace('"rate_bps":8000000', '"rate_bps":8000001', 1)

    def m_pass_rate(d):
        c = case(d, "pass_rate_is_not_rate")
        c["request"] = c["request"].replace('"pass_rate_bps":9000000', '"pass_rate_bps":3000000', 1)

    def m_containment(d):
        c = case(d, "subdivision_in_its_country")
        s = next(x for x in c["outcome"]["candidates"] if not x["removed"])
        s["removed"] = [{"limit": "footprint", "value": ["XA", "XB"]}]

    def m_tier_direction(d):
        c = case(d, "tier_other_direction_ignored")
        s = next(x for x in c["outcome"]["candidates"] if not x["removed"])
        s["removed"] = [{"limit": "tier", "value": 1000000}]

    def m_break_even(d):
        c = case(d, "unknown_audience")
        next(x for x in c["outcome"]["candidates"] if x["break_even"] is not None)["break_even"] -= 1

    def m_removal_value(d):
        c = case(d, "unknown_audience")
        next(r for x in c["outcome"]["candidates"] for r in x["removed"] if r["limit"] == "tier")["value"] += 1

    def m_decided_by(d):
        case(d, "audience_at_break_even")["outcome"]["decision"]["decided_by"] = "price"

    def m_request_digest(d):
        case(d, "unknown_audience")["outcome"]["request"] = "00" * 32

    def m_unit_basis(d):
        case(d, "bytes_sent_two_way")["outcome"] = {"outcome": "read", "digest": "00" * 32, "seller": "00" * 32, "entries": 1}

    def m_reason_gone(d):
        d["lists"].remove(case(d, "footprints_overlap"))
        d["lists"].remove(case(d, "country_and_subdivision_across_entries"))

    def m_decision_gone(d):
        d["comparisons"].remove(case(d, "entry_breaks_tie"))

    def evidence_text(c, magic):
        return next(i for i, t in enumerate(c["evidence"]) if t.startswith(magic))

    def flip(t, at):
        b = bytearray(bytes.fromhex(t))
        b[at] ^= 1
        return b.hex()

    def m_record_body(d):
        c = case(d, "checked_class_before_figure")
        i = evidence_text(c, "574d4831")
        c["evidence"][i] = flip(c["evidence"][i], 100)

    def m_record_sig(d):
        c = case(d, "checked_class_before_figure")
        i = evidence_text(c, "574d4831")
        c["evidence"][i] = flip(c["evidence"][i], len(c["evidence"][i]) // 2 - 1)

    def m_room_gone(d):
        c = case(d, "checked_class_before_figure")
        del c["evidence"][evidence_text(c, "57524431")]

    def m_verdict(d):
        c = case(d, "label_on_borrowed_readings")
        next(x for x in c["outcome"]["candidates"] if x["evidence"] and x["evidence"].startswith("refuted"))["evidence"] = "absent"

    def m_rank(d):
        c = case(d, "checked_class_before_figure")
        next(x for x in c["outcome"]["candidates"] if x["energy_rank"] == 0)["energy_rank"] = 2

    def m_damaged_gives(d):
        # A plan that states the record a damaged copy names as given.
        c = case(d, "record_signature_fails")
        named_by = [hashlib.sha256(bytes.fromhex(t)[:-64]).hexdigest() for t in c["evidence"]]
        c["outcome"]["evidence"] = sorted(set(named_by))

    return [m_price_unsigned, m_digest, m_twin_digest, m_entries, m_instant, m_revive, m_texts_order, m_audience, m_rate,
            m_pass_rate, m_containment, m_tier_direction, m_break_even, m_removal_value, m_decided_by, m_request_digest,
            m_unit_basis, m_reason_gone, m_decision_gone, m_record_body, m_record_sig, m_room_gone, m_verdict, m_rank,
            m_damaged_gives]


WEAKENINGS = [
    # §2 and the list's form (§4.1, §4.2, rules 1 to 4).
    "region_grammar", "members", "kind", "version", "seller_form", "currency_form", "sig_form", "entries_empty",
    "direction_form", "bearer_form", "footprint_empty", "footprint_bound", "max_bps_form", "unit_form", "divisor_form",
    "energy_zero", "energy_class", "energy_pair", "window", "footprint_order", "footprint_redundant", "order", "mergeable",
    "ppm_pair", "ppm_bound", "readings_pair", "readings_bound", "readings_order", "readings_digest", "mergeable_readings",
    "mergeable_ppm",
    # Rules 5 and 6.
    "unit_known", "unit_basis", "fixed_divisor", "room_price_bound", "readings_unit", "ambiguous", "ambiguous_containment",
    # §4.4.
    "canonical", "signature", "body_digest",
    # §5.
    "not_yet", "expired", "supersede", "contradiction", "no_revive",
    # §6.1.
    "request_members", "request_kind", "request_direction", "request_to", "request_rate", "request_pass_rate",
    "request_bearers", "request_bearer_class", "request_currency", "request_quantities", "request_max_price",
    "request_audience", "request_sort", "request_confirmed", "request_evidence_age",
    # §6.2.
    "direction", "candidate_order", "currency", "bearer", "unit", "units_never_converted", "footprint", "footprint_exact",
    "footprint_upward", "rate", "pass_rate", "tier", "tier_direction", "tier_holds", "unit_price",
    # §6.3.
    "audience_default", "ceiling", "cost_bound", "energy_bound", "energy_share",
    # §6.4.
    "price_key", "energy_key", "seller_key", "entry_key", "class_first", "energy_null_rank", "energy_figure", "null_last",
    "label_decides", "uncertainty", "uncertainty_product", "uncertainty_ppm_first", "uncertainty_counter",
    "refuted_kept",
    # §4.6 and §6.4's evidence.
    "evidence_given", "evidence_room_given", "evidence_signature", "evidence_any_copy", "evidence_room_signature",
    "evidence_role", "evidence_role_compositor", "evidence_signer", "evidence_class", "evidence_unmetered",
    "evidence_rules_reversed", "evidence_class_before_signer", "evidence_rule_major", "evidence_first_missing",
    "evidence_ppm", "evidence_ppm_max", "evidence_ppm_last", "evidence_ppm_over", "evidence_figure", "evidence_round",
    "evidence_figure_over", "evidence_figure_before_ppm", "evidence_media_only", "evidence_media_out",
    "evidence_layer_8", "evidence_window", "evidence_window_edge", "evidence_window_upper", "evidence_window_first",
    "evidence_window_last", "evidence_confirmed", "evidence_confirmed_class", "evidence_precedence",
    "evidence_window_before_figure", "evidence_confirmed_before_window", "evidence_window_before_absent",
    "evidence_survivors_only",
    # §6.5.
    "c_star_survivors", "break_even_survivors", "break_even_zero",
]


# ------------------------------------------------------------------ what no vector carries

def direct(doc):
    """§4.5's text of 1,048,577 bytes, §4.1's list of 1,025 entries, §6.1's request of 65,537 bytes
    and §4.6's evidence text of 1,048,577 bytes, each against the largest that is admitted; the
    `tier` limit's value with two lower ceilings that cover a rung; and §7.2's pass rate."""
    fails = []
    base = next(c for c in doc["lists"] if c["name"] == "unicast_tiers")
    comp = next(c for c in doc["comparisons"] if c["name"] == "unknown_audience")
    text = base["text"]
    pad = MAX_TEXT - len(text.encode("utf-8"))
    at_bound = list_outcome(text + " " * pad)
    if not same(at_bound, base["outcome"]):
        fails.append("a list of 1,048,576 bytes, whitespace added, did not read as itself: " + json.dumps(at_bound))
    past = list_outcome(text + " " * (pad + 1))
    if past != {"outcome": "refuse", "reason": "rung_prices_malformed"}:
        fails.append("a list of 1,048,577 bytes was not refused malformed: " + json.dumps(past))

    def many(n):
        v = json.loads(text)
        e = v["entries"][0]
        codes = [f"X{chr(65 + i // 676 % 26)}-{chr(65 + i // 26 % 26)}{chr(65 + i % 26)}" for i in range(n)]
        # Rule 4: one tier, ordered by footprint, each entry at its own price.
        v["entries"] = [dict(e, footprint=[c], price=k + 1) for k, c in enumerate(codes)]
        return json.dumps(v)
    # Unsigned, so a list that is not malformed is refused for its signature and nothing else.
    if list_outcome(many(MAX_ENTRIES)) != {"outcome": "refuse", "reason": "rung_prices_signature_invalid"}:
        fails.append("a list of 1,024 entries was refused before its signature")
    if list_outcome(many(MAX_ENTRIES + 1)) != {"outcome": "refuse", "reason": "rung_prices_malformed"}:
        fails.append("a list of 1,025 entries was not refused malformed")
    req = comp["request"]
    pad = MAX_REQUEST - len(req.encode("utf-8"))
    at_bound = compare(comp["texts"], comp["evidence"], comp["at_us"], req + " " * pad)
    if not same(at_bound, comp["outcome"]):
        fails.append("a request of 65,536 bytes, whitespace added, did not give its plan")
    past = compare(comp["texts"], comp["evidence"], comp["at_us"], req + " " * (pad + 1))
    if past != {"outcome": "refuse", "reason": "rung_request_invalid"}:
        fails.append("a request of 65,537 bytes was not refused: " + json.dumps(past))
    # §6.2 `tier`'s value, which no vector distinguishes: with two lower ceilings covering a rung
    # of 1,000,000, each entry above them is removed at the least, 2,000,000, which survives.
    lists = [base["text"]]
    req = json.loads(comp["request"])
    req.update(rate_bps=1000000, quantities=[{"quantity": 500000000, "unit": "bytes"}])
    plan = compare(lists, [], comp["at_us"], json.dumps(req))
    got = [(c["max_bps"], [(x["limit"], x["value"]) for x in c["removed"]]) for c in plan.get("candidates", [])]
    want = [(2000000, []), (8000000, [("tier", 2000000)]), (MAX, [("tier", 2000000)])]
    if got != want:
        fails.append("§6.2's tier value is not the least ceiling that covers the rung: " + json.dumps(got))
    if tier(json.loads(base["text"])["entries"], "downlink", "bidirectional", "bytes", "XB", 1000000) != 0:
        fails.append("`tier` does not name the least ceiling that covers a rung of 1,000,000")
    # §7.2, which no vector carries: a pass rate rounded up, from an exact division.
    if pass_rate(3, 3, 3) != 2667 or pass_rate(1000, 2, 1000) != 4000:
        fails.append("§7.2's pass rate")
    try:
        pass_rate(1001, 2, 1000)
        fails.append("§7.2's division of bytes_sent by count is exact, and 1001 / 2 was taken")
    except ValueError:
        pass
    # §4.6: a text of 64 to 1,048,576 bytes names a digest, and no other.
    if named(["00" * (MAX_TEXT + 1)]) != {} or len(named(["00" * MAX_TEXT])) != 1 or named(["00" * 63]) != {} \
            or len(named(["00" * 64])) != 1:
        fails.append("an evidence text of 1,048,577 bytes, or of 63, named a digest, or one of 1,048,576 or 64 named none")
    return fails


def main(argv):
    global DROPPED
    if len(argv) != 2:
        print(__doc__)
        return 2
    doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8"))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    fails = check(doc)
    if extra:
        fails.append(f"files the corpus does not hold: {extra}")
    print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): {len(fails)} failure(s)")
    b = direct(doc)
    print(f"what no vector carries: {len(b)} failure(s)")
    fails += b
    caught = 0
    for m in mutations(doc):
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append(f"mutation {m.__name__} was not caught")
    print(f"{caught} of {len(mutations(doc))} mutated copies fail")
    weak = 0
    for w in WEAKENINGS:
        DROPPED = w
        try:
            if check(doc, quiet=True, full=False):
                weak += 1
            else:
                fails.append(f"a verifier without its {w} check passes the corpus")
        finally:
            DROPPED = None
    print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught")
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))