open-standards/wai/claim-conformance/claim_verify.py

#!/usr/bin/env python3
"""Second verifier for claim-conformance/ — typed claim receipts, the soft binding class,
and recovery under the ancestor rule (extensions/jwp-receipts.md §2.4.3-§2.4.12, §2.6,
§3.1-§3.2), and the arithmetic of a measurement report (energy-measurement.md §3.1, §6.2),
computed with Python's integers.

Written from the extension text. It shares no code with wai-rs (`crate::claim_receipt`),
which generates the corpus: the canonical form is Python's JSON serialiser with members
ordered by UTF-16 code units, BLAKE3 is written from its specification (`_blake3.py`),
Ed25519 is WAI's strict rule written from RFC 8032 (`_ed25519.py`, SPEC §7 *Signatures*;
the verifiers of the other corpora call it too), and OER receipts named by an `energy`
claim are read, verified and their figures read by the OER reference verifier
(`../oer-conformance/oer2_verify.py`), under oer-receipts' own rules.

    python3 claim_verify.py [vectors.json]   verify every vector and the canonical fixture
    python3 claim_verify.py fixture          print the fixture's canonical bytes,
                                             signature and link hash
    python3 claim_verify.py brackets         print the figure and least declaration of
                                             each bracket wai-rs pins
    python3 claim_verify.py verdicts <file>  print `name<TAB>verdict` for every vector of a
                                             corpus-format file, ignoring any declared
                                             outcome (the input to differential.py)
    python3 claim_verify.py consumption <file>
                                             steps 1-4 over the `consumption` claims of the
                                             AI usage corpus (live-conformance/usage)
    python3 claim_verify.py bearer <file>    steps 1-4 over the `transmission` and
                                             `reconciliation` claims of the bearer corpus
                                             (live-conformance/bearers)

Exit status 0 only if every vector reports its declared outcome, and the fixture and the
bracket arithmetic this script computes equal the literals pinned in wai-rs.

A corpus-format file is read as the OER reference reads a vector document (`OER.load`,
oer-receipts §4 "The document"), since it holds OER receipts: a file that reading refuses
has no verdicts, and `verdicts` exits 1 with no output. No integer in a claim is converted
from more than twenty digits, so no verdict depends on an interpreter's limit on integer
digits.
"""
import hashlib, json, os, re, struct, sys, unicodedata
from decimal import localcontext

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "oer-conformance"))
import oer2_verify as OER  # noqa: E402  (the OER reference verifier: grades + receipt checks)
sys.path.insert(0, os.path.join(HERE, "..", "staged-conformance"))
import staged_verify as SV  # noqa: E402  (the strict JSON and manifest reader http_verify.py uses too)
sys.path.insert(0, os.path.join(HERE, "..", "staged-measure-conformance"))
import measure_verify as MV  # noqa: E402  (staged-measurement §3's counted bytes, written from that text)
import int_payload_verify as IPV  # noqa: E402  (the integer payload parsers; measure_verify puts them on the path)
sys.path.insert(0, HERE)

from cryptography.hazmat.primitives import serialization  # noqa: E402
from cryptography.hazmat.primitives.asymmetric import ed25519  # noqa: E402

# ---------------------------------------------------------------- BLAKE3 (hash mode)

from _blake3 import blake3  # noqa: E402  (written from its specification; shared with staged_verify.py)


# ------------------------------------------------ Ed25519 (RFC 8032), §3.1 step 3's rule
#
# §3.1 step 3 is WAI's one rule for every signature (SPEC §7 *Signatures*): the key is the
# canonical encoding of a point of large order, S < L, R is not of small order, and the
# cofactorless equation holds with R compared byte for byte. `_ed25519.py` implements it
# from RFC 8032's definitions; the `cryptography` package alone would accept keys that are
# not canonical encodings and keys and R of small order.

import _ed25519 as ED  # noqa: E402


def check_signature(pk: bytes, sig: bytes, msg: bytes):
    """§3.1 step 3, in its order: the key (`bad_signer_key`), then S < L, R, and the
    equation (`signature_verify_failed`)."""
    why = ED.verify(pk, msg, sig)
    need(why != "bad_key", 3, "bad_signer_key")
    need(why is None, 3, "signature_verify_failed")


# ------------------------------------------------------------- canonical form (RFC 8785)

MAX_INT = 2 ** 53 - 1


def jcs(v):
    if v is None:
        return "null"
    if v is True or v is False:
        return "true" if v else "false"
    if isinstance(v, int):
        if not 0 <= v <= MAX_INT:
            raise ValueError("integer outside the canonical subset")
        return str(v)
    if isinstance(v, str):
        return json.dumps(v, ensure_ascii=False)
    if isinstance(v, list):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    if isinstance(v, dict):
        keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
        return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"
    raise ValueError("value outside the canonical subset")


def link_hash(claim: dict) -> str:
    return blake3(b"moq-jwp:claim-link\x01" + jcs(claim).encode()).hex()


def signing_input(claim: dict) -> bytes:
    return b"moq-jwp:claim\x01" + jcs({k: v for k, v in claim.items() if k != "sig"}).encode()


MAX_DEPTH = 32  # §3.1 step 1: arrays and objects nested deeper, the receipt being level 1


class OutsideSubset:
    """A JSON number that is not an unsigned integer written without a sign, fraction or
    exponent: whatever its value, not one a claim may carry (`field_type`, never
    `malformed`). Kept as a marker rather than a float so that no reading depends on how a
    float is rounded, or on whether it overflows."""


OUTSIDE_SUBSET = OutsideSubset()


def _scalar_text(s):
    """Whether `s` is Unicode text: no surrogate code point, which is what an escape that
    leaves a surrogate unpaired decodes to here."""
    return not any("\ud800" <= ch <= "\udfff" for ch in s)


def _check_tree(v, level):
    if isinstance(v, str):
        if not _scalar_text(v):
            raise ValueError("an escape that leaves a surrogate unpaired")
    elif isinstance(v, (list, dict)):
        if level + 1 > MAX_DEPTH:
            raise ValueError("nested too deeply")
        for x in (v.values() if isinstance(v, dict) else v):
            _check_tree(x, level + 1)


def parse_strict(text):
    """§3.1 step 1's reading of a JSON text; ValueError is `malformed`. Refuses what RFC
    8259 does, a duplicate member name, a string or member name holding an unpaired
    surrogate, and nesting deeper than MAX_DEPTH. Never refuses a number: one outside the
    claim subset is OUTSIDE_SUBSET."""
    def pairs(ps):
        d = {}
        for k, v in ps:
            if not _scalar_text(k):
                raise ValueError("an escape that leaves a surrogate unpaired")
            if k in d:
                raise ValueError("duplicate member " + k)
            d[k] = v
        return d

    def no_constant(c):
        raise ValueError("not JSON: " + c)

    def integer(s):
        # A plain unsigned integer beyond 20 digits exceeds every u64, as wai-rs reads it.
        return OUTSIDE_SUBSET if s.startswith("-") or len(s) > 20 else int(s)

    if not isinstance(text, str):
        raise TypeError("a claim is given as its wire form")
    try:
        v = json.loads(text, object_pairs_hook=pairs, parse_constant=no_constant,
                       parse_float=lambda s: OUTSIDE_SUBSET, parse_int=integer)
    except RecursionError:
        raise ValueError("nested too deeply")
    _check_tree(v, 0)
    return v


# -------------------------------------------------------------------- the extension

ENVELOPE = ["kind", "class", "subject_type", "subject", "seq", "issued_at", "body",
            "signer_pubkey", "signer_id", "parent_receipt_hash", "sig"]
ADMITS = {"energy": {"object"}, "lawful-basis": {"object", "receipt"}, "log-inclusion": {"receipt"},
          "soft-binding": {"object"}, "binding-recovery": {"object"},
          "measurement": {"object", "receipt"}, "operating-point": {"object", "receipt"},
          "stage": {"object"}, "delivery-gap": {"object"}, "reconstruct-session": {"object"},
          "consumption": {"object"}, "transmission": {"receipt"}, "reconciliation": {"receipt"}}
FIGURE = ["figure_profile", "figure_index", "group", "joules_micro", "energy_provenance", "energy_uncertainty",
          "energy_counter_filtering", "energy_calibration_ref"]
BODY = {
    "energy": ["joules_micro", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
               "energy_calibration_ref", "coverage", "grade", "method", "work_units", "work_unit", "oer_receipt"],
    "log-inclusion": ["log", "profile", "entry", "log_receipt"],
    "soft-binding": ["alg", "registry", "binding_type", "value", "resolution_endpoint"],
    "binding-recovery": ["alg", "binding_type", "value", "declaration", "resolution_endpoint", "observed", "attests"],
    "measurement": FIGURE + ["bracket", "build", "power_reader"],
    "operating-point": FIGURE + ["vocabulary", "op_requested", "op_committed", "op_delivered", "prediction"],
    "stage": ["vocabulary", "op_requested", "op_committed", "op_delivered", "presented", "units_by_stage", "tier",
              "stage_hash", "objects", "conflicts", "deadline_misses", "bytes_envelopes"],
    "delivery-gap": ["track_alias", "group_id", "object_id", "reason", "observed_at"],
    # §2.4.13: ai-usage §6.1.
    "consumption": ["objects", "usage", "seen", "grant"],
    # §2.4.14, §2.4.15: bearer-receipts §4.1, §5.1.
    "transmission": ["bearer", "bearer_id", "energy", "attests"],
    "reconciliation": ["bearer", "bearer_id", "leaf_count", "received", "received_root", "attests"],
    "reconstruct-session": ["session_nonce", "media", "host", "decoders", "units", "leaf_count", "output_root",
                            "primary_units", "fallback_units", "switches", "switches_total", "deadline_us", "missed",
                            "missed_units", "stage_sets", "framing", "counted_primary", "counted_fallback",
                            "counted_discarded", "uncounted", "transferred", "receipt_bytes", "classical_declared",
                            "classical_source", "energy"],
}
# §2.4.10: one reported layer object, its units, the enumerations this extension defines
# for it, the tiers of staged-delivery §5, and the reasons of staged-delivery §4.2.
STAGE_OBJECT = ["layer", "object", "units", "bytes", "status", "evidence", "reason"]
STAGE_UNITS = ["from", "count"]
STAGE_STATUS = ("applied", "partial", "late", "dropped", "refused")
STAGE_EVIDENCE = ("delivered-verified", "inclusion-verified", "none")
STAGE_TIERS = ("decode-equivalence", "entropy-consistency", "none")
STAGE_REASONS = ("refines-malformed", "capability-not-refinement", "base-absent", "base-via-fallback",
                 "base-not-admitted", "refinement-unattested", "payload-malformed", "cost-over-limit",
                 "chain-over-limit", "layer-conflict", "prev-mismatch", "layer-gap", "geometry-mismatch",
                 "deadline-missed")
STAGE_VOCABULARY = "urn:wai:staged-delivery:1"
GAP_REASONS = ("delivery-timeout", "discretion", "not-received")  # §2.4.11
BRACKET = ["idle", "busy", "window_us", "completions", "in_flight", "opening"]
SUMMARY = ["n", "mean_uw", "sd_uw", "live_intervals", "longest_live_run"]
BUILD = ["profile", "toolchain", "target"]
PREDICTION = ["predictor", "descriptor_hash", "predicted_joules_micro"]
# energy-measurement §6.1: profile -> (subject_type, carries several figures, an Ed25519
# key signs the figure, names a group).
PROFILES = {"moq-jwp:object": ("object", True, True, True), "moq-jwp:claim-link": ("receipt", False, True, False),
            "wai:energy-binding-id": ("receipt", False, False, False)}
for _n in ["video-receipt-id", "video-manifest-receipt-id", "video-ssai-receipt-id", "video-keys-receipt-id",
           "video-package-receipt-id", "video-channel-receipt-id", "world-receipt-id", "film-receipt-id",
           "haptic-receipt-id", "score-receipt-id", "splat4d-receipt-id", "audio-receipt-id", "ternary-receipt-id",
           "quantum-receipt-id", "quantum-calibration-id", "quantum-job-id", "quantum-mitigate-id",
           "quantum-decode-id", "quantum-compile-id", "quantum-rearrange-id", "quantum-fabric-id",
           "phasor-receipt-id"]:
    PROFILES["wai:" + _n] = ("receipt", False, True, False)
for _n in ["asset-receipt-id", "quantum-qbom-id"]:
    PROFILES["wai:" + _n] = ("receipt", True, True, False)
DECLARATION = ["record", "vocabulary", "fragment", "purposes"]
PURPOSE = ["purpose", "basis", "data"]
DECISION = ["record", "vocabulary", "declaration", "purpose", "outcome"]
UNCERTAINTY = ["relative_ppm", "window_us"]
# energy-measurement.md §2's acquisition classes, each with the strongest oer-receipts.md
# §2 grade it supports. Grades and their ranking come from the OER reference verifier.
STRONGEST = {"CalibratedInstrument": "measured", "OnChipCounter": "measured", "HwShunt": "measured",
             "ModelBased": "derived", "Estimator": "projected"}
