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);
and, under a tariff of version 7, a group by ../jwp-conformance/jwp_verify.py (jwp-receipts §3),
its publisher's declarations and schedules by ../claim-conformance/claim_verify.py (jwp-receipts
§3.1 steps 1 to 5, the octets a schedule's entry sends) and carousel_verify.py (step 6, carousel
§4.3 and bearer-receipts §4.3), a node's claims of its own instance by cache_authority_verify.py
(cache-authority §5.3) and its settlement declarations by settlement_verify.py (settlement §3.6),
an aggregate receipt by telemetry_aggregate_verify.py's reader (telemetry-aggregation §7.2, §6.1)
and an agent's consumption claims by claim_verify.py. 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);
- in version 7, its transmission prices per bearer class, unit and ceiling (§4.10), the
  declarations and schedules of a group that stand (carousel §6.2 steps 3 and 4), a node's own
  instances and what bills them, each entry's pass rate and the price whose ceiling is the least
  at or above it, and the transmissions' report (§8.3); an aggregate receipt's collector and
  count (§4.11); and an agent's consumption claims, each object one unit (§4.12);
- 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
import carousel_verify as CAR  # noqa: E402  (it puts ../claim-conformance and ../jwp-conformance on the path)
import cache_authority_verify as CAV  # noqa: E402
import telemetry_aggregate_verify as TAV  # noqa: E402
import claim_verify as CV  # noqa: E402
import jwp_verify as JV  # 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)
GROUP, AGGREGATED, CONSUMPTION = 6, 7, 8                          # §6.3: version 7
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
GROUP_RECEIPT, TRANSMISSION_CLAIM, SCHEDULE_CLAIM = 20, 21, 22    # §5.2, version 7
CACHE_CLAIM, AGGREGATE_RECEIPT, CONSUMPTION_CLAIM = 23, 24, 25    # §5.2, version 7
PRIVATE = 128                                                    # §5.2: 128 to 255, private and provisional use
KIND_NAMES = {ROOM: "room", DELIVERY: "delivery", PACKAGE: "package", ADS: "ads", JOB: "job", GROUP: "group",
              AGGREGATED: "aggregated", CONSUMPTION: "consumption"}
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",
          7: "transmissions"}                                    # §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, 7)                                # §4.5 to §4.10: the versions that price jobs
WORK_VERSIONS = (3, 4, 5, 6, 7)                                  # §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
V7 = 7                                                           # §4.10: version 6, and a transmission per class, unit and ceiling
SPENDING = (V4, V5, V6, V7)                                      # §4.7 to §4.10: the versions that bill what the work spends
DECODE_FRAMES = (V5, V6, V7)                                     # §4.8 to §4.10: a decode held to its frames
ENDPOINTS = (V6, V7)                                             # §4.9, §4.10: per endpoint, by a usage, and re-derived jobs
TRANSMISSION_UNITS = [("bytes_sent", 10 ** 9), ("pass_us", 10 ** 6)]   # §4.10: carousel §5.4's units and their divisors
BEARERS = {1: "bidirectional", 2: "multicast", 3: "unidirectional"}    # §4.10: bearer-receipts §2's classes, in order
MAX_SAFE = 2 ** 53 - 1                                           # §4.10: a ceiling's and a price's bound (rung-price §4.2, §4.3)
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
UNITS_7 = dict(UNITS_4)                                          # §4.11, §4.12, §6.3: version 7's
UNITS_7[AGGREGATED] = [("aggregated_impressions", 1)]
UNITS_7[CONSUMPTION] = [("consumed_objects", 1)]


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


def scopes(version):
    """§6.1: the scopes an invoice of `version` states."""
    return 7 if version == V7 else 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, 7):
        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"] in ENDPOINTS else b""
                unit = r.u(1)
                usage = r.take(r.u(1)) if t["version"] in ENDPOINTS else b""
                t["job"].append((cap, op, unit, r.u(8), r.u(8), endpoint, usage))
        t["transmission"] = []                           # (bearer, unit, max_bps, price)
        t["aggregated"] = t["consumption"] = 0
        if t["version"] == V7:                                                       # §4.10 to §4.12
            t["transmission"] = [(r.u(1), r.u(1), r.u(8), r.u(8)) for _ in range(r.u(1))]
            t["aggregated"], t["consumption"] = r.u(8), r.u(8)
        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"] in ENDPOINTS 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"])
    def transmission_price(p):
        """§4.10: a shared bearer's class, a unit of the two, a ceiling of 1 to 2^53 − 1, a price within it."""
        return p[0] in (2, 3) and p[1] in (1, 2) and 1 <= p[2] <= MAX_SAFE and p[3] <= MAX_SAFE
    top = CONSUMPTION if t["version"] == V7 else 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 ("transmission-rules" in SKIP or (all(transmission_price(p) for p in t["transmission"])
                                                   and ascending([p[:3] for p in t["transmission"]])))
             and ("unit-price-bound" in SKIP or (t["aggregated"] <= MAX_SAFE and t["consumption"] <= MAX_SAFE))
             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]
    if kind == GROUP:
        return t["transmission"][unit - 1][3]
    if kind == AGGREGATED:
        return t["aggregated"]
    if kind == CONSUMPTION:
        return t["consumption"]
    return {ROOM: t["room"], DELIVERY: t["delivery"], PACKAGE: t["package"], ADS: t["ads"]}[kind][unit - 1]


def divisor(t, kind, unit):
    if kind == GROUP:
        return TRANSMISSION_UNITS[t["transmission"][unit - 1][1] - 1][1]
    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] * (7 - 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
        if x["kind"] == GROUP:
            return (i["version"] == V7 or "kind-version" in SKIP) 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"] in ENDPOINTS or "kind-version" in SKIP) and THIRD_PARTY <= k <= RENDITION_RECORD)
                     or ((i["version"] == V7 or "kind-version" in SKIP) and GROUP_RECEIPT <= k <= CONSUMPTION_CLAIM) 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 in ENDPOINTS 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 in ENDPOINTS 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}


def tier(t, bearer, unit, rate):
    """§4.10: the price, from 1, of `bearer`'s class and `unit` whose ceiling is the least at or
    above `rate`; None when the rate is above every one."""
    fits = [(p[2], k + 1) for k, p in enumerate(t["transmission"]) if p[0] == bearer and p[1] == unit and p[2] >= rate]
    if not fits:
        return None
    return max(fits)[1] if "tier" in SKIP else min(fits)[1]


def days_from_civil(y, mo, d):
    """Days from 1970-01-01 to a date of the proleptic Gregorian calendar, for every year RFC 3339
    writes, 0000 among them, in integers (a March-based year, so a leap day is its year's last)."""
    y -= mo <= 2
    era = y // 400
    yoe = y - era * 400
    doy = (153 * (mo - 3 if mo > 2 else mo + 9) + 2) // 5 + d - 1
    return era * 146097 + yoe * 365 + yoe // 4 - yoe // 100 + doy - 719468


def rfc3339_seconds(s):
    """jwp-receipts §2.4.3's date-time: seconds, its offset applied and any fraction dropped."""
    m = re.match(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.[0-9]+)?(?:[Zz]|([+-])([0-9]{2}):([0-9]{2}))\Z", s)
    y, mo, d, h, mi, sec = (int(x) for x in m.groups()[:6])
    t = days_from_civil(y, mo, d) * 86400 + h * 3600 + mi * 60 + sec
    if m.group(7):
        off = int(m.group(8)) * 3600 + int(m.group(9)) * 60
        t += -off if m.group(7) == "+" else off
    return t


def held_parent(s):
    """§4.10, §4.12: `parent`, jwp-receipts §3.1 steps 1 to 4, held as its signer's previous claim."""
    held = {}
    if s["parent"] is not None:
        try:
            p = CV.verify_claim(s["parent"])
        except CV.Reject as r:
            raise Refuse(r.reason, ["parent"])
        held[p["signer_pubkey"]] = p
    return held


def group_source(s, cache=False):
    """§5.1, §4.10: a group, the object receipts given under it, and its publisher's transmission
    and schedule claims in their chain's order, `parent` held before the first: verified as carousel
    §6.2 steps 1 to 4 verify them, and refused at the first group, object, parent or claim that is
    refused, or the first claim set aside. A cache source is a node's own instances of that group:
    its cache-transmission and schedule claims, as cache-authority §7.2 steps 1, 2 and 4 verify them,
    with its settlement declarations."""
    members = {"kind", "group", "objects", "claims", "parent"} | ({"declarations"} if cache else set())
    if (set(s) != members or not isinstance(s["objects"], list)
            or not isinstance(s["claims"], list) or not all(isinstance(c, str) for c in s["claims"])
            or not (s["parent"] is None or isinstance(s["parent"], str)) or (cache and not s["claims"])):
        raise Refuse("malformed", [])
    try:                                                         # jwp-receipts §3 steps 1 and 2
        g = JV.read_group(s["group"])
        JV.verify_group(g)
    except JV.Refuse as r:
        raise Refuse(r.reason, ["group"])
    objs = []
    for i, x in enumerate(s["objects"]):
        if not isinstance(x, dict) or set(x) != {"receipt", "envelope"} or not (x["envelope"] is None or is_hex(x["envelope"])):
            raise Refuse("malformed", [])
        env = None if x["envelope"] is None else bytes.fromhex(x["envelope"])
        try:
            o = JV.read_object(x["receipt"])
            if env is None:
                JV.inclusion(o, g)
            else:
                JV.accept(o, g, env)
        except JV.Refuse as r:
            raise Refuse(r.reason, ["objects", i])
        objs.append((o, env))
    held = held_parent(s)
    claims = []
    for i, w in enumerate(s["claims"]):                          # carousel §6.2 step 2
        try:
            c = CV.verify_claim(w)
            previous = held.get(c["signer_pubkey"])
            held[c["signer_pubkey"]] = c
            if "group-chain" not in SKIP:
                CV.check_chain(c, previous)
            one_node = not claims or claims[0]["signer_pubkey"] == c["signer_pubkey"] or "cache-one-node" in SKIP
            if not cache and c["class"] == "transmission":
                CAR.check_transmission(c, g)
            elif not cache and c["class"] == "schedule":
                CAR.check_schedule(c, g, objs)
            elif cache and one_node and c["class"] == "cache-transmission":
                CAV.check_cache_transmission(c, g, [o for o, _ in objs])
            elif cache and one_node and c["class"] == "schedule" and c["signer_pubkey"] != g["signer_pubkey"]:
                CAR.check_schedule(c, g, objs, node=True)
            else:
                raise CV.Reject(6, "context_mismatch")
        except CV.Reject as r:
            raise Refuse(r.reason, ["claims", i])
        claims.append(c)

    declaring = "cache-transmission" if cache else "transmission"

    def by_instance(cls):
        out = {}
        for i, c in enumerate(claims):
            if c["class"] == cls:
                out.setdefault(c["body"]["bearer_id"], []).append(i)
        return out
    aside, declared, contradicted = {}, {}, set()                # carousel §6.2 steps 3 and 4
    for bid, idx in by_instance(declaring).items():
        bodies = [claims[i]["body"] for i in idx]
        if all(b == bodies[0] for b in bodies):
            declared[bid] = ({"bidirectional": 1, "multicast": 2, "unidirectional": 3}[bodies[0]["bearer"]], idx)
        else:
            contradicted.add(bid)
            aside.update((i, "bearer_contradicted") for i in idx)
    links = {bytes.fromhex(CV.link_hash(c)) for c in claims if c["class"] == declaring} if cache else set()
    acks = []
    if cache:                                                    # settlement §4.4
        if not isinstance(s["declarations"], list) or not all(is_hex(x) for x in s["declarations"]):
            raise Refuse("malformed", [])
        for i, x in enumerate(s["declarations"]):
            try:
                a = SV.read_ack(bytes.fromhex(x))
            except SV.Refuse as r:
                raise Refuse(r.reason, ["declarations", i])
            if ("cache-declaration" not in SKIP and (a["instance"] == ZERO or a["upstream"].hex() != claims[0]["signer_pubkey"]
                                                     or a["instance"] not in links)):
                raise Refuse("context_mismatch", ["declarations", i])
            acks.append(a)
    schedules = []
    for bid in sorted(by_instance("schedule")):
        idx = by_instance("schedule")[bid]
        bodies = [claims[i]["body"] for i in idx]
        moved = any(not CAR.extends(b, a) for a, b in zip(bodies, bodies[1:]))       # carousel §4.5
        if moved and not cache and "schedule-extends" not in SKIP:
            reason = "schedule_contradicted"
        elif bid in contradicted:
            reason = "bearer_contradicted"
        elif bid not in declared:
            reason = "bearer_not_declared"
        elif declared[bid][0] == 1:
            reason = "schedule_on_bidirectional"
        elif moved and "schedule-extends" not in SKIP:           # cache-authority §5.6: a node's order
            reason = "schedule_contradicted"
        else:
            reason = None
        if reason:
            aside.update((i, reason) for i in idx)
            continue
        entries = []
        sent = CAV.positions(claims[declared[bid][1][0]]["body"]["sent"]) if cache else None
        for e in bodies[-1]["objects"]:                          # the last schedule's: each before it is its first passes
            if cache and e["leaf_index"] not in sent and "cache-sent" not in SKIP:
                continue                                         # cache-authority §7.2 step 6: only what its claim says it sent
            done = CV.schedule_bytes(e, CV.schedule_blocks(e))
            entries.append({"leaf_index": e["leaf_index"], "bytes_sent": done, "pass_us": e["count"] * e["duration_ms"] * 1000,
                            "rate": -(-8000 * (done // e["count"]) // e["duration_ms"]),         # rung-price §7.2, rounded up
                            "first": e["first_ms"] * 1000,
                            "last": (e["first_ms"] + (e["count"] - 1) * e["period_ms"] + e["duration_ms"]) * 1000})
        decl = [bytes.fromhex(CV.link_hash(claims[i])) for i in declared[bid][1]]
        settled = [a["digest"] for a in acks if a["instance"] in decl]
        if cache and entries and not settled and "cache-settled" not in SKIP:
            raise Refuse("record_missing", ["claims", declared[bid][1][0]])
        schedules.append({"bearer_id": bid, "bearer": declared[bid][0], "entries": entries,
                          "claims": [bytes.fromhex(CV.link_hash(claims[i])) for i in idx],
                          "declarations": decl, "settled": settled})
    if aside and "group-standing" not in SKIP:
        k = min(aside)
        raise Refuse(aside[k], ["claims", k])
    instances = []
    for bid in sorted(declared):
        e = claims[declared[bid][1][0]]["body"]["energy"]
        if e is None:
            instances.append(figure(0, None))
        else:
            tag = TAGS[e["energy_provenance"]]
            ppm = e["energy_uncertainty"]["relative_ppm"] if e["energy_uncertainty"] is not None else 0
            instances.append(figure(e["joules_micro"], (tag, ppm)))
    return {"kind": GROUP, "link": JV.link(g), "cache": cache,
            "publisher": bytes.fromhex(claims[0]["signer_pubkey"] if cache else g["signer_pubkey"]),
            "declarations": [bytes.fromhex(CV.link_hash(c)) for c in claims if c["class"] == declaring],
            "instances": instances, "bearers": sorted(declared), "schedules": schedules,
            "settled": [(a["digest"], a["start"], a["end"]) for a in acks]}


def time_precision(task):
    """DAP §4.2: a task read through its `time_precision`, or None when it does not read so or the
    field is 0 (telemetry-aggregation §4.3)."""
    try:
        r = TAV.Rd(task)
        r.opaque(1)
        r.opaque(2)
        r.opaque(2)
        return r.u(8) or None
    except ValueError:
        return None


def leader_statement(b):
    """§4.11: a leader's statement read whole by telemetry-aggregation §6.1, its role the leader's
    (§8.1 step 5 there) and its batch of at least one time unit (step 7)."""
    if "aggregated-statement" in SKIP:                           # the fields at their places alone
        st = {"role": b[5] if len(b) > 5 else 2, "batch": (int.from_bytes(b[70:78], "big"), int.from_bytes(b[78:86], "big")),
              "count": int.from_bytes(b[86:94], "big")}
    else:
        try:
            st = TAV.read_statement(b)
        except TAV.Refuse:
            raise Refuse("statement_invalid", ["leader"])
    if st["role"] != 2 and "aggregated-role" not in SKIP:
        raise Refuse("wrong_signer", ["leader"])
    if st["batch"][1] == 0 and "aggregated-duration" not in SKIP:
        raise Refuse("batch_invalid", ["leader"])
    return st


def aggregated_source(s):
    """§5.1, §4.11: an aggregate receipt, read as data: its layout, its collector's signature, its
    task through its `time_precision`, which places its batch in time (§5.3), its leader's statement,
    and the sum of its result held to that statement's count. No other step of impression-aggregates
    §8.1, which a holder of the receipt and its catalogue runs."""
    if set(s) != {"kind", "receipt"} or not is_hex(s["receipt"]):
        raise Refuse("malformed", [])
    b = bytes.fromhex(s["receipt"])
    try:
        rc = TAV.read_receipt(b)
    except TAV.Refuse as r:
        raise Refuse(r.reason, ["receipt"])
    why = ED.verify(rc["signer"], b"wai:telemetry-aggregate/receipt\x01" + rc["body"], rc["sig"])
    if why is not None and "aggregated-signature" not in SKIP:
        raise Refuse(why, ["receipt"])
    tp = time_precision(rc["task"])
    if tp is None and "aggregated-task" not in SKIP:
        raise Refuse("task_invalid", ["task"])
    st = leader_statement(rc["leader"])
    if sum(rc["result"]) != st["count"] and "aggregated-count" not in SKIP:
        raise Refuse("count_mismatch", ["result"])
    return {"kind": AGGREGATED, "digest": hashlib.sha256(rc["body"]).digest(), "collector": rc["signer"], "task": rc["task"],
            "batch": st["batch"], "from": st["batch"][0] * (tp or 1) * 10 ** 6,
            "to": (st["batch"][0] + st["batch"][1]) * (tp or 1) * 10 ** 6, "count": st["count"]}


def consumption_source(s):
    """§5.1, §4.12: an agent's consumption claims in their chain's order, `parent` held before the
    first: jwp-receipts §3.1 steps 1 to 5; step 6 is the holder of the objects', and bills nothing."""
    if (set(s) != {"kind", "claims", "parent"} or not isinstance(s["claims"], list) or not s["claims"]
            or not all(isinstance(c, str) for c in s["claims"]) or not (s["parent"] is None or isinstance(s["parent"], str))):
        raise Refuse("malformed", [])
    held = held_parent(s)
    agent, claims = None, []
    for i, w in enumerate(s["claims"]):
        try:
            c = CV.verify_claim(w)
            previous = held.get(c["signer_pubkey"])
            held[c["signer_pubkey"]] = c
            if "group-chain" not in SKIP:
                CV.check_chain(c, previous)
            if c["class"] != "consumption":
                raise CV.Reject(6, "context_mismatch")
            agent = agent or c["signer_pubkey"]
            if agent != c["signer_pubkey"] and "consumption-one-agent" not in SKIP:
                raise CV.Reject(6, "context_mismatch")
        except CV.Reject as r:
            raise Refuse(r.reason, ["claims", i])
        claims.append({"link": bytes.fromhex(CV.link_hash(c)), "objects": len(c["body"]["objects"]),
                       "issued_s": rfc3339_seconds(c["issued_at"])})
    return {"kind": CONSUMPTION, "agent": bytes.fromhex(agent), "claims": claims}


_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") in ("job", "group", "cache", "aggregated", "consumption") 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", "group-chain", "group-standing",
                                     "group-version", "schedule-extends", "cache-sent", "cache-settled", "cache-declaration",
                                     "cache-one-node", "aggregated-signature", "aggregated-count", "aggregated-statement",
                                     "aggregated-role", "aggregated-duration", "aggregated-task", "consumption-one-agent"})], 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)
    if s.get("kind") in ("group", "cache", "aggregated", "consumption") and (version == V7 or "group-version" in SKIP):
        if s["kind"] == "aggregated":
            return aggregated_source(s)
        if s["kind"] == "consumption":
            return consumption_source(s)
        return group_source(s, s["kind"] == "cache")
    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"])]
    if f["kind"] == GROUP:                         # what a chain bills once; the group receipt is evidence (§4.10)
        declared = CACHE_CLAIM if f["cache"] else TRANSMISSION_CLAIM
        return [(declared, d) for d in f["declarations"]] + [(SCHEDULE_CLAIM, d) for x in f["schedules"] for d in x["claims"]]
    if f["kind"] == AGGREGATED:
        return [(AGGREGATE_RECEIPT, f["digest"])]
    if f["kind"] == CONSUMPTION:
        return [(CONSUMPTION_CLAIM, x["link"]) for x in f["claims"]]
    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"]}
    elif f["kind"] == GROUP:
        out |= {(GROUP_RECEIPT, f["link"])} | {(ACK, d) for d, _, _ in f["settled"]}
    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"
    elif f["kind"] == GROUP:                     # §5.3: every pass of every entry billed within the period
        if not in_scope(t, GROUP, f["publisher"]):
            return "outside_scope"
        if not all(within(e["first"], e["last"]) for x in f["schedules"] for e in x["entries"]):
            return "outside_period"
        if not all(within(a, b) for _, a, b in f["settled"]):
            return "outside_period"
    elif f["kind"] == AGGREGATED:                # §5.3: its batch within the period, whose chain's periods do not overlap
        if not in_scope(t, AGGREGATED, f["collector"]):
            return "outside_scope"
        if "aggregated-period" not in SKIP and not within(f["from"], f["to"]):
            return "outside_period"
    elif f["kind"] == CONSUMPTION:               # §5.3: each claim's `issued_at` within the period, which excludes its end
        if not in_scope(t, CONSUMPTION, f["agent"]):
            return "outside_scope"
        if "period" not in SKIP and not all(start <= x["issued_s"] * 10 ** 6 < end for x in f["claims"]):
            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"] == AGGREGATED:     # impression-aggregates §8.2: batches of one task sharing a time unit
        return (f["task"] == g["task"] and "aggregated-overlap" not in SKIP
                and overlapping((f["batch"][0], f["batch"][0] + f["batch"][1]), (g["batch"][0], g["batch"][0] + g["batch"][1])))
    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
    if kind == GROUP_RECEIPT and version == V7 and "group-again" not in SKIP:
        return False                               # §4.10, §6.9: a group sent again on a new instance
    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, instances = 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 f["kind"] == GROUP and "instance-once" not in SKIP:     # §5.2: one party's one transmission
            for bid in f["bearers"]:
                instance = bid if "instance-group" in SKIP else (f["link"], bid)
                who = instances.get(instance)
                if who is not None:
                    return k, "record_duplicate" if who == f["publisher"] or "instance-party" in SKIP else "bearer_contradicted"
                instances[instance] = f["publisher"]
        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)
        elif f["kind"] == GROUP:                                                     # §4.10, §6.5
            for x in f["schedules"]:
                cites = [(GROUP_RECEIPT, f["link"])] + [(SCHEDULE_CLAIM, d) for d in x["claims"]]
                cites += [(CACHE_CLAIM if f["cache"] else TRANSMISSION_CLAIM, d) for d in x["declarations"]]
                cites += [(ACK, d) for d in x["settled"]]
                for e in x["entries"]:
                    for unit, n in ((1, e["bytes_sent"]), (2, e["pass_us"])):
                        k = tier(t, x["bearer"], unit, e["rate"])
                        if k is not None:
                            add(GROUP, f["publisher"], k, n, cites)
        elif f["kind"] == AGGREGATED:                                                # §4.11
            add(AGGREGATED, f["collector"], 1, f["count"], [(AGGREGATE_RECEIPT, f["digest"])])
        elif f["kind"] == CONSUMPTION:                                               # §4.12
            for x in f["claims"]:
                add(CONSUMPTION, f["agent"], 1, x["objects"], [(CONSUMPTION_CLAIM, x["link"])])
        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"]]
        elif f["kind"] == GROUP and "transmission-energy" not in SKIP:              # §6.6: each declared instance once
            figs += [(7, x) for x in f["instances"]]
    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, 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."""
    if x["kind"] == GROUP:
        unit = TRANSMISSION_UNITS[t["transmission"][x["unit"] - 1][1] - 1][0]
    elif x["kind"] == JOB:
        unit = JOB_UNITS[t["job"][x["unit"] - 1][2] - 1]
    else:
        unit = units(t["version"])[x["kind"]][x["unit"] - 1][0]
    out = {"source_kind": KIND_NAMES[x["kind"]], "source": x["source"].hex(), "unit": unit,
           "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"] in ENDPOINTS:                  # §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")
    if x["kind"] == GROUP:                             # §7.3: its bearer class and its ceiling
        out["bearer"] = BEARERS[t["transmission"][x["unit"] - 1][0]]
        out["max_bps"] = t["transmission"][x["unit"] - 1][2]
    return out


def transmission_report(facts, t):
    """§8.3: each entry of each standing schedule of each group source, with the price it is billed
    at in each unit, or None above every ceiling."""
    out = []
    for f in facts:
        if f["kind"] != GROUP:
            continue
        entries = [{"bearer_id": x["bearer_id"], "bearer": BEARERS[x["bearer"]], "leaf_index": e["leaf_index"],
                    "bytes_sent": e["bytes_sent"], "pass_us": e["pass_us"], "rate_bps": e["rate"] if e["rate"] <= MAX_SAFE else None,
                    "bytes_sent_price": tier(t, x["bearer"], 1, e["rate"]), "pass_us_price": tier(t, x["bearer"], 2, e["rate"])}
                   for x in f["schedules"] for e in x["entries"]]
        out.append({"group": f["link"].hex(), "publisher": f["publisher"].hex(), "cache": f["cache"], "entries": entries})
    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": dict({"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)},
                        **({"transmission": transmission_report(facts, t)} if inv["version"] == V7 else {})),
    }


# ---------------------------------------------------------------- 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 name in MONTH:                                           # §11: any byte of a month's invoice
        if name in names:
            for at in (5, 72, 137, -40, -1):
                m = copy.deepcopy(doc)
                m["vectors"][names[name]]["invoice"] = flip_hex(doc["vectors"][names[name]]["invoice"], at)
                out.append(("%s: a byte of its invoice at %d" % (name, at), m))
    if "version_7_week_5" in names:                              # §4.10 to §4.12
        k = names["version_7_week_5"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        aggregated = next(n for n, l in enumerate(lines) if l["source_kind"] == "aggregated")
        consumed = next(n for n, l in enumerate(lines) if l["source_kind"] == "consumption")
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][aggregated]["quantity"] += 1
        out.append(("version_7_week_5: its aggregated line a count above its receipts'", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][aggregated]["unit"] = "presented_impressions"
        out.append(("version_7_week_5: its aggregated line named a presented unit", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][consumed]["quantity"] -= 1
        out.append(("version_7_week_5: its consumption line an object short", m))
        m = copy.deepcopy(doc)
        m["sources"]["aggregated_r0"]["receipt"] = doc["sources"]["aggregated_r12"]["receipt"]
        out.append(("aggregated_r0: the other batch's receipt given in its place", m))
        m = copy.deepcopy(doc)
        m["sources"]["consumption_week_5"]["claims"] = doc["sources"]["consumption_week_5"]["claims"][:1]
        out.append(("consumption_week_5: its second claim left out", m))
        m = copy.deepcopy(doc)
        m["sources"]["cache_week_5"]["declarations"] = []
        out.append(("cache_week_5: its settlement declaration left out", m))
        m = copy.deepcopy(doc)
        m["sources"]["group_week_5"]["claims"] = doc["sources"]["group_week_5"]["claims"][:2]
        out.append(("group_week_5: its extending schedule left out", m))
        m = copy.deepcopy(doc)
        m["sources"]["aggregated_r0"]["receipt"] = doc["sources"]["aggregated_statement_of_the_helper"]["receipt"]
        out.append(("aggregated_r0: the helper's statement where its leader's belongs", m))
        k = names["version_7_aggregated_overlap_in_the_next_invoice"]
        m = copy.deepcopy(doc)
        m["vectors"][k]["sources"] = ["aggregated_r12"]
        out.append(("version_7_aggregated_overlap_in_the_next_invoice: another batch of the period given in its place", m))
        k = names["version_7_a_node_billing_the_publishers_instance"]
        m = copy.deepcopy(doc)
        m["vectors"][k]["sources"] = ["group_week_5", "cache_week_5"]
        out.append(("version_7_a_node_billing_the_publishers_instance: the node on an instance of its own", m))
    if "version_7_week_1" in names:                              # §4.10: a transmission per class, unit and ceiling
        k = names["version_7_week_1"]
        lines = doc["vectors"][k]["outcome"]["lines"]
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][0]["max_bps"] = lines[1]["max_bps"]
        out.append(("version_7_week_1: its first line named another ceiling", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["lines"][0]["bearer"] = "unidirectional"
        out.append(("version_7_week_1: its multicast line named a broadcast", m))
        m = copy.deepcopy(doc)
        entries = m["vectors"][k]["outcome"]["reports"]["transmission"][0]["entries"]
        entries[0]["pass_us_price"] += 1
        out.append(("version_7_week_1: its report's first entry at another price", m))
        m = copy.deepcopy(doc)
        m["sources"]["group_week_1"]["claims"] = doc["sources"]["group_week_1"]["claims"][:2]
        out.append(("group_week_1: its multicast's declaration and schedule left out", m))
        m = copy.deepcopy(doc)
        m["sources"]["group_week_1"]["claims"][1] = doc["sources"]["group_week_1"]["claims"][1].replace('"duration_ms":4000', '"duration_ms":40')
        out.append(("group_week_1: a pass shortened, its signature unchanged", m))
        k = names["version_7_week_3"]
        m = copy.deepcopy(doc)
        m["vectors"][k]["outcome"]["reports"]["transmission"][0]["entries"][0]["bytes_sent_price"] = 2
        out.append(("version_7_week_3: an entry above every ceiling reported at the highest", m))
        m = copy.deepcopy(doc)
        m["vectors"][k]["previous"] = None
        out.append(("version_7_week_3: the invoice before it not given", m))
        m = copy.deepcopy(doc)
        m["sources"]["group_week_3"]["parent"] = None
        out.append(("group_week_3: given without the claim its first follows", 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",
          "transmission-rules", "group-version", "group-chain", "group-standing", "tier", "group-again", "transmission-energy",
          "unit-price-bound", "schedule-extends", "cache-sent", "cache-settled", "cache-declaration", "cache-one-node",
          "aggregated-signature", "aggregated-count", "aggregated-overlap", "aggregated-statement", "aggregated-role",
          "aggregated-duration", "aggregated-task", "aggregated-period", "instance-once", "instance-party", "instance-group",
          "consumption-one-agent"]


MONTH = ["version_7_week_1", "version_7_week_2", "version_7_week_3", "version_7_week_4", "version_7_week_5"]   # §11


def invoice_body(inv, c, t):
    """§6.1: every byte of an invoice of these figures but `sig`."""
    def u(n, w):
        return n.to_bytes(w, "little")
    o = INVOICE_MAGIC + bytes([t["version"]]) + t["seller"] + t["buyer"] + t["digest"] + inv["prev"]
    o += u(inv["start"], 8) + u(inv["end"], 8) + t["currency"]
    o += u(len(c["records"]), 4) + b"".join(bytes([k]) + i for k, i in c["records"])
    o += u(len(c["lines"]), 2)
    for x in c["lines"]:
        o += bytes([x["kind"]]) + x["source"] + bytes([x["unit"]]) + u(x["quantity"], 8) + u(x["price"], 8) + u(x["amount"], 8)
        o += u(x["cited"], 4) + x["cites"]
    o += bytes([len(c["energy"])]) + b"".join(bytes([s, g]) + u(n, 4) + u(j, 8) + u(p, 4) for s, g, n, j, p in c["energy"])
    return o + b"".join(u(n, 4) for n in c["unmetered"][:scopes(t["version"])]) + u(c["amount"], 8)


def month_recomputes(doc):
    """§11: each of the month's invoices written again from its tariff, the invoice before it and
    its sources, byte for byte up to its signature, which verify() checks."""
    fails = []
    for name in MONTH:
        v = next(x for x in doc["vectors"] if x["name"] == name)
        d = document(doc, v)
        ib = bytes.fromhex(d["invoice"])
        t, inv = read_tariff(bytes.fromhex(d["tariff"])), read_invoice(ib, ["invoice"])
        prev = None if d["previous"] is None else read_invoice(bytes.fromhex(d["previous"]), ["previous"])
        facts = [verify_source(s, [i for k, i in t["scope"] if k == 4], t["version"]) for s in d["sources"]]
        hold(t, facts, prev, inv["start"], inv["end"])
        if invoice_body(inv, compute(t, facts), t) != ib[:-64]:
            fails.append("%s: the invoice written again differs" % name)
    return fails


def aggregates_stand(doc, corpus_dir):
    """§4.11: the ledger reads a receipt as data, so what it bills is held here to the text that
    verifies it: impression-aggregates §8.1 accepts each receipt, by the collector the corpus names,
    and the line's quantity is their report counts' sum."""
    import impression_aggregate_verify as IAV
    with open(os.path.join(corpus_dir, "..", "impressions", "vectors.json"), encoding="utf-8") as fh:
        flights = json.load(fh)
    catalogue = bytes.fromhex(flights["fixtures"]["catalogues"]["flight"])
    fails, count = [], 0
    for name in ("aggregated_r0", "aggregated_r12"):
        try:
            got, _, _ = IAV.verify_receipt(bytes.fromhex(doc["sources"][name]["receipt"]), catalogue, None)
        except IAV.Refuse as e:
            fails.append("%s: impression-aggregates §8.1 refuses it, %s" % (name, e.reason))
            continue
        count += got["report_count"]
        if got["collector"] != doc["keys"]["collector"]:
            fails.append("%s: another collector" % name)
    w5 = next(v for v in doc["vectors"] if v["name"] == "version_7_week_5")
    line = next(x for x in w5["outcome"]["lines"] if x["source_kind"] == "aggregated")
    if line["quantity"] != count:
        fails.append("the aggregated line is not its receipts' report counts")
    return fails


def main(corpus_dir):
    doc = json.load(open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8"))
    fails = check(doc)
    remade = month_recomputes(doc)
    print("%s the month's invoices are written again byte for byte (%d)" % ("PASS" if not remade else "FAIL", len(MONTH)))
    fails += remade
    stood = aggregates_stand(doc, corpus_dir)
    print("%s each aggregate receipt billed stands by impression-aggregates §8.1" % ("PASS" if not stood else "FAIL"))
    fails += stood
    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", "bearer_contradicted"] 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]))