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

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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library. Each source
is verified by the second implementation of the corpus of its own extension, each written from
its text alone: a room-session receipt by live_metering_verify.py (live-metering §5), a
settlement statement by settlement_verify.py (settlement §8), a package receipt by
../pack-conformance/pack_verify.py (packaging §9), an ads room by live_ads_verify.py with
ad_presentation_verify.py (live-ads §9, ad-presentation §9), and a stream stitched by others by
stitched_presentation_verify.py (stitched-presentation §7), whose statement the ledger reports as
it is; and, under a tariff of version 2 or later, a media job by route_record_verify.py (route-record
§7, its hop receipts by the OER/2 reference verifier, its third-party records in version 6) and its
output receipts by _int_motion_receipt.py (integer-payloads §15), and in version 6 an upscale's by
upscale_verify.py (exact-upscale §7) and a tone mapping's by tonemap_verify.py (exact-tonemap §8).
Everything else is here:

- the tariff (§4): reading, rules, both signatures, its scope, its job prices per capability
  (version 2) or per capability and operation (versions 3 and 4), and in version 4 its price of
  a room's bytes out, of the work units of a package a run made, and of a package's storage by
  byte-days (§4.7), and in version 6 its prices of an endpoint and of a usage (§4.9);
- the invoice (§6): reading, rules, signature, chain, and what a chain bills once (§6.9): in
  versions 3 and 4 a job's work, its route record and hop receipts, and never its output's
  receipt; in version 4 a package's work once per run that made it, its receipt cited by each,
  and its storage for each period it is held;
- a job's outputs held to the objects its served hop states it delivered (§5.1), by their
  content hashes, computed with ../claim-conformance/_blake3.py;
- what each source contributes (§5.2), its placement in time (§5.3), and the checks across
  sources (§7.2 step 6);
- every line, citation, energy row and amount recomputed (§6.2 to §6.6), compared in §7.2 step
  7's order;
- the two reports (§8).

For every vector it requires the outcome the corpus states; of the corpus, that every reason of
§10 appears in it and that no other file is there. 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 receipt_ledger_verify.py <live-conformance/ledger>
    python3 -I receipt_ledger_verify.py --document <verification document>

Exit status 0 only if every vector verifies, every mutated copy fails and every weakened verifier
fails. Python 3.8 or later; the standard library, and the `cryptography` package that the OER/2
reference verifier checks a job's hop receipts with.
"""

import copy
import hashlib
import io
import json
import os
import re
import struct
import sys
import zlib

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
sys.path.insert(0, os.path.join(HERE, "..", "pack-conformance"))
import _ed25519 as ED  # noqa: E402
import live_rooms_verify as LRV  # noqa: E402
import live_metering_verify as LMV  # noqa: E402
import settlement_verify as SV  # noqa: E402
import live_ads_verify as LAV  # noqa: E402
import ad_presentation_verify as APV  # noqa: E402
import pack_verify as PV  # noqa: E402
import stitched_presentation_verify as STV  # noqa: E402
import route_record_verify as RRV  # noqa: E402  (it puts ../oer-conformance on the path)
import oer2_verify as OER  # noqa: E402
import _int_motion_receipt as IMR  # noqa: E402
import upscale_verify as UV  # noqa: E402
import tonemap_verify as TV  # noqa: E402
from _blake3 import blake3  # noqa: E402
from decimal import Decimal, ROUND_HALF_EVEN, localcontext, MAX_PREC, MAX_EMAX, MIN_EMIN  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "receipt-ledger.md")

TARIFF_MAGIC, INVOICE_MAGIC = b"WLT1", b"WLI1"                  # §4.1, §6.1
TARIFF_DOMAIN = b"wai:ledger-tariff\x01"                         # §4.2
ACCEPT_DOMAIN = b"wai:ledger-accept\x01"                         # §4.2
INVOICE_DOMAIN = b"wai:ledger-invoice\x01"                       # §6.7
CITES_DOMAIN = b"wai:ledger-cites\x01"                           # §6.5
ZERO = bytes(32)
U64 = 2 ** 64 - 1
ROOM, DELIVERY, PACKAGE, ADS, JOB = 1, 2, 3, 4, 5               # §6.3 source kinds (5: version 2)
SESSION, HOP, STATEMENT, ACK, PACK, AD_RECEIPT, CLAIM, POD_INPUT, STITCHED_CLAIM = 1, 2, 3, 4, 5, 6, 7, 8, 9   # §5.2
OER_RECEIPT, VIDEO_RECEIPT, ROUTE_RECORD = 14, 15, 16           # §5.2, version 2 (10 to 13 are other texts')
THIRD_PARTY, UPSCALE_RECORD, RENDITION_RECORD = 17, 18, 19       # §5.2, version 6
PRIVATE = 128                                                    # §5.2: 128 to 255, private and provisional use
KIND_NAMES = {ROOM: "room", DELIVERY: "delivery", PACKAGE: "package", ADS: "ads", JOB: "job"}
STITCHED = "stitched"                                            # §5.1: a source kind billed as ads
SCOPES = {1: "room_hops", 2: "delivery_hops", 3: "packaging", 4: "ad_decisions", 5: "jobs", 6: "unbilled_hops"}   # §6.6
CONTENT_DOMAIN = b"moq-jwp:object\x01"                         # route-record §2, jwp-receipts §2.1
DAY_US = 86400 * 10 ** 6                                         # §4.7: a byte-day's day
JOB_UNITS = ["jobs", "subpixels", "work_units", "usage"]         # §4.5; the third from version 4 (§4.7), the fourth in 6 (§4.9)
USAGE = 4                                                        # §4.9: the job unit of a third party's usage
JOB_VERSIONS = (2, 3, 4, 5, 6)                                   # §4.5 to §4.9: the versions that price jobs
WORK_VERSIONS = (3, 4, 5, 6)                                     # §4.6, §6.9: priced per operation, billed by the work
V4 = 4                                                           # §4.7: bytes out, work units, storage by byte-days
V5 = 5                                                           # §4.8: version 4, and a decode held to its frames
V6 = 6                                                           # §4.9: version 5, per endpoint, by a usage, and re-derived jobs
SPENDING = (V4, V5, V6)                                          # §4.7 to §4.9: the versions that bill what the work spends
DECODE_FRAMES = (V5, V6)                                         # §4.8, §4.9: a decode held to its frames
TOKEN = re.compile(r"\A[A-Za-z0-9._:/-]{1,128}\Z")                # route-record §2: an endpoint's form
UNIT_NAME = re.compile(r"\A[a-z][a-z0-9_]{0,63}\Z")               # route-record §2: a unit's form
UPSCALE_OP, TONEMAP_OP = "media.upscale", "media.tonemap"        # §4.9
CAPABILITY = re.compile(r"\Awai\.[a-z0-9_.]{0,124}\Z")
OPERATION = re.compile(r"\A[a-z][a-z0-9]*\.[a-z][a-z0-9]*\Z")    # §4.6, route-record §2
LABELS = {2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
TAGS = {v: k for k, v in LABELS.items()}
CLASS_NAMES = ["live", "fetch", "fill", "ancillary"]
UNITS = {                                                        # §6.3: name, divisor
    ROOM: [("publisher_us", 10 ** 6)] + [("subscriber_us_%d" % c, 10 ** 6) for c in range(9)]
    + [("composition_us", 10 ** 6), ("recorded_us", 10 ** 6), ("events_sent", 1)],
    DELIVERY: [(n, 10 ** 9) for n in CLASS_NAMES],
    PACKAGE: [("packages", 1), ("entries", 1), ("stored_bytes", 10 ** 9)],
    ADS: [("presented_impressions", 1), ("billable_impressions", 1)],
}
UNITS_4 = dict(UNITS)                                            # §4.7, §6.3: version 4's
UNITS_4[ROOM] = UNITS[ROOM] + [("bytes_out", 10 ** 9)]
UNITS_4[PACKAGE] = [None, None, ("byte_days", 10 ** 9)]           # a package source bills its storage alone


def units(version):
    return UNITS_4 if version in SPENDING else UNITS


def scopes(version):
    """§6.1: the scopes an invoice of `version` states."""
    return 6 if version in SPENDING else 5 if version in JOB_VERSIONS else 4

SKIP = set()


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


def sha(*parts):
    h = hashlib.sha256()
    for p in parts:
        h.update(p)
    return h.digest()


class R:
    """A cursor over little-endian fields; running out is `truncated`."""

    def __init__(self, b, p=0):
        self.b, self.p = b, p

    def take(self, n):
        if self.p + n > len(self.b):
            raise ValueError("truncated")
        s = self.b[self.p:self.p + n]
        self.p += n
        return s

    def u(self, n):
        return int.from_bytes(self.take(n), "little")


def head(b, magic):
    """§4.3's first steps."""
    if len(b) < 4:
        return "truncated"
    if b[:4] != magic:
        return "malformed"
    if len(b) < 5:
        return "truncated"
    if b[4] not in (1, 2, 3, 4, 5, 6):
        return "unknown_version"
    return None


def capitals(c):
    return len(c) == 3 and all(0x41 <= x <= 0x5A for x in c)


def ascending(xs):
    return all(a < b for a, b in zip(xs, xs[1:]))


def verify_sig(key, domain, body, sig):
    return None if "signatures" in SKIP else ED.verify(key, domain + body, sig)


# ---------------------------------------------------------------- §4 the tariff

def read_tariff(b):
    bad = head(b, TARIFF_MAGIC)
    if bad:
        raise Refuse(bad, ["tariff"])
    r = R(b, 5)
    try:
        t = {"seller": r.take(32), "buyer": r.take(32), "start": r.u(8), "end": r.u(8), "currency": r.take(3)}
        # §4.1: version 4 states a fourteenth room price, of bytes out; versions 1 to 3 read it as 0.
        t["room"] = [r.u(4) for _ in range(14 if b[4] in SPENDING else 13)] + ([] if b[4] in SPENDING else [0])
        t["delivery"] = [r.u(8) for _ in range(4)]
        # §4.1: version 4 states one package price, per 10^9 byte-days held; the others read as 0.
        t["package"] = [0, 0, r.u(8)] if b[4] in SPENDING else [r.u(8) for _ in range(3)]
        t["ads"] = [r.u(8) for _ in range(2)]
        t["version"] = b[4]
        t["job"] = []                                    # (capability, operation, unit, divisor, price, endpoint, usage)
        if t["version"] in JOB_VERSIONS:                                             # §4.5, §4.6
            for _ in range(r.u(2)):
                cap = r.take(r.u(1))
                op = r.take(r.u(1)) if t["version"] in WORK_VERSIONS else b""
                endpoint = r.take(r.u(1)) if t["version"] == V6 else b""
                unit = r.u(1)
                usage = r.take(r.u(1)) if t["version"] == V6 else b""
                t["job"].append((cap, op, unit, r.u(8), r.u(8), endpoint, usage))
        t["scope"] = [(r.u(1), r.take(32)) for _ in range(r.u(2))]
        body_end = r.p
        seller_sig, buyer_sig = r.take(64), r.take(64)
    except ValueError:
        raise Refuse("truncated", ["tariff"])
    if r.p != len(b):
        raise Refuse("malformed", ["tariff"])
    def job_price(j):
        try:
            cap, op, endpoint, usage = j[0].decode("utf-8"), j[1].decode("utf-8"), j[5].decode("utf-8"), j[6].decode("utf-8")
        except UnicodeDecodeError:
            return False
        named = OPERATION.match(op) is not None and len(j[1]) <= 128 if t["version"] in WORK_VERSIONS else op == ""
        units_priced = USAGE if t["version"] == V6 else 3 if t["version"] in SPENDING else 2
        ends = endpoint == "" or TOKEN.match(endpoint) is not None or "price-endpoint" in SKIP
        unit_named = (UNIT_NAME.match(usage) is not None if j[2] == USAGE else usage == "") or "price-usage" in SKIP
        return CAPABILITY.match(cap) is not None and named and ends and unit_named and 1 <= j[2] <= units_priced and j[3] >= 1
    def overlap_free():
        """§4.9: a price of any endpoint and a price of one, for the same work, would bill a job at both."""
        anys = {(j[0], j[1], j[2], j[6]) for j in t["job"] if j[5] == b""}
        return not any(j[5] != b"" and (j[0], j[1], j[2], j[6]) in anys for j in t["job"])
    top = 5 if t["version"] in JOB_VERSIONS else 4
    rules = (t["start"] < t["end"] and capitals(t["currency"]) and t["seller"] != ZERO and t["buyer"] != ZERO
             and t["seller"] != t["buyer"] and t["scope"] and ascending(t["scope"])
             and all(1 <= k <= top and i != ZERO for k, i in t["scope"])
             and len(t["job"]) <= 255 and (ascending([(j[0], j[1], j[5], j[2], j[6]) for j in t["job"]]) or "price-order" in SKIP)
             and all(job_price(j) for j in t["job"]) and (overlap_free() or "any-beside-one" in SKIP))
    if not rules and "tariff-rules" not in SKIP:
        raise Refuse("tariff_invalid", ["tariff"])
    body = b[:body_end]
    bad = verify_sig(t["seller"], TARIFF_DOMAIN, body, seller_sig)
    if bad:
        raise Refuse(bad, ["tariff", "seller"])
    bad = None if "buyer-signature" in SKIP else verify_sig(t["buyer"], ACCEPT_DOMAIN, body, buyer_sig)
    if bad:
        raise Refuse(bad, ["tariff", "buyer"])
    t["digest"] = sha(body)
    return t


def price(t, kind, unit):
    if kind == JOB:
        return t["job"][unit - 1][4]
    return {ROOM: t["room"], DELIVERY: t["delivery"], PACKAGE: t["package"], ADS: t["ads"]}[kind][unit - 1]


def divisor(t, kind, unit):
    return t["job"][unit - 1][3] if kind == JOB else units(t["version"])[kind][unit - 1][1]


# ---------------------------------------------------------------- §6 the invoice

def read_invoice(b, at):
    bad = head(b, INVOICE_MAGIC)
    if bad:
        raise Refuse(bad, at)
    r = R(b, 5)
    try:
        i = {"seller": r.take(32), "buyer": r.take(32), "tariff": r.take(32), "prev": r.take(32), "start": r.u(8),
             "end": r.u(8), "currency": r.take(3)}
        i["records"] = [(r.u(1), r.take(32)) for _ in range(r.u(4))]
        i["lines"] = []
        for _ in range(r.u(2)):
            i["lines"].append({"kind": r.u(1), "source": r.take(32), "unit": r.u(1), "quantity": r.u(8), "price": r.u(8),
                               "amount": r.u(8), "cited": r.u(4), "cites": r.take(32)})
        i["energy"] = [(r.u(1), r.u(1), r.u(4), r.u(8), r.u(4)) for _ in range(r.u(1))]
        i["version"] = b[4]
        v2 = i["version"] in JOB_VERSIONS
        n = scopes(i["version"])
        i["unmetered"] = [r.u(4) for _ in range(n)] + [0] * (6 - n)
        i["amount"] = r.u(8)
        sig = r.take(64)
    except ValueError:
        raise Refuse("truncated", at)
    if r.p != len(b):
        raise Refuse("malformed", at)
    if any(e[1] == 0 for e in i["energy"]) and "row-of-no-class" not in SKIP:
        raise Refuse("classes_summed", at)
    if any(k >= PRIVATE for k, _ in i["records"]) and "private-kind" not in SKIP:
        raise Refuse("private_kind", at)
    keys = [(x["kind"], x["source"], x["unit"]) for x in i["lines"]]
    def unit_ok(x):
        if x["kind"] == JOB:
            return v2 and x["unit"] >= 1
        u = units(i["version"])
        return x["kind"] in u and 1 <= x["unit"] <= len(u[x["kind"]]) and u[x["kind"]][x["unit"] - 1] is not None
    rules = (i["start"] < i["end"] and capitals(i["currency"]) and ascending(i["records"])
             and all(1 <= k <= 9 or (v2 and OER_RECEIPT <= k <= ROUTE_RECORD)
                     or ((i["version"] == V6 or "kind-version" in SKIP) and THIRD_PARTY <= k <= RENDITION_RECORD) for k, _ in i["records"])
             and ascending(keys)
             and all(unit_ok(x) and x["quantity"] > 0 and x["cited"] > 0 for x in i["lines"])
             and ascending([(e[0], e[1]) for e in i["energy"]])
             and all(1 <= e[0] <= scopes(i["version"]) and 2 <= e[1] <= 5 and e[2] > 0 for e in i["energy"]))
    if not rules and "invoice-rules" not in SKIP:
        raise Refuse("invoice_invalid", at)
    bad = verify_sig(i["seller"], INVOICE_DOMAIN, b[:-64], sig)
    if bad:
        raise Refuse(bad, at)
    i["digest"] = sha(b[:-64])
    return i


# ---------------------------------------------------------------- §5 sources

def figure(joules, cls):
    """(tag or None, joules, declared ppm): a figure as §6.6 counts it."""
    if cls is None or joules == 0:
        return (None, 0, 0)
    return (cls[0], joules, cls[1] if cls[0] in (4, 5) else 0)


def room_source(s):
    keys = {"kind", "room", "receipt", "hops", "descriptors", "compositions", "takes", "logs"}
    if set(s) != keys or not isinstance(s["room"], dict) or set(s["room"]) != {"descriptor", "grants", "key_log_head", "chains"}:
        raise Refuse("malformed", [])
    doc = {"rooms": {"r": s["room"]}}
    v = {"room": "r", "receipt": s["receipt"], "hops": s["hops"], "descriptors": s["descriptors"],
         "compositions": s["compositions"], "takes": s["takes"], "logs": s["logs"]}
    got = LMV.verify_vector(v, doc, LMV.Cache())
    if got["outcome"] != "accept":
        raise Refuse(got["reason"], got["at"])
    d = LRV.read_descriptor(bytes.fromhex(s["room"]["descriptor"]))
    rc = LMV.read_receipt(bytes.fromhex(s["receipt"]), d["room"])
    u = rc["units"]
    # A receipt of version 2 states its bytes out and their price (live-metering §4.4, §4.6);
    # version 1 states neither, read as 0.
    figures = [u["publisher_us"]] + list(u["subscriber_us"]) + [u["composition_us"], u["recorded_us"], u["events_sent"],
                                                                  u.get("bytes_out", 0)]
    prices = list(rc["prices"]) + [0] * (14 - len(rc["prices"]))
    return {"kind": ROOM, "digest": rc["digest"], "room": d["room"], "t0": d["t0"], "start": rc["start"], "end": rc["end"],
            "units": figures, "currency": rc["currency"], "prices": prices,
            "hops": [(h["record"], figure(h["joules"], h["class"])) for h in rc["hops"]]}


def delivery_source(s):
    if set(s) != {"kind", "statement", "records", "proofs", "hops", "rooms"}:
        raise Refuse("malformed", [])
    doc = {"records": [{"record": r["record"], "items": r["items"]} for r in s["records"]],
           "proofs": [{"proof": p} for p in s["proofs"]], "hops": [{"hop": h} for h in s["hops"]],
           "rooms": {str(k): r for k, r in enumerate(s["rooms"])}}
    v = {"statement": s["statement"], "rooms": [str(k) for k in range(len(s["rooms"]))], "withhold": [],
         "extra": {"records": [], "proofs": []}}
    got = SV.statement_outcome(doc, v)
    if got["outcome"] != "accept":
        raise Refuse(got["reason"], got["at"])
    st = SV.read_statement(bytes.fromhex(s["statement"]))
    given = {}
    for r in s["records"]:
        b = bytes.fromhex(r["record"])
        given[sha(b[:-64])] = (b, [bytes.fromhex(x) for x in r["items"]])
    acks = []
    for d in st["records"]:
        b, lists = given[d]
        a = SV.read_ack(b)
        a["item_lists"] = [SV.read_items(x) for x in lists]
        acks.append(a)
    hops = {}
    for h in s["hops"]:
        b = bytes.fromhex(h)
        hops[sha(b[:-64])] = b
    figs = []
    for d in st["hops"]:
        x = LMV.read_hop(hops[d], hops[d][5:37])
        figs.append((d, figure(x["joules"], x["class"])))
    return {"kind": DELIVERY, "digest": st["digest"], "statement": st, "acks": acks, "hops": figs}


def package_source(s):
    if set(s) != {"kind", "receipt", "indexes"} or not isinstance(s["receipt"], str):
        raise Refuse("malformed", [])
    try:
        r = PV.read_receipt(s["receipt"].encode("utf-8"))
        PV.verify_receipt_v2(r, [bytes.fromhex(x) for x in s["indexes"]])
    except PV.Refused as e:
        raise Refuse(e.code, ["receipt"])
    j = r["json"]
    cls = None
    if j["energy_provenance"] is not None:
        tag = TAGS[j["energy_provenance"]]
        cls = (tag, j["energy_uncertainty"]["relative_ppm"] if tag in (4, 5) else 0)
    return {"kind": PACKAGE, "digest": bytes.fromhex(j["receipt_hash"]), "signer": bytes.fromhex(j["signer_pubkey"]),
            "entries": j["total_entries"], "stored": j["stored_bytes"],
            "figure": figure(j["joules_micro"], cls), "held": None}


def ads_source(s):
    keys = {"kind", "descriptor", "grants", "key_log", "chains", "policy", "pt0_ms", "receipts", "pods", "claims",
            "creatives", "issuers"}
    if set(s) != keys:
        raise Refuse("malformed", [])
    holder = {"creatives": s["creatives"], "issuers": s["issuers"]}
    APV.held(holder)
    v = dict(s, name="source")
    got, _ = LAV.ads(v, judge=lambda vv, d, p: APV.Judge(holder, vv, p))
    if got["outcome"] == "gap":
        raise Refuse("source_gap", [])
    if got["outcome"] != "accept":
        raise Refuse(got["reason"], got["at"])
    d = LRV.read_descriptor(bytes.fromhex(s["descriptor"]))
    claims_required = "claim_facts" in got
    breaks = {bytes.fromhex(b["digest"]): b["t"] for b in got["breaks"]}
    claim_digests = [None if c["digest"] is None else bytes.fromhex(c["digest"]) for c in got.get("claim_facts", [])]
    receipts = []
    for o, fact in zip(s["receipts"], got["receipt_facts"]):
        r = LAV.read_receipt(o)
        imps = []
        for k, i in enumerate(r["impressions"]):
            billable = i["quartiles_reached"] >= 1
            if claims_required:
                c = fact["impressions"][k]
                counts = c["verdict"] == "presented"
                verdict = c["verdict"]
                claim = claim_digests[c["claim"]] if counts and c["claim"] is not None else None
            else:
                counts, verdict, claim = billable, "beacons", None
            imps.append({"slot": i["slot"], "ad_id": i["ad_id"], "price_micros": i["price_micros"],
                         "quartiles_reached": i["quartiles_reached"], "billable": billable, "counts": counts,
                         "verdict": verdict, "claim": claim})
        cls = None
        if r["class"] is not None:
            tag = TAGS[r["class"]["label"]]
            cls = (tag, r["class"].get("ppm", 0))
        receipts.append({"identity": bytes.fromhex(fact["identity"]), "decider": r["signer_pubkey"], "cue": r["cue"],
                         "session": r["session"], "t": breaks[r["cue"]], "spend": r["budget_spent_micros"],
                         "counted_spend": fact.get("presented_spend_micros", r["budget_spent_micros"]),
                         "impressions": imps, "figure": figure(r["joules_micro"], cls)})
    return {"kind": ADS, "room": d["room"], "t0": d["t0"], "claims_required": claims_required, "receipts": receipts,
            "claims": [c for c in claim_digests if c is not None]}


def stitched_source(s, signers):
    """§5.1: a stitched source, verified as stitched-presentation §7 verifies, with the tariff's
    ad scope as its signers (§4.4). That verification refuses nothing as a whole."""
    keys = {"kind", "records", "claims", "manifests", "responses", "receipts", "creatives", "issuers"}
    if set(s) != keys:
        raise Refuse("malformed", [])
    creatives, issuers, _ = STV.held({"creatives": s["creatives"], "issuers": s["issuers"], "signers": []})
    records = [bytes.fromhex(x) for x in s["records"]]
    inp = {"records": records, "claims": [bytes.fromhex(x) for x in s["claims"]],
           "manifests": [bytes.fromhex(x) for x in s["manifests"]], "responses": [bytes.fromhex(x) for x in s["responses"]],
           "receipts": s["receipts"], "creatives": creatives, "issuers": issuers, "signers": list(signers)}
    got = STV.verify(inp)
    claims = got["claim_facts"]
    accepted = []
    for b, f in zip(records, got["record_facts"]):
        if f["verdict"] != "accepted":
            continue
        ads = [(a["presentations"], [bytes.fromhex(claims[j]["digest"]) for j in a["own_claims"]
                                      if claims[j]["verdict"] == "presented" and claims[j]["digest"] is not None])
               for a in f["ads"]]
        accepted.append({"digest": bytes.fromhex(f["digest"]), "signer": STV.read_record(b)["signer"], "ads": ads})
    return {"kind": STITCHED, "records": accepted, "claims": [bytes.fromhex(c["digest"]) for c in claims if c["digest"] is not None],
            "statement": got}


# A source's verification is a pure function of the source, and the corpus verifies the same
# sources under every mutated copy and every weakened verifier (none of whose checks is a
# source's own): each is verified once.
def oer_identity(rid):
    """§5.2: the 32 bytes an OER/2 receipt id names."""
    return bytes.fromhex(rid.split(":", 1)[1])


def micro(v):
    """§5.1: a figure in joules as whole microjoules, the nearest, a half to the even one; None
    when it is not a non-negative quantity (oer-receipts §4)."""
    try:
        d = OER.dec(v)
    except TypeError:
        return None
    with localcontext() as c:
        c.prec, c.Emax, c.Emin = MAX_PREC, MAX_EMAX, MIN_EMIN
        n = (d * 1000000).to_integral_value(rounding=ROUND_HALF_EVEN)
    return int(n) if n >= 0 else None


def job_figure(r):
    """§5.1: the billed hop's figure as §6.6 counts it, or None when no row can state it."""
    if OER.is_unmetered(r):
        return (None, 0, 0)
    e = r.get("energy") if isinstance(r.get("energy"), dict) else {}
    c = e.get("class")
    if not OER.class_valid(c):
        return None
    j = micro(e.get("joules_total"))
    if j is None or j < 1 or j > U64:
        return None
    tag = TAGS[c["label"]]
    return (tag, j, c["relative_ppm"] if tag in (4, 5) else 0)


def content_hash(b):
    """route-record §2: an object's content hash (jwp-receipts §2.1)."""
    return blake3(CONTENT_DOMAIN + b)


def animation(b):
    """§4.8, 2: an animated PNG of 8-bit RGB frames, each the whole picture, read whole. The
    chunks under a still's rules (packaging §5: every type four letters and its CRC; IHDR first,
    8-bit RGB without interlace; an optional PLTE before the image data; the IDATs consecutive;
    ancillary chunks read past; any other critical chunk refused; IEND last, empty, nothing after
    it), and the animation's: one acTL of 8 bytes before any fcTL and the image data, of at least
    one frame; each fcTL of 26 bytes, the whole picture at offsets 0, dispose 0 to 2, blend 0, the
    first before the IDATs, the others after them, each followed by at least one fdAT; the fcTL
    and fdAT sequence numbers 0, 1, 2, ... in order, an fdAT at least 4 bytes; as many fcTLs as
    acTL counts. Its width, height and each frame's zlib stream, the IDATs' data for the first and
    its fdATs' after their sequence numbers for each other; ValueError where it is not one."""
    if b[:8] != b"\x89PNG\r\n\x1a\n":
        raise ValueError("not a PNG signature")
    at, size, count, frames, seq = 8, None, None, [], 0
    ended, plte, idat = False, False, 0               # idat: 0 none yet, 1 in the run, 2 past it
    while at < len(b):
        if ended:
            raise ValueError("bytes after IEND")
        if at + 12 > len(b):
            raise ValueError("a chunk past the end")
        n = struct.unpack_from(">I", b, at)[0]
        kind, data = b[at + 4:at + 8], b[at + 8:at + 8 + n]
        if at + 12 + n > len(b):
            raise ValueError("a chunk past the end")
        if not all(65 <= c <= 90 or 97 <= c <= 122 for c in kind):
            raise ValueError("a chunk type that is not four letters")
        if zlib.crc32(kind + data) != struct.unpack_from(">I", b, at + 8 + n)[0]:
            raise ValueError("a chunk CRC")
        if idat == 1 and kind != b"IDAT":
            idat = 2
        if kind == b"IHDR":
            if size is not None or at != 8 or n != 13:
                raise ValueError("IHDR is not one 13-byte chunk first")
            w, h = struct.unpack(">II", data[:8])
            if data[8:] != bytes([8, 2, 0, 0, 0]) or w == 0 or h == 0 or 3 * w * h > 1 << 26:
                raise ValueError("not 8-bit RGB without interlace, or a size past 2^26 samples")
            size = (w, h)
        elif size is None:
            raise ValueError("a chunk before IHDR")
        elif kind == b"PLTE":
            if plte or idat or n == 0 or n % 3 or n > 768:
                raise ValueError("PLTE: more than one, after the image data, or not 1 to 256 entries")
            plte = True
        elif kind == b"acTL":
            if count is not None or frames or idat or n != 8:
                raise ValueError("acTL is not one 8-byte chunk before the frames")
            count = struct.unpack(">I", data[:4])[0]
            if count == 0:
                raise ValueError("acTL counts no frame")
        elif kind == b"fcTL":
            if count is None or n != 26:
                raise ValueError("fcTL before acTL, or not 26 bytes")
            q, fw, fh, x, y = struct.unpack(">IIIII", data[:20])
            if q != seq:
                raise ValueError("a sequence number out of order")
            seq += 1
            if (fw, fh, x, y) != (size[0], size[1], 0, 0):
                raise ValueError("a frame that is not the whole picture")
            if data[24] > 2 or data[25] != 0:
                raise ValueError("a frame's dispose or blend operation")
            if len(frames) == 1 and not idat:
                raise ValueError("a second frame before the image data")
            if len(frames) > 1 and not frames[-1]:
                raise ValueError("a frame with no fdAT")
            frames.append([])
        elif kind == b"IDAT":
            if idat == 2:
                raise ValueError("IDAT chunks that are not consecutive")
            if len(frames) != 1:
                raise ValueError("image data that is not the first frame's")
            idat = 1
            frames[0].append(data)
        elif kind == b"fdAT":
            if len(frames) < 2 or n < 4:
                raise ValueError("fdAT before a frame after the first, or shorter than 4 bytes")
            if struct.unpack(">I", data[:4])[0] != seq:
                raise ValueError("a sequence number out of order")
            seq += 1
            frames[-1].append(data[4:])
        elif kind == b"IEND":
            if n:
                raise ValueError("IEND with data")
            ended = True
        elif not kind[0] & 0x20:
            raise ValueError("an unknown critical chunk")
        at += 12 + n
    if size is None or not ended:
        raise ValueError("no IHDR or no IEND")
    if count is None:
        raise ValueError("no acTL: not an animation")
    if not idat:
        raise ValueError("no image data")
    if len(frames) > 1 and not frames[-1]:
        raise ValueError("a frame with no fdAT")
    if len(frames) != count:
        raise ValueError("acTL counts another number of frames")
    return size[0], size[1], [b"".join(f) for f in frames]


def frame_samples(z, w, h):
    """§4.8, 2: a frame's zlib stream read as a still's image data is (packaging §5): the still
    of that one IDAT, read by the packaging verifier. Its RGB8 bytes; ValueError otherwise."""
    def chunk(kind, data):
        return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data))
    still = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">II", w, h) + bytes([8, 2, 0, 0, 0])) + chunk(b"IDAT", z) + chunk(b"IEND", b"")
    return PV.png_frame(still)[2]


HEX_BYTES = re.compile(r"(?:[0-9a-f]{2})*")
PNG_FAILS = (ValueError, zlib.error, struct.error, IndexError)


def is_hex(v):
    return isinstance(v, str) and HEX_BYTES.fullmatch(v) is not None


def upscale_output(x, i):
    """§4.9: an upscale's output: its record proved by exact-upscale §7 from the input, a WAI1 of one
    8-bit RGB PNG, and the prior; and the picture's samples the output it states. The link hash, the
    subpixels and the picture."""
    if not isinstance(x, dict) or set(x) != {"upscale", "object", "prior", "picture"} or not isinstance(x["upscale"], str) \
            or not all(is_hex(x[k]) for k in ("object", "prior", "picture")):
        raise Refuse("malformed", [])

    def bad(*at):
        return Refuse("output_invalid", ["outputs", i] + list(at))
    try:
        u = UV.read(x["upscale"], "wai-upscale", UV.UPSCALE, UV.upscale_fields)
    except UV.Refuse:
        raise bad()
    obj, prior, picture = bytes.fromhex(x["object"]), bytes.fromhex(x["prior"]), bytes.fromhex(x["picture"])
    if "upscale-input" in SKIP:
        rgb = b""
    else:
        try:
            m, _, payload = UV.STV.read_wai1(obj)
            if m["model_requirement"]["capability"] != "wai.image.png":
                raise ValueError("another capability")
            rgb = TV.png_decode(payload, False)[3]
        except (UV.STV.NotWai1,) + PNG_FAILS:
            raise bad("object")
    case = {"claims": [], "upscales": [x["upscale"]], "presentations": [], "objects": [x["object"]],
            "decoded": [{"object": content_hash(obj).hex(), "samples": rgb.hex()}], "priors": ["given"], "residuals": [], "targets": [],
            "policy": {"dual_layer": "when-given", "assistive": "require-proof", "second_layer": "accept-stated", "upscale": "require-proof"}}
    if "upscale-proof" not in SKIP:
        try:
            UV.verify(case, {}, {"given": prior})
        except UV.Refuse:
            raise bad()
    f = UV.SR.FACTOR[u["derivation"]]
    ow, oh = f * u["w"], f * u["h"]
    try:
        pw, ph, _, samples = TV.png_decode(picture, False)
    except PNG_FAILS:
        raise bad("picture")
    if "upscale-picture" not in SKIP and ((pw, ph) != (ow, oh) or hashlib.sha256(samples).digest() != u["output"]):
        raise bad("picture")
    return u["link"], ow * oh * 3, picture


def read_proof(pr):
    """§5.1: an inclusion proof's form, or None."""
    if not (isinstance(pr, dict) and set(pr) == {"index", "siblings"} and isinstance(pr["index"], int) and not isinstance(pr["index"], bool)
            and 0 <= pr["index"] <= 2 ** 53 - 1 and isinstance(pr["siblings"], list)
            and all(isinstance(h, str) and re.fullmatch(r"[0-9a-f]{64}", h) for h in pr["siblings"])):
        return None
    return pr["index"], [bytes.fromhex(h) for h in pr["siblings"]]


def tonemap_output(x, i, committed):
    """§4.9: a tone mapping's output: its record proved by exact-tonemap §8 from the envelope, and each
    rendition the bytes its picture states. The link hash, the subpixels, each rendition's content hash
    and its proof."""
    want = {"tonemap", "envelope", "renditions"} | ({"proofs"} if committed else set())
    if not isinstance(x, dict) or set(x) != want or not isinstance(x["tonemap"], str) or not is_hex(x["envelope"]) \
            or not isinstance(x["renditions"], list) or not all(is_hex(b) for b in x["renditions"]):
        raise Refuse("malformed", [])
    if committed:
        if not isinstance(x["proofs"], list) or len(x["proofs"]) != len(x["renditions"]):
            raise Refuse("malformed", [])
        proofs = [read_proof(p) for p in x["proofs"]]
        if any(p is None for p in proofs):
            raise Refuse("malformed", [])
    else:
        proofs = [None] * len(x["renditions"])

    def bad(*at):
        return Refuse("output_invalid", ["outputs", i] + list(at))
    try:
        r = TV.read_record(x["tonemap"])
    except TV.Refuse:
        raise bad()
    env = bytes.fromhex(x["envelope"])
    if "tonemap-proof" not in SKIP:
        held = {"envelopes": {content_hash(env).hex(): env}, "frames": {}}
        if TV.run_case({"records": [x["tonemap"]], "policy": "require-proof"}, held)["outcome"] != "accept":
            raise bad()
    renditions = [bytes.fromhex(b) for b in x["renditions"]]
    if len(renditions) != len(r["pictures"]) and "tonemap-count" not in SKIP:
        raise bad("renditions")
    if "tonemap-renditions" not in SKIP:
        for k, (b, p) in enumerate(zip(renditions, r["pictures"])):
            if hashlib.sha256(b).hexdigest() != p["output"]:
                raise bad("renditions", k)
    w, h = r["base"]["width"], r["base"]["height"]
    return bytes.fromhex(r["_link"]), len(r["pictures"]) * w * h * 3, [content_hash(b) for b in renditions], proofs


def job_source(s, version):
    members = {"kind", "request", "pricebook", "policy", "record", "receipts", "keys", "outputs"}
    third = "third_party" in s and (version == V6 or "third-party-version" in SKIP)    # §4.9
    if third:
        members.add("third_party")
    if set(s) != members or s["receipts"] is None or s["keys"] is None or not isinstance(s["outputs"], list):
        raise Refuse("malformed", [])
    doc = {k: s[k] for k in members - {"kind", "outputs"}}
    got = RRV.verify(doc)
    if got["outcome"] != "accept":
        raise Refuse(got["reason"], got["at"])
    records = got.get("third_party", [])
    rec = json.loads(s["record"])
    texts = [OER.load(io.StringIO(t)) for t in s["receipts"]]
    billed, fig, endpoint, usage = None, None, b"", None
    for a in rec["attempts"]:
        if a["billed"] is True:
            cap = next(c["capability"] for c in rec["candidates"] if c["endpoint"] == a["endpoint"])
            endpoint = a["endpoint"].encode("utf-8")
            # §4.9: a failed or late hop's record states what the provider billed, and no line reads it.
            mine = records[:1] if "billed-usage" in SKIP else [r for r in records if r["hop"] == a["hop"]]
            if mine and mine[0]["usage"] is not None:
                usage = (mine[0]["usage"]["unit"].encode("utf-8"), mine[0]["usage"]["quantity"], bytes.fromhex(mine[0]["digest"]))
            j = next(i for i, t in enumerate(texts) if isinstance(t, dict) and t.get("receipt_id") == a["receipt"])
            fig = job_figure(texts[j]) if "job-figure" not in SKIP else (None, 0, 0)
            if fig is None:
                raise Refuse("job_figure_invalid", ["receipts", j])
            billed = (cap.encode("utf-8"), oer_identity(a["receipt"]))
            break
    unbilled = []                                  # §6.6, version 4: every hop's figure
    if version in SPENDING and "unbilled-hops" not in SKIP:
        for a in rec["attempts"]:
            if a["billed"] is True:
                continue
            j = next(i for i, t in enumerate(texts) if isinstance(t, dict) and t.get("receipt_id") == a["receipt"])
            f = job_figure(texts[j])
            if f is None:
                raise Refuse("job_figure_invalid", ["receipts", j])
            unbilled.append(f)
    # §5.1: the objects the served hop states it delivered, when its record states them (version
    # 2); under version 4 a record that states none holds every output to none.
    if rec["version"] == 2:
        delivered = {bytes.fromhex(h) for a in rec["attempts"] if a["outcome"] == "served" for h in a["outputs"]}
    elif version in SPENDING:
        delivered = set()
    else:
        delivered = None
    if "output-delivered" in SKIP:
        delivered = None
    # §5.1: under a route record of version 3 the served hop states what it delivered by its
    # count and root, and each output proves its place by its inclusion proof (route-record
    # §6.1); one no attempt served holds every output to none.
    served_att = next((a for a in rec["attempts"] if a["outcome"] == "served"), None)
    committed = rec["version"] == 3 and served_att is not None
    if rec["version"] == 3 and not committed and "output-delivered" not in SKIP:
        delivered = set()
    count, root = (served_att["delivered"]["count"], bytes.fromhex(served_att["delivered"]["root"])) if committed else (0, b"")
    held = "output-delivered" not in SKIP and (delivered is not None or committed)

    def holds(h, proof, place):
        """Whether the object `h` is one the hop delivered, at `place` where one is required."""
        if committed:
            if "output-proof" in SKIP:
                return True
            return proof is not None and (place is None or proof[0] == place) and RRV.inclusion_proof(h, proof[0], proof[1], count, root)
        if place is not None:
            return place < len(stated) and stated[place] == h
        return h in delivered

    seen = set()
    outputs, packages, upscales, renditions = [], [], [], []
    hexs = re.compile(r"(?:[0-9a-f]{2})*")
    # §4.8, version 5: a decode delivers one animated PNG of its frames, and its codestream is
    # what it read; the served hop states exactly those animations, in order, and nothing else.
    decode = version in DECODE_FRAMES and rec["operation"] == "media.decode" and "decode-frames" not in SKIP
    # §4.9, version 6: an upscale and a tone mapping are given in their own forms.
    of_six = version == V6 or "form-version" in SKIP
    upscale = of_six and rec["operation"] == UPSCALE_OP
    tonemap = of_six and rec["operation"] == TONEMAP_OP
    stated = [bytes.fromhex(h) for a in rec["attempts"] if a["outcome"] == "served" for h in a["outputs"]] if rec["version"] == 2 else []
    nxt = 0
    for i, x in enumerate(s["outputs"]):
        # One record's pictures are many objects, each with its proof. A picture that repeats within
        # its record is one object its hop states once, and is billed as a picture (§4.9).
        if tonemap:
            link, n, objects, proofs = tonemap_output(x, i, committed)
            renditions.append((link, n))
            own = set()
            for k, h in enumerate(objects):
                fresh = h not in own
                own.add(h)
                if "output-delivered" in SKIP:
                    continue
                if not holds(h, proofs[k], None) or (fresh and h in seen):
                    raise Refuse("output_not_delivered", ["outputs", i, "renditions", k])
                seen.add(h)
            continue
        # Under a record of version 3 each output gives `proof`, and under one of version 1 or 2
        # none (§5.1).
        proof = None
        if committed:
            if not isinstance(x, dict) or "proof" not in x:
                raise Refuse("malformed", [])
            x = dict(x)
            proof = read_proof(x.pop("proof"))
            if proof is None:
                raise Refuse("malformed", [])
        if decode:
            if not isinstance(x, dict) or set(x) != {"receipt", "codestream", "frames"} or not isinstance(x["receipt"], str) \
                    or not isinstance(x["codestream"], str) or not hexs.fullmatch(x["codestream"]) \
                    or not isinstance(x["frames"], str) or not hexs.fullmatch(x["frames"]):
                raise Refuse("malformed", [])
            try:
                r = IMR.read(x["receipt"])
            except IMR.Bad:
                raise Refuse("output_invalid", ["outputs", i])
            if "output" not in SKIP and IMR.rederived(r, bytes.fromhex(x["codestream"])) is not None:
                raise Refuse("output_invalid", ["outputs", i])
            anim = bytes.fromhex(x["frames"])
            try:
                w, h, parts = animation(anim)
            except ValueError:
                raise Refuse("output_invalid", ["outputs", i, "frames"])
            if (w, h) != (r["w"], r["h"]) or len(parts) != len(r["frames"]):
                raise Refuse("output_invalid", ["outputs", i, "frames"])
            for k, z in enumerate(parts):
                try:
                    rgb = frame_samples(z, w, h)
                except ValueError:
                    raise Refuse("output_invalid", ["outputs", i, "frames", k])
                if IMR.frame_hash(k, rgb) != r["frames"][k]:
                    raise Refuse("output_invalid", ["outputs", i, "frames", k])
            outputs.append((IMR.receipt_hash(r), r["work"]))
            if "output-delivered" not in SKIP:
                if not holds(content_hash(anim), proof, nxt):
                    raise Refuse("output_not_delivered", ["outputs", i, "frames"])
                nxt += 1
            continue
        if upscale:
            link, n, obj = upscale_output(x, i)
            upscales.append((link, n))
        elif version in SPENDING and isinstance(x, dict) and set(x) == {"package", "indexes"}:
            # §4.7: a package a run of `package.build` made, its object its receipt's text.
            if not isinstance(x["package"], str) or not isinstance(x["indexes"], list) \
                    or not all(isinstance(h, str) and hexs.fullmatch(h) for h in x["indexes"]):
                raise Refuse("malformed", [])
            try:
                pr = PV.read_receipt(x["package"].encode("utf-8"))
                PV.verify_receipt_v2(pr, [bytes.fromhex(h) for h in x["indexes"]])
            except PV.Refused:
                raise Refuse("output_invalid", ["outputs", i])
            obj = x["package"].encode("utf-8")
            packages.append((bytes.fromhex(pr["json"]["receipt_hash"]), pr["json"]["work_units"] if "run-work" not in SKIP else 0))
        else:
            if not isinstance(x, dict) or set(x) != {"receipt", "codestream"} or not isinstance(x["receipt"], str) \
                    or not isinstance(x["codestream"], str) or not hexs.fullmatch(x["codestream"]):
                raise Refuse("malformed", [])
            try:
                r = IMR.read(x["receipt"])
            except IMR.Bad:
                raise Refuse("output_invalid", ["outputs", i])
            if "output" not in SKIP and IMR.rederived(r, bytes.fromhex(x["codestream"])) is not None:
                raise Refuse("output_invalid", ["outputs", i])
            obj = bytes.fromhex(x["codestream"])
            outputs.append((IMR.receipt_hash(r), r["work"]))
        if held:
            h = content_hash(obj)
            if not holds(h, proof, None) or h in seen:
                raise Refuse("output_not_delivered", ["outputs", i])
            seen.add(h)
    if decode and "output-delivered" not in SKIP and nxt != (count if committed else len(stated)):
        raise Refuse("output_not_delivered", ["outputs"])     # §4.8, 3: an object past the animations
    return {"kind": JOB, "digest": bytes.fromhex(RRV.digest(rec)), "router": bytes.fromhex(rec["signer"]), "job": rec["job"],
            "operation": rec["operation"].encode("utf-8"), "accepted": rec["accepted"], "hops": [oer_identity(a["receipt"]) for a in rec["attempts"]], "billed": billed,
            "figure": fig, "outputs": outputs, "packages": packages, "unbilled": unbilled, "endpoint": endpoint, "usage": usage,
            "third_party": [bytes.fromhex(r["digest"]) for r in records], "upscales": upscales, "renditions": renditions}


_SOURCES = {}


def verify_source(s, signers, version=1):
    """§5.1: the source verified by its own extension, a stitched source with `signers`, a job
    source under a tariff of version 2 or 3, or Refuse with that extension's place."""
    key = json.dumps([s, sorted(k.hex() for k in signers), version if isinstance(s, dict) and s.get("kind") == "job" else version in JOB_VERSIONS,
                      sorted(SKIP & {"job-figure", "output", "output-delivered", "unbilled-hops", "run-work", "decode-frames", "output-proof",
                                     "third-party-version", "form-version", "billed-usage", "upscale-input", "upscale-proof", "upscale-picture",
                                     "tonemap-proof", "tonemap-count", "tonemap-renditions"})], sort_keys=True)
    if key not in _SOURCES:
        try:
            _SOURCES[key] = read_source(s, signers, version)
        except Refuse as e:
            _SOURCES[key] = e
    got = _SOURCES[key]
    if isinstance(got, Refuse):
        raise Refuse(got.reason, got.at)
    return got


def read_source(s, signers, version):
    if not isinstance(s, dict):
        raise Refuse("malformed", [])
    if s.get("kind") == "stitched":
        return stitched_source(s, signers)
    if s.get("kind") == "job" and version in JOB_VERSIONS:
        return job_source(s, version)
    readers = {"room": room_source, "delivery": delivery_source, "package": package_source, "ads": ads_source}
    if s.get("kind") not in readers:
        raise Refuse("malformed", [])
    return readers[s["kind"]](s)


def heads(f):
    """§5.2."""
    if f["kind"] == ROOM:
        return [(SESSION, f["digest"])]
    if f["kind"] == DELIVERY:
        return [(STATEMENT, f["digest"])]
    if f["kind"] == PACKAGE:
        return [(PACK, f["digest"])]
    if f["kind"] == STITCHED:
        return [(POD_INPUT, r["digest"]) for r in f["records"]]
    if f["kind"] == JOB:
        return [(ROUTE_RECORD, f["digest"])]
    return [(AD_RECEIPT, r["identity"]) for r in f["receipts"]]


def contributed(f):
    """§5.2: every record a source contributes."""
    out = set(heads(f))
    if f["kind"] == ROOM:
        out |= {(HOP, d) for d, _ in f["hops"]}
    elif f["kind"] == DELIVERY:
        out |= {(ACK, d) for d in f["statement"]["records"]} | {(HOP, d) for d in f["statement"]["hops"]}
    elif f["kind"] == ADS:
        out |= {(CLAIM, c) for c in f["claims"]}
    elif f["kind"] == STITCHED:
        out |= {(STITCHED_CLAIM, c) for c in f["claims"]}
    elif f["kind"] == JOB:
        out |= {(OER_RECEIPT, h) for h in f["hops"]} | {(VIDEO_RECEIPT, h) for h, _ in f["outputs"]}
        out |= {(PACK, h) for h, _ in f["packages"]}
        out |= {(THIRD_PARTY, d) for d in f["third_party"]}
        out |= {(UPSCALE_RECORD, h) for h, _ in f["upscales"]} | {(RENDITION_RECORD, h) for h, _ in f["renditions"]}
    return out


def in_scope(t, kind, i):
    return "scope" in SKIP or (kind, i) in t["scope"]


def check_source(t, start, end, f):
    """§7.2 step 5 after the source's own verification: scope, one tariff (§3), period (§5.3)."""
    def within(a, b):
        return "period" in SKIP or (start <= a and b <= end)
    if f["kind"] == ROOM:
        if not in_scope(t, 1, f["room"]):
            return "outside_scope"
        # §3: the thirteen prices, and the price of bytes out (§4.7), 0 where either states none.
        prices = f["prices"] if "conflict-bytes" not in SKIP else f["prices"][:13] + t["room"][13:]
        if f["currency"] != bytes(3) and (f["currency"] != t["currency"] or prices != t["room"]) and "conflict" not in SKIP:
            return "tariff_conflict"
        if not within(f["t0"] + f["start"], f["t0"] + f["end"]):
            return "outside_period"
    elif f["kind"] == DELIVERY:
        st = f["statement"]
        if not in_scope(t, 2, st["party"]):
            return "outside_scope"
        if st["currency"] != bytes(3) and (st["currency"] != t["currency"] or st["prices"] != t["delivery"]) and "conflict" not in SKIP:
            return "tariff_conflict"
        if not within(st["start"], st["end"]):
            return "outside_period"
    elif f["kind"] == PACKAGE:
        if not in_scope(t, 3, f["signer"]):
            return "outside_scope"
    elif f["kind"] == STITCHED:
        return None     # accepted only under the tariff's signers (§4.4); it states no time (§5.3)
    elif f["kind"] == JOB:
        if not in_scope(t, 5, f["router"]):
            return "outside_scope"
        if "period" not in SKIP and not start <= f["accepted"] < end:
            return "outside_period"
    else:
        if not all(in_scope(t, 4, r["decider"]) for r in f["receipts"]):
            return "outside_scope"
        if "period" not in SKIP and not all(start <= f["t0"] + r["t"] < end for r in f["receipts"]):
            return "outside_period"
    return None


def overlapping(a, b):
    return a[0] < b[1] and b[0] < a[1]


def bills_again(f, g, version):
    """§5.3: whether f bills what g, an earlier source, billed, under a tariff of `version`."""
    if f["kind"] != g["kind"]:
        return False
    if f["kind"] == ROOM:
        return f["room"] == g["room"] and overlapping((f["start"], f["end"]), (g["start"], g["end"]))
    if f["kind"] == DELIVERY:
        a, b = f["statement"], g["statement"]
        return a["party"] == b["party"] and overlapping((a["start"], a["end"]), (b["start"], b["end"]))
    if f["kind"] == ADS:
        return f["room"] == g["room"] and any((x["cue"], x["session"]) == (y["cue"], y["session"])
                                             for x in f["receipts"] for y in g["receipts"])
    if f["kind"] == JOB:            # version 2; in version 3 two records of one job are two runs of it
        return version == 2 and f["router"] == g["router"] and f["job"] == g["job"]
    return False


def billed_once(version, kind):
    """§6.9: whether a chain bills a record of `kind` once. In versions 3 and 4 a job's output
    receipt is the evidence of what it delivered, which a re-run delivers again; its work, the route
    record and the hop receipts, is billed once. In version 4 a package receipt is listed again by
    each invoice that bills its storage (§4.7)."""
    if kind == PACK and version in SPENDING and "package-again" not in SKIP:
        return False
    if kind in (UPSCALE_RECORD, RENDITION_RECORD) and "evidence-again" not in SKIP:
        return False                               # §4.9: what re-derives an output is its evidence
    return not (kind == VIDEO_RECEIPT and (version in WORK_VERSIONS or "output-once" in SKIP))


def hold(t, facts, previous, start, end):
    """§4.7: under version 4, a package the previous invoice lists is held through [start, end)."""
    if t["version"] not in SPENDING:
        return
    held = {i for k, i in (previous["records"] if previous else []) if k == PACK}
    for k, f in enumerate(facts):
        if f["kind"] == PACKAGE:     # a copy: a source's facts are shared by every vector that gives it
            facts[k] = dict(f, held=end - start if f["digest"] in held and "storage-held" not in SKIP else None)


def check_across(facts, previous, version=1):
    """§7.2 step 6: (the index, the reason) of the first source refused, or None."""
    billed = {r for r in (previous["records"] if previous else []) if billed_once(version, r[0])}
    seen, jobs = set(), set()
    for k, f in enumerate(facts):
        hs = heads(f)
        if "duplicate" not in SKIP and (any(h in seen for h in hs) or len(set(hs)) != len(hs)):
            return k, "record_duplicate"
        if f["kind"] == JOB:                       # §5.2: one hop, one job; in version 2, one output too
            own = {r for r in contributed(f) - set(hs) if billed_once(version, r[0])}
            if "job-duplicate" not in SKIP and own & jobs:
                return k, "record_duplicate"
            jobs |= own
        if "overlap" not in SKIP and any(bills_again(f, g, version) for g in facts[:k]):
            return k, "record_overlap"
        if "rebilled" not in SKIP and contributed(f) & billed:
            return k, "record_rebilled"
        seen |= set(hs)
    return None


# ---------------------------------------------------------------- §6.2–§6.6 the figures

def free(c, grant):
    return c in (2, 3) and grant == ZERO


def cites_digest(cited):
    return sha(CITES_DOMAIN, b"".join(bytes([k]) + i for k, i in sorted(cited)))


def compute(t, facts):
    """The records, lines (with their cited records), rows, unmetered counts, amount and the sums
    per (scope, tag); None when a figure does not fit its field."""
    records = set()
    for f in facts:
        records |= contributed(f)
    q = {}

    def add(kind, source, unit, n, cites):
        if n == 0:
            return
        e = q.setdefault((kind, source, unit), [0, set()])
        e[0] += n
        e[1] |= set(cites)
    for f in facts:
        if f["kind"] == ROOM:
            for unit in range(1, 15 if t["version"] in SPENDING and "bytes-out" not in SKIP else 14):
                add(ROOM, f["room"], unit, f["units"][unit - 1], [(SESSION, f["digest"])])
        elif f["kind"] == DELIVERY:
            st = f["statement"]
            for c in range(1, 5):
                n = sum(x["bytes"] for x in st["lines"] if x["kind"] == 1 and x["class"] == c and not free(c, x["payer"]))
                if n == 0:
                    continue
                cites = [(STATEMENT, f["digest"])]
                cites += [(ACK, a["digest"]) for a in f["acks"]
                          if a["mode"] == 1 and any(e["class"] == c and not free(c, e["grant"]) for e in a["entries"])]
                add(DELIVERY, st["party"], c, n, cites)
        elif f["kind"] == PACKAGE and t["version"] in SPENDING:          # §4.7: the product held, its storage alone
            if f["held"] is not None:
                add(PACKAGE, f["signer"], 3, f["stored"] * f["held"], [(PACK, f["digest"])])   # byte-microseconds
        elif f["kind"] == PACKAGE:
            cite = [(PACK, f["digest"])]
            add(PACKAGE, f["signer"], 1, 1, cite)
            add(PACKAGE, f["signer"], 2, f["entries"], cite)
            add(PACKAGE, f["signer"], 3, f["stored"], cite)
        elif f["kind"] == STITCHED:
            for r in f["records"]:
                cites = [(POD_INPUT, r["digest"])] + [(STITCHED_CLAIM, c) for _, cs in r["ads"] for c in cs]
                add(ADS, r["signer"], 1, sum(p for p, _ in r["ads"]), cites)
        elif f["kind"] == JOB:                                                       # §6.3, §6.5
            if f["billed"] is None:
                continue
            cap, hop = f["billed"]
            for k, (pcap, pop, unit, _, _, pend, pusage) in enumerate(t["job"]):
                if pcap != cap or (t["version"] in WORK_VERSIONS and pop != f["operation"] and "operation" not in SKIP):
                    continue
                if pend and pend != f["endpoint"] and "endpoint" not in SKIP:    # §4.9
                    continue
                cites = [(ROUTE_RECORD, f["digest"]), (OER_RECEIPT, hop)]
                if unit == 1:
                    n = 1
                elif unit == 2:
                    cites += [(VIDEO_RECEIPT, h) for h, w in f["outputs"] if w > 0]
                    cites += [(UPSCALE_RECORD, h) for h, w in f["upscales"] if w > 0]
                    cites += [(RENDITION_RECORD, h) for h, w in f["renditions"] if w > 0]
                    n = sum(w for _, w in f["outputs"] + f["upscales"] + f["renditions"])
                elif unit == 3:                             # §4.7: the work units of its packages
                    cites += [(PACK, h) for h, w in f["packages"] if w > 0]
                    n = sum(w for _, w in f["packages"])
                else:                                       # §4.9: what the billed hop's record states
                    u = f["usage"]
                    if u is None or (u[0] != pusage and "usage-unit" not in SKIP):
                        n = 0
                    else:
                        n = u[1]
                        cites.append((THIRD_PARTY, u[2]))
                add(JOB, f["router"], k + 1, n, cites)
        else:
            unit = 1 if f["claims_required"] else 2
            for r in f["receipts"]:
                counted = [i for i in r["impressions"] if i["counts"]]
                cites = [(AD_RECEIPT, r["identity"])] + [(CLAIM, i["claim"]) for i in counted if i["claim"] is not None]
                add(ADS, r["decider"], unit, len(counted), cites)
    lines, total = [], 0
    for (kind, source, unit) in sorted(q):
        n, cited = q[(kind, source, unit)]
        if t["version"] in SPENDING and kind == PACKAGE and unit == 3:
            n //= DAY_US                           # §4.7: byte-days, rounded down once per line
            if n == 0:
                continue
        p = price(t, kind, unit)
        a = n * p // divisor(t, kind, unit)
        if n > U64 or a > U64:
            return None
        total += a
        lines.append({"kind": kind, "source": source, "unit": unit, "quantity": n, "price": p, "amount": a,
                      "cited": len(cited), "cites": cites_digest(cited), "cited_records": sorted(cited)})
    figs, seen = [], set()
    for f in facts:
        if f["kind"] == ROOM:
            for d, fig in f["hops"]:
                if d not in seen:
                    seen.add(d)
                    figs.append((1, fig))
    for f in facts:
        if f["kind"] == DELIVERY:
            for d, fig in f["hops"]:
                if d not in seen:
                    seen.add(d)
                    figs.append((2, fig))
    for f in facts:
        if f["kind"] == PACKAGE and t["version"] not in SPENDING:   # §6.6: in version 4, its run's hop states it
            figs.append((3, f["figure"]))
        elif f["kind"] == ADS:
            figs += [(4, r["figure"]) for r in f["receipts"]]
        elif f["kind"] == JOB:
            if f["figure"] is not None:
                figs.append((5, f["figure"]))
            figs += [(6, x) for x in f["unbilled"]]
    rows, unmetered, sums = rows_of(figs)
    if rows is None or total > U64:
        return None
    return {"records": sorted(records), "lines": lines, "energy": rows, "unmetered": unmetered, "amount": total, "sums": sums}


def rows_of(figs):
    unmetered = [0, 0, 0, 0, 0, 0]
    acc = {}
    for s, (tag, joules, ppm) in figs:
        if tag is None:
            unmetered[s - 1] += 1
            continue
        e = acc.setdefault((s, tag), [0, 0, 0])
        e[0] += 1
        e[1] += joules
        e[2] = max(e[2], ppm)
    rows = [(s, tag) + tuple(acc[(s, tag)]) for s, tag in sorted(acc)]
    if any(r[3] > U64 for r in rows):
        return None, unmetered, {}
    return rows, unmetered, {k: v[1] for k, v in acc.items()}


def summed_across(row, sums):
    """§6.6: a row whose joules sum its class and another class of its scope, not its own."""
    s, tag, _, joules, _ = row
    own = sums.get((s, tag))
    if own is None or own == joules:
        return False
    others = [j for (sc, tg), j in sums.items() if sc == s and tg != tag]
    for m in range(1, 1 << len(others)):
        if own + sum(j for i, j in enumerate(others) if m >> i & 1) == joules:
            return True
    return False


def check_figures(inv, c):
    """§7.2 step 7, in its order."""
    got = {(x["kind"], x["source"], x["unit"]): x for x in c["lines"]}
    for x in inv["lines"]:
        g = got.get((x["kind"], x["source"], x["unit"]))
        if "quantity" not in SKIP and (g is None or x["quantity"] > g["quantity"]):
            return "quantity_not_recorded"
    if "summed" not in SKIP and any(summed_across(r, c["sums"]) for r in inv["energy"]):
        return "classes_summed"
    for x in inv["lines"]:
        g = got.get((x["kind"], x["source"], x["unit"]))
        if "citations" not in SKIP and g is not None and (x["cited"] != g["cited"] or x["cites"] != g["cites"]):
            return "citation_mismatch"
    strip = [{k: v for k, v in x.items() if k != "cited_records"} for x in c["lines"]]
    if "mismatch" not in SKIP and (inv["records"] != c["records"] or inv["lines"] != strip or inv["energy"] != c["energy"]
                                   or inv["unmetered"] != c["unmetered"] or inv["amount"] != c["amount"]):
        return "invoice_mismatch"
    return None


# ---------------------------------------------------------------- §8 the reports

def rows_json(rows):
    return [{"class": LABELS[r[1]], "figures": r[2], "joules": r[3], "max_ppm": r[4]} for r in rows]


def ads_report(f):
    receipts, nc = [], {}
    tot = {"b": 0, "c": 0, "bs": 0, "cs": 0}
    for r in f["receipts"]:
        b = sum(1 for i in r["impressions"] if i["billable"])
        c = sum(1 for i in r["impressions"] if i["counts"])
        for i in r["impressions"]:
            if i["billable"] and not i["counts"]:
                nc[i["verdict"]] = nc.get(i["verdict"], 0) + 1
        tot["b"] += b
        tot["c"] += c
        tot["bs"] += r["spend"]
        tot["cs"] += r["counted_spend"]
        receipts.append({
            "identity": r["identity"].hex(), "decider": r["decider"].hex(), "cue": r["cue"].hex(), "session": r["session"].hex(),
            "impressions": [{k: i[k] for k in ("slot", "ad_id", "price_micros", "quartiles_reached", "billable", "counts", "verdict")}
                            for i in r["impressions"]],
            "billable_impressions": b, "billable_spend_micros": r["spend"], "counted_impressions": c,
            "counted_spend_micros": r["counted_spend"]})
    return {"room": f["room"].hex(), "claims_required": f["claims_required"], "receipts": receipts,
            "billable_impressions": tot["b"], "counted_impressions": tot["c"], "billable_spend_micros": min(tot["bs"], U64),
            "counted_spend_micros": min(tot["cs"], U64), "not_counted": dict(sorted(nc.items()))}


def overlaps_of(facts):
    """§8.2: by (content hash, downstream, first party, first record, second party, second record),
    the bytes their items share."""
    acks = [a for f in facts if f["kind"] == DELIVERY for a in f["acks"] if a["mode"] == 1]
    out = {}
    for x in acks:
        for y in acks:
            if not (x["upstream"] < y["upstream"] and x["downstream"] == y["downstream"]
                    and overlapping((x["start"], x["end"]), (y["start"], y["end"]))):
                continue
            for a in [i for l in x["item_lists"] for i in l]:
                for b in [i for l in y["item_lists"] for i in l]:
                    if a[0] == b[0] and overlapping(a[1:], b[1:]):
                        k = (a[0], x["downstream"], x["upstream"], x["digest"], y["upstream"], y["digest"])
                        out[k] = min(out.get(k, 0) + min(a[2], b[2]) - max(a[1], b[1]), U64)
    return out


def delivery_report(facts, c):
    parties = {}
    for f in facts:
        if f["kind"] == DELIVERY:
            parties.setdefault(f["statement"]["party"], []).append(f)
    rows = []
    for party in sorted(parties):
        fs = parties[party]
        classes = [[0, 0] for _ in range(4)]
        free_bytes = stated = 0
        declared = [0, 0, 0]
        figs, seen = [], set()
        currency = next((f["statement"]["currency"] for f in fs if f["statement"]["currency"] != bytes(3)), None)
        for f in fs:
            st = f["statement"]
            for k in range(4):
                classes[k][0] += st["classes"][k][0]
                classes[k][1] += st["classes"][k][1]
            free_bytes += sum(x["bytes"] for x in st["lines"] if x["kind"] == 1 and free(x["class"], x["payer"]))
            stated += st["amount"]
            for k in range(3):
                declared[k] += st["unack"][k]
            for d, fig in f["hops"]:
                if d not in seen:
                    seen.add(d)
                    figs.append((2, fig))
        energy, unmetered, _ = rows_of(figs)
        rows.append({
            "party": party.hex(), "statements": [f["digest"].hex() for f in fs],
            "currency": None if currency is None else currency.decode("ascii"),
            "classes": {CLASS_NAMES[k]: {"objects": classes[k][0], "bytes": classes[k][1]} for k in range(4)},
            "free_bytes": free_bytes,
            "declared": {"declarations": declared[0], "objects": declared[1], "bytes": declared[2]},
            "stated_amount": stated,
            "invoiced_amount": sum(x["amount"] for x in c["lines"] if x["kind"] == DELIVERY and x["source"] == party),
            "energy": rows_json(energy or []), "unmetered": unmetered[1]})
    ov = overlaps_of(facts)
    overlaps = [{"content_hash": h.hex(), "downstream": d.hex(), "bytes": ov[(h, d, p1, r1, p2, r2)],
                 "first": {"party": p1.hex(), "record": r1.hex()}, "second": {"party": p2.hex(), "record": r2.hex()}}
                for (h, d, p1, r1, p2, r2) in sorted(ov)]
    return {"parties": rows, "overlaps": overlaps}


# ---------------------------------------------------------------- §7 verification

def unhex_field(doc, k):
    v = doc.get(k)
    if not isinstance(v, str) or not re.fullmatch(r"(?:[0-9a-f]{2})*", v):
        raise Refuse("malformed", [k])
    return bytes.fromhex(v)


def verify(doc):
    """§7.2: the outcome of a verification document."""
    try:
        tb, ib = unhex_field(doc, "tariff"), unhex_field(doc, "invoice")
        pb = None if doc.get("previous") is None else unhex_field(doc, "previous")
        sources = doc.get("sources")
        if not isinstance(sources, list):
            raise Refuse("malformed", ["source"])
        t = read_tariff(tb)                                                          # step 1
        inv = read_invoice(ib, ["invoice"])                                          # step 2
        if "binding" not in SKIP and (inv["tariff"] != t["digest"] or inv["seller"] != t["seller"] or inv["version"] != t["version"]
                                      or inv["buyer"] != t["buyer"] or inv["currency"] != t["currency"]):   # step 3
            raise Refuse("tariff_mismatch", ["invoice"])
        if "outside-tariff" not in SKIP and not (t["start"] <= inv["start"] and inv["end"] <= t["end"]):
            raise Refuse("outside_tariff", ["invoice"])
        prev = None
        if inv["prev"] != ZERO:                                                      # step 4
            if "previous-given" not in SKIP and (pb is None or sha(pb[:-64]) != inv["prev"]):
                raise Refuse("record_missing", ["previous"])
            if pb is not None:
                prev = read_invoice(pb, ["previous"])
                if prev["tariff"] != inv["tariff"] or prev["seller"] != inv["seller"] or prev["buyer"] != inv["buyer"]:
                    raise Refuse("tariff_mismatch", ["previous"])
                if "previous-overlap" not in SKIP and prev["end"] > inv["start"]:
                    raise Refuse("invoice_overlap", ["previous"])
        facts = []
        for k, s in enumerate(sources):                                              # step 5
            try:
                f = verify_source(s, [i for k, i in t["scope"] if k == 4], t["version"])
            except Refuse as e:
                raise Refuse(e.reason, ["source", k] + list(e.at))
            bad = check_source(t, inv["start"], inv["end"], f)
            if bad:
                raise Refuse(bad, ["source", k])
            facts.append(f)
        across = check_across(facts, prev, t["version"])                             # step 6
        if across:
            raise Refuse(across[1], ["source", across[0]])
        hold(t, facts, prev, inv["start"], inv["end"])
        c = compute(t, facts)                                                        # step 7
        if c is None:
            raise Refuse("invoice_mismatch", ["invoice"])
        bad = check_figures(inv, c)
        if bad:
            raise Refuse(bad, ["invoice"])
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}
    return accepted(inv, facts, c, t)


def line_json(x, t):
    """§7.3: a line, a job line naming its job price's unit and capability."""
    out = {"source_kind": KIND_NAMES[x["kind"]], "source": x["source"].hex(),
           "unit": JOB_UNITS[t["job"][x["unit"] - 1][2] - 1] if x["kind"] == JOB else units(t["version"])[x["kind"]][x["unit"] - 1][0],
           "quantity": x["quantity"], "price": x["price"], "amount": x["amount"],
           "cites": [[k, i.hex()] for k, i in x["cited_records"]]}
    if x["kind"] == JOB:
        out["capability"] = t["job"][x["unit"] - 1][0].decode("utf-8")
        if t["version"] in WORK_VERSIONS:
            out["operation"] = t["job"][x["unit"] - 1][1].decode("utf-8")
        if t["version"] == V6:                         # §7.3: its endpoint, empty for any; a usage's unit
            out["endpoint"] = t["job"][x["unit"] - 1][5].decode("utf-8")
            if t["job"][x["unit"] - 1][2] == USAGE:
                out["usage_unit"] = t["job"][x["unit"] - 1][6].decode("utf-8")
    return out


def accepted(inv, facts, c, t):
    unmetered, coverage = {}, {}
    for s in range(1, scopes(inv["version"]) + 1):
        n = inv["unmetered"][s - 1]
        unmetered[SCOPES[s]] = n
        coverage[SCOPES[s]] = "partial" if n > 0 else "whole" if any(r[0] == s for r in inv["energy"]) else "none"
    return {
        "outcome": "accept",
        "digest": inv["digest"].hex(),
        "tariff": inv["tariff"].hex(),
        "currency": inv["currency"].decode("ascii"),
        "start": inv["start"],
        "end": inv["end"],
        "records": len(inv["records"]),
        "lines": [line_json(x, t) for x in c["lines"]],
        "energy": [{"scope": SCOPES[r[0]], "class": LABELS[r[1]], "figures": r[2], "joules": r[3], "max_ppm": r[4]}
                   for r in inv["energy"]],
        "unmetered": unmetered,
        "coverage": coverage,
        "amount": inv["amount"],
        "reports": {"ads": [ads_report(f) for f in facts if f["kind"] == ADS],
                    "stitched": [f["statement"] for f in facts if f["kind"] == STITCHED],
                    "delivery": delivery_report(facts, c)},
    }


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

def document(doc, v):
    return {"tariff": v["tariff"], "invoice": v["invoice"], "previous": v["previous"],
            "sources": [doc["sources"][n] for n in v["sources"]]}


def check(doc, quiet=False):
    """The failures: every vector whose stated outcome is not the one this verifier reaches."""
    fails = []
    for v in doc.get("vectors", []):
        try:
            got = verify(document(doc, v))
        except Exception as e:  # a crash on a mutated copy is a failure to verify it
            got = {"outcome": "crash", "error": "%s: %s" % (type(e).__name__, e)}
        ok = got == v.get("outcome")
        if not quiet:
            what = got["outcome"] + (" %s at %s" % (got["reason"], got["at"]) if got["outcome"] == "refuse" else "")
            print(("PASS" if ok else "FAIL") + " %s: %s" % (v.get("name"), what))
        if not ok:
            fails.append(v.get("name"))
    return fails


def reasons_listed():
    """§10's table, in order."""
    text = open(EXTENSION, encoding="utf-8").read()
    at = text.index("\n## 10. Reasons")
    end = text.find("\n## ", at + 1)
    return [m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", text[at:end], re.M)]


def mutations(doc):
    """Copies of the corpus, one defect each, that a verifier must reject."""
    def flip_hex(h, at):
        b = bytearray(bytes.fromhex(h))
        b[at % len(b)] ^= 0x01
        return b.hex()
    out = []
    for k, v in enumerate(doc["vectors"]):
        if v["outcome"]["outcome"] != "accept":
            continue
        for field, at in (("tariff", 6), ("tariff", -1), ("invoice", 40), ("invoice", -70), ("invoice", -1)):
            m = copy.deepcopy(doc)
            m["vectors"][k][field] = flip_hex(v[field], at)
            out.append(("%s: %s byte %d flipped" % (v["name"], field, at), m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["sources"] = v["sources"][:-1]
        out.append(("%s: its last source dropped" % v["name"], m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["amount"] += 1
        out.append(("%s: its stated amount raised" % v["name"], m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][0]["cites"] = []
        out.append(("%s: its first line's citations dropped" % v["name"], m))
    m = copy.deepcopy(doc)
    rep = m["vectors"][0]["outcome"]["reports"]
    rep["delivery"]["overlaps"] = []
    out.append(("a_day: its overlaps dropped", m))
    m = copy.deepcopy(doc)
    m["vectors"][0]["outcome"]["reports"]["ads"][0]["not_counted"] = {}
    out.append(("a_day: its ad report's not-counted figures dropped", m))
    if "job_int_motion" in doc["sources"]:
        m = copy.deepcopy(doc)
        m["sources"]["job_int_motion"]["outputs"] = []
        out.append(("jobs: the int_motion job's output receipt dropped", m))
        m = copy.deepcopy(doc)
        k = next(i for i, v in enumerate(doc["vectors"]) if v["name"] == "jobs_routed_and_served")
        m["vectors"][k]["outcome"]["coverage"]["jobs"] = "whole"
        out.append(("jobs_routed_and_served: its jobs' coverage stated whole", m))
    names = {v["name"]: i for i, v in enumerate(doc["vectors"])}
    if "jobs_priced_per_operation" in names:                                         # version 3
        k = names["jobs_priced_per_operation"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        dec = next(i for i, x in enumerate(lines) if x.get("operation") == "media.decode" and x["unit"] == "subpixels")
        enc = next(i for i, x in enumerate(lines) if x.get("operation") == "media.encode" and x["unit"] == "subpixels")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][dec]["price"] = lines[enc]["price"]
        out.append(("jobs_priced_per_operation: the decode priced as the encode", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][dec]["operation"] = "media.encode"
        out.append(("jobs_priced_per_operation: the decode line named an encode", m))
        k = names["one_output_under_two_hops"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        sp = next(i for i, x in enumerate(lines) if x["unit"] == "subpixels")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][sp]["quantity"] //= 2
        out.append(("one_output_under_two_hops: the shared output's subpixels billed once", m))
        k = names["a_rerun_billed"]
        m = copy.deepcopy(doc)
        m["vectors"][k]["previous"] = None
        out.append(("a_rerun_billed: the invoice before it not given", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["sources"] = ["job_rerun_replays_its_hop"]
        out.append(("a_rerun_billed: its re-run replaced by one replaying the billed hop", m))
    if "version_4_by_the_byte_and_the_work" in names:                               # version 4, §4.7, §6.6
        k = names["version_4_by_the_byte_and_the_work"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        bo = next(i for i, x in enumerate(lines) if x["unit"] == "bytes_out")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][bo]["quantity"] -= 1
        out.append(("version_4_by_the_byte_and_the_work: its bytes out stated one less", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["energy"] = [e for e in doc["vectors"][k]["outcome"]["energy"] if e["scope"] != "unbilled_hops"]
        out.append(("version_4_by_the_byte_and_the_work: the unbilled hops' rows dropped", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["coverage"]["unbilled_hops"] = "whole"
        out.append(("version_4_by_the_byte_and_the_work: the unbilled hops' coverage stated whole", m))
        wu = next(i for i, x in enumerate(lines) if x["unit"] == "work_units")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][wu]["unit"] = "entries"
        out.append(("version_4_by_the_byte_and_the_work: its work units named entries", m))
        k = names["version_4_storage_held_through_a_period"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        bd = next(i for i, x in enumerate(lines) if x["unit"] == "byte_days")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][bd]["quantity"] += 1
        out.append(("version_4_storage_held_through_a_period: a byte-day more", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["previous"] = None
        out.append(("version_4_storage_held_through_a_period: the invoice before it not given", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_v2_encode"]["outputs"] = doc["sources"]["job_v2_another_output"]["outputs"]
        out.append(("job_v2_encode: its output replaced by another clip's receipt", m))
    if "job_v5_decode" in doc["sources"]:                         # §4.8: a decode held to its frames
        for other, what in (("job_v5_frames_swapped", "its first two frames swapped"), ("job_v5_a_frame_altered", "a frame with one sample changed"),
                            ("job_v5_a_frame_missing", "its last frame left out")):
            m = copy.deepcopy(doc)
            m["sources"]["job_v5_decode"]["outputs"][0]["frames"] = doc["sources"][other]["outputs"][0]["frames"]
            out.append((f"job_v5_decode: {what}", m))
    if "job_tp_alpha" in doc["sources"]:                          # §4.9: what a third party bills is signed
        m = copy.deepcopy(doc)
        m["sources"]["job_tp_alpha"]["third_party"] = doc["sources"]["job_tp_beta"]["third_party"]
        out.append(("job_tp_alpha: another job's record", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_tp_failover"]["third_party"] = doc["sources"]["job_tp_failover"]["third_party"][1:]
        out.append(("job_tp_failover: its failed hop's record left out", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_tp_alpha"]["third_party"][0] = doc["sources"]["job_tp_alpha"]["third_party"][0].replace('"quantity":8', '"quantity":80')
        out.append(("job_tp_alpha: its record's usage raised, its signature unchanged", m))
    if "job_upscale" in doc["sources"]:                           # §4.9: re-derived, and of what the hop delivered
        m = copy.deepcopy(doc)
        o = m["sources"]["job_upscale"]["outputs"]
        o[0]["picture"], o[1]["picture"] = o[1]["picture"], o[0]["picture"]
        out.append(("job_upscale: its two pictures swapped", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_upscale"]["outputs"][0]["object"] = doc["sources"]["job_upscale"]["outputs"][1]["object"]
        out.append(("job_upscale: its first record given the second's input", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_upscale"]["outputs"][0]["proof"]["index"] = 1
        out.append(("job_upscale: its first picture proved at the second place", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_tonemap"]["outputs"][0]["proofs"][0]["index"] = 1
        out.append(("job_tonemap: its first rendition proved at the second place", m))
        m = copy.deepcopy(doc)
        r = m["sources"]["job_tonemap"]["outputs"][0]["renditions"]
        r[0], r[1] = r[1], r[0]
        out.append(("job_tonemap: its first two renditions swapped", m))
    if "job_v3r_encode" in doc["sources"]:                        # §5.1: an output proved at its place
        m = copy.deepcopy(doc)
        m["sources"]["job_v3r_encode"]["outputs"][0]["proof"]["index"] = 0
        out.append(("job_v3r_encode: its proof stating another place", m))
        m = copy.deepcopy(doc)
        m["sources"]["job_v3r_encode"]["outputs"][0]["proof"]["siblings"][0] = "00" * 32
        out.append(("job_v3r_encode: its proof's sibling replaced", m))
        m = copy.deepcopy(doc)
        m["sources"]["package_run_v3r"]["outputs"][0]["proof"]["index"] = 1
        out.append(("package_run_v3r: its receipt proved at the second place", m))
    for k, v in enumerate(doc["vectors"]):
        if v["outcome"]["outcome"] == "refuse":
            m = copy.deepcopy(doc)
            m["vectors"][k]["outcome"]["reason"] = "invoice_mismatch" if v["outcome"]["reason"] != "invoice_mismatch" else "malformed"
            out.append(("%s: its stated reason changed" % v["name"], m))
            break
    return out


CHECKS = ["signatures", "buyer-signature", "tariff-rules", "invoice-rules", "row-of-no-class", "binding", "outside-tariff",
          "previous-given", "previous-overlap", "scope", "conflict", "period", "duplicate", "overlap", "rebilled", "quantity",
          "summed", "citations", "mismatch", "job-figure", "output", "job-duplicate", "private-kind", "operation", "output-once",
          "output-delivered", "unbilled-hops", "storage-held", "package-again", "run-work", "bytes-out", "conflict-bytes",
          "decode-frames", "output-proof", "price-endpoint", "price-usage", "price-order", "any-beside-one",
          "kind-version", "third-party-version", "form-version", "billed-usage", "endpoint", "usage-unit", "evidence-again",
          "upscale-input", "upscale-proof", "upscale-picture", "tonemap-proof", "tonemap-count", "tonemap-renditions"]


def main(corpus_dir):
    doc = json.load(open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8"))
    fails = check(doc)
    files = sorted(os.listdir(corpus_dir))
    if files != ["README.md", "vectors.json"]:
        fails.append("the corpus holds other files: %s" % files)
    seen = {v["outcome"].get("reason") for v in doc["vectors"]}
    listed = reasons_listed()
    missing = [r for r in listed + ["truncated", "malformed", "unknown_version", "bad_key", "bad_signature", "record_missing",
                                     "classes_summed"] if r not in seen]
    if missing:
        fails.append("no vector is refused %s" % missing)
    print("%s every reason of §10 refuses a vector (%d listed)" % ("PASS" if not missing else "FAIL", len(listed)))
    weak = []
    for name, m in mutations(doc):
        if not check(m, quiet=True):
            weak.append(name)
    print("%s every mutated copy of the corpus fails (%d copies)" % ("PASS" if not weak else "FAIL", len(mutations(doc))))
    fails += ["the mutated copy %r verifies" % w for w in weak]
    undetected = []
    for c in CHECKS:
        SKIP.add(c)
        try:
            if not check(doc, quiet=True):
                undetected.append(c)
        finally:
            SKIP.discard(c)
    print("%s the corpus catches the verifier with each of its checks dropped (%d checks)" % ("PASS" if not undetected else "FAIL", len(CHECKS)))
    fails += ["dropping the check %r goes unnoticed" % c for c in undetected]
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


def check_document(path):
    """One verification document (§7.1), such as a ledger service serves with an invoice: its
    outcome, printed. Status 0 only when it is accepted."""
    doc = json.load(open(path, encoding="utf-8"))
    got = verify(doc)
    print(json.dumps(got, indent=1))
    return 0 if got["outcome"] == "accept" else 1


if __name__ == "__main__":
    if len(sys.argv) == 3 and sys.argv[1] == "--document":
        sys.exit(check_document(sys.argv[2]))
    if len(sys.argv) != 2:
        print("usage: receipt_ledger_verify.py <live-conformance/ledger> | --document FILE", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))