FILTERING = {"On", "Off", "Unknown"}  # energy-measurement.md §2.4
RESOLUTION_FLOOR = 288676  # energy-measurement.md §2.3: joules_micro x relative_ppm, at least
# Unicode's White_Space property: what a calibration reference must hold something besides
# (energy-measurement.md §2.5, "non-empty"). Written out, because str.isspace() also
# counts U+001C to U+001F, which are not White_Space.
WHITE_SPACE = ("\t\n\x0b\x0c\r \x85\xa0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007"
               "\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000")
COVERAGE = {"Whole", "Partial"}
GRADES = OER.GRADES

# §2.4.12: a session's host, decoder entries, switches, missed presentations, stage sets
# and energy, the enumerations it defines, and delivery-telemetry.md §5's switch reasons.
SESSION_HOST = ["profile", "target", "build", "threads"]
SESSION_DECODER = ["capability", "rendition", "object", "first_unit", "path", "decoder", "variant", "pin", "tier", "module_status", "leaves"]
SESSION_SWITCH = ["unit", "from", "to", "reason"]
SESSION_MISSED = ["unit", "layers", "late_us"]
SESSION_STAGE_SET = ["first_unit", "claim"]
SESSION_ENERGY = ["joules_micro", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
                  "energy_calibration_ref"]
HOST_PROFILES = ("native-ffi", "browser-baseline", "browser-gpu", "edge-wasi")
PATHS = ("primary", "fallback")
FRAMINGS = ("registered",)
CLASSICAL_SOURCES = ("wai2-table",)
MODULE_STATUSES = ("pinned", "fetched")
TIER_RANK = {"none": 0, "entropy-consistency": 1, "decode-equivalence": 2}  # SPEC §3.1
SWITCH_REASONS = ("pin-malformed", "form-not-registered", "form-unreadable", "form-required", "prior-revoked",
                  "prior-not-authorised", "prior-not-held", "digest-superseded", "set-malformed", "part-mismatch",
                  "set-incomplete", "missing-capability", "module-unverified", "module-revoked", "trap",
                  "fuel-exhausted", "memory-exhausted", "deadline-threshold", "unit-hash-mismatch",
                  "budget-exhausted", "recovered")
# The capability registry (../registry/capabilities.json): each capability's kind, status
# and replicate class, for a session's tiers, re-decodes and classical rendition.
with open(os.path.join(HERE, "..", "registry", "capabilities.json"), encoding="utf-8") as _f:
    REGISTRY = {c["capability"]: c for c in json.load(_f)["capabilities"]}
# The codec glossary (../glossary/codecs.json): the capabilities an entry of era `classical`
# links, from which a session's classical rendition is read (§2.4.12).
with open(os.path.join(HERE, "..", "glossary", "codecs.json"), encoding="utf-8") as _f:
    CLASSICAL = {link["capability"] for e in json.load(_f)["entries"] if e["era"] == "classical"
                 for link in (e.get("wai") or [])}


class Reject(Exception):
    def __init__(self, step, reason):
        super().__init__(f"reject step={step} reason={reason}")
        self.step, self.reason = step, reason


def is_int(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT


def is_hex(v, nbytes=None):
    if not isinstance(v, str) or len(v) % 2 or not re.fullmatch(r"[0-9a-f]*", v):
        return False
    return nbytes is None or len(v) == 2 * nbytes


def is_token(s):
    return (isinstance(s, str) and s != "" and len(s.encode()) <= 2048
            and not any(c.isspace() or unicodedata.category(c) == "Cc" for c in s))


def is_uri(s):
    return is_token(s) and re.fullmatch(r"[A-Za-z][A-Za-z0-9+.\-]*:.+", s, re.S) is not None


def is_https(s):
    """§2.6.2: `https://`, then an authority (up to the first `/`, `?` or `#`) with no `@`,
    that is a non-empty host — the text before the first `:`, or what an IP literal's
    brackets enclose — optionally followed by `:` and a port of ASCII digits."""
    if not isinstance(s, str) or not s.startswith("https://"):
        return False
    authority = re.split(r"[/?#]", s[len("https://"):], maxsplit=1)[0]
    if not is_uri(s) or "@" in authority:
        return False
    if authority.startswith("["):
        host, closed, after = authority[1:].partition("]")
        if not closed:
            return False
    else:
        host, colon, port = authority.partition(":")
        after = colon + port
    return host != "" and (after == "" or (after[0] == ":" and all(c in "0123456789" for c in after[1:])))


def registry_form(alg):
    labels = alg.split(".")
    return len(labels) >= 2 and all(re.fullmatch(r"[A-Za-z0-9_+\-]+", l) for l in labels)


def days_in(year, month):
    """Days in a month of the proleptic Gregorian calendar (RFC 3339 Appendix C)."""
    if month == 2:
        return 29 if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) else 28
    return 30 if month in (4, 6, 9, 11) else 31


def is_rfc3339(s):
    """RFC 3339 §5.6 date-time with §5.7's day-of-month rule; ASCII digits only (a regex
    `\\d` would also take any Unicode decimal digit). A second of 60 is accepted at any
    time of day: whether a leap second occurred then is not a property of the syntax."""
    m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(\.[0-9]+)?"
                     r"([Zz]|[+-]([0-9]{2}):([0-9]{2}))", s)
    if not m:
        return False
    yr, mo, da, hh, mi, ss = (int(m.group(i)) for i in (1, 2, 3, 4, 5, 6))
    if not (1 <= mo <= 12 and 1 <= da <= days_in(yr, mo) and hh <= 23 and mi <= 59 and ss <= 60):
        return False
    return m.group(9) is None or (int(m.group(9)) <= 23 and int(m.group(10)) <= 59)


def is_oer_id(s):
    return isinstance(s, str) and re.fullmatch(r"(sha256|blake3):[0-9a-f]{64}", s) is not None


def members(obj, required, step):
    if any(k not in obj for k in required):
        raise Reject(step, "missing_required_field")
    if any(k not in required for k in obj):
        raise Reject(step, "unknown_member")


def need(ok, step, reason="field_type"):
    if not ok:
        raise Reject(step, reason)


def check_body(cls, b):
    """§3.1 step 2: the members, then each member's type and encoding together, in the
    member table's order, a nested object's members checked where the table reaches it.
    Where every failure left is `field_type` the checks are grouped; the order shows only
    where a nested object's missing or unknown member could be reported instead."""
    need(isinstance(b, dict), 2)
    if cls == "lawful-basis":
        # §2.4.6: `record` is read before anything else, and names the table.
        if "record" not in b:
            raise Reject(2, "missing_required_field")
        need(isinstance(b["record"], str), 2)
        if b["record"] == "declaration":
            members(b, DECLARATION, 2)
            need(is_uri(b["vocabulary"]), 2)
            need(b["fragment"] is None or is_hex(b["fragment"], 32), 2)
            need(isinstance(b["purposes"], list), 2)
            for p in b["purposes"]:
                need(isinstance(p, dict), 2)
                members(p, PURPOSE, 2)
                need(is_token(p["purpose"]), 2)
                need(p["basis"] == "" or is_token(p["basis"]), 2)  # an empty basis is step 4's
                need(isinstance(p["data"], list) and all(is_token(d) for d in p["data"]), 2)
        elif b["record"] == "decision":
            members(b, DECISION, 2)
            need(all(isinstance(b[k], str) for k in ("vocabulary", "purpose", "outcome")), 2)
            need(is_hex(b["declaration"], 32), 2)
            need(is_uri(b["vocabulary"]) and is_token(b["purpose"]) and is_token(b["outcome"]), 2)
        else:
            raise Reject(2, "field_type")
        return
    members(b, BODY[cls], 2)
    if cls == "reconstruct-session":
        check_session_body(b)
        return
    if cls == "stage":
        # In the member table's order: `vocabulary` before the objects, whose members are
        # checked where the table reaches them.
        need(is_uri(b["vocabulary"]), 2)
        for k in ("op_requested", "op_committed", "op_delivered", "presented"):
            pt = b[k]
            need(pt is not None or k != "presented", 2)
            if pt is not None:
                need(isinstance(pt, dict) and all(is_token(n) and is_int(x) for n, x in pt.items()), 2)
        need(isinstance(b["units_by_stage"], list) and all(is_int(n) for n in b["units_by_stage"]), 2)
        need(b["tier"] in STAGE_TIERS, 2)
        need(b["stage_hash"] is None or is_hex(b["stage_hash"], 32), 2)
        need(isinstance(b["objects"], list), 2)
        for o in b["objects"]:
            need(isinstance(o, dict), 2)
            members(o, STAGE_OBJECT, 2)
            need(is_int(o["layer"]) and is_hex(o["object"], 32), 2)
            need(isinstance(o["units"], dict), 2)
            members(o["units"], STAGE_UNITS, 2)
            need(is_int(o["units"]["from"]) and is_int(o["units"]["count"]), 2)
            need(is_int(o["bytes"]) and o["status"] in STAGE_STATUS and o["evidence"] in STAGE_EVIDENCE, 2)
            need(o["reason"] is None or isinstance(o["reason"], str), 2)
        need(all(is_int(b[k]) for k in ("conflicts", "deadline_misses", "bytes_envelopes")), 2)
        return
    if cls in ("transmission", "reconciliation"):
        # bearer-receipts §4.1, §5.1: the bearer's class from its enumeration, the instance
        # an absolute URI; a reconciliation's bitmap a byte string of any length (its length
        # is a class rule, step 4); a transmission's energy as a session's.
        need(b["bearer"] in BEARERS, 2)
        need(is_uri(b["bearer_id"]), 2)
        if cls == "reconciliation":
            need(is_int(b["leaf_count"]), 2)
            need(is_hex(b["received"]), 2)
            need(is_hex(b["received_root"], 32), 2)
        else:
            e = b["energy"]
            if e is not None:
                nested(e, SESSION_ENERGY)
                need(is_int(e["joules_micro"]), 2)
                need(isinstance(e["energy_provenance"], str), 2)
                u = e["energy_uncertainty"]
                if u is not None:
                    nested(u, UNCERTAINTY)
                    need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
                need(e["energy_counter_filtering"] is None or isinstance(e["energy_counter_filtering"], str), 2)
                need(e["energy_calibration_ref"] is None or isinstance(e["energy_calibration_ref"], str), 2)
        need(isinstance(b["attests"], str), 2)
        return
    if cls == "consumption":
        # ai-usage §6.1: the arrays' lengths and a term's are in the type column.
        need(isinstance(b["objects"], list) and 1 <= len(b["objects"]) <= 1024
             and all(is_hex(h, 32) for h in b["objects"]), 2)
        need(isinstance(b["usage"], list) and 1 <= len(b["usage"]) <= 3, 2)
        for e in b["usage"]:
            need(isinstance(e, dict), 2)
            members(e, ["vocabulary", "term"], 2)
            need(e["vocabulary"] in CONSUMPTION_VOCABULARIES, 2)
            need(e["term"] is None or (isinstance(e["term"], str) and 1 <= len(e["term"].encode()) <= 255), 2)
        need(isinstance(b["seen"], list) and len(b["seen"]) <= 64 and all(is_hex(h, 32) for h in b["seen"]), 2)
        need(b["grant"] is None or is_hex(b["grant"], 32), 2)
        return
    if cls == "delivery-gap":
        need(all(is_int(b[k]) for k in ("track_alias", "group_id", "object_id")), 2)
        need(b["reason"] in GAP_REASONS, 2)
        need(isinstance(b["observed_at"], str) and is_rfc3339(b["observed_at"]), 2)
        return
    if cls in ("measurement", "operating-point"):
        check_figure_header(b)
        if cls == "measurement":
            br = b["bracket"]
            if br is not None:
                need(isinstance(br, dict), 2)
                members(br, BRACKET, 2)
                for side in ("idle", "busy"):
                    need(isinstance(br[side], dict), 2)
                    members(br[side], SUMMARY, 2)
                    need(all(is_int(br[side][k]) for k in SUMMARY), 2)
                need(all(is_int(br[k]) for k in ("window_us", "completions", "in_flight")), 2)
                need(br["opening"] in ("from-rest", "running"), 2)
            bd = b["build"]
            need(isinstance(bd, dict), 2)
            members(bd, BUILD, 2)
            need(bd["profile"] in ("release", "debug"), 2)
            need(bd["toolchain"] is None or isinstance(bd["toolchain"], str), 2)
            need(bd["target"] is None or is_token(bd["target"]), 2)
            need(b["power_reader"] is None or isinstance(b["power_reader"], str), 2)
        else:
            need(is_uri(b["vocabulary"]), 2)
            for k in ("op_requested", "op_committed", "op_delivered"):
                pt = b[k]
                if pt is not None:
                    need(isinstance(pt, dict), 2)
                    for name, val in pt.items():
                        need(is_token(name) and is_int(val), 2)
            pr = b["prediction"]
            if pr is not None:
                need(isinstance(pr, dict), 2)
                members(pr, PREDICTION, 2)
                need(is_uri(pr["predictor"]), 2)
                need(is_hex(pr["descriptor_hash"], 32), 2)
                need(is_int(pr["predicted_joules_micro"]), 2)
        return
    if cls == "energy":
        # In the member table's order: a nested object's member check can report another
        # reason than a type check, so the order is the one every verifier uses.
        need(is_int(b["joules_micro"]), 2)
        need(isinstance(b["energy_provenance"], str), 2)
        u = b["energy_uncertainty"]
        if u is not None:
            need(isinstance(u, dict), 2)
            members(u, UNCERTAINTY, 2)
            need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
        need(b["energy_counter_filtering"] is None or isinstance(b["energy_counter_filtering"], str), 2)
        need(b["energy_calibration_ref"] is None or isinstance(b["energy_calibration_ref"], str), 2)
        need(all(isinstance(b[k], str) for k in ("coverage", "grade", "method")), 2)
        need(is_int(b["work_units"]), 2)
        need(isinstance(b["work_unit"], str), 2)
        need(b["oer_receipt"] is None or isinstance(b["oer_receipt"], str), 2)
        need(is_uri(b["method"]) and is_token(b["work_unit"]), 2)
        need(b["oer_receipt"] is None or is_oer_id(b["oer_receipt"]), 2)
    elif cls == "log-inclusion":
        need(all(isinstance(b[k], str) for k in ("log", "profile", "entry")), 2)
        need(is_hex(b["log_receipt"]), 2)
        need(is_uri(b["log"]) and is_uri(b["profile"]) and is_token(b["entry"]), 2)
    else:  # soft-binding, binding-recovery
        need(isinstance(b["alg"], str) and b["binding_type"] in ("watermark", "fingerprint"), 2)
        need(is_hex(b["value"]) and isinstance(b["resolution_endpoint"], str), 2)
        if cls == "soft-binding":
            need(isinstance(b["registry"], str), 2)
            need(is_token(b["alg"]) and is_uri(b["registry"]) and is_uri(b["resolution_endpoint"]), 2)
        else:
            need(is_hex(b["declaration"], 32), 2)
            need(b["observed"] is None or is_hex(b["observed"], 32), 2)
            need(isinstance(b["attests"], str), 2)
            need(is_token(b["alg"]) and is_uri(b["resolution_endpoint"]), 2)


def nested(v, required):
    """A nested object a table fixes: an object, then its members, missing then unknown."""
    need(isinstance(v, dict), 2)
    members(v, required, 2)
    return v


def check_session_body(b):
    """§2.4.12, at §3.1 step 2: each member's type and encoding in the table's order, the
    members of the host, of each decoder entry, switch, missed presentation and stage set,
    and of `energy`, where the table reaches them."""
    need(is_hex(b["session_nonce"], 32), 2)
    need(isinstance(b["media"], str) and re.fullmatch(r"[a-z0-9-]{1,32}", b["media"]) is not None, 2)
    h = nested(b["host"], SESSION_HOST)
    need(h["profile"] in HOST_PROFILES, 2)
    need(h["target"] is None or is_token(h["target"]), 2)
    bd = nested(h["build"], BUILD)
    need(bd["profile"] in ("release", "debug"), 2)
    need(bd["toolchain"] is None or isinstance(bd["toolchain"], str), 2)
    need(bd["target"] is None or is_token(bd["target"]), 2)
    need(is_int(h["threads"]), 2)
    need(isinstance(b["decoders"], list), 2)
    for d in b["decoders"]:
        nested(d, SESSION_DECODER)
        need(is_token(d["capability"]), 2)
        need(d["rendition"] is None or is_int(d["rendition"]), 2)
        need(is_hex(d["object"], 32), 2)
        need(is_int(d["first_unit"]), 2)
        need(d["path"] in PATHS, 2)
        need(isinstance(d["decoder"], str), 2)
        need(d["variant"] is None or is_token(d["variant"]), 2)
        need(d["pin"] is None or is_hex(d["pin"], 32), 2)
        need(d["tier"] is None or isinstance(d["tier"], str), 2)
        need(d["module_status"] is None or isinstance(d["module_status"], str), 2)
        need(is_int(d["leaves"]), 2)
    need(all(is_int(b[k]) for k in ("units", "leaf_count")), 2)
    need(b["output_root"] is None or is_hex(b["output_root"], 32), 2)
    need(all(is_int(b[k]) for k in ("primary_units", "fallback_units")), 2)
    need(isinstance(b["switches"], list), 2)
    for w in b["switches"]:
        nested(w, SESSION_SWITCH)
        need(is_int(w["unit"]) and w["from"] in PATHS and w["to"] in PATHS and isinstance(w["reason"], str), 2)
    need(is_int(b["switches_total"]), 2)
    need(b["deadline_us"] is None or is_int(b["deadline_us"]), 2)
    need(is_int(b["missed"]), 2)
    need(isinstance(b["missed_units"], list), 2)
    for x in b["missed_units"]:
        nested(x, SESSION_MISSED)
        need(all(is_int(x[k]) for k in SESSION_MISSED), 2)
    need(isinstance(b["stage_sets"], list), 2)
    for x in b["stage_sets"]:
        nested(x, SESSION_STAGE_SET)
        need(is_int(x["first_unit"]) and is_hex(x["claim"], 32), 2)
    need(b["framing"] in FRAMINGS, 2)
    need(all(is_int(b[k]) for k in ("counted_primary", "counted_fallback", "counted_discarded", "uncounted",
                                     "transferred", "receipt_bytes")), 2)
    need(b["classical_declared"] is None or is_int(b["classical_declared"]), 2)
    need(b["classical_source"] is None or is_token(b["classical_source"]), 2)
    e = b["energy"]
    if e is not None:
        nested(e, SESSION_ENERGY)
        need(is_int(e["joules_micro"]), 2)
        need(isinstance(e["energy_provenance"], str), 2)
        u = e["energy_uncertainty"]
        if u is not None:
            nested(u, UNCERTAINTY)
            need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
        need(e["energy_counter_filtering"] is None or isinstance(e["energy_counter_filtering"], str), 2)
        need(e["energy_calibration_ref"] is None or isinstance(e["energy_calibration_ref"], str), 2)


def decoder_ok(s):
    """A decoder identifier of §2.4.12: builtin:<crate>@<version> or module:sha256:<hex>."""
    if s.startswith("module:sha256:"):
        return re.fullmatch(r"[0-9a-f]{64}", s[len("module:sha256:"):]) is not None
    if not s.startswith("builtin:") or "@" not in s:
        return False
    krate, version = s[len("builtin:"):].split("@", 1)
    return (re.fullmatch(r"[a-z0-9_-]{1,64}", krate) is not None
            and re.fullmatch(r"[0-9][0-9A-Za-z.+-]{0,63}", version) is not None)


def replicate(cap):
    """A capability's replicate class as the registry states it, None for one it does not
    register or classify."""
    c = REGISTRY.get(cap)
    return c["replicate"] if c else None


def class_tier(cap):
    """SPEC §3.1 *Tier*: the tier a capability's class gives — decode-equivalence for an
    integer-exact neural decode, entropy-consistency for an entropy-exact one, none for
    every other class — or None for a capability with no class."""
    r = replicate(cap)
    if r is None:
        return None
    return {"neural-int-exact": "decode-equivalence", "neural-entropy-exact": "entropy-consistency"}.get(r, "none")


def selectable(entry):
    """SPEC §7.1: not a companion or a refinement, and a capability a sink may dispatch —
    one the registry does not list, or a payload that is not a candidate."""
    if entry.get("role") in ("companion", "refinement"):
        return False
    c = REGISTRY.get(entry["capability"])
    return c is None or (c["kind"] == "payload" and c["status"] != "candidate")


def session_rules(b):
    """§2.4.12's rules, at §3.1 step 4, in order."""
    ds = b["decoders"]
    need(b["primary_units"] + b["fallback_units"] == b["units"], 4, "session_path_sum")
    need(sum(d["leaves"] for d in ds) == b["leaf_count"] and b["leaf_count"] >= b["units"]
         and (b["units"] == 0) == (b["leaf_count"] == 0), 4, "session_leaf_sum")
    need((b["output_root"] is not None) == (b["leaf_count"] > 0), 4, "session_root_presence")
    t = b["host"]["threads"]
    need(t > 0 and t & (t - 1) == 0, 4, "session_host_malformed")
    for d in ds:
        builtin = d["decoder"].startswith("builtin:")
        tier_ok = (d["pin"] is None and d["tier"] is None) or (
            d["pin"] is not None and d["tier"] is not None and d["tier"] in TIER_RANK)
        status_ok = builtin if d["module_status"] is None else (not builtin and d["module_status"] in MODULE_STATUSES)
        need(decoder_ok(d["decoder"]) and (d["rendition"] is None or d["rendition"] <= 65535) and tier_ok
             and status_ok and d["leaves"] != 0, 4, "session_decoder_malformed")
    keys = [tuple(d[k] for k in SESSION_DECODER if k != "leaves") for d in ds]
    need(len(set(keys)) == len(keys), 4, "session_decoder_malformed")
    for d in ds:
        ct = class_tier(d["capability"])
        need(d["tier"] is None or ct is None or TIER_RANK[d["tier"]] <= TIER_RANK[ct], 4, "session_tier_exceeds_class")
    sw = b["switches"]
    need(len(sw) == min(b["switches_total"], 1024)
         and all(w["from"] != w["to"] and w["reason"] in SWITCH_REASONS for w in sw)
         and all(sw[i]["unit"] <= sw[i + 1]["unit"] for i in range(len(sw) - 1)), 4, "session_switch_invalid")
    mu = b["missed_units"]
    need(b["missed"] <= b["leaf_count"] and (b["deadline_us"] is not None or b["missed"] == 0)
         and len(mu) == min(b["missed"], 256)
         and all(1 <= x["layers"] <= 65535 and x["late_us"] >= 1 for x in mu)
         and all((mu[i]["unit"], mu[i]["layers"]) < (mu[i + 1]["unit"], mu[i + 1]["layers"]) for i in range(len(mu) - 1)),
         4, "session_missed_invalid")
    ss = b["stage_sets"]
    need(all((ss[i]["first_unit"], ss[i]["claim"]) < (ss[i + 1]["first_unit"], ss[i + 1]["claim"])
             for i in range(len(ss) - 1)), 4, "session_stage_sets_invalid")
    need(b["counted_discarded"] <= b["counted_primary"] + b["counted_fallback"], 4, "session_counted_invalid")
    cd, cs = b["classical_declared"], b["classical_source"]
    need((cd is None and cs is None) or (cd is not None and cs in CLASSICAL_SOURCES), 4, "session_classical_invalid")
    e = b["energy"]
    if e is not None:
        need(e["joules_micro"] != 0, 4, "energy_zero_figure")
        acquisition(e)


# ------------------------------------------- §2.4.12 step 6: what a session is checked against

def out_leaf(p):
    """staged-measurement Appendix C: a presentation's leaf."""
    return hashlib.sha256(b"wai:out-leaf\x01" + p["unit"].to_bytes(8, "big") + p["layers"].to_bytes(2, "big")
                          + p["decoder"].to_bytes(2, "big") + bytes.fromhex(p["out_digest"])).digest()


def out_mth(leaves):
    """RFC 6962's Merkle tree hash, under the output tree's node domain."""
    if len(leaves) == 1:
        return leaves[0]
    k = 1
    while k * 2 < len(leaves):
        k *= 2
    return hashlib.sha256(b"wai:out-node\x01" + out_mth(leaves[:k]) + out_mth(leaves[k:])).digest()


def ctx_int(v, bits=64):
    if not (isinstance(v, int) and not isinstance(v, bool) and 0 <= v < 2 ** bits):
        raise ValueError("bad integer in a session context")
    return v


def ctx_hash(v):
    if not is_hex(v, 32):
        raise ValueError("bad hash in a session context")
    return v


def presentations(v):
    if not isinstance(v, list):
        raise ValueError("outputs must be an array")
    return [{"unit": ctx_int(p["unit"]), "layers": ctx_int(p["layers"], 16), "decoder": ctx_int(p["decoder"], 16),
             "out_digest": ctx_hash(p["out_digest"])} for p in v]


def finals(ps):
    """Each unit presented, with its presentation at its largest layer count."""
    out = {}
    for p in ps:
        if p["unit"] not in out or p["layers"] > out[p["unit"]]["layers"]:
            out[p["unit"]] = p
    return out


def check_session_outputs(c, ps, redecoded):
    b = c["body"]
    need(c["class"] == "reconstruct-session", 6, "context_mismatch")
    keys = [(p["unit"], p["layers"]) for p in ps]
    need(len(ps) == b["leaf_count"] and len(set(keys)) == len(keys), 6, "session_output_mismatch")
    if ps:
        ordered = sorted(ps, key=lambda p: (p["unit"], p["layers"]))
        need(out_mth([out_leaf(p) for p in ordered]).hex() == b["output_root"], 6, "session_output_mismatch")
    else:
        need(b["output_root"] is None, 6, "session_output_mismatch")
    ds = b["decoders"]
    bad = "session_output_inconsistent"
    need(all(p["decoder"] < len(ds) and p["layers"] >= 1 for p in ps), 6, bad)
    fin = finals(ps)
    need(len(fin) == b["units"], 6, bad)
    need(all(d["leaves"] == sum(1 for p in ps if p["decoder"] == i) for i, d in enumerate(ds)), 6, bad)
    nxt = 0
    for p in sorted(ps, key=lambda p: (p["layers"], p["unit"])):
        need(p["decoder"] <= nxt, 6, bad)
        if p["decoder"] == nxt:
            nxt += 1
    primary = sum(1 for p in fin.values() if ds[p["decoder"]]["path"] == "primary")
    need(primary == b["primary_units"] and len(fin) - primary == b["fallback_units"], 6, bad)
    need(all((x["unit"], x["layers"]) in set(keys) for x in b["missed_units"]), 6, bad)
    for i, p in enumerate(ps):
        if replicate(ds[p["decoder"]]["capability"]) in ("neural-int-exact", "standard-defined") and i in redecoded:
            need(redecoded[i] == p["out_digest"], 6, "session_output_irreproducible")


def objects_seen(v):
    if not isinstance(v, list):
        raise ValueError("objects must be an array")
    out = []
    for o in v:
        ctx_hash(o["content_hash"])
        if o["path"] not in PATHS or not isinstance(o["presented"], bool):
            raise ValueError("bad object")
        out.append({"content_hash": o["content_hash"], "path": o["path"],
                    "counted": None if o["counted"] is None else ctx_int(o["counted"]),
                    "transferred": ctx_int(o["transferred"]), "presented": o["presented"]})
    return out


def check_session_objects(c, objs):
    """§2.4.12 against the objects received: each (content hash, path) once; the entry
    object and each entry's object among them; the counted sums; and an object presented
    exactly when a decoder entry names it."""
    b = c["body"]
    need(c["class"] == "reconstruct-session", 6, "context_mismatch")
    keys = [(o["content_hash"], o["path"]) for o in objs]
    need(len(set(keys)) == len(keys), 6, "session_object_duplicate")
    need(any(o["content_hash"] == c["subject"] for o in objs), 6, "session_entry_not_received")
    received = {o["content_hash"] for o in objs}
    need(all(d["object"] in received for d in b["decoders"]), 6, "session_source_not_received")
    sums = {"primary": 0, "fallback": 0}
    discarded = uncounted = transferred = 0
    for o in objs:
        n = o["counted"]
        transferred += o["transferred"]
        if n is None:
            uncounted += 1
        else:
            sums[o["path"]] += n
            if not o["presented"]:
                discarded += n
    need((sums["primary"], sums["fallback"], discarded, uncounted, transferred)
         == (b["counted_primary"], b["counted_fallback"], b["counted_discarded"], b["uncounted"], b["transferred"]),
         6, "session_counted_mismatch")
    named = {d["object"] for d in b["decoders"]}
    need(all(o["presented"] == (o["content_hash"] in named) for o in objs), 6, "session_presented_mismatch")


def wai2_read(data):
    """SPEC §7.1: a WAI2 envelope's manifest and each rendition's payload length, or None
    for bytes that are not one. The manifest is read as staged_verify.py and
    http_verify.py read one: strict JSON, then the members SPEC §7.1 requires with their
    types."""
    if len(data) < 8 or data[:4] != b"WAI2":
        return None
    ml = struct.unpack_from("<I", data, 4)[0]
    if ml > SV.MANIFEST_MAX or 8 + ml > len(data):
        return None
    try:
        m = SV.strict_json(data[8:8 + ml], SV.MANIFEST_MAX_VALUES)
        SV.check_manifest_v2(m)
    except (SV.NotJSON, SV.Schema):
        return None
    rs = m["renditions"]
    o = 8 + ml
    if o + 2 > len(data) or struct.unpack_from("<H", data, o)[0] != len(rs):
        return None
    o += 2
    if o + 8 * len(rs) > len(data):
        return None
    lengths, expected = [], 0
    for i in range(len(rs)):
        off, ln = struct.unpack_from("<II", data, o + 8 * i)
        if off != expected:
            return None
        lengths.append(ln)
        expected += ln
    if o + 8 * len(rs) + expected > len(data):
        return None
    return m, lengths


def plaintext_v2(m):
    """SPEC §6.1, §7.1: a WAI2 manifest's payloads are plaintext when its `security` is
    absent or null, or a security object whose `encryption` is absent, null or "none". A
    `security` that is not a security object (not an object, an `encryption` that is not a
    string, a `kid` that is not an integer from 0 to 2^64 - 1) is not plaintext."""
    sec = m.get("security")
    if sec is None:
        return True
    if not isinstance(sec, dict):
        return False
    enc, kid = sec.get("encryption"), sec.get("kid")
    if enc is not None and not isinstance(enc, str):
        return False
    if kid is not None and not (isinstance(kid, int) and not isinstance(kid, bool) and 0 <= kid < 2 ** 64):
        return False
    return enc is None or enc == "none"


def classical_rendition(m):
    """§2.4.12: the first rendition, in manifest order, that is selectable and of a
    classical capability whose registry media is the manifest's `media` and is not text."""
    for i, e in enumerate(m["renditions"]):
        c = REGISTRY.get(e["capability"])
        if (selectable(e) and e["capability"] in CLASSICAL and c is not None
                and c["media"] == m["media"] and c["media"] != "text"):
            return i
    return None


def envelopes(v):
    if not isinstance(v, list):
        raise ValueError("envelopes must be an array")
    out = []
    for e in v:
        if not is_hex(e["envelope"]):
            raise ValueError("bad envelope")
        out.append(bytes.fromhex(e["envelope"]))
    return out


def counted_of(e):
    """staged-measurement §3: an envelope's counted bytes, or None where it is not
    countable, for whatever reason (measure_verify.py's reading of the byte rule)."""
    try:
        return MV.count_any(e)["total"]
    except Exception:  # noqa: BLE001  (every way an object fails the byte rule is "not countable")
        return None


def payload_units(cap, p):
    """§2.4.12: the units a base payload holds where WAI reads them from its format: a
    WIV1 clip's frames, one for a WIH1 picture, a WIA1 clip or a WIS1 tensor; None for a
    capability with no pinned integer format, or a payload not in its own."""
    m = IPV.CAP_MAGIC.get(cap)
    if m is None:
        return None
    try:
        if m == b"WIV1":
            return IPV.parse_wiv1(p, None)["n_frames"]
        {b"WIH1": IPV.parse_wih1, b"WIA1": IPV.parse_wia1, b"WIS1": IPV.parse_wis1}[m](p)
        return 1
    except Exception:  # noqa: BLE001  (a payload that does not parse has no units to read)
        return None


def read_envelope(data):
    """A WAI1 or WAI2 envelope whose payloads are plaintext, read: ("one", manifest,
    payload) or ("multi", manifest, payloads); None for any other bytes (inert)."""
    if data[:4] == b"WAI2":
        r = wai2_read(data)
        if r is None or not plaintext_v2(r[0]):
            return None
        m, lengths = r
        o = 8 + struct.unpack_from("<I", data, 4)[0] + 2 + 8 * len(lengths)
        payloads = []
        for ln in lengths:
            payloads.append(data[o:o + ln])
            o += ln
        return ("multi", m, payloads)
    if data[:4] == b"WAI1":
        try:
            m, _, payload = SV.read_wai1(data)
        except SV.NotWai1:
            return None
        return ("one", m, payload) if SV.plaintext(m) else None
    return None


def read_any(data):
    """A held envelope, read: (read_envelope's form, sealed). An encrypted one (SPEC §6.4)
    is read too — its manifest and rendition table are in the clear — with sealed true,
    since its payloads are not; None for any other bytes."""
    r = read_envelope(data)
    if r is not None:
        return r, False
    if not encrypted_envelope(data):
        return None
    if data[:4] == b"WAI2":
        m, lengths = wai2_read(data)
        o = 8 + struct.unpack_from("<I", data, 4)[0] + 2 + 8 * len(lengths)
        payloads = []
        for ln in lengths:
            payloads.append(data[o:o + ln])
            o += ln
        return ("multi", m, payloads), True
    m, _, payload = SV.read_wai1(data)
    return ("one", m, payload), True


def base_units(cap, fallback, p):
    """§2.4.12: the units a WAI1 base's payload holds, read in the payload's own format: of
    the formats pinned for the manifest's capability and its fallback, which decodes the
    same payload (SPEC §4 step 5), the one whose magic the payload starts with."""
    for c in (cap, fallback):
        if isinstance(c, str) and IPV.CAP_MAGIC.get(c) == p[:4]:
            return payload_units(c, p)
    return None


def layer_units(ref, n):
    """A well-formed refines and its base's units where read: (layers, span), or None where
    its span lies outside those units or its runs do not sum to the units it covers
    (staged-delivery §3, §4.1 step 2)."""
    if n is not None:
        try:
            SV.check_units(ref, n)
        except SV.Malformed:
            return None
    span = ref["span"] if ref["span"] is not None else (None if n is None else (0, n))
    return ref["layer"] + 1, span


def dispatchable(cap):
    """SPEC §4: a capability a sink decodes on its own — one the registry does not list,
    or a payload that is not a candidate."""
    c = REGISTRY.get(cap)
    return c is None or (c["kind"] == "payload" and c["status"] != "candidate")


def companion_only(cap):
    c = REGISTRY.get(cap)
    return c is not None and c["kind"] == "companion"


def digest_matches(e, payload):
    """SPEC §7.1: a rendition's payload matches its stated sha256, if it states one."""
    h = e.get("sha256")
    return h is None or (SV.is_hash(h) and hashlib.sha256(payload).hexdigest() == h)


def digest_well_formed(e):
    """A rendition's sha256 absent or well formed (64 lowercase hex digits), read from the
    entry alone: a malformed one matches no payload, so no sink decodes the rendition. Of an
    encrypted envelope, whose payloads are not read, this is what an entry gives of its
    digest (§2.4.12 rule 4)."""
    h = e.get("sha256")
    return h is None or SV.is_hash(h)


def encrypted_envelope(data):
    """SPEC §6.4: an envelope a sink holding its key decrypts and dispatches — well formed,
    with a security object naming an encryption."""
    if data[:4] == b"WAI2":
        r = wai2_read(data)
        if r is None:
            return False
        sec = r[0].get("security")
        return isinstance(sec, dict) and plaintext_v2(r[0]) is False and isinstance(sec.get("encryption"), str) and (
            sec.get("kid") is None or (isinstance(sec["kid"], int) and not isinstance(sec["kid"], bool) and 0 <= sec["kid"] < 2 ** 64))
    if data[:4] == b"WAI1":
        try:
            m, _, _ = SV.read_wai1(data)
        except SV.NotWai1:
            return False
        return m.get("security") is not None and not SV.plaintext(m)
    return False


def component_roles(rs):
    """SPEC §5 / §7.1's component rules over a WAI2 table (staged_verify.component_roles,
    which http_verify.py reads too): None where every component conforms, else the first
    break's code. A sink offers the envelope's refinements only when it conforms."""
    return SV.component_roles(rs)


def check_session_envelopes(c, envs, objs, ps):
    """§2.4.12 against the envelopes the session received: every WAI2 one, and any WAI1
    one the verifier holds, with the objects and the presentations where those are held.
    In order: each among the objects received; each object listed with an envelope's
    content hash states its counted bytes; each envelope a decoder entry names held,
    readable or encrypted; each entry naming what its envelope gives, of an encrypted one a
    rendition whose sha256 is absent or well formed; each held layer an entry names one a
    sink can apply; each held layer's entry at the first unit of an entry naming its base;
    each presentation through a held entry at the layer count its envelope gives, and of a
    unit it covers; then the classical length over the held WAI2 envelopes, readable or
    encrypted, whose rendition tables are in the clear. Of an encrypted envelope the rules
    read the manifest and the table, and nothing that needs a payload: whether a payload
    matches a well-formed sha256, a base's units and a layer's WIR1 payload are the
    signer's statement."""
    b = c["body"]
    need(c["class"] == "reconstruct-session", 6, "context_mismatch")
    if objs is not None:
        received = {o["content_hash"] for o in objs}
        need(all(SV.content_hash(e) in received for e in envs), 6, "session_source_not_received")
        for e in envs:
            h, n = SV.content_hash(e), counted_of(e)
            need(all(o["counted"] == n for o in objs if o["content_hash"] == h), 6, "session_counted_mismatch")
    by_hash, inert, encrypted, held_envs, exploded = {}, set(), set(), [], []
    for data in envs:
        got = read_any(data)
        if got is None:
            inert.add(SV.content_hash(data))
        else:
            r, sealed = got
            h = SV.content_hash(data)
            if sealed:
                encrypted.add(h)
            if r[0] == "multi":
                # Each rendition is also its exploded envelope, the same object (SPEC §7.1,
                # staged-delivery §9).
                try:
                    each = SV.explode_each(data)
                except Exception:  # noqa: BLE001  (an envelope that does not explode has no aliases)
                    each = []
                exploded += [(SV.content_hash(x), h, j) for j, x in enumerate(each) if x is not None]
            by_hash.setdefault(h, r)
            held_envs.append(r)
    need(not any(d["object"] in inert and d["object"] not in by_hash for d in b["decoders"]), 6, "session_envelope_unreadable")
    # An encrypted envelope's payloads are not read: its digests, its units and a layer's
    # WIR1 payload are its signer's statement.
    sealed = lambda h: h in encrypted  # noqa: E731
    # Each entry as the envelopes resolve it: one naming a held WAI2 rendition by its exploded
    # envelope's content hash, with no rendition, names that rendition, unless the exploded
    # envelope's own bytes are held (a separate delivery).
    resolved = []
    for d in b["decoders"]:
        hit = next(((w, j) for x, w, j in exploded if x == d["object"]), None)
        if d["rendition"] is None and d["object"] not in by_hash and hit is not None:
            resolved.append(hit)
        else:
            resolved.append((d["object"], d["rendition"]))
    refinement_only = lambda cap: cap in SV.UNIT_KIND  # noqa: E731
    # A layer is decided by its capability; any other is a base, a refines on it ignored.
    layer_entry = lambda e: refinement_only(e["capability"])  # noqa: E731
    layer_one = lambda m: refinement_only(m["model_requirement"]["capability"])  # noqa: E731
    for d0, (o, rn) in zip(b["decoders"], resolved):
        d = dict(d0, object=o, rendition=rn)
        r = by_hash.get(d["object"])
        if r is not None and r[0] == "multi":
            rs, i = r[1]["renditions"], d["rendition"]
            gives = (i is not None and i < len(rs) and rs[i]["capability"] == d["capability"]
                     and (selectable(rs[i]) or layer_entry(rs[i]))
                     and (digest_well_formed(rs[i]) if sealed(d["object"]) else digest_matches(rs[i], r[2][i])))
        elif r is not None:
            mr = r[1]["model_requirement"]
            gives = (d["rendition"] is None and d["capability"] in (mr["capability"], mr.get("fallback"))
                     and (layer_one(r[1]) or (dispatchable(d["capability"]) and not companion_only(mr["capability"]))))
        else:
            gives = d["rendition"] is None
        need(gives, 6, "session_source_mismatch")

    def component(m, payloads, j, seal):
        e = m["renditions"][j]
        return {"capability": e["capability"], "units": None if seal else payload_units(e["capability"], payloads[j]),
                "is_base": selectable(e)}

    def base_by_hash(h):
        r = by_hash.get(h)
        if r is not None and r[0] == "one":
            mr = r[1]["model_requirement"]
            return {"capability": mr["capability"], "units": None if sealed(h) else base_units(mr["capability"], mr.get("fallback"), r[2]),
                    "is_base": not layer_one(r[1])}
        hit = next(((w, j) for x, w, j in exploded if x == h), None)
        if hit is None or by_hash[hit[0]][0] != "multi":
            return None
        return component(by_hash[hit[0]][1], by_hash[hit[0]][2], hit[1], sealed(hit[0]))

    def names_of(hash_form, pair):
        hashes = [hash_form] if hash_form is not None else []
        pairs = [pair] if pair is not None else []
        for x, w, j in exploded:
            if hash_form == x:
                pairs.append((w, j))
            if pair == (w, j):
                hashes.append(x)
        return hashes, pairs

    ds = b["decoders"]

    def layer_of(cap, ref, base, names, payload):
        """A layer of capability cap over the base (where held) named by names: ("layer",
        layers, span, first units of the entries it is applied over), or ("na",)."""
        n = None
        if base is not None:
            if not base["is_base"] or base["capability"] not in SV.ADMIT.get(cap, ()):
                return ("na",)
            n = base["units"]
        hashes, pairs = names
        naming = [e for e, (o, rn) in zip(ds, resolved)
                  if e["object"] in hashes or o in hashes or (rn is not None and (o, rn) in pairs)]
        over = [e["first_unit"] for e in naming
                if e["capability"] in SV.ADMIT.get(cap, ()) and (base is None or e["capability"] == base["capability"])]
        if naming and not over:
            return ("na",)
        lu = layer_units(ref, n)
        if lu is None:
            return ("na",)
        if cap not in SV.UNIT_KIND:
            return ("na",)
        if payload is not None:
            try:
                w = SV.parse_wir1(payload, SV.UNIT_KIND[cap])
            except SV.Refused:
                return ("na",)
            if lu[1] is not None and w["n_units"] != lu[1][1]:
                return ("na",)
        return ("layer", lu[0], lu[1], over)

    def given(k):
        """What entry k's held envelope is: ("base", units or None), ("layer", layers, span
        or None, first units), or ("na",) for a layer a sink cannot apply; None where the
        envelope is not held."""
        obj, rendition = resolved[k]
        r = by_hash.get(obj)
        if r is None:
            return None
        if r[0] == "one":
            m, payload = r[1], r[2]
            mr = m["model_requirement"]
            if not layer_one(m):
                return ("base", None if sealed(obj) else base_units(mr["capability"], mr.get("fallback"), payload))
            try:
                ref = SV.parse_refines(m.get("refines"))
            except SV.Malformed:
                return ("na",)
            if mr.get("fallback") is not None:
                return ("na",)
            return layer_of(mr["capability"], ref, base_by_hash(ref["base"]), names_of(ref["base"], None), None if sealed(obj) else payload)
        i = rendition
        if i is None:
            return None
        m, payloads = r[1], r[2]
        rs = m["renditions"]
        if i >= len(rs):
            return None
        e = rs[i]
        if not layer_entry(e):
            return ("base", None if sealed(obj) else payload_units(e["capability"], payloads[i]))
        # SPEC §7.1: a sink offers a WAI2's refinements only when every component conforms.
        if component_roles(rs) is not None:
            return ("na",)
        ref = SV.parse_refines_ids(e["refines"])
        bj = next(j for j, y in enumerate(rs) if y.get("id") == ref["base"])
        return layer_of(e["capability"], ref, component(m, payloads, bj, sealed(obj)), names_of(None, (obj, bj)), None if sealed(obj) else payloads[i])

    need(not any((given(k) or ("",))[0] == "na" for k in range(len(ds))), 6, "session_layer_not_applicable")
    for k, d in enumerate(ds):
        g = given(k)
        if g is not None and g[0] == "layer":
            need(not g[3] or d["first_unit"] in g[3], 6, "session_first_unit_mismatch")
    if ps is not None:
        for p in ps:
            g = given(p["decoder"]) if p["decoder"] < len(ds) else None
            if g is not None and g[0] in ("base", "layer"):
                need(p["layers"] == (1 if g[0] == "base" else g[1]), 6, "session_layers_mismatch")
        for p in ps:
            g = given(p["decoder"]) if p["decoder"] < len(ds) else None
            covers = None
            if g is not None and g[0] == "base" and g[1] is not None:
                covers = (0, g[1])
            elif g is not None and g[0] == "layer" and g[2] is not None:
                covers = g[2]
            if covers is not None:
                f = ds[p["decoder"]]["first_unit"] + covers[0]
                need(f <= p["unit"] < f + covers[1], 6, "session_unit_outside_object")
    # 9. The classical length, over the held WAI2 envelopes, readable or encrypted.
    decoded = {d["capability"] for d in b["decoders"]}
    total = None
    for r in held_envs:
        if r[0] != "multi":
            continue
        m = r[1]
        i = classical_rendition(m)
        if i is not None and m["renditions"][i]["capability"] not in decoded:
            total = (total or 0) + len(r[2][i])
    need(total == b["classical_declared"], 6, "session_classical_mismatch")


def check_session_stage_claims(c, wires, ps, objs):
    """§2.4.12 against the stage claims. A sink names one for each epoch of a stage set,
    and bases over the same units of the title each have their own: each claim's
    identity and objects first, then with the outputs each claim's units and its count
    from below, and each unit of a set from above, by the objects listed as applied or
    partial by the claims whose sets hold it."""
    if not isinstance(wires, list) or not all(isinstance(w, str) for w in wires):
        raise ValueError("stage_claims must be an array of wire forms")
    b = c["body"]
    need(c["class"] == "reconstruct-session", 6, "context_mismatch")
    received = {o["content_hash"] for o in objs} if objs is not None else None
    claims = []
    for w in wires:
        try:
            s = verify_claim(w)
        except Reject:
            raise Reject(6, "context_invalid")
        link = link_hash(s)
        named = [x["first_unit"] for x in b["stage_sets"] if x["claim"] == link]
        need(s["class"] == "stage" and s["signer_pubkey"] == c["signer_pubkey"] and named, 6, "context_mismatch")
        arrived = [o for o in s["body"]["objects"] if o["status"] != "dropped" and o["bytes"] > 0]
        if received is not None:
            need(s["subject"] in received and all(o["object"] in received for o in arrived), 6, "session_stage_disagrees")
        claims.append((s, named))
    # Each entry naming a claim's base, or a layer object it lists, has a first unit at
    # which stage_sets names a claim listing that object.
    at = {}
    for s, named in claims:
        for h in [s["subject"]] + [o["object"] for o in s["body"]["objects"]]:
            at.setdefault(h, set()).update(named)
    need(all(d["first_unit"] in at[d["object"]] for d in b["decoders"] if d["object"] in at), 6, "session_first_unit_mismatch")
    if ps is None:
        return
    fin = finals(ps)

    def in_set(f, n):
        return [(u - f, p["layers"]) for u, p in fin.items() if f <= u < f + n]

    sets = []
    for s, named in claims:
        ubs = s["body"]["units_by_stage"]
        n = sum(ubs)
        for f in named:
            presented = in_set(f, n)
            need(len(presented) == n, 6, "session_stage_disagrees")
            for st in range(1, len(ubs)):
                need(sum(1 for _, l in presented if l > st) >= sum(ubs[st:]), 6, "session_stage_disagrees")
            sets.append((f, n, [o for o in s["body"]["objects"] if o["status"] in ("applied", "partial")]))
    for u, p in fin.items():
        holding = [(f, applied) for f, n, applied in sets if f <= u < f + n]
        if not holding:
            continue
        for layer in range(1, p["layers"]):
            need(any(o["layer"] == layer and o["units"]["from"] <= u - f < o["units"]["from"] + o["units"]["count"]
                     for f, applied in holding for o in applied), 6, "session_stage_disagrees")


def check_figure_header(b):
    """§2.4.8's header at step 2, in its table's order: `figure_profile` a term, as it is
    read; the index an integer or null; the group a hash or null; then the figure and the
    class's members as §2.4.5 types them."""
    need(is_token(b["figure_profile"]), 2)
    need(b["figure_index"] is None or is_int(b["figure_index"]), 2)
    need(b["group"] is None or is_hex(b["group"], 32), 2)
    need(is_int(b["joules_micro"]), 2)
    need(isinstance(b["energy_provenance"], str), 2)
    u = b["energy_uncertainty"]
    if u is not None:
        need(isinstance(u, dict), 2)
        members(u, UNCERTAINTY, 2)
        need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
    need(b["energy_counter_filtering"] is None or isinstance(b["energy_counter_filtering"], str), 2)
    need(b["energy_calibration_ref"] is None or isinstance(b["energy_calibration_ref"], str), 2)


# ------------------------------------------- a measurement report's arithmetic (§3.1, §6.2)

def summary_live(s):
    """energy-measurement §3.1: a rail is live over a bracket only if it advanced in two
    consecutive intervals, in at least five, and in at least four of every five — facts
    that describe one sequence of `n` intervals. A power sample advances only where it
    differs from the one before it, so a bracket whose samples are all equal (`sd_uw` 0)
    advanced once at most, and is never live whatever facts it carries."""
    n, live, run = s["n"], s["live_intervals"], s["longest_live_run"]
    return s["sd_uw"] > 0 and run <= live <= n and run >= 2 and live >= 5 and live * 5 >= n * 4


def work(br):
    c, n = br["completions"], br["in_flight"]
    return (2 * c + n, 2) if br["opening"] == "from-rest" else (c, 1)


def bracket_figure(br):
    """F = floor((2·M·T·Wd + 10^6·Wn) / (2·10^6·Wn)), or None."""
    m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
    t, c, n = br["window_us"], br["completions"], br["in_flight"]
    if m <= 0 or t == 0 or c == 0 or n == 0:
        return None
    wn, wd = work(br)
    f = (2 * m * t * wd + 10 ** 6 * wn) // (2 * 10 ** 6 * wn)
    return f if 0 < f <= MAX_INT else None


def k_and_p(br, f):
    m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
    t, n = br["window_us"], br["in_flight"]
    wn, wd = work(br)
    ni, nb, si, sb = br["idle"]["n"], br["busy"]["n"], br["idle"]["sd_uw"], br["busy"]["sd_uw"]
    k = 12 * ni * nb * wn ** 4 * f * f
    p = 12 * wn ** 2 * t ** 2 * wd ** 2 * (si * si * nb + sb * sb * ni) + 4 * ni * nb * m * m * t * t * wd * wd * n * n \
        + 10 ** 12 * ni * nb * wn ** 4
    return k, p


def least_ppm(br, f):
    """The least p with p^2·K >= P, or None above 2^32 - 1."""
    k, p = k_and_p(br, f)
    if k == 0:
        return None
    q = -(-p // k)                       # p^2 >= ceil(P / K)
    r = __import__("math").isqrt(q)
    while r * r * k < p:
        r += 1
    while r > 1 and (r - 1) * (r - 1) * k >= p:
        r -= 1
    return r if 1 <= r <= 2 ** 32 - 1 else None


def bracket_rules(br, joules, label, u):
    """§2.4.8 rule 5, in its order."""
    hw = label in ("OnChipCounter", "CalibratedInstrument")
    need(br["idle"]["n"] >= 2 and br["busy"]["n"] >= 2, 4, "measurement_too_few_samples")
    need(not hw or (summary_live(br["idle"]) and summary_live(br["busy"])), 4, "measurement_rail_not_live")
    need(br["window_us"] != 0 and br["completions"] != 0 and br["in_flight"] != 0, 4, "measurement_empty_burst")
    m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
    need(m > 0, 4, "measurement_no_marginal")
    need(not hw or m >= br["idle"]["sd_uw"], 4, "measurement_below_attribution_floor")
    need(bracket_figure(br) == joules, 4, "measurement_figure_mismatch")
    if hw:
        need(u["window_us"] == br["window_us"], 4, "measurement_window_mismatch")
        k, p = k_and_p(br, joules)
        need(u["relative_ppm"] ** 2 * k >= p, 4, "measurement_declaration_below_bracket")


def figure_rules(c):
    """§2.4.8 rules 1-3: a non-zero figure, a profile for the subject type with an index
    and a group exactly where it has them, and the class."""
    b = c["body"]
    need(b["joules_micro"] != 0, 4, "energy_zero_figure")
    prof = PROFILES.get(b["figure_profile"])
    need(prof is not None and prof[0] == c["subject_type"], 4, "figure_unknown_profile")
    need((b["figure_index"] is not None) == prof[1], 4, "figure_index_mismatch")
    need((b["group"] is not None) == prof[3], 4, "figure_group_mismatch")
    acquisition(b)


def acquisition(b):
    """§2.4.5 rule 2: the class is one energy-measurement.md §2 names, not the legacy
    HwShunt, and carries exactly the evidence §2.3-§2.5 give it: a refined hardware class
    a non-zero declared uncertainty whose relative part fits in 32 bits (§2.7) and which
    reaches the resolution floor, an on-chip counter a filtering state and no calibration
    reference, a calibrated instrument a calibration reference that is not all white space
    and no filtering state; every other class none of the three."""
    label = b["energy_provenance"]
    need(label in STRONGEST, 4, "energy_unknown_acquisition")
    need(label != "HwShunt", 4, "energy_legacy_acquisition")
    u, f, ref = b["energy_uncertainty"], b["energy_counter_filtering"], b["energy_calibration_ref"]
    ev = "energy_acquisition_evidence"
    if label in ("OnChipCounter", "CalibratedInstrument"):
        need(u is not None and 0 < u["relative_ppm"] <= 2 ** 32 - 1 and u["window_us"] > 0, 4, ev)
        if label == "OnChipCounter":
            need(f in FILTERING and ref is None, 4, ev)
        else:
            need(f is None and ref is not None and ref.strip(WHITE_SPACE) != "", 4, ev)
        need(b["joules_micro"] * u["relative_ppm"] >= RESOLUTION_FLOOR, 4, ev)
    else:
        need(u is None and f is None and ref is None, 4, ev)


def class_rules(c):
    """§3.1 step 4 — §2.4.5-§2.4.7 and §2.6."""
    cls, b = c["class"], c["body"]
    if cls == "energy":
        need(b["joules_micro"] != 0, 4, "energy_zero_figure")
        acquisition(b)
        need(b["coverage"] in COVERAGE, 4, "energy_unknown_coverage")
        need(b["grade"] in GRADES, 4, "energy_unknown_grade")
        need(GRADES.index(b["grade"]) <= GRADES.index(STRONGEST[b["energy_provenance"]]), 4,
             "energy_grade_exceeds_acquisition")
        need(b["work_units"] != 0, 4, "energy_zero_work")
    elif cls == "lawful-basis" and b["record"] == "declaration":
        need(len(b["purposes"]) > 0, 4, "lawful_basis_no_purposes")
        seen = set()
        for p in b["purposes"]:
            need(p["basis"] != "", 4, "lawful_basis_purpose_without_basis")
            need(p["purpose"] not in seen, 4, "lawful_basis_duplicate_purpose")
            seen.add(p["purpose"])
    elif cls == "measurement":
        figure_rules(c)
        # Rule 4: a hardware figure is a reading, and its report names the reader.
        need(b["energy_provenance"] not in ("OnChipCounter", "CalibratedInstrument") or b["power_reader"] is not None,
             4, "measurement_no_power_reader")
        if b["bracket"] is not None:
            bracket_rules(b["bracket"], b["joules_micro"], b["energy_provenance"], b["energy_uncertainty"])
    elif cls == "operating-point":
        figure_rules(c)
        pts = [b[k] for k in ("op_requested", "op_committed", "op_delivered")]
        need(all(p is None or len(p) > 0 for p in pts), 4, "operating_point_empty_point")
        req, com, dlv = pts
        need(not (dlv is not None and com is None), 4, "operating_point_delivered_alone")
        need(not ((req is not None or com is not None) and dlv is None), 4, "operating_point_incomplete")
        need(dlv is not None or b["prediction"] is not None, 4, "operating_point_empty")
    elif cls == "stage":
        stage_rules(b)
    elif cls == "reconstruct-session":
        session_rules(b)
    elif cls == "consumption":
        consumption_rules(c)
    elif cls == "transmission":
        # bearer-receipts §4.2.
        need(b["attests"] == "transmitted", 4, "transmission_not_transmitted")
        e = b["energy"]
        if e is not None:
            need(e["joules_micro"] != 0, 4, "energy_zero_figure")
            acquisition(e)
    elif cls == "reconciliation":
        # bearer-receipts §5.2: object i is bit 0x80 >> (i mod 8) of byte i div 8.
        need(b["attests"] == "received", 4, "reconciliation_not_received")
        n, bits = b["leaf_count"], bytes.fromhex(b["received"])
        bad = "reconciliation_bitmap_malformed"
        need(n >= 1 and len(bits) == (n + 7) // 8, 4, bad)
        need(not any(bits[i // 8] & (0x80 >> (i % 8)) for i in range(n, 8 * len(bits))), 4, bad)
        need(any(bits), 4, bad)
    elif cls == "log-inclusion":
        need(b["log_receipt"] != "", 4, "log_inclusion_no_evidence")
    elif cls in ("soft-binding", "binding-recovery"):
        need(registry_form(b["alg"]), 4, "soft_binding_alg_not_registry_form")
        need(b["value"] != "", 4, "soft_binding_empty_value")
        need(is_https(b["resolution_endpoint"]), 4, "soft_binding_endpoint_not_https")
        if cls == "binding-recovery":
            need(b["attests"] == "ancestor", 4, "recovery_not_ancestor")
            need(b["observed"] != c["subject"], 4, "recovery_of_hard_bound_object")


# §2.4.14, §2.4.15: bearer-receipts §2's bearer classes.
BEARERS = ("bidirectional", "multicast", "unidirectional")

# §2.4.13: ai-usage §3, §6.2. The vocabularies, and the terms each defines.
CONSUMPTION_VOCABULARIES = ("aipref", "cawg", "tdmrep")
AIPREF_LABELS = ("train-ai", "ai-use", "search")  # draft-ietf-aipref-vocab-08 §6.1
CAWG_KEYS = ("cawg.data_mining", "cawg.ai_inference", "cawg.ai_generative_training", "cawg.ai_training")
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def did_key(pk):
    """§7.1's Ed25519 did:key: `did:key:z` and base58btc of 0xed 0x01 and the key."""
    b = b"\xed\x01" + pk
    n, out = int.from_bytes(b, "big"), ""
    while n:
        n, r = divmod(n, 58)
        out = B58[r] + out
    return "did:key:z" + "1" * (len(b) - len(b.lstrip(b"\x00"))) + out


def cawg_key(k):
    """ai-usage §3.4: components of letters, digits, `_` and `-`, separated by single dots."""
    parts = k.split(".")
    return 1 <= len(k.encode()) <= 255 and all(p and re.fullmatch(r"[A-Za-z0-9_-]+", p) for p in parts)


def consumption_rules(c):
    """ai-usage §6.2, in its order."""
    b = c["body"]
    asc = lambda xs: all(x < y for x, y in zip(xs, xs[1:]))  # noqa: E731
    need(asc(b["objects"]), 4, "consumption_objects_unordered")
    need(c["subject"] == b["objects"][0], 4, "consumption_subject_mismatch")
    need(asc([e["vocabulary"] for e in b["usage"]]), 4, "consumption_usage_unordered")
    for e in b["usage"]:
        t, v = e["term"], e["vocabulary"]
        known = t is None or (t in AIPREF_LABELS if v == "aipref" else t == "tdm-reservation" if v == "tdmrep"
                              else t in CAWG_KEYS or (cawg_key(t) and not t.startswith("cawg.")))
        need(known, 4, "consumption_term_unknown")
    need(asc(b["seen"]), 4, "consumption_seen_unordered")
    need(c["signer_id"] == did_key(bytes.fromhex(c["signer_pubkey"])), 4, "consumption_signer_not_principal")


def stage_layers(pt):
    """The layers a stage point names: exactly the member `wai-stage:layers`, 0 to 255."""
    if len(pt) != 1 or "wai-stage:layers" not in pt or pt["wai-stage:layers"] > 255:
        return None
    return pt["wai-stage:layers"]


STAGE_REFUSED_UNCHECKED = ("refines-malformed", "capability-not-refinement", "base-absent")  # §4.1 steps 1-2


def refused_unchecked(o):
    """An object the sink refused before it checked its refines against the set: its
    units are as its refines states them, and it is no layer of the set's."""
    return o["status"] == "refused" and o["reason"] in STAGE_REFUSED_UNCHECKED


def merge(spans):
    """Half-open intervals, sorted, merged, none empty."""
    out = []
    for a, e in sorted(x for x in spans if x[0] < x[1]):
        if out and a <= out[-1][1]:
            out[-1] = (out[-1][0], max(out[-1][1], e))
        else:
            out.append((a, e))
    return out


def length(v):
    return sum(e - a for a, e in v)


def intersect(v, span):
    a, e = span
    return [(max(x, a), min(y, e)) for x, y in v if max(x, a) < min(y, e)]


def subtract(v, w):
    out = []
    for a, e in v:
        at = a
        for x, y in w:
            if y <= at or x >= e:
                continue
            if x > at:
                out.append((at, x))
            at = max(at, y)
        if at < e:
            out.append((at, e))
    return out


def covered(objs, n):
    """How many of the units 0..n the objects' spans cover between them."""
    spans = sorted((min(o["units"]["from"], n), min(o["units"]["from"] + o["units"]["count"], n)) for o in objs)
    total, reach = 0, 0
    for a, e in spans:
        a = max(a, reach)
        if e > a:
            total += e - a
            reach = e
    return total


def delivered_stage(b, n):
    """§2.4.10: the highest k such that, at every layer from 1 to k, the objects listed
    with delivered-verified evidence — an object refused before its refines was checked
    excepted — cover every one of the n units."""
    k = 0
    for layer in range(1, 256):
        objs = [o for o in b["objects"] if o["layer"] == layer and o["evidence"] == "delivered-verified"
                and not refused_unchecked(o)]
        if covered(objs, n) < n:
            break
        k = layer
    return k


def stage_rules(b):
    """§2.4.10's rules, in order: the points, the stage hash, the units, the objects, the
    units against the objects, the delivered stage."""
    pts = [b[k] for k in ("op_requested", "op_committed", "op_delivered")]
    need(b["vocabulary"] == STAGE_VOCABULARY, 4, "stage_point_malformed")
    need(all(stage_layers(p) is not None for p in pts + [b["presented"]] if p is not None), 4, "stage_point_malformed")
    req, com, dlv = pts
    need(not ((req is not None or com is not None) and dlv is None), 4, "stage_point_incomplete")
    need((b["stage_hash"] is not None) == (b["tier"] == "decode-equivalence"), 4, "stage_hash_without_tier")
    ubs = b["units_by_stage"]
    n = sum(ubs)
    need(0 < len(ubs) <= 256 and n > 0 and ubs[-1] != 0, 4, "stage_units_malformed")
    lowest = next(i for i, c in enumerate(ubs) if c != 0)
    need(stage_layers(b["presented"]) == lowest and b["conflicts"] <= n and b["deadline_misses"] <= n,
         4, "stage_units_malformed")
    for o in b["objects"]:
        u, st, ev, r = o["units"], o["status"], o["evidence"], o["reason"]
        in_set = refused_unchecked(o) or u["from"] + u["count"] <= n
        least = 2 if st == "partial" else 1  # a partial object applied to some units of its span, not all
        need(1 <= o["layer"] <= 255 and u["count"] >= least and in_set, 4, "stage_objects_malformed")
        if st in ("applied", "late", "dropped"):
            need(r is None, 4, "stage_objects_malformed")
        else:
            need(r in STAGE_REASONS, 4, "stage_objects_malformed")
        if st in ("applied", "partial") and dlv is not None:
            need(ev == "delivered-verified", 4, "stage_objects_malformed")  # §10.1 over a base verified first
        if st == "dropped":
            need(ev != "delivered-verified", 4, "stage_objects_malformed")
        need((st == "dropped") == (o["bytes"] == 0), 4, "stage_objects_malformed")
    keys = [(o["layer"], o["units"]["from"], o["object"]) for o in b["objects"]]
    need(all(x < y for x, y in zip(keys, keys[1:])), 4, "stage_objects_malformed")
    need(sum(o["bytes"] for o in b["objects"]) < b["bytes_envelopes"], 4, "stage_objects_malformed")
    # Two objects of one layer are never both applied to a unit.
    for layer in {o["layer"] for o in b["objects"]}:
        spans = sorted((o["units"]["from"], o["units"]["from"] + o["units"]["count"]) for o in b["objects"]
                       if o["layer"] == layer and o["status"] == "applied")
        need(all(spans[i + 1][0] >= max(e for _, e in spans[:i + 1]) for i in range(len(spans) - 1)),
             4, "stage_objects_malformed")
    # The units against the objects, per unit: an applied object of layer s lies within the
    # spans of layer s - 1's applied and partial objects (every unit for layer 1), a partial
    # one meets them, and the units at stage s or above are no more than the units layer s's
    # applied objects cover and, of the rest below, at most each partial object's span less
    # one.
    reaching = [o for o in b["objects"] if o["status"] in ("applied", "partial")]
    top = max([o["layer"] for o in reaching] + [len(ubs) - 1])
    below = [(0, n)]
    for st in range(1, top + 1):
        layer = [o for o in reaching if o["layer"] == st]
        span = lambda o: (min(o["units"]["from"], n), min(o["units"]["from"] + o["units"]["count"], n))
        applied = merge([span(o) for o in layer if o["status"] == "applied"])
        free, partial_most = [], 0
        for o in layer:
            inside = intersect(below, span(o))
            if o["status"] == "applied":
                need(length(inside) >= o["units"]["count"], 4, "stage_units_unsupported")
            else:
                need(len(inside) > 0, 4, "stage_units_unsupported")
                open_ = subtract(inside, applied)
                partial_most += min(length(open_), o["units"]["count"] - 1)
                free += open_
        reach = length(applied) + min(length(merge(free)), partial_most)
        need(sum(ubs[st:]) <= reach, 4, "stage_units_unsupported")
        below = merge([span(o) for o in layer])
    if dlv is not None:
        need(stage_layers(dlv) == delivered_stage(b, n), 4, "stage_delivered_disagrees")


def verify_claim(text):
    """§3.1 steps 1-4. Returns the claim; raises Reject."""
    try:
        c = parse_strict(text)
    except ValueError:
        raise Reject(1, "malformed")
    need(isinstance(c, dict), 1, "malformed")
    if "kind" not in c:
        raise Reject(1, "missing_required_field")
    need(c["kind"] == "claim", 1, "wrong_kind")
    members(c, ENVELOPE, 1)
    need(isinstance(c["class"], str), 1)
    need(c["class"] != "delivery", 1, "class_reserved")
    need(c["class"] in ADMITS, 1, "unknown_class")
    need(c["subject_type"] in ("object", "receipt"), 1)
    need(is_hex(c["subject"], 32), 1)
    need(is_int(c["seq"]), 1)
    need(isinstance(c["issued_at"], str) and is_rfc3339(c["issued_at"]), 1)
    need(is_hex(c["signer_pubkey"], 32), 1)
    need(isinstance(c["signer_id"], str), 1)
    need(c["parent_receipt_hash"] is None or is_hex(c["parent_receipt_hash"], 32), 1)
    need(is_hex(c["sig"], 64), 1)
    need(c["subject_type"] in ADMITS[c["class"]], 1, "subject_type_not_admitted")
    check_body(c["class"], c["body"])
    check_signature(bytes.fromhex(c["signer_pubkey"]), bytes.fromhex(c["sig"]), signing_input(c))
    class_rules(c)
    return c


def check_chain(c, previous):
    """§3.1 step 5."""
    parent = c["parent_receipt_hash"]
    if previous is None:
        need(parent is None, 5, "parent_not_held")
        return
    need(parent is not None, 5, "chain_break")
    need(previous["signer_pubkey"] == c["signer_pubkey"] and parent == link_hash(previous)
         and c["seq"] > previous["seq"], 5, "chain_break")


def match(alg, declared, observed, exact_algs):
    need(alg in exact_algs, 6, "recovery_no_matcher")
    need(declared == observed, 6, "recovery_value_mismatch")


def dig(v, *path):
    """The member at `path` through JSON objects, or None."""
    for k in path:
        v = v.get(k) if isinstance(v, dict) else None
    return v


def figure_rounds_to_micro(figure, uj):
    """§2.4.5 check 4: whether an OER receipt's joules_total is `uj` microjoules to the
    nearest microjoule, a figure half-way between two admitting either.

    The figure is read as the OER reference reads a quantity for its own composition
    checks (`OER.dec`: a decimal string of at most 34 significand digits, or an integer
    within I-JSON's range) and compared in its exact context, so no digit is rounded
    away: 0.0120004 followed by twenty-five nines is below half-way, although rounding it
    to 28 digits would make it a tie. Check 2 has already refused a receipt whose figure
    is anything else; such a value is no figure here either."""
    try:
        with localcontext(OER.EXACT):
            return abs(OER.dec(figure) * 1000000 - uj) * 2 <= 1
    except TypeError:
        return False


def check_context(c, ctx):
    """§3.1 step 6."""
    b = c["body"]
    if "object" in ctx:
        o = ctx["object"]
        # The corpus gives a delivered object's microjoules as an integer within 64 bits, as
        # wai-rs's harness requires; anything else (`-0`, which is read as a double, among
        # them) is a malformed corpus, not a verdict.
        j = o.get("joules_micro")
        if not (isinstance(j, int) and not isinstance(j, bool) and 0 <= j < 2 ** 64):
            raise ValueError("bad object.joules_micro")
        need(c["class"] == "energy" and o["content_hash"] == c["subject"], 6, "context_mismatch")
        need(o["joules_micro"] == b["joules_micro"], 6, "energy_disagrees_with_delivery")
        label = class_of_json(o)  # the label in the object's leaf, where its leaf is labelled
        need(label is None or label == class_of_body(b), 6, "energy_class_disagrees_with_delivery")
    if "figure" in ctx:
        check_figure(c, figure_view(ctx["figure"]))
    elif c["class"] == "measurement":
        # §2.4.8: a report is its figure signer's only once step 6 has run against the
        # figure (check 5); checked without it, it is not accepted.
        raise Reject(6, "figure_not_held")
    if "oer" in ctx:
        r, keys = ctx["oer"]["receipt"], ctx["oer"]["keys"]
        need(c["class"] == "energy" and b["oer_receipt"] is not None
             and dig(r, "receipt_id") == b["oer_receipt"], 6, "context_mismatch")
        # The reference reports every failure as a reason (oer-receipts §7); an exception
        # here is a defect in it, and is left to surface as one rather than read as a verdict.
        need(OER.verify_receipt(r, keys) == [], 6, "energy_oer_invalid")
        need(dig(r, "subject", "output") == c["subject"], 6, "context_mismatch")
        need(dig(r, "provenance", "how") == b["grade"], 6, "energy_disagrees_with_oer")
        need(figure_rounds_to_micro(dig(r, "energy", "joules_total"), b["joules_micro"]), 6, "energy_disagrees_with_oer")
    if "gap_object" in ctx:
        # §2.4.11: the object receipt at the claim's coordinates, as the verifier holds it
        # once verified (by inclusion at least).
        g = ctx["gap_object"]
        need(c["class"] == "delivery-gap", 6, "context_mismatch")
        need(all(g[k] == b[k] for k in ("track_alias", "group_id", "object_id")), 6, "context_mismatch")
        need(g["content_hash"] == c["subject"], 6, "gap_subject_mismatch")
    # §2.4.12: a session against its outputs, the objects it received, the WAI2 envelopes
    # it received and the stage claims it names, in that order.
    ps = presentations(ctx["outputs"]) if "outputs" in ctx else None
    if ps is not None:
        red = ctx.get("redecoded", [])
        if not isinstance(red, list):
            raise ValueError("redecoded must be an array")
        found = {}
        for x in red:
            found[ctx_int(x["index"])] = ctx_hash(x["out_digest"])
        check_session_outputs(c, ps, found)
    objs = objects_seen(ctx["objects"]) if "objects" in ctx else None
    if objs is not None:
        check_session_objects(c, objs)
    if "envelopes" in ctx:
        check_session_envelopes(c, envelopes(ctx["envelopes"]), objs, ps)
    if "stage_claims" in ctx:
        check_session_stage_claims(c, ctx["stage_claims"], ps, objs)
    if "declaration" in ctx:
        # In §3.1 step 6's order: the claim must be one that names a declaration; the
        # declaration must itself pass steps 1-4 (context_invalid) — its signer_pubkey is
        # only its signer's once that is established; then it must be of the class named.
        recovery = c["class"] == "binding-recovery"
        decision = c["class"] == "lawful-basis" and b["record"] == "decision"
        # §2.6.3: the declaration is context only together with the publisher key of a
        # verified delivery receipt for its subject. A corpus that gives one without the
        # other is malformed, not a verdict.
        if recovery and "publisher" not in ctx:
            raise ValueError("a recovery record's declaration is context only with a publisher key")
        need(recovery or decision, 6, "context_mismatch")
        if not isinstance(ctx["declaration"], str):
            raise TypeError("a declaration is given as its wire form")
        try:
            d = verify_claim(ctx["declaration"])
        except Reject:
            raise Reject(6, "context_invalid")
        db = d["body"]
        if recovery:
            need(d["class"] == "soft-binding", 6, "context_mismatch")
            need(b["declaration"] == link_hash(d) and c["subject"] == d["subject"]
                 and b["alg"] == db["alg"] and b["binding_type"] == db["binding_type"],
                 6, "recovery_declaration_mismatch")
            need(d["signer_pubkey"] == ctx["publisher"], 6, "recovery_signer_mismatch")
            match(db["alg"], db["value"], b["value"], ctx.get("exact_match_algs", []))
        else:
            need(d["class"] == "lawful-basis" and db["record"] == "declaration", 6, "context_mismatch")
            need(b["declaration"] == link_hash(d) and c["subject"] == d["subject"]
                 and c["subject_type"] == d["subject_type"] and b["vocabulary"] == db["vocabulary"],
                 6, "decision_declaration_mismatch")
            need(any(p["purpose"] == b["purpose"] for p in db["purposes"]), 6, "decision_purpose_not_declared")


def class_of_json(m):
    """A class as energy-measurement §2.7 reads one from a receipt's JSON members: None
    for no class; (label, ppm, window, filtering, reference) otherwise, an on-chip
    counter's absent filtering state read as Unknown. A label that cannot be read is a
    malformed corpus (ValueError), not a verdict."""
    label = m.get("energy_provenance")
    u, f, r = m.get("energy_uncertainty"), m.get("energy_counter_filtering"), m.get("energy_calibration_ref")
    if label is None:
        if u is not None or f is not None or r is not None:
            raise ValueError("evidence without a class")
        return None
    if label in ("OnChipCounter", "CalibratedInstrument"):
        if not isinstance(u, dict):
            raise ValueError("a refined class without its uncertainty")
        ppm, win = u["relative_ppm"], u["window_us"]
        if label == "OnChipCounter":
            return (label, ppm, win, "Unknown" if f is None else f, None)
        return (label, ppm, win, None, r)
    if label not in STRONGEST:
        raise ValueError("unknown class " + str(label))
    return (label, None, None, None, None)


def class_of_body(b):
    """The class a claim's body restates, in class_of_json's form (step 4 has admitted it)."""
    return class_of_json(b)


def figure_view(f):
    """A `figure` context as (profile, identity, group, {index: (joules, class)}, signer),
    or None for a receipt that does not verify. A `claim` is an energy claim's wire form,
    viewed under `moq-jwp:claim-link`; a `view` is given as the verifier holds it."""
    if "claim" in f:
        if not isinstance(f["claim"], str):
            raise TypeError("a figure claim is given as its wire form")
        try:
            e = verify_claim(f["claim"])
        except Reject:
            return None
        figs = {}
        if e["class"] == "energy":
            figs[None] = (e["body"]["joules_micro"], class_of_body(e["body"]))
        elif e["class"] == "reconstruct-session" and e["body"]["energy"] is not None:
            # energy-measurement §6.1: a session's figure is its `energy`, with its class.
            figs[None] = (e["body"]["energy"]["joules_micro"], class_of_json(e["body"]["energy"]))
        return ("moq-jwp:claim-link", link_hash(e), None, figs, e["signer_pubkey"])
    v = f["view"]
    figs = {x["index"]: (x["joules_micro"], class_of_json(x)) for x in v["figures"]}
    return (v["profile"], v["identity"], v["group"], figs, v["signer"])


def check_figure(c, view):
    """§2.4.8 / §2.4.9 against the figure, in order."""
    b = c["body"]
    need(c["class"] in ("measurement", "operating-point"), 6, "context_mismatch")
    need(view is not None, 6, "context_invalid")
    profile, identity, group, figs, signer = view
    need(profile == b["figure_profile"] and identity == c["subject"] and group == b["group"]
         and b["figure_index"] in figs, 6, "context_mismatch")
    joules, cls = figs[b["figure_index"]]
    need(joules == b["joules_micro"], 6, "figure_disagrees")
    need(cls is not None, 6, "figure_unclassified")
    need(cls == class_of_body(b), 6, "figure_class_disagrees")
    if c["class"] == "measurement":
        need(PROFILES[b["figure_profile"]][2], 6, "figure_signer_unknown")
        need(signer == c["signer_pubkey"], 6, "figure_signer_mismatch")


def held_previous(ctx):
    """§3.1 step 5's previous claim: one that does not itself pass steps 1-4 is not held —
    failing steps 1-3 it is not known to be its signer's, failing step 4 it breaks its
    class's rules."""
    if "previous" not in ctx:
        return None
    if not isinstance(ctx["previous"], str):
        raise TypeError("a previous claim is given as its wire form")
    try:
        return verify_claim(ctx["previous"])
    except Reject:
        return None


def recover(ctx):
    """§3.2. Returns the ancestor; raises Reject."""
    d = verify_claim(ctx["declaration"])
    need(d["class"] == "soft-binding", 6, "context_mismatch")
    need(d["signer_pubkey"] == ctx["publisher"], 6, "recovery_signer_mismatch")
    need(ctx.get("observed_hash") != d["subject"], 6, "recovery_of_hard_bound_object")
    match(d["body"]["alg"], d["body"]["value"], ctx["observed_value"], ctx.get("exact_match_algs", []))
    return d["subject"]


def verdict(v):
    """This verifier's verdict on one vector: `accept link_hash=<hex>` for an accepted
    claim, `accept ancestor=<hex>` for a recovery, or `reject step=<n> reason=<label>`."""
    ctx = v.get("context", {})
    try:
        if v["procedure"] == "recover":
            return "accept ancestor=" + recover(ctx)
        c = verify_claim(v["receipt"])
        check_chain(c, held_previous(ctx))
        check_context(c, ctx)
        return "accept link_hash=" + link_hash(c)
    except Reject as r:
        return f"reject step={r.step} reason={r.reason}"


def run_vector(v):
    got = verdict(v)
    if got.startswith("accept link_hash="):
        # An accepted claim must also reproduce its pinned link hash, when one is pinned.
        pinned = v["expected"].get("link_hash")
        return "accept" if pinned in (None, got[len("accept link_hash="):]) else got
    return got


def verdicts(path):
    """`name<TAB>verdict` for every vector in a corpus-format file, ignoring any declared
    outcome: the input to a differential comparison with wai-rs (`differential.py`).
    A vector this script cannot interpret is `error`; an exception of any other kind is
    reported as `crash: <type>`, which a conforming run never prints."""
    try:
        doc = OER.load(open(path, encoding="utf-8"))
    except ValueError as e:
        print(f"{path}: not read: {e}", file=sys.stderr)
        return 1
    for v in doc["vectors"]:
        try:
            got = verdict(v)
        except UnicodeError as e:
            got = "crash: " + type(e).__name__
        except (ValueError, KeyError, TypeError):
            got = "error"
        except Exception as e:  # noqa: BLE001 -- reported, never swallowed
            got = "crash: " + type(e).__name__
        print(f"{v['name']}\t{got}")
    return 0


def expected_of(e):
    if e["outcome"] == "accept":
        return "accept ancestor=" + e["ancestor"] if "ancestor" in e else "accept"
    return f"reject step={e['step']} reason={e['reason']}"


# ------------------------------------------------------------------ canonical fixture

FIXTURE_SEED = bytes([0x22] * 32)


def fixture():
    """The canonical fixture wai-rs pins: computed here, from the members and the seed."""
    sk = ed25519.Ed25519PrivateKey.from_private_bytes(FIXTURE_SEED)
    pk = sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex()
    c = {
        "signer_id": "did:example:météo\u001f",
        "body": {"work_unit": "subpixel", "oer_receipt": None, "energy_calibration_ref": None, "joules_micro": 4096,
                 "grade": "measured", "energy_uncertainty": {"window_us": 1000000, "relative_ppm": 20000},
                 "method": "https://transaction.science/wai/extensions/energy-measurement",
                 "coverage": "Whole", "energy_counter_filtering": "Unknown", "work_units": 49152,
                 "energy_provenance": "OnChipCounter"},
        "seq": 7, "parent_receipt_hash": None, "kind": "claim",
        "subject": blake3(b"moq-jwp:object\x01" + b"claim-conformance: canonical fixture object").hex(),
        "issued_at": "2026-09-27T12:00:00.5+02:00", "class": "energy", "subject_type": "object",
        "signer_pubkey": pk,
    }
    c["sig"] = sk.sign(signing_input(c)).hex()
    return c, jcs(c), link_hash(c)


def fixture_matches_wai_rs():
    """The literals `check_canonical_form` pins in wai-rs equal what this script computes."""
    src_path = os.path.join(HERE, "..", "wai-rs", "src", "claim_receipt.rs")
    try:
        src = open(src_path, encoding="utf-8").read()
    except OSError as e:
        return [f"cannot read {src_path}: {e}"]
    c, canon, link = fixture()
    fails = []
    m = re.search(r'const CANONICAL: &str = "((?:[^"\\]|\\.)*)";', src)
    pinned = re.sub(r'\\(["\\])', r"\1", m.group(1)) if m else None
    if pinned != canon:
        fails.append("canonical bytes differ from wai-rs")
    m = re.search(r'const LINK: &str = "([0-9a-f]{64})";', src)
    if not m or m.group(1) != link:
        fails.append("link hash differs from wai-rs")
    if f'"sig": "{c["sig"]}"' not in src:
        fails.append("signature differs from wai-rs")
    return fails


def bracket_fixture():
    """The brackets wai-rs pins in `energy_report::pinned_brackets`, read from its source,
    with the figure and least declaration this script computes for each:
    [(name, values, F, p)]."""
    src_path = os.path.join(HERE, "..", "wai-rs", "src", "energy_report.rs")
    src = open(src_path, encoding="utf-8").read()
    body = src[src.index("fn pinned_brackets()"):]
    body = body[:body.index("\n}\n")]
    out = []
    num = r"(x|[0-9_]+)"
    pat = (r'\("(\w+)", v\(s\(' + num + r', ' + num + r', ' + num + r'\), s\(' + num + r', ' + num + r', ' + num
           + r'\), ' + num + r', ' + num + r', ' + num + r', (FromRest|Running)\), ' + num + r', ' + num + r'\)')
    for m in re.finditer(pat, body):
        g = [MAX_INT if x == "x" else int(x.replace("_", "")) if x not in ("FromRest", "Running") else x for x in m.groups()[1:]]
        br = {"idle": {"n": g[0], "mean_uw": g[1], "sd_uw": g[2]}, "busy": {"n": g[3], "mean_uw": g[4], "sd_uw": g[5]},
              "window_us": g[6], "completions": g[7], "in_flight": g[8],
              "opening": "from-rest" if g[9] == "FromRest" else "running"}
        out.append((m.group(1), br, g[10], g[11]))
    return out


def brackets_match_wai_rs():
    """The figures and least declarations wai-rs pins equal what this script computes, and
    one ppm less is not supported."""
    fails, cases = [], bracket_fixture()
    if len(cases) < 7:
        return ["fewer pinned brackets than expected were read from energy_report.rs"]
    for name, br, f, p in cases:
        got_f = bracket_figure(br)
        got_p = least_ppm(br, f) if got_f is not None else None
        if got_f != f or got_p != p:
            fails.append(f"{name}: computed F={got_f} p={got_p}, pinned F={f} p={p}")
    return fails


def consumption_claims(path):
    """§3.1 steps 1-4 over the `consumption` claims of the AI usage corpus
    (live-conformance/usage/vectors.json, ai-usage §10): a claim stated refused at one of those
    steps is refused there with the stated reason, and every other claim passes them."""
    with open(path, encoding="utf-8") as f:
        doc = json.load(f)
    fail = 0
    for c in doc["claims"]:
        o = c["outcome"]
        want = (o["step"], o["reason"]) if o["outcome"] == "refuse" and o["step"] <= 4 else None
        try:
            verify_claim(c["wire"])
            got = None
        except Reject as r:
            got = (r.step, r.reason)
        fail += got != want
        print(f"{'PASS' if got == want else 'FAIL'} {c['name']}  expected={want}  got={got}")
    print(f"{len(doc['claims'])} consumption claims, {fail} failure(s)")
    return 1 if fail else 0


def bearer_claims(path):
    """§3.1 steps 1-4 over the `transmission` and `reconciliation` claims of the bearer corpus
    (live-conformance/bearers/vectors.json, bearer-receipts §11): a claim the corpus states
    refused at one of those steps is refused there with the stated reason, and every other
    claim of those classes passes them. Steps 5 and 6 are the bearer verifier's."""
    with open(path, encoding="utf-8") as f:
        doc = json.load(f)
    fail = n = 0
    for case in doc["cases"]:
        if case["outcome"]["outcome"] != "verdict":
            continue
        for wire, stated in zip(case["claims"], case["outcome"]["claims"]):
            m = re.fullmatch(r"reject step=(\d) reason=(\S+)", stated)
            want = (int(m[1]), m[2]) if m and int(m[1]) <= 4 else None
            try:
                c = verify_claim(wire)
                if c["class"] not in ("transmission", "reconciliation"):
                    continue
                got = None
            except Reject as r:
                got = (r.step, r.reason)
            n += 1
            fail += got != want
            print(f"{'PASS' if got == want else 'FAIL'} {case['name']}  expected={want}  got={got}")
    print(f"{n} bearer claims, {fail} failure(s)")
    return 1 if fail or n == 0 else 0


def main(argv):
    if argv[1:2] == ["brackets"]:
        for name, br, _, _ in bracket_fixture():
            f = bracket_figure(br)
            print(name, f, least_ppm(br, f))
        return 0
    if argv[1:2] == ["verdicts"] and len(argv) == 3:
        return verdicts(argv[2])
    if argv[1:2] == ["consumption"] and len(argv) == 3:
        return consumption_claims(argv[2])
    if argv[1:2] == ["bearer"] and len(argv) == 3:
        return bearer_claims(argv[2])
    if argv[1:2] == ["fixture"]:
        c, canon, link = fixture()
        print("sig      ", c["sig"])
        print("canonical", canon)
        print("link_hash", link)
        return 0
    fail = 0
    # BLAKE3's published values for the empty input and "abc", before anything relies on it.
    b3 = blake3(b"").hex() == "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262" and \
        blake3(b"abc").hex() == "6437b3ac38465133ffb63b75273a8db548c558465d79db03fd359c6cd5bd9d85"
    print(("PASS" if b3 else "FAIL") + " BLAKE3 reproduces its published values")
    fail += not b3
    ff = fixture_matches_wai_rs()
    print(("PASS" if not ff else "FAIL") + " canonical fixture equals the literals pinned in wai-rs " + ",".join(ff))
    fail += bool(ff)
    bm = brackets_match_wai_rs()
    print(("PASS" if not bm else "FAIL") + " bracket arithmetic equals the literals pinned in wai-rs " + ",".join(bm))
    fail += bool(bm)
    rl = SV.reader_limit_failures(lambda b: wai2_read(b) is not None, True)
    print(("PASS" if not rl else "FAIL") + " wai2_read refuses a manifest past the reader limits (SPEC §2) " + "; ".join(rl))
    fail += bool(rl)
    rd = OER.check_canonical_form()  # the OER reference's own fixtures, its reading of -0 among them
    print(("PASS" if not rd else "FAIL") + " the OER reference's canonical form and number reading hold " + ",".join(rd))
    fail += bool(rd)
    path = argv[1] if len(argv) > 1 else os.path.join(HERE, "vectors.json")
    doc = OER.load(open(path, encoding="utf-8"))
    for v in doc["vectors"]:
        want, got = expected_of(v["expected"]), run_vector(v)
        ok = want == got
        fail += not ok
        print(f"{'PASS' if ok else 'FAIL'} {v['name']}  expected={want}  got={got}")
    print(f"{len(doc['vectors'])} vectors, {fail} failure(s)")
    return 1 if fail else 0


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