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

#!/usr/bin/env python3
"""Differential check: wai-rs and claim_verify.py give the same verdict on inputs the corpus
does not hold.

The corpus pins one declared outcome per vector, which shows both verifiers agree on those
vectors and nothing more. This script generates inputs around the edges where two readings
of jwp-receipts §3.1-§3.2 could part — JSON numbers and strings at the limits of what a
parser accepts, nesting, calendar dates, URI authorities, acquisition classes with every
combination of evidence around energy-measurement §2.3-§2.5's rules, OER figures near a
half-microjoule and at the edges of oer-receipts §4's quantities, OER receipts whose sums
and floors sit at §6's tolerances, OER receipts holding numbers §4 forbids in a body
(`-0`, fractions, integers beyond I-JSON's range) in members of every kind, context
receipts that do not themselves verify, and bodies of every class with one defect, or two
at different positions, where the order §3.1 step 2 reads a body in decides the reason;
measurement reports whose brackets sit at the least declaration and one ppm below it, at a
half microjoule, at the ratio thresholds, with a flat idle bracket, with products far past
128 bits and with every pattern of liveness facts; operating-point claims with every
presence combination of their points and prediction; reports checked against figure views
that agree and disagree in each member; energy claims against labelled objects; stage
claims around every rule of §2.4.10 — points, the stage hash, the units, the objects'
order, statuses, evidence, reasons and bytes, and the delivered stage they give — and
delivery-gap claims against object receipts at their coordinates and elsewhere; and
reconstruct-session claims built from random presentation lists, around every rule of
§2.4.12 — units, leaves and paths off by one, switches and missed presentations at and past
their caps, a null deadline with misses, presences that disagree, decoder indices out of
range, leaves moved between entries, entries out of first-use order, every acquisition
class — with their outputs, re-decodes, objects, entry envelopes and stage claims that
reach their stages or miss them, and sessions holding a base's and its layers' envelopes
— WAI1 envelopes or a WAI2 envelope's components, a base named by its exploded
envelope or by its envelope and rendition, a base whose fallback is of another format,
some layers a sink cannot apply —
presented within and past the layers and units each envelope gives, through entries at
their base's first unit or another, with objects whose figures are each envelope's or
told otherwise —
writes them in the corpus's format with no declared outcome, runs both verifiers'
`verdicts` mode, and reports every input on which they differ.

Some inputs cannot share a file with the rest: a number whose nearest double is infinite,
`NaN`, an unpaired surrogate, a repeated member name or nesting past 64 levels makes a
whole file unreadable (oer-receipts §4 "The document"), for both verifiers or — if they
differ — for one. Each of those is written to a file of its own with a readable twin beside
it, and the two verifiers must either both refuse the file or give the same verdicts on it.

    python3 differential.py <wai_claim_vectors binary> [--random N] [--seed S] [--keep FILE]

Keys are the corpus's fixed seeds, so every signed input is reproducible. Exit status 0
only if the two verdict lists are identical, every file of its own is read or refused
alike, and neither verifier printed `crash`; any difference, whatever step it is at,
exits 1.

An energy claim that names an OER receipt reaches the OER verifier of its own language
(§2.4.5 check 2): wai-rs's `crate::oer`, and the reference `oer2_verify.py`. A difference
there is still a difference in the claim verdict, and is reported like any other; the
listing marks it, since its cause is in the two OER verifiers rather than the claim rules.
Those two are compared on their own by `../oer-conformance/fuzz_differential.py`.
"""
import argparse, base64, copy, hashlib, io, json, os, random, struct, subprocess, sys, tempfile
from decimal import Decimal, localcontext

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import claim_verify as CV  # noqa: E402
SV = CV.SV  # the strict JSON, manifest and envelope readers claim_verify.py uses

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

OER = CV.OER


def key(seed):
    return ed25519.Ed25519PrivateKey.from_private_bytes(bytes([seed] * 32))


def pk(k):
    return k.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex()


PUBLISHER, METER, OPERATOR, VIEWER, RESOLVER, RELAY = (key(s) for s in (0x11, 0x22, 0x33, 0x44, 0x55, 0x66))
OER_SIGNER = "ed25519:claim-conformance-meter"
WM_ALG = "org.example.watermark.1"


def sign(c, k):
    """`c` with its signature made under `k` over §2.4.3's signing input."""
    c = copy.deepcopy(c)
    c.pop("sig", None)
    c["signer_pubkey"] = pk(k)
    c["sig"] = k.sign(CV.signing_input(c)).hex()
    return c


RAW = {}


def raw(spelling):
    """A JSON value written into the input file exactly as `spelling` — `-0`, an integer
    beyond 64 bits, a string with an unpaired surrogate escape — where json.dumps would
    write another spelling or none. A placeholder string until text_of() writes it."""
    k = f"\u0000raw{len(RAW)}\u0000"
    RAW[k] = spelling
    return k


def text_of(v, **kw):
    """json.dumps(v, ensure_ascii=True), every raw() placeholder written as its spelling."""
    t = json.dumps(v, ensure_ascii=True, **kw)
    if "\\u0000raw" in t:
        for k, spelling in RAW.items():
            t = t.replace(json.dumps(k), spelling)
    return t


def seal_oer(template, joules_total, terms=None, floor=None, edit=None):
    """`template` recording `joules_total`, re-sealed and signed as oer-receipts §4-§5
    define. The breakdown is `terms` (compute_j first) or, by default, `joules_total` as
    compute_j with the other terms as the template has them (zero); `floor`, if given, is
    (bits_committed, temperature_k, floor_j); `edit`, if given, changes the receipt last.

    The content address is computed over the receipt as the file will be read: its text,
    read as §4 reads a document (`OER.load`). A receipt whose body holds a number §4
    forbids is sealed all the same; both verifiers must refuse it before its address."""
    r = copy.deepcopy(template)
    r.pop("chain", None)
    r.pop("receipt_id", None)
    r["energy"]["joules_total"] = joules_total
    b = r["energy"]["breakdown"]
    for k, t in zip(list(b), terms if terms is not None else [joules_total]):
        b[k] = t
    if floor is not None:
        r["floor"]["bits_committed"], r["energy"]["temperature_k"], r["floor"]["floor_j"] = floor
    if edit is not None:
        edit(r)
    body = OER.canonical_bytes(OER.load(io.StringIO(text_of(r))))
    prev = "sha256:" + "0" * 64
    rh = OER.h("sha-256", prev.encode() + body)
    r["chain"] = {"prev_hash": prev, "receipt_hash": rh, "signer": OER_SIGNER,
                  "signature": METER.sign(rh.encode()).hex()}
    r["receipt_id"] = OER.h("sha-256", body)
    return r


class Inputs:
    def __init__(self, corpus):
        self.by_name = {v["name"]: v for v in corpus["vectors"]}
        self.out = []

    def receipt(self, name):
        return json.loads(self.by_name[name]["receipt"])

    def wire(self, name):
        return self.by_name[name]["receipt"]

    def add(self, name, receipt=None, context=None, procedure="claim"):
        v = {"name": f"{len(self.out):05d}_{name}", "procedure": procedure}
        if receipt is not None:
            v["receipt"] = receipt
        if context:
            v["context"] = context
        self.out.append(v)


# ------------------------------------------------------------------ input families

NUMBER_TOKENS = [
    "0", "-0", "1", "-1", "9007199254740991", "9007199254740992", "18446744073709551615",
    "18446744073709551616", "1" + "0" * 30, "1" + "0" * 308, "1" + "0" * 309, "1" + "0" * 400,
    "-" + "1" + "0" * 400, "0.0", "-0.0", "1.0", "1e0", "1E2", "1e-2", "1e308", "1e309", "1e400",
    "-1e400", "1E400", "1e-400", "1.7976931348623157e308", "1.7976931348623158e308",
    "1.7976931348623159e308", "1.8e308", "2e308", "1e99999999999999999999",
    "179769313486231580793728971405303415079934132710037826936173778980444968292764750946649017977587207096330286416692887910946555547851940402630657488671505820681908902000708383676273854845817711531764475730270069855571366959622842914819860834936475292719074168444365510704342711559699508093042880177904174497791",
    "01", "+1", ".5", "1.", "1e", "1e+", "--1", "0x10", "NaN", "Infinity", "-Infinity", "1_0",
    "１", "١",
]

STRING_BODIES = [
    "\\ud800", "\\udc00", "\\udbff\\udfff", "\\ud800\\u0041", "\\ud83d\\ude00", "\\ud83d", "a\\u0000b",
    "\\u00e9", "\\u00E9", "\\/", "\x01", "\x7f", "
", "", "￿", "﷐", "\\ud800x",
    "\\uDFFF", "\\u", "\\u12", "\\x41", "\\'", "tab\there",
]


def numbers(ins):
    wire = ins.wire("energy_accept")
    for t in NUMBER_TOKENS:
        ins.add("seq_" + t[:24], wire.replace('"seq":0', '"seq":' + t, 1))
        ins.add("work_units_" + t[:24], wire.replace('"work_units":49152', '"work_units":' + t, 1))


def strings(ins):
    wire = ins.wire("energy_accept")
    for s in STRING_BODIES:
        ins.add("signer_id_" + s.encode("unicode_escape").decode()[:24],
                wire.replace('"did:example:meter"', '"' + s + '"', 1))
        ins.add("member_name_" + s.encode("unicode_escape").decode()[:24],
                wire.replace('{"body":', '{"' + s + '":1,"body":', 1))
    # A validly signed claim whose signer_id holds characters outside the BMP and a control
    # character, sent with every such character escaped: canonicalisation undoes escapes.
    c = sign(dict(ins.receipt("energy_accept"), signer_id="did:example:m\U0001F600é\u001f"), METER)
    escaped = json.dumps(c, ensure_ascii=True, separators=(", ", ": "))
    ins.add("escaped_astral_signer_id", escaped)
    ins.add("canonical_astral_signer_id", CV.jcs(c))
    ins.add("bom_prefixed", "" + wire)
    ins.add("trailing_garbage", wire + " x")
    ins.add("trailing_whitespace", wire + " \n\t\r")
    ins.add("nbsp_whitespace", " " + wire)
    ins.add("empty", "")


def nesting(ins):
    wire = ins.wire("energy_accept")
    for d in list(range(1, 40)) + [63, 64, 65, 126, 127, 128, 129, 200, 999, 1001, 5000]:
        ins.add(f"nesting_{d}", wire.replace('"did:example:meter"', "[" * d + "]" * d, 1))
    ins.add("nesting_obj_40", wire.replace('"did:example:meter"', '{"a":' * 40 + "1" + "}" * 40, 1))


def dates(ins, rng, n):
    base = ins.receipt("energy_accept")
    fixed = ["2026-02-28T00:00:00Z", "2026-02-29T00:00:00Z", "2028-02-29T00:00:00Z", "2100-02-29T00:00:00Z",
             "2000-02-29T00:00:00Z", "1900-02-29T00:00:00Z", "0000-02-29T00:00:00Z", "0004-02-29T00:00:00Z",
             "2026-04-31T00:00:00Z", "2026-04-30T23:59:60Z", "2026-12-31T23:59:60+14:00", "2026-01-00T00:00:00Z",
             "2026-00-10T00:00:00Z", "2026-13-01T00:00:00Z", "2026-09-27T24:00:00Z", "2026-09-27T12:60:00Z",
             "2026-09-27T12:00:61Z", "2026-09-27T12:00:00.Z", "2026-09-27T12:00:00.000000001Z",
             "2026-09-27t12:00:00z", "2026-09-27T12:00:00-00:00", "2026-09-27T12:00:00+24:00",
             "2026-09-27T12:00:00+23:60", "2026-09-27 12:00:00Z", "2026-9-27T12:00:00Z",
             "２０２６-09-27T12:00:00Z", "2026-09-2٧T12:00:00Z", "2026-09-27T12:00:00Ｚ",
             "+2026-09-27T12:00:00Z", "2026-09-27T12:00:00Z\n", "2026-09-31T12:00:00Z", "2026-06-31T12:00:00Z",
             "2026-11-31T12:00:00Z", "2026-01-31T12:00:00Z", "2026-03-31T12:00:00Z"]
    cases = list(fixed)
    for _ in range(n):
        y, mo, d = rng.randrange(0, 10000), rng.randrange(0, 14), rng.randrange(0, 33)
        cases.append(f"{y:04d}-{mo:02d}-{d:02d}T{rng.randrange(0, 25):02d}:{rng.randrange(0, 61):02d}:"
                     f"{rng.randrange(0, 62):02d}{rng.choice(['Z', '+05:30', '-12:00', '.5Z', 'z'])}")
    for t in cases:
        ins.add("issued_at_" + t.encode("unicode_escape").decode()[:32], CV.jcs(sign(dict(base, issued_at=t), METER)))


def endpoints(ins, rng, n):
    base = ins.receipt("soft_binding_accept")
    fixed = ["https://resolver.example.org/matches", "https://:443/m", "https://:/m", "https://resolver.example.org:/m",
             "https://resolver.example.org:8443/m", "https://resolver.example.org:x/m", "https://resolver.example.org:1:2/m",
             "https://[2001:db8::1]:8443/m", "https://[2001:db8::1]/m", "https://[]/m", "https://[/m", "https://[::1/m",
             "https://::1]/m", "https://[::1]x/m", "https://[::1]:/m", "https://[::1]:x/m", "https://h]:1/m",
             "https://[email protected]/m", "https://@/m", "https://resolver.example.org",
             "https://resolver.example.org?q", "https://resolver.example.org#f", "https://?q", "https://#f", "https://",
             "https:///m", "HTTPS://resolver.example.org/m", "http://resolver.example.org/m", "https:resolver.example.org",
             "https://résolver.example.org/m", "https://resolver.example.org:８/m", "https://1.2.3.4:5/m",
             "https://%72esolver.example.org/m", "https://a/b@c"]
    alphabet = "ab:[]@/?#1%."
    cases = list(fixed) + ["https://" + "".join(rng.choice(alphabet) for _ in range(rng.randrange(0, 9))) for _ in range(n)]
    for e in cases:
        body = dict(base["body"], resolution_endpoint=e)
        ins.add("endpoint_" + e.encode("unicode_escape").decode()[:32], CV.jcs(sign(dict(base, body=body), PUBLISHER)))


FIGURES = ["0.0120005", "0.012000500000000000000000000000000", "0.0120004999999999999999999999999999",
           "0.0120005000000000000000000000000000", "0.012000500000000000000000000000000000000000000",
           "0.0120004999999999999999999999999999999999999999", "0.012000", "0.0120004", "0.0120006", "1.20005e-2",
           "120005e-7", "1.20005E-2", "1.20005e-02", "1.20005e-002", "1.20005e-0002", "12000.5e-6",
           "0.0000000000000000000000000000000000000000000000120005e43",
           "+0.0120005", " 0.0120005", "0.0120005 ", "00.0120005", ".0120005", "0.0120005e", "0_0120005", "1_2",
           "０.0120005", "NaN", "Infinity", "-0.0120005", "0", "-0", "0e0", "1e-999", "1e999", "1e-1000",
           "12000500000000000000000000000000000000000000000e-51", "0.00000050", "0.0000005", "0.0000015", "0.0000025",
           "18446744073709.5516155", "18446744073709.551615", "9007199254.740991", "9007199254.7409915"]
# JSON numbers: §4 admits integers within I-JSON's range in a body and nothing else, so every
# fraction, `-0` and integer beyond the range here must make the receipt malformed. Among
# them, doubles half-way between two shortest decimals, which a formatter could write two
# ways — no receipt body may carry one, so no content address may depend on it.
NUMERIC_FIGURES = [0.0120005, 0.012, 0.0120004, 0.0120006, 0.0000025, 1.5e-6, 2.5e-6, 12, 0, -0.0120005, 1e-300,
                   9007199254.740991, 1.2000500000000001e-2, True, False, None, -0.0,
                   1e37, 9.999999999999999e37, 1e38, 1e-38, 9.9e-39, 1.7976931348623157e308, 5e-324,
                   2 ** 53 - 1, 2 ** 53, -(2 ** 53 - 1), -(2 ** 53), 2 ** 63 - 1, 2 ** 64 - 1, 2 ** 64, 10 ** 30,
                   75.036163330078125, 2365.15167236328125, 5.9604644775390625e-8, 983123238.6575775]
# §4's decimal strings at their limits: 34 significand digits and a three-digit exponent.
RANGE_FIGURES = ["9" * 34, "9" * 35, "0." + "0" * 32 + "1", "0." + "0" * 33 + "1", "-" + "9" * 34, "1e999", "1e-999",
                 "9.999999999999999999999999999999999e999", "9.9999999999999999999999999999999999e999",
                 "0.0120004" + "9" * 26, "0.0120004" + "9" * 27, "0.0120005" + "0" * 26, "0.0120005" + "0" * 26 + "1"]


LABELS = ["OnChipCounter", "CalibratedInstrument", "ModelBased", "Estimator", "HwShunt", "onchipcounter",
          "HWSHUNT", "Measured", "", " OnChipCounter", "OnChipCounter\u200b"]
UNCERTAINTIES = [None, {"relative_ppm": 20000, "window_us": 1000000}, {"relative_ppm": 0, "window_us": 1},
                 {"relative_ppm": 1, "window_us": 0}, {"relative_ppm": 2 ** 32 - 1, "window_us": 1},
                 {"relative_ppm": 2 ** 32, "window_us": 1}, {"relative_ppm": 2 ** 53 - 1, "window_us": 2 ** 53 - 1},
                 {"relative_ppm": 288676, "window_us": 1}, {"relative_ppm": 288675, "window_us": 1},
                 {"relative_ppm": 144338, "window_us": 1}, {"relative_ppm": 144337, "window_us": 1},
                 {"relative_ppm": 20000}, {"window_us": 1}, {}, {"relative_ppm": 20000, "window_us": 1, "k": 2},
                 {"relative_ppm": "20000", "window_us": 1}, [20000, 1], "20000", 20000, True]
FILTERINGS = [None, "On", "Off", "Unknown", "unknown", "", "Maybe", 0, False]
# A calibration reference is blank when it is White_Space only (energy-measurement §2.5, as
# the reference implementation trims it); U+001C to U+001F, U+200B and U+FEFF are not.
CALIBRATION_REFS = [None, "cert 7", "", " ", "\t", "\u3000", "\u0085", "\u00a0", "\u2028", "\u001c", "\u001f",
                    "\u200b", "\ufeff", " x ", "https://calibration.example.org/certificates/0001", 7]
CLASS_FIGURES = [1, 2, 14, 15, 12000, 2 ** 53 - 1, 0]


def classes(ins, rng, n):
    """§2.4.5 rule 2 and its interaction with rule 5: energy claims with every acquisition
    label, uncertainty, filtering state and calibration reference above in fixed
    combinations — each field varied alone from a valid counter, a valid instrument and a
    ModelBased figure — and `n` random combinations of all four with a figure and a grade.
    Each is signed, so its verdict is decided at step 2, 4 or later, never by the signature."""
    base = ins.receipt("energy_accept")
    bases = {
        "OnChipCounter": dict(base["body"], energy_provenance="OnChipCounter",
                              energy_uncertainty={"relative_ppm": 20000, "window_us": 1000000},
                              energy_counter_filtering="Unknown", grade="measured"),
        "CalibratedInstrument": dict(base["body"], energy_provenance="CalibratedInstrument",
                                     energy_uncertainty={"relative_ppm": 5000, "window_us": 1000000},
                                     energy_calibration_ref="cert 7", grade="measured"),
        "ModelBased": dict(base["body"], oer_receipt=None),
    }
    cases = []
    for bname, b in bases.items():
        b = dict(b, oer_receipt=None)
        cases += [(f"{bname}_label_{i}", dict(b, energy_provenance=x)) for i, x in enumerate(LABELS)]
        cases += [(f"{bname}_uncertainty_{i}", dict(b, energy_uncertainty=x)) for i, x in enumerate(UNCERTAINTIES)]
        cases += [(f"{bname}_filtering_{i}", dict(b, energy_counter_filtering=x)) for i, x in enumerate(FILTERINGS)]
        cases += [(f"{bname}_calibration_{i}", dict(b, energy_calibration_ref=x)) for i, x in enumerate(CALIBRATION_REFS)]
        cases += [(f"{bname}_figure_{i}", dict(b, joules_micro=x)) for i, x in enumerate(CLASS_FIGURES)]
        cases += [(f"{bname}_grade_{g}", dict(b, grade=g)) for g in OER.GRADES + ["certified"]]
    for i in range(n):
        cases.append((f"random_{i}", dict(base["body"], oer_receipt=None, energy_provenance=rng.choice(LABELS),
                                          energy_uncertainty=rng.choice(UNCERTAINTIES),
                                          energy_counter_filtering=rng.choice(FILTERINGS),
                                          energy_calibration_ref=rng.choice(CALIBRATION_REFS),
                                          joules_micro=rng.choice(CLASS_FIGURES), grade=rng.choice(OER.GRADES))))
    for name, body in cases:
        ins.add("class_" + name, CV.jcs(sign(dict(base, body=body), METER)))


# §3.1 step 2 reads a body's members, then each member's type and encoding together in its
# table's order, a nested object's members where the table reaches it and a lawful-basis
# body's `record` first. One defect shows where each is caught; two at different positions
# show the order. Per accepted corpus claim: its signer and, in table order, each position
# (a member, or a member of a nested object) with the values put there. DELETE removes the
# member; a position ending in EXTRA adds a member the table does not have. Values stay in
# what the canonical form can write, so every input is signed — numbers at the edges of
# what a claim may hold are the `numbers` family's.
DELETE = object()
EXTRA = "+"
BODY_DEFECTS = {
    "lawful_basis_declaration_accept": ("OPERATOR", [
        (("record",), ["consent", "decision", 5, DELETE]),
        (("vocabulary",), [5, None, "w3id.org/dpv", "", "https://w3id.org/ dpv", DELETE]),
        (("fragment",), [5, "zz", "AB" * 32, "ab", DELETE]),
        (("purposes",), ["x", [], [5], {}]),
        (("purposes", 0, "purpose"), [5, "dpv:A B", "", "dpv:\u3000", DELETE]),
        (("purposes", 0, "basis"), [5, None, "", "a b", DELETE]),
        (("purposes", 0, "data"), ["x", [5], [""], ["a b"], [], DELETE]),
        (("purposes", 0, EXTRA), [1]),
        (("purposes", 1, "purpose"), [5, "dpv:A B", "dpv:ServiceProvision", DELETE]),
        (("purposes", 1, "basis"), [5, "", "a\tb", DELETE]),
        (("purposes", 1, "data"), [[5], ["x y"], DELETE]),
        (("purposes", 1, EXTRA), [1]),
        ((EXTRA,), [1]),
    ]),
    "lawful_basis_decision_accept": ("VIEWER", [
        (("record",), ["declaration", "x", 5, DELETE]),
        (("vocabulary",), [5, "dpv", DELETE]),
        (("declaration",), [5, "zz", "AB" * 32, DELETE]),
        (("purpose",), [5, "", "a b", DELETE]),
        (("outcome",), [5, "", DELETE]),
        ((EXTRA,), [1]),
    ]),
    "energy_on_chip_counter_accept": ("METER", [
        (("joules_micro",), ["1", 0, True, None, DELETE]),
        (("energy_provenance",), [5, "Measured", "HwShunt", DELETE]),
        (("energy_uncertainty",), ["x", [1, 1], None, {}, DELETE]),
        (("energy_uncertainty", "relative_ppm"), ["1", 0, DELETE]),
        (("energy_uncertainty", "window_us"), [None, 0, DELETE]),
        (("energy_uncertainty", EXTRA), [2]),
        (("energy_counter_filtering",), [5, None, "Maybe", DELETE]),
        (("energy_calibration_ref",), [5, "cert", DELETE]),
        (("coverage",), [5, "Most", DELETE]),
        (("grade",), [5, "certified", DELETE]),
        (("method",), [5, "x", "", DELETE]),
        (("work_units",), ["1", 0, DELETE]),
        (("work_unit",), [5, "", "a b", DELETE]),
        (("oer_receipt",), [5, "md5:1", DELETE]),
        ((EXTRA,), [1]),
    ]),
    "soft_binding_accept": ("PUBLISHER", [
        (("alg",), [5, "", "a b", "bare", DELETE]),
        (("registry",), [5, "x", DELETE]),
        (("binding_type",), [5, "hash", DELETE]),
        (("value",), [5, "zz", "ABCD", "", DELETE]),
        (("resolution_endpoint",), [5, "x", "http://x/", "https://u@x/", DELETE]),
        ((EXTRA,), [1]),
    ]),
    "soft_binding_span_objects_accept": ("PUBLISHER", [
        (("alg",), [5, "", "bare", DELETE]),
        (("value",), ["zz", "", DELETE]),
        (("span",), ["x", [7], None, {}, DELETE]),
        (("span", "track_alias"), ["7", None, DELETE]),
        (("span", "start_ms"), ["1", None, 180000, DELETE]),
        (("span", "end_ms"), [119999, "x", DELETE]),
        (("span", "first_group"), [None, 45, "40", DELETE]),
        (("span", "last_group"), [None, 39, DELETE]),
        (("span", "first_object"), [None, 2, DELETE]),
        (("span", "last_object"), [None, 0, DELETE]),
        (("span", EXTRA), [1]),
        ((EXTRA,), [1]),
    ]),
    "binding_recovery_accept": ("RESOLVER", [
        (("alg",), [5, "", "bare", DELETE]),
        (("binding_type",), [5, "hash", DELETE]),
        (("value",), [5, "zz", "", DELETE]),
        (("declaration",), [5, "zz", DELETE]),
        (("resolution_endpoint",), [5, "x", "http://x/", DELETE]),
        (("observed",), [5, "zz", DELETE]),
        (("attests",), [5, "object", DELETE]),
        ((EXTRA,), [1]),
    ]),
    "log_inclusion_accept": ("PUBLISHER", [
        (("log",), [5, "x", DELETE]),
        (("profile",), [5, "", DELETE]),
        (("entry",), [5, "", "a b", DELETE]),
        (("log_receipt",), [5, "zz", "", DELETE]),
        ((EXTRA,), [1]),
    ]),
    "stage_accept": ("VIEWER", [
        (("vocabulary",), [5, "staged-delivery", "urn:example:stages:1", DELETE]),
        (("op_requested",), ["x", {}, {"wai-stage:layers": "2"}, {"a b": 1}, {"wai-stage:layers": 256}, None, DELETE]),
        (("op_delivered",), [None, {"wai-stage:layers": 1}, {"wai-stage:layers": 3}, DELETE]),
        (("presented",), [None, 5, {}, {"wai-stage:layers": 1}, DELETE]),
        (("units_by_stage",), ["x", [], [0, 0, 4, 0], [4], [0, "4"], [1, 1, 2], DELETE]),
        (("tier",), [5, "none", "exact", DELETE]),
        (("stage_hash",), [None, "zz", "AB" * 32, DELETE]),
        (("objects",), ["x", [], [5], DELETE]),
        (("objects", 0, "layer"), ["1", 0, 256, DELETE]),
        (("objects", 0, "units"), ["x", {}, DELETE]),
        (("objects", 0, "units", "count"), [0, 4, DELETE]),
        (("objects", 0, "units", EXTRA), [1]),
        (("objects", 0, "status"), [5, "skipped", "late", "partial", "dropped", DELETE]),
        (("objects", 0, "evidence"), ["verified", "none", "inclusion-verified", DELETE]),
        (("objects", 0, "reason"), [5, "too-late", "deadline-missed", DELETE]),
        (("objects", 0, EXTRA), [1]),
        (("objects", 4, "status"), ["dropped", "refused"]),
        (("deadline_misses",), [5, "1", DELETE]),
        (("bytes_envelopes",), [0, DELETE]),
        ((EXTRA,), [1]),
    ]),
    "gap_accept": ("RELAY", [
        (("track_alias",), ["8", None, DELETE]),
        (("group_id",), [9, DELETE]),
        (("reason",), [5, "timeout", "discretion", DELETE]),
        (("observed_at",), [5, "2026-02-29T00:00:00Z", "now", DELETE]),
        ((EXTRA,), [1]),
    ]),
}


def edit_at(body, path, value):
    """Put `value` at `path` in `body` (DELETE removes the member there; a path ending in
    EXTRA adds the member `note`). False if an earlier edit left no object on the path."""
    *parents, last = path
    node = body
    for p in parents:
        if isinstance(p, int):
            if not (isinstance(node, list) and p < len(node)):
                return False
        elif not (isinstance(node, dict) and p in node):
            return False
        node = node[p]
    if not isinstance(node, dict):
        return False
    if last == EXTRA:
        node["note"] = value
    elif value is DELETE:
        node.pop(last, None)
    else:
        node[last] = value
    return True


def bodies(ins):
    keys = {"OPERATOR": OPERATOR, "VIEWER": VIEWER, "METER": METER, "PUBLISHER": PUBLISHER, "RESOLVER": RESOLVER,
            "RELAY": RELAY}
    for name, (signer, positions) in BODY_DEFECTS.items():
        base = ins.receipt(name)
        singles = [(path, i, value) for path, values in positions for i, value in enumerate(values)]
        cases = [[s] for s in singles]
        for a, first in enumerate(singles):
            for second in singles[a + 1:]:
                p, q = first[0], second[0]
                if p[:len(q)] != q and q[:len(p)] != p:  # different positions, neither inside the other
                    cases.append([first, second])
        for case in cases:
            body = copy.deepcopy(base["body"])
            if not all(edit_at(body, path, value) for path, _, value in case):
                continue
            label = "+".join(".".join(map(str, path)) + f"#{i}" for path, i, _ in case)
            ins.add(f"body_{name[:-len('_accept')]}_{label}", CV.jcs(sign(dict(base, body=body), keys[signer])))


def stages(ins, rng, n):
    """§2.4.10 and §2.4.11 around their rules: stage bodies built at random — a set of
    one to five units, points that agree or not, objects of random layers, spans (some
    past the set, refused before their span was checked or not), statuses, evidence,
    reasons and bytes, in order or not, a tier with or without a hash — signed by the
    sink, so each is decided at step 2, 4 or later; and the corpus's gap
    claim against object receipts at its coordinates and elsewhere, naming its object or
    another."""
    base = ins.receipt("stage_accept")
    hashes = [CV.blake3(b"differential: layer object %d" % i).hex() for i in range(8)]
    for i in range(n):
        units = rng.randint(1, 5)
        top = rng.randint(0, 3)
        ubs = [0] * (top + 1)
        for _ in range(units):
            ubs[rng.randint(0, top)] += 1
        if rng.random() < 0.1:
            ubs.append(rng.choice([0, 1]))
        objects = []
        for _ in range(rng.randint(0, 6)):
            frm = rng.randint(0, units)
            st = rng.choice(["applied", "applied", "partial", "late", "dropped", "refused"])
            objects.append({
                "layer": rng.choice([1, 1, 2, 2, 3, 0, 256]) if rng.random() < 0.2 else rng.randint(1, max(1, top)),
                "object": rng.choice(hashes),
                "units": {"from": frm, "count": rng.choice([1, units - frm, 0, units + 1, units + 7]) if units > frm else rng.choice([0, 1])},
                "bytes": (0 if st == "dropped" else rng.choice([700, 2400])) if rng.random() < 0.8 else rng.choice([0, 700]),
                "status": st,
                "evidence": rng.choice(["delivered-verified", "delivered-verified", "inclusion-verified", "none"]),
                "reason": (None if st in ("applied", "late", "dropped") else "deadline-missed")
                if rng.random() < 0.8 else rng.choice([None, "layer-conflict", "too-late", "refines-malformed", "base-absent"]),
            })
        if rng.random() < 0.8:
            objects.sort(key=lambda o: (o["layer"], o["units"]["from"], o["object"]))
        lowest = next((j for j, c in enumerate(ubs) if c), 0)
        pt = lambda k: {"wai-stage:layers": k}
        delivered = rng.choice([None, pt(rng.randint(0, 3)), pt(lowest)])
        committed = rng.choice([None, pt(2)]) if delivered is not None else rng.choice([None, None, pt(2)])
        tier = rng.choice(["decode-equivalence", "decode-equivalence", "entropy-consistency", "none"])
        has_hash = (tier == "decode-equivalence") != (rng.random() < 0.1)
        body = dict(base["body"], op_requested=rng.choice([None, committed]), op_committed=committed,
                    op_delivered=delivered, presented=pt(lowest if rng.random() < 0.9 else rng.randint(0, 3)),
                    units_by_stage=ubs, tier=tier, stage_hash=base["body"]["stage_hash"] if has_hash else None,
                    objects=objects, conflicts=rng.choice([0, 0, 1, units + 1]),
                    deadline_misses=rng.choice([0, 0, 1, units + 1]),
                    bytes_envelopes=rng.choice([5000, 5000, 5000, 0]) + max(0, sum(o["bytes"] for o in objects) - rng.choice([0, 1])))
        ins.add(f"stage_random_{i}", CV.jcs(sign(dict(base, body=body), VIEWER)))
    gap = ins.wire("gap_accept")
    g = json.loads(gap)
    for name, ctx in [
        ("at", {"track_alias": 8, "group_id": 3, "object_id": 0, "content_hash": g["subject"]}),
        ("other_object", {"track_alias": 8, "group_id": 3, "object_id": 0, "content_hash": hashes[0]}),
        ("other_track", {"track_alias": 7, "group_id": 3, "object_id": 0, "content_hash": g["subject"]}),
        ("other_group", {"track_alias": 8, "group_id": 4, "object_id": 0, "content_hash": g["subject"]}),
        ("other_index", {"track_alias": 8, "group_id": 3, "object_id": 1, "content_hash": hashes[1]}),
    ]:
        ins.add("gap_context_" + name, gap, {"gap_object": ctx})
    ins.add("gap_context_to_a_stage_claim", ins.wire("stage_accept"),
            {"gap_object": {"track_alias": 8, "group_id": 3, "object_id": 0, "content_hash": g["subject"]}})


SESSION_CAPS = [("wai.video.int_hyper", "primary", True), ("wai.video.av1", "fallback", False),
                ("wai.video.int_refine", "primary", False), ("wai.image.png", "fallback", False),
                ("wai.neural.int_hyper", "primary", True), ("wai.image.int_refine", "primary", False)]
SESSION_ENERGY = [
    None,
    {"joules_micro": 250000, "energy_provenance": "OnChipCounter",
     "energy_uncertainty": {"relative_ppm": 20000, "window_us": 1000000}, "energy_counter_filtering": "Unknown",
     "energy_calibration_ref": None},
    {"joules_micro": 1, "energy_provenance": "OnChipCounter", "energy_uncertainty": {"relative_ppm": 288676, "window_us": 1},
     "energy_counter_filtering": "Off", "energy_calibration_ref": None},
    {"joules_micro": 1, "energy_provenance": "OnChipCounter", "energy_uncertainty": {"relative_ppm": 288675, "window_us": 1},
     "energy_counter_filtering": "On", "energy_calibration_ref": None},
    {"joules_micro": 250000, "energy_provenance": "OnChipCounter",
     "energy_uncertainty": {"relative_ppm": 20000, "window_us": 1000000}, "energy_counter_filtering": None,
     "energy_calibration_ref": None},
    {"joules_micro": 9000, "energy_provenance": "CalibratedInstrument",
     "energy_uncertainty": {"relative_ppm": 5000, "window_us": 1000000}, "energy_counter_filtering": None,
     "energy_calibration_ref": "\u3000"},
    {"joules_micro": 9000, "energy_provenance": "CalibratedInstrument",
     "energy_uncertainty": {"relative_ppm": 5000, "window_us": 1000000}, "energy_counter_filtering": None,
     "energy_calibration_ref": "urn:example:certificate"},
    {"joules_micro": 9000, "energy_provenance": "ModelBased", "energy_uncertainty": None,
     "energy_counter_filtering": None, "energy_calibration_ref": None},
    {"joules_micro": 9000, "energy_provenance": "ModelBased", "energy_uncertainty": {"relative_ppm": 1, "window_us": 1},
     "energy_counter_filtering": None, "energy_calibration_ref": None},
    {"joules_micro": 9000, "energy_provenance": "Estimator", "energy_uncertainty": None,
     "energy_counter_filtering": None, "energy_calibration_ref": None},
    {"joules_micro": 9000, "energy_provenance": "HwShunt", "energy_uncertainty": None,
     "energy_counter_filtering": None, "energy_calibration_ref": None},
    {"joules_micro": 0, "energy_provenance": "ModelBased", "energy_uncertainty": None,
     "energy_counter_filtering": None, "energy_calibration_ref": None},
    {"joules_micro": 9000, "energy_provenance": "onchipcounter", "energy_uncertainty": None,
     "energy_counter_filtering": None, "energy_calibration_ref": None},
]


def session_stage_claim(rng, n_set, at_stage_1, signer, subject, late=0, tier="none"):
    """A stage claim (§2.4.10) that passes its own rules: `n_set` units, `at_stage_1` of
    them at stage 1 through one applied first layer, the rest at stage 0; and, with
    `late`, a first layer over the next `late` units that arrived late. At `tier`
    `decode-equivalence` it carries a stage hash, a clip digest of its units."""
    ubs = [n_set - at_stage_1, at_stage_1] if at_stage_1 else [n_set]
    objects = [] if not at_stage_1 else [{
        "layer": 1, "object": CV.blake3(b"differential: a stage layer").hex(), "units": {"from": 0, "count": at_stage_1},
        "bytes": 700, "status": "applied", "evidence": "none", "reason": None}]
    if late:
        objects.append({"layer": 1, "object": CV.blake3(b"differential: a late stage layer").hex(),
                        "units": {"from": at_stage_1, "count": late}, "bytes": 650, "status": "late", "evidence": "none",
                        "reason": None})
    body = {"vocabulary": CV.STAGE_VOCABULARY, "op_requested": None, "op_committed": None, "op_delivered": None,
            "presented": {"wai-stage:layers": 0 if ubs[0] else 1}, "units_by_stage": ubs, "tier": tier,
            "stage_hash": CV.blake3(b"differential: a clip digest").hex() if tier == "decode-equivalence" else None,
            "objects": objects, "conflicts": 0, "deadline_misses": 0, "bytes_envelopes": 5700}
    c = {"kind": "claim", "class": "stage", "subject_type": "object", "subject": subject, "seq": 0,
         "issued_at": "2026-09-27T12:00:00Z", "body": body, "signer_id": "did:example:sink", "parent_receipt_hash": None}
    return CV.jcs(sign(c, signer))


def restate(body, objects):
    """The session's counted sums and transfer restated from `objects`, as a sink states
    them from the objects it lists."""
    body["counted_primary"] = sum(o["counted"] for o in objects if o["counted"] is not None and o["path"] == "primary")
    body["counted_fallback"] = sum(o["counted"] for o in objects if o["counted"] is not None and o["path"] == "fallback")
    body["counted_discarded"] = sum(o["counted"] for o in objects if o["counted"] is not None and not o["presented"])
    body["uncounted"] = sum(1 for o in objects if o["counted"] is None)
    body["transferred"] = sum(o["transferred"] for o in objects)


def encrypted_envelope(wai2):
    """The model session's WAI2 envelope with a `security` member that encrypts it: inert
    (SPEC §6.4)."""
    ml = int.from_bytes(wai2[4:8], "little")
    m = json.loads(wai2[8:8 + ml])
    m["security"] = {"encryption": "mls-sframe", "kid": 1}
    return wai2_with(json.dumps(m, separators=(",", ":")), wai2[8 + ml:])


def sessions(ins, rng, n):
    """§2.4.12 around its rules: a random presentation list — units with gaps, one to three
    layers each, through one to four decoder entries — and the session it gives, then a
    defect or two in the body or the contexts, signed by the sink."""
    base = ins.receipt("session_accept")
    wai2 = model_envelope(ins, b"WAI2")
    envelope, wai2_hash = wai2.hex(), SV.content_hash(wai2)
    encrypted = encrypted_envelope(wai2)
    reasons = list(CV.SWITCH_REASONS)
    for i in range(n):
        # The entry object: a WAI1 envelope of one capability or another, so that an
        # entry naming it has its capability or not.
        base_cap = rng.choice(["wai.video.int_hyper", "wai.video.int_motion"])
        base_env = SV.wai1(json.dumps({"wai": "1.0", "media": "video", "intent": "replicate",
                                       "model_requirement": {"capability": base_cap}, "conditioning": {"kind": "none"}},
                                      separators=(",", ":")).encode(), b"\x33" * 16)
        entry = SV.content_hash(base_env)
        pool = rng.sample(SESSION_CAPS, rng.randint(1, 4))
        ps = []
        for u in range(rng.randint(1, 12)):
            if rng.random() < 0.1:
                continue
            for layers in range(1, rng.randint(1, 3) + 1):
                if layers == 1 or rng.random() < 0.9:
                    ps.append({"unit": u, "layers": layers, "decoder": rng.randrange(len(pool)),
                               "out_digest": CV.blake3(b"differential: output %d %d %d" % (i, u, layers)).hex()})
        # Entries in first-use order, as an honest sink lists them.
        order = []
        for p in sorted(ps, key=lambda p: (p["layers"], p["unit"])):
            if p["decoder"] not in order:
                order.append(p["decoder"])
        for p in ps:
            p["decoder"] = order.index(p["decoder"])
        decoders = []
        for j, old in enumerate(order):
            cap, path, pinned = pool[old]
            # The envelope's first rendition is wai.video.int_hyper: an entry naming rendition 0
            # with that capability is bound by it, any other is not.
            rendition = rng.choice([None, 0]) if cap == "wai.video.int_hyper" or rng.random() < 0.2 else None
            # Its object: the WAI2 envelope for a rendition, else the entry object or a
            # layer of its own, and now and then the other.
            if rendition is not None:
                source = wai2_hash if rng.random() < 0.9 else entry
            else:
                source = rng.choice([entry, entry, CV.blake3(b"differential: a source %d %d" % (i, j)).hex(), wai2_hash])
            # Its first unit: the first unit decoded through it, as an honest sink states it,
            # or one off; a rendition of one frame covers that unit alone.
            used = [p["unit"] for p in ps if p["decoder"] == j]
            first = min(used) if used else 0
            decoders.append({"capability": cap, "rendition": rendition, "object": source,
                             "first_unit": rng.choice([first, first, first, first + 1, max(0, first - 1), 0]), "path": path,
                             "decoder": "builtin:[email protected]", "variant": None,
                             "pin": CV.blake3(b"differential: a pin").hex() if pinned else None,
                             "tier": "decode-equivalence" if pinned else None, "module_status": None,
                             "leaves": sum(1 for p in ps if p["decoder"] == j)})
        fin = CV.finals(ps)
        primary = sum(1 for p in fin.values() if decoders[p["decoder"]]["path"] == "primary")
        ordered = sorted(ps, key=lambda p: (p["unit"], p["layers"]))
        root = CV.out_mth([CV.out_leaf(p) for p in ordered]).hex() if ps else None
        deadline = rng.choice([None, 40000])
        late = rng.sample(ordered, min(len(ordered), rng.randint(0, 3))) if deadline else []
        late = sorted(({"unit": p["unit"], "layers": p["layers"], "late_us": rng.randint(1, 5000)} for p in late),
                      key=lambda x: (x["unit"], x["layers"]))
        switches = sorted(({"unit": rng.randint(0, 12), "from": "primary", "to": "fallback", "reason": rng.choice(reasons)}
                           for _ in range(rng.randint(0, 3))), key=lambda w: w["unit"])
        def obj(h, path="primary", counted=None, presented=True):
            return {"content_hash": h, "path": path, "counted": counted, "transferred": rng.choice([700, 1200, 3000]),
                    "presented": presented}
        objects = [obj(CV.blake3(b"differential: object %d %d" % (i, k)).hex(), rng.choice(["primary", "primary", "fallback"]),
                       rng.choice([None, 700, 1200, 0]), rng.random() < 0.8) for k in range(rng.randint(0, 5))]
        if rng.random() < 0.85:
            # The entry object's counted bytes: its envelope's, or a figure that is not.
            objects.insert(rng.randint(0, len(objects)), obj(entry, counted=CV.counted_of(base_env) if rng.random() < 0.9 else 5000))
        # The envelopes held, chosen here so the objects received can hold them or not.
        envs = rng.choice([None, None, [], [envelope], [envelope] * 2, [base_env.hex(), envelope], [base_env.hex()]])
        for h in sorted({d["object"] for d in decoders} | ({wai2_hash} if envs and envelope in envs else set())):
            if h != entry and rng.random() < 0.9 and all(o["content_hash"] != h for o in objects):
                held = h == wai2_hash and rng.random() < 0.9
                objects.append(obj(h, counted=CV.counted_of(wai2) if held else rng.choice([None, 700])))
        stage_wires, stage_sets = [], []
        if fin and rng.random() < 0.5:
            # A stage claim over n_set units from a presented unit f: its first `at1` units
            # at stage 1 through one applied first layer, so a unit of the set presented at
            # two layers or more lies within that layer's span or not, and at three layers
            # beyond any layer it lists (§2.4.12 step 6, both bounds).
            units = sorted(fin)
            f = rng.choice(units)
            n_set = rng.randint(1, 3)
            reached = sum(1 for u, p in fin.items() if f <= u < f + n_set and p["layers"] > 1)
            at1 = min(n_set, max(0, reached + rng.choice([0, 0, 1, -1])))
            signer = VIEWER if rng.random() < 0.9 else RELAY
            # Its base: one no entry names, or the entry object, which entries name and decode
            # at whatever step their capability, pin and path give; at decode-equivalence the
            # claim carries a stage hash (§2.4.12 step 6, against the stage claims, item 5).
            base_hash = CV.blake3(b"differential: a base %d" % i).hex() if rng.random() < 0.7 else entry
            stage_tier = rng.choice(["none", "none", "decode-equivalence"])
            w = session_stage_claim(rng, n_set, at1, signer, base_hash, tier=stage_tier)
            if rng.random() < 0.05:
                w = w.replace('"tier":"none"', '"tier":"nonе"')
            stage_wires.append(w)
            stage_sets.append({"first_unit": f, "claim": CV.link_hash(json.loads(w)) if rng.random() < 0.95 else "00" * 32})
            # An entry naming the set's base has the unit the session names the set at, as an
            # honest sink states it, so a stage hash is checked against the step and the path
            # of each entry that decoded the base (item 5) and not refused before it (item 4).
            if base_hash == entry and rng.random() < 0.8:
                for d in decoders:
                    if d["object"] == entry:
                        d["first_unit"] = f
            # The set's base and its layer object among the objects received, or not.
            if rng.random() < 0.85:
                objects.append(obj(base_hash, counted=4000))
            if at1 and rng.random() < 0.85:
                objects.append(obj(CV.blake3(b"differential: a stage layer").hex(), counted=700))
        # Presented exactly when an entry names it, as an honest sink states it.
        named = {d["object"] for d in decoders}
        for o in objects:
            o["presented"] = o["content_hash"] in named
        counted = lambda path: sum(o["counted"] for o in objects if o["counted"] is not None and o["path"] == path)
        body = dict(base["body"], decoders=decoders, units=len(fin), leaf_count=len(ps), output_root=root,
                    primary_units=primary, fallback_units=len(fin) - primary, switches=switches,
                    switches_total=len(switches) + rng.choice([0, 0, 0, 2]), deadline_us=deadline,
                    missed=len(late) + rng.choice([0, 0, 0, 1]) if deadline else 0, missed_units=late,
                    stage_sets=stage_sets, counted_primary=counted("primary"), counted_fallback=counted("fallback"),
                    counted_discarded=sum(o["counted"] for o in objects if o["counted"] is not None and not o["presented"]),
                    uncounted=sum(1 for o in objects if o["counted"] is None),
                    transferred=sum(o["transferred"] for o in objects),
                    energy=copy.deepcopy(rng.choice(SESSION_ENERGY)))
        if rng.random() < 0.5:
            body["classical_declared"], body["classical_source"] = None, None
        ctx = {"outputs": copy.deepcopy(ps), "objects": objects}
        # One or two defects, in the body or in a context.
        inert = None
        for _ in range(rng.choice([0, 1, 1, 2])):
            k = rng.randrange(33)
            ds = body["decoders"]
            if k == 0:
                body["primary_units"] += rng.choice([1, -1]) if body["primary_units"] else 1
            elif k == 1:
                body["units"] += 1
            elif k == 2 and len(ds) > 1 and ds[0]["leaves"] > 1:
                ds[0]["leaves"] -= 1
                ds[1]["leaves"] += 1
            elif k == 3:
                body["leaf_count"] += rng.choice([1, -1]) if body["leaf_count"] else 1
            elif k == 4:
                body["output_root"] = rng.choice([None, CV.blake3(b"differential: another root").hex()])
            elif k == 5 and late:
                late[0]["layers"] += 3
            elif k == 6:
                body["switches_total"] = max(0, len(switches) - 1)
            elif k == 7:
                body["switches"] = [{"unit": u, "from": "primary", "to": "fallback", "reason": "recovered"} for u in range(1025)]
                body["switches_total"] = 1025
            elif k == 8 and ps:
                body["deadline_us"], body["missed"] = 40000, 257
                body["missed_units"] = [{"unit": u, "layers": 1, "late_us": 1} for u in range(257)]
            elif k == 9:
                body["deadline_us"], body["missed"] = None, max(1, body["missed"])
            elif k == 10:
                body["classical_declared"], body["classical_source"] = rng.choice([(700, None), (None, "wai2-table"), (700, "referenced")])
            elif k == 11 and ctx["outputs"]:
                ctx["outputs"][rng.randrange(len(ctx["outputs"]))]["decoder"] = len(ds) + rng.choice([0, 1])
            elif k == 12 and len(ds) > 1:
                ds[0], ds[1] = ds[1], ds[0]
            elif k == 13:
                body["counted_discarded"] += 1
            elif k == 14 and ctx["objects"]:
                o = ctx["objects"][0]
                o["counted"] = None if o["counted"] is not None else 5
            elif k == 15:
                body["host"] = dict(body["host"], threads=rng.choice([0, 3, 6, 2 ** 52]))
            elif k == 16 and ds:
                ds[-1]["decoder"] = rng.choice(["builtin:wai", "builtin:wai@x", "module:sha256:" + "a" * 64, "builtin:Wai@1"])
            elif k == 17 and ds:
                ds[-1]["pin"] = rng.choice([ds[-1]["pin"], CV.blake3(b"differential: a pin").hex()])
                ds[-1]["tier"] = rng.choice(["none", "decode-equivalence", "entropy-consistency", "bit-exact", None])
            elif k == 18 and ds:
                ds.append(dict(ds[0]))
            elif k == 19 and ctx["outputs"]:
                ctx["outputs"].pop(rng.randrange(len(ctx["outputs"])))
            elif k == 20 and ctx["outputs"]:
                ctx["outputs"][0]["out_digest"] = CV.blake3(b"differential: not this output").hex()
            elif k == 21 and stage_sets:
                stage_sets.append(dict(stage_sets[0]))
            elif k == 22:
                body["framing"] = rng.choice(["registered", "wai-measure/0"])
            elif k == 23 and ds:
                ds[0]["module_status"] = rng.choice(["pinned", "fetched", "cached"])
            elif k == 24:
                body["counted_primary"] += rng.choice([1, -1]) if body["counted_primary"] else 1
            elif k == 25 and switches:
                body["switches"] = [dict(switches[0], to="primary")] + switches[1:]
            elif k == 26 and objects:
                # An object listed twice, or a fetch listed apart from the one before it, on
                # the other path; the sums restated to agree.
                o = dict(rng.choice(objects))
                if rng.random() < 0.3:
                    o["path"] = "fallback" if o["path"] == "primary" else "primary"
                objects.append(o)
                restate(body, objects)
            elif k == 27 and objects:
                o = rng.choice(objects)
                o["presented"] = not o["presented"]
                restate(body, objects)
            elif k == 28:
                inert = rng.choice(["encrypted", "bytes"])
            elif k == 29 and ds:
                d = rng.choice(ds)
                d["first_unit"] = d["first_unit"] + 1 if d["first_unit"] == 0 or rng.random() < 0.5 else d["first_unit"] - 1
            elif k == 30 and objects:
                # A held envelope's object told a figure other than its counted bytes.
                o = rng.choice(objects)
                o["counted"] = rng.choice([None, 0, (o["counted"] or 0) + 1, max(0, (o["counted"] or 0) - 1)])
                restate(body, objects)
            elif k == 31 and ds:
                # The decode path an entry states (§2.4.12 rules 13 and 14): exact, float, a
                # term this revision does not register, a lowering a registration names
                # (whm1/1 is the model session's integer capabilities' own), or none.
                rng.choice(ds)["variant"] = rng.choice([None, "exact", "float", "fp16", "whm1/1", "dred-int/1"])
            elif k == 32 and ds:
                # A float path under a pin of either tier, naming the set read with no pin
                # verified, or naming no set, through a capability the registry classifies or
                # one it does not.
                d = rng.choice(ds)
                d["variant"] = rng.choice(["float", "exact", "fp16", "dred-int/1"])
                d["pin"], d["tier"] = rng.choice([(CV.blake3(b"differential: a pin").hex(), "decode-equivalence"),
                                                  (CV.blake3(b"differential: a pin").hex(), "entropy-consistency"),
                                                  (CV.blake3(b"differential: a pin").hex(), "none"), (None, None)])
                if rng.random() < 0.6:
                    d["capability"] = rng.choice(["org.example.picture", "org.example.picture", "wai.image.jpegai",
                                                  "wai.audio.opus.dred.int"])
        if rng.random() < 0.5:
            ctx["redecoded"] = [{"index": j, "out_digest": rng.choice([p["out_digest"], CV.blake3(b"differential: redecoded").hex()])}
                                for j, p in enumerate(ps) if rng.random() < 0.3]
        if envs is not None:
            ctx["envelopes"] = [{"envelope": e} for e in envs]
            n2 = sum(1 for e in envs if e == envelope)
            if n2:
                body["classical_declared"], body["classical_source"] = rng.choice(
                    [(700, "wai2-table"), (700 * n2, "wai2-table"), (701, "wai2-table"), (None, None)])
        if inert:
            # Inert bytes received — an encrypted envelope, or bytes of neither magic — listed
            # as an object neither counted nor presented, held or not, and named by an entry
            # or not.
            data = encrypted if inert == "encrypted" else b"differential: inert bytes %d" % i
            h = SV.content_hash(data)
            if rng.random() < 0.8:
                ctx.setdefault("envelopes", []).append({"envelope": data.hex()})
            if body["decoders"] and rng.random() < 0.3:
                d = rng.choice(body["decoders"])
                d["object"], d["rendition"] = h, None
            if "objects" in ctx:
                named = {d["object"] for d in body["decoders"]}
                ctx["objects"].append(obj(h, counted=None, presented=h in named))
                restate(body, ctx["objects"])
        if stage_wires:
            ctx["stage_claims"] = stage_wires
        if rng.random() < 0.1:
            del ctx["outputs"]
        if rng.random() < 0.1:
            del ctx["objects"]
        ins.add(f"session_random_{i}", CV.jcs(sign(dict(base, subject=entry, body=body), VIEWER)), ctx)


def session_stage_bounds(ins, rng, n):
    """§2.4.12 step 6 against a stage claim, both ways, on sessions that pass steps 1-5: a
    set of one to six units whose first layer applied to some and arrived late for others,
    and presentations of its units at one to three layers. A unit presented at two layers
    lies within an arrived first layer or not, one at three layers never has an arrived
    second layer, and the units presented at two layers or more reach the units the claim
    counts at stage 1 or fall short. The objects received hold the set's base and its
    layers, or lack either."""
    base = ins.receipt("session_accept")
    dec = [dict(base["body"]["decoders"][0], leaves=0), dict(base["body"]["decoders"][3], leaves=0)]
    for i in range(n):
        n_set = rng.randint(1, 6)
        at1 = rng.randint(0, n_set)
        late = rng.randint(0, n_set - at1)
        ps = []
        for u in range(n_set):
            top = 1
            if u < at1 + late and rng.random() < 0.8:
                top = rng.choice([2, 2, 3])
            elif rng.random() < 0.15:
                top = 2
            for layers in range(1, top + 1):
                ps.append({"unit": u, "layers": layers, "decoder": 0 if layers == 1 else 1,
                           "out_digest": CV.blake3(b"differential: bounds %d %d %d" % (i, u, layers)).hex()})
        base_hash = CV.blake3(b"differential: a bounded base %d" % i).hex()
        decoders = [dict(dec[0], object=base_hash, leaves=sum(1 for p in ps if p["decoder"] == 0))]
        if any(p["decoder"] == 1 for p in ps):
            decoders.append(dict(dec[1], object=CV.blake3(b"differential: a stage layer").hex(), leaves=sum(1 for p in ps if p["decoder"] == 1)))
        if rng.random() < 0.15:
            # The base's entry, or its layer's, at another first unit than the set's.
            rng.choice(decoders)["first_unit"] += 1
        # At decode-equivalence the claim states a stage hash (item 5), which holds while the
        # base's entry decodes it at decode-equivalence and each entry runs an exact path;
        # now and then the base's entry is unpinned (naming the set it read with the tier
        # none, or with pin and tier null as a sink of an earlier text writes it), or an
        # entry names another path.
        tier = rng.choice(["none", "none", "decode-equivalence"])
        if tier == "decode-equivalence" and rng.random() < 0.5:
            d = rng.choice(decoders)
            if d is decoders[0] and rng.random() < 0.5:
                d["pin"], d["tier"] = rng.choice([(CV.blake3(b"differential: a pin").hex(), "none"), (None, None)])
            else:
                d["variant"] = rng.choice(["float", "fp16", "exact", "whm1/1"])
        w = session_stage_claim(rng, n_set, rng.choice([at1, at1, max(0, at1 - 1), min(n_set, at1 + 1)]), VIEWER, base_hash,
                                late=late if late and at1 + late <= n_set else 0, tier=tier)
        objects = []
        for h, cnt in [(base_hash, 4000), (CV.blake3(b"differential: a stage layer").hex(), 700),
                       (CV.blake3(b"differential: a late stage layer").hex(), 650)]:
            if rng.random() < 0.9:
                objects.append({"content_hash": h, "path": "primary", "counted": cnt, "transferred": cnt,
                                "presented": h in {d["object"] for d in decoders}})
        entry = objects[0]["content_hash"] if objects else base_hash
        ordered = sorted(ps, key=lambda p: (p["unit"], p["layers"]))
        body = dict(base["body"], decoders=decoders, units=n_set, leaf_count=len(ps), primary_units=n_set, fallback_units=0,
                    output_root=CV.out_mth([CV.out_leaf(p) for p in ordered]).hex(), switches=[], switches_total=0,
                    deadline_us=None, missed=0, missed_units=[], stage_sets=[{"first_unit": 0, "claim": CV.link_hash(json.loads(w))}],
                    counted_primary=sum(o["counted"] for o in objects), counted_fallback=0,
                    counted_discarded=sum(o["counted"] for o in objects if not o["presented"]), uncounted=0,
                    transferred=sum(o["transferred"] for o in objects), classical_declared=None, classical_source=None,
                    energy=None)
        ctx = {"outputs": ps, "objects": objects, "stage_claims": [w]}
        if rng.random() < 0.1:
            del ctx["objects"]
        ins.add(f"session_bounds_{i}", CV.jcs(sign(dict(base, subject=entry, body=body), VIEWER)), ctx)


def epoch_stage_claim(rng, i, k, n_set, subject):
    """A stage claim (§2.4.10) that passes its own rules, as the claim of one epoch of a
    set of `n_set` units: a first layer of any status over a span, perhaps a second over
    other units, and perhaps a second layer applied or not within an applied first; its
    units at each stage as many as its applied layers can have reached, or fewer."""
    def obj(layer, j, frm, cnt, status):
        h = CV.blake3(b"differential: epoch layer %d %d %d %d" % (i, k, layer, j)).hex()
        return {"layer": layer, "object": h, "units": {"from": frm, "count": cnt},
                "bytes": 0 if status == "dropped" else 600 + 10 * j + layer, "status": status, "evidence": "none",
                "reason": {"partial": "layer-gap", "refused": "geometry-mismatch"}.get(status)}
    objects = []
    cut = rng.randint(1, n_set)
    st1 = rng.choice(["applied", "applied", "partial", "late", "refused", "dropped"])
    if st1 == "partial" and cut < 2:
        st1 = "applied"
    objects.append(obj(1, 0, 0, cut, st1))
    if cut < n_set and rng.random() < 0.6:
        st = rng.choice(["applied", "late", "refused", "dropped"])
        objects.append(obj(1, 1, cut, n_set - cut, st))
    reach1 = sum(o["units"]["count"] if o["status"] == "applied" else o["units"]["count"] - 1 if o["status"] == "partial" else 0
                 for o in objects)
    reach2 = 0
    if st1 == "applied" and rng.random() < 0.6:
        st = rng.choice(["applied", "applied", "late", "refused"])
        objects.append(obj(2, 0, 0, 1, st))
        reach2 = 1 if st == "applied" else 0
    s1 = rng.randint(0, reach1)
    s2 = rng.randint(0, min(reach2, s1))
    ubs = [n_set - s1, s1 - s2, s2] if s2 else ([n_set - s1, s1] if s1 else [n_set])
    objects.sort(key=lambda o: (o["layer"], o["units"]["from"], o["object"]))
    body = {"vocabulary": CV.STAGE_VOCABULARY, "op_requested": None, "op_committed": None, "op_delivered": None,
            "presented": {"wai-stage:layers": next(j for j, c in enumerate(ubs) if c)}, "units_by_stage": ubs, "tier": "none",
            "stage_hash": None, "objects": objects, "conflicts": 0, "deadline_misses": 0,
            "bytes_envelopes": 5000 + sum(o["bytes"] for o in objects)}
    c = {"kind": "claim", "class": "stage", "subject_type": "object", "subject": subject, "seq": k,
         "issued_at": "2026-09-27T12:00:00Z", "body": body, "signer_id": "did:example:sink", "parent_receipt_hash": None}
    return CV.jcs(sign(c, VIEWER)), objects


def session_stage_epochs(ins, rng, n):
    """§2.4.12 step 6 from above across epochs: one to three stage claims of one set, as a
    sink that forgot its base signs one per epoch, each listing first and second layers of
    every status; sometimes one named one unit further on. The session presents each unit
    at one layer and some at two or three, within what one of the claims holding the unit
    applied or past it, and receives the arrived layers or misses one."""
    base = ins.receipt("session_accept")
    dec = [dict(base["body"]["decoders"][0], leaves=0), dict(base["body"]["decoders"][3], leaves=0)]
    for i in range(n):
        n_set = rng.randint(2, 6)
        base_hash = CV.blake3(b"differential: an epoch base %d" % i).hex()
        claims = [epoch_stage_claim(rng, i, k, n_set, base_hash) for k in range(rng.randint(1, 3))]
        named = [(1 if k and rng.random() < 0.15 else 0, w) for k, (w, _) in enumerate(claims)]
        ps = []
        for u in range(n_set + 1):
            top = rng.choice([1, 1, 2, 2, 2, 3])
            for layers in range(1, top + 1):
                ps.append({"unit": u, "layers": layers, "decoder": 0 if layers == 1 else 1,
                           "out_digest": CV.blake3(b"differential: epochs %d %d %d" % (i, u, layers)).hex()})
        decoders = [dict(dec[0], object=base_hash, leaves=sum(1 for p in ps if p["decoder"] == 0))]
        objects = [{"content_hash": base_hash, "path": "primary", "counted": 4000, "transferred": 4000, "presented": True}]
        seen = set()
        for _, objs in claims:
            for o in objs:
                if o["status"] != "dropped" and o["object"] not in seen and rng.random() < 0.95:
                    seen.add(o["object"])
                    objects.append({"content_hash": o["object"], "path": "primary", "counted": o["bytes"],
                                    "transferred": o["bytes"], "presented": o["status"] in ("applied", "partial")})
        if any(p["decoder"] == 1 for p in ps):
            # The refinement's entry names a layer the session received, or its base.
            layer = next((o["content_hash"] for o in objects[1:] if o["presented"]), base_hash)
            decoders.append(dict(dec[1], object=layer, leaves=sum(1 for p in ps if p["decoder"] == 1)))
        # Presented exactly when an entry names it, as an honest sink states it.
        for o in objects:
            o["presented"] = o["content_hash"] in {d["object"] for d in decoders}
        if rng.random() < 0.15:
            # The base's entry, or a layer's, at another first unit than the claims'.
            rng.choice(decoders)["first_unit"] += rng.choice([1, 2])
        units = n_set + 1
        ordered = sorted(ps, key=lambda p: (p["unit"], p["layers"]))
        sets = sorted({(f, CV.link_hash(json.loads(w))) for f, w in named})
        body = dict(base["body"], decoders=decoders, units=units, leaf_count=len(ps), primary_units=units, fallback_units=0,
                    output_root=CV.out_mth([CV.out_leaf(p) for p in ordered]).hex(), switches=[], switches_total=0,
                    deadline_us=None, missed=0, missed_units=[], stage_sets=[{"first_unit": f, "claim": h} for f, h in sets],
                    counted_primary=sum(o["counted"] for o in objects), counted_fallback=0,
                    counted_discarded=sum(o["counted"] for o in objects if not o["presented"]), uncounted=0,
                    transferred=sum(o["transferred"] for o in objects), classical_declared=None, classical_source=None,
                    energy=None)
        ctx = {"outputs": ps, "objects": objects, "stage_claims": [w for w, _ in claims]}
        ins.add(f"session_epochs_{i}", CV.jcs(sign(dict(base, subject=base_hash, body=body), VIEWER)), ctx)


def wiv1(frames, keyframe):
    """A WIV1 clip of `frames` frames, 16 by 16, built as the corpus generator builds its
    own: `keyframe` its WIH1 keyframe, its residual stream filler that nothing decodes."""
    block = 1 if frames == 1 else 16
    nb = (16 // block) ** 2
    return CV.IPV.wiv1_from_json({"n_frames": frames, "h": 16, "w": 16, "block": block, "q": 1, "offset": 0,
                                  "cdf": [0, 32768, 65536], "mvs": [[]] + [[[0, 0]] * nb for _ in range(frames - 1)],
                                  "rans_b64": base64.b64encode(b"\x33" * 8).decode()}, keyframe=keyframe)


def wai2_of(entries, payloads):
    """A WAI2 envelope of the manifest entries `entries`, each with its payload's SHA-256
    unless it states its own, over `payloads`."""
    m = {"wai": "1.1", "media": "video", "intent": "replicate",
         "renditions": [dict({"sha256": hashlib.sha256(p).hexdigest()}, **e) for e, p in zip(entries, payloads)], "target": None}
    t = json.dumps(m, separators=(",", ":")).encode()
    table, o = b"", 0
    for p in payloads:
        table += o.to_bytes(4, "little") + len(p).to_bytes(4, "little")
        o += len(p)
    return b"WAI2" + len(t).to_bytes(4, "little") + t + len(payloads).to_bytes(2, "little") + table + b"".join(payloads)


def wir1_units(n, _cache={}):
    """A WIR1 payload of `n` units that parses (staged-delivery Appendix A): the corpus's
    four-unit layer's header and table, its first unit's stream for every unit. Nothing
    decodes it."""
    if "four" not in _cache:
        raise RuntimeError("wir1_units before session_held read the corpus's layer")
    p = _cache["four"]
    w = SV.parse_wir1(p, "rgb8")
    cdf_len = struct.unpack_from("<H", p, 40)[0]
    head = p[:8] + struct.pack("<I", n) + p[12:44]
    tables = p[44:44 + 4 * cdf_len * (n if w["per_unit"] else 1)] if not w["per_unit"] else p[44:44 + 4 * cdf_len] * n
    s0 = w["streams"][0]
    return head + tables + struct.pack("<I", len(s0)) * n + s0 * n


def session_held(ins, rng, n):
    """§2.4.12 step 6 against the envelopes with the outputs and the objects, around the
    rules that read what a held envelope gives: a base of one to six frames — a WAI1
    envelope whose manifest may name a fallback the base's entry decodes through, or the
    base component of a WAI2 envelope — and layers refining it at layer 1 or 2 over a span
    or over all of its units, as WAI1 envelopes or as refinement components of that WAI2
    envelope, now and then naming another base or one that is a layer, at layer 0, with an
    empty span or one past the base, or with no `refines`; presentations through each at
    one to three layers, of units within what the envelope covers and past it, through
    entries whose `first_unit` is the base's first unit or one off, the base's and the
    layers' each; the objects each envelope's figures — counted bytes, length, fetched once
    or more, presented exactly when an entry names it — or one told otherwise, listed
    twice, or inert bytes received beside them; the base held or not."""
    model = ins.by_name["session_held_layer_accept"]
    base = json.loads(model["receipt"])
    envs = [bytes.fromhex(e["envelope"]) for e in model["context"]["envelopes"]]
    whole = [bytes.fromhex(e["envelope"]) for e in ins.by_name["session_unit_whole_layer_accept"]["context"]["envelopes"]][1]
    base_m, _, base_p = SV.read_wai1(envs[0])
    keyframe = CV.IPV.parse_wiv1(base_p)["wih1"]
    one_m, _, one_p = SV.read_wai1(envs[1])  # a layer coding one unit
    _, _, four_p = SV.read_wai1(whole)  # and one coding four
    wir1_units.__defaults__[0]["four"] = four_p
    jtext = lambda m: json.dumps(m, separators=(",", ":")).encode()
    dec = {d["capability"]: d for d in base["body"]["decoders"]}
    encrypted = encrypted_envelope(model_envelope(ins, b"WAI2"))
    HYPER, REFINE = "wai.video.int_hyper", "wai.video.int_refine"
    for i in range(n):
        frames = rng.choice([1, 2, 3, 4, 4, 6])
        first = rng.choice([0, 0, 1, 2, 5])
        form = rng.choice(["wai1", "wai1", "wai2", "alias"])
        listed_only = []  # objects received and listed whose bytes the verifier does not hold
        # The layers: at layer 1 or 2, over a span or the whole base, and now and then one a
        # sink cannot apply.
        specs = []
        for k in range(rng.choice([0, 1, 1, 2, 3])):
            level = 1 if k == 0 or rng.random() < 0.5 else 2
            frm = rng.randint(0, frames)
            count = rng.randint(1, max(1, frames - min(frm, frames - 1)))
            span = None if rng.random() < 0.3 else {"from": frm, "count": count}
            bad = rng.choice(["layer0", "empty", "absent", "past", "on_layer"]) if rng.random() < 0.12 else None
            specs.append((level, span, bad))

        def refines(base_ref, prev, level, span, bad, on_layer):
            r = {"base": base_ref, "prev": prev, "layer": 0 if bad == "layer0" else level}
            if bad == "empty":
                r["span"] = {"from": 0, "count": 0}
            elif bad == "past":
                r["span"] = {"from": frames, "count": 2}
            elif span is not None:
                r["span"] = span
            if bad == "on_layer" and on_layer is not None:
                r["base"] = r["prev"] = on_layer
                r["layer"] = 1
            return r

        # A layer's WIR1 payload: a unit for each unit it covers, or now and then one fewer.
        payload_of = lambda span: wir1_units(max(1, (span["count"] if span is not None and span["count"] else frames)  # noqa: E731
                                                - (1 if rng.random() < 0.08 else 0)))
        refine_cap = lambda: "wai.image.int_refine" if rng.random() < 0.05 else REFINE  # noqa: E731
        if form in ("wai1", "alias"):
            if form == "wai1":
                # The base's capability and fallback: one format, or a fallback of another
                # format than the capability's, or a companion-only capability.
                cap, fallback = rng.choice([(HYPER, None), (HYPER, None), ("wai.video.int_motion", None), (HYPER, "wai.video.av1"),
                                            ("wai.neural.int_hyper", HYPER), ("wai.audio.int_codec", "wai.video.int_motion"),
                                            ("wai.meta.hdr10plus", HYPER)])
                mr = {"capability": cap, "fallback": fallback} if fallback else {"capability": cap}
                base_env = SV.wai1(jtext(dict(base_m, model_requirement=mr)), wiv1(frames, keyframe) if rng.random() < 0.95 else b"\x33" * 16)
                subject = layer_base = SV.content_hash(base_env)
                held_envs = [base_env]
                base_entry = dict(dec[HYPER], capability=cap, rendition=None, object=subject, first_unit=first)
                if fallback and rng.random() < 0.5:
                    base_entry["capability"] = fallback  # decoded through the fallback
            else:
                # The base a WAI2 envelope's component, which the layers refine by its
                # exploded envelope, and which its entry names in either form.
                w = wai2_of([{"capability": HYPER, "kind": "int_hyper", "id": "b"}], [wiv1(frames, keyframe)])
                x = SV.explode_each(w)[0]
                layer_base = SV.content_hash(x)
                if rng.random() < 0.7:
                    subject, held_envs = SV.content_hash(w), [w] + ([x] if rng.random() < 0.3 else [])
                    base_entry = dict(dec[HYPER], rendition=0, object=subject, first_unit=first)
                else:
                    # The exploded envelope's own bytes held, or only the WAI2 the entry
                    # resolves to, or neither.
                    subject = layer_base
                    held_envs = ([x] if rng.random() < 0.5 else []) + ([w] if rng.random() < 0.8 else [])
                    listed_only = [] if x in held_envs else [x]
                    base_entry = dict(dec[HYPER], rendition=None, object=subject, first_unit=first)
            if base_entry["capability"] != HYPER:
                base_entry.update(pin=None, tier=None)  # no pin above its capability's class
            layer_objects, layer_caps = [], []
            for k, (level, span, bad) in enumerate(specs):
                ref_base = layer_base if rng.random() < 0.9 else CV.blake3(b"differential: another base %d" % i).hex()
                prev = ref_base if level == 1 or not layer_objects else layer_objects[-1]
                lm = copy.deepcopy(one_m)
                lc = refine_cap()
                lm["model_requirement"] = dict(lm["model_requirement"], capability=lc)
                if bad == "absent":
                    lm.pop("refines", None)
                else:
                    lm["refines"] = refines(ref_base, prev, level, span, bad, layer_objects[0] if layer_objects else None)
                e = SV.wai1(jtext(lm), payload_of(span))
                held_envs.append(e)
                layer_objects.append(SV.content_hash(e))
                layer_caps.append(lc)
            layer_entries = [dict(dec[REFINE], capability=lc, rendition=None, object=h) for h, lc in zip(layer_objects, layer_caps)]
        else:
            comps, payloads = [{"capability": HYPER, "kind": "int_hyper", "id": "b"}], [wiv1(frames, keyframe)]
            if rng.random() < 0.05:
                comps[0]["role"] = "companion"
            for k, (level, span, bad) in enumerate(specs):
                base_id = "b" if rng.random() < 0.92 else rng.choice(["c", "B", "r0"])
                prev = base_id if level == 1 or k == 0 else f"r{k - 1}"
                e = {"capability": refine_cap(), "kind": "refine", "id": f"r{k}", "role": "refinement"}
                if bad != "absent":
                    e["refines"] = refines(base_id, prev, level, span, bad, "r0" if k else None)
                # Now and then a refinement that does not conform, or does not match its digest.
                flaw = rng.choice(["no_sha", "no_role", "wrong_sha"]) if rng.random() < 0.08 else None
                if flaw == "no_sha":
                    e["sha256"] = None
                elif flaw == "no_role":
                    del e["role"]
                elif flaw == "wrong_sha":
                    e["sha256"] = hashlib.sha256(b"differential: another payload").hexdigest()
                comps.append(e)
                payloads.append(payload_of(span))
            env = wai2_of(comps, payloads)
            subject = SV.content_hash(env)
            held_envs = [env]
            base_entry = dict(dec[HYPER], rendition=0, object=subject, first_unit=first)
            each = SV.explode_each(env)
            if rng.random() < 0.25 and each[0] is not None:
                # The base component named by its exploded envelope, its own bytes held or
                # not (then the entry resolves to the WAI2's rendition).
                x = each[0]
                if rng.random() < 0.5:
                    held_envs.append(x)
                else:
                    listed_only.append(x)
                subject = SV.content_hash(x)
                base_entry = dict(dec[HYPER], rendition=None, object=subject, first_unit=first)
            layer_entries = []
            for k in range(len(specs)):
                xk = each[k + 1] if k + 1 < len(each) else None
                if xk is not None and rng.random() < 0.25:
                    # A refinement named by its exploded envelope's content hash.
                    layer_entries.append(dict(dec[REFINE], capability=comps[k + 1]["capability"], rendition=None, object=SV.content_hash(xk)))
                    if rng.random() < 0.3:
                        held_envs.append(xk)
                    else:
                        listed_only.append(xk)
                else:
                    layer_entries.append(dict(dec[REFINE], capability=comps[k + 1]["capability"], rendition=k + 1, object=SV.content_hash(env)))
        for d in layer_entries:
            d["first_unit"] = first if rng.random() < 0.9 else rng.choice([0, first + 1])
        # Each unit at one layer through the base, and some at two or three through a layer
        # of that level — one covering the unit, or any.
        units = list(range(first, first + frames))
        if rng.random() < 0.2:
            units.append(rng.choice([first + frames, max(0, first - 1)]))
        ps = []
        for u in sorted(set(units)):
            ps.append({"unit": u, "layers": 1, "decoder": 0})
            for level in (1, 2):
                cands = [j for j, (lv, _, _) in enumerate(specs) if lv == level]
                cover = [j for j in cands if specs[j][1] is None or specs[j][1]["from"] <= u - first < specs[j][1]["from"] + specs[j][1]["count"]]
                pick = cover if cover and rng.random() < 0.85 else cands
                if not pick or rng.random() < 0.4:
                    break
                ps.append({"unit": u, "layers": level + 1, "decoder": 1 + rng.choice(pick)})
        if rng.random() < 0.15 and specs:
            # A presentation through an entry whose envelope gives another layer count.
            p = rng.choice(ps)
            p["decoder"] = 1 + rng.randrange(len(specs)) if p["layers"] == 1 or rng.random() < 0.6 else 0
        for p in ps:
            p["out_digest"] = CV.blake3(b"differential: held %d %d %d" % (i, p["unit"], p["layers"])).hex()
        # Entries in first-use order, the unused layers named by none.
        order = []
        for p in sorted(ps, key=lambda p: (p["layers"], p["unit"])):
            if p["decoder"] not in order:
                order.append(p["decoder"])
        for p in ps:
            p["decoder"] = order.index(p["decoder"])
        all_entries = [base_entry] + layer_entries
        decoders = []
        for j, old in enumerate(order):
            d = dict(all_entries[old])
            d["leaves"] = sum(1 for p in ps if p["decoder"] == j)
            decoders.append(d)
        if rng.random() < 0.1:
            decoders[0]["first_unit"] = rng.choice([first + 1, max(0, first - 1)])
        held = list(held_envs)
        named = {d["object"] for d in decoders}
        listed = held + [e for e in listed_only if e not in held]
        objects = [{"content_hash": SV.content_hash(e), "path": "primary", "counted": CV.counted_of(e),
                    "transferred": len(e) * rng.choice([1, 1, 1, 2, 3]), "presented": SV.content_hash(e) in named}
                   for e in listed]
        # One or two defects among the objects and envelopes.
        inert = None
        for _ in range(rng.choice([0, 0, 1, 1, 2])):
            k = rng.randrange(6)
            o = rng.choice(objects)
            if k == 0:
                objects.append(dict(o))
            elif k == 1:
                o["presented"] = not o["presented"]
            elif k == 2:
                o["counted"] = rng.choice([None, 0, (o["counted"] or 0) + 1, max(0, (o["counted"] or 0) - 1)])
            elif k == 3 and len(held) > 1:
                held = held[1:]  # the base not held
            elif k == 4:
                inert = rng.choice([encrypted, b"differential: inert bytes %d" % i])
            elif k == 5:
                objects.append(dict(o, path="fallback"))  # fetched again on the other path
        envelopes = list(held)
        if inert is not None:
            ih = SV.content_hash(inert)
            envelopes.append(inert)
            if rng.random() < 0.3 and len(decoders) > 1:
                decoders[-1]["object"] = ih
            objects.append({"content_hash": ih, "path": "primary", "counted": None, "transferred": len(inert),
                            "presented": ih in {d["object"] for d in decoders}})
        rng.shuffle(envelopes)
        fin = CV.finals(ps)
        ordered = sorted(ps, key=lambda p: (p["unit"], p["layers"]))
        body = dict(base["body"], decoders=decoders, units=len(fin), leaf_count=len(ps), primary_units=len(fin),
                    fallback_units=0, output_root=CV.out_mth([CV.out_leaf(p) for p in ordered]).hex())
        restate(body, objects)
        ctx = {"outputs": ps, "objects": objects, "envelopes": [{"envelope": e.hex()} for e in envelopes]}
        if rng.random() < 0.05:
            del ctx["outputs"]
        ins.add(f"session_held_{form}_{i}", CV.jcs(sign(dict(base, subject=subject, body=body), VIEWER)), ctx)


def session_encrypted(ins, rng, n):
    """§2.4.12 against an encrypted envelope, whose manifest and rendition table are in the
    clear (SPEC §6.4): the model session with its WAI2 envelope encrypted — its security
    well formed or not — and the entry decoding it naming a rendition in the table, out
    of it or none, its capability or another; the named rendition's sha256 kept, absent,
    well formed and unmatched, or malformed (uppercase, 63 digits, not hex), which matches
    no payload whether or not the envelope is encrypted; the classical length stated as
    the table gives it, or not; the plaintext envelope held beside it or not."""
    model = ins.by_name["session_envelope_encrypted"]
    base = json.loads(model["receipt"])
    enc = bytes.fromhex(model["context"]["envelopes"][0]["envelope"])
    ml = int.from_bytes(enc[4:8], "little")
    m0, rest = json.loads(enc[8:8 + ml]), enc[8 + ml:]
    plain = model_envelope(ins, b"WAI2")
    for i in range(n):
        m = copy.deepcopy(m0)
        m["security"] = rng.choice([m0["security"], m0["security"], {"encryption": "mls-sframe", "kid": -1}, {"encryption": 5}, {"encryption": "none"}])
        r0 = m["renditions"][0]
        good = r0.get("sha256") or hashlib.sha256(b"a payload").hexdigest()
        form = rng.choice(["keep", "keep", "keep", "absent", "unmatched", "upper", "short", "not_hex"])
        if form == "absent":
            r0.pop("sha256", None)
        elif form != "keep":
            r0["sha256"] = {"unmatched": hashlib.sha256(b"another payload").hexdigest(), "upper": good.upper(),
                            "short": good[:63], "not_hex": "z" * 64}[form]
        e = wai2_with(json.dumps(m, separators=(",", ":")), rest)
        body = copy.deepcopy(base["body"])
        d = body["decoders"][1]
        d["object"] = SV.content_hash(e)
        d["rendition"] = rng.choice([0, 0, 1, 2, 9, None])
        d["capability"] = rng.choice(["wai.video.int_hyper", "wai.video.int_hyper", "wai.video.avc", "wai.audio.opus"])
        if d["capability"] != "wai.video.int_hyper":
            d["pin"], d["tier"] = None, None
        envs = [e] + ([plain] if rng.random() < 0.3 else [])
        body["classical_declared"] = rng.choice([700, 700, None, 700 * len(envs), 701])
        body["classical_source"] = None if body["classical_declared"] is None else "wai2-table"
        ctx = dict(model["context"], envelopes=[{"envelope": x.hex()} for x in envs])
        ins.add(f"session_encrypted_{i}", CV.jcs(sign(dict(base, body=body), VIEWER)), ctx)


def session_roles_sweep(ins):
    """SPEC §7.1's component rules, which decide whether a WAI2 offers its refinements, over
    every registered capability and an unregistered string: a refinement beside a companion
    of each capability bound to the base, and beside an HDR10+ descriptor and an
    enhancement layer bound to a rendition of each capability."""
    model = ins.by_name["session_roles_companion_and_refinement_accept"]
    base = json.loads(model["receipt"])
    w = bytes.fromhex(model["context"]["envelopes"][0]["envelope"])
    _, m, payloads = CV.read_envelope(w)
    b, comp, ref = m["renditions"]
    for cap in sorted(CV.REGISTRY) + ["org.example.meta"]:
        cases = [
            ("companion", [b, dict(comp, capability=cap), ref], [payloads[0], payloads[1], payloads[2]], 2),
            ("hdr_over", [b, {"capability": cap, "kind": "k", "id": "x"}, dict(comp, companion_of=cap, companion_of_id=None), ref],
             [payloads[0], payloads[0], payloads[1], payloads[2]], 3),
            ("lcevc_over", [b, {"capability": cap, "kind": "k", "id": "x"}, dict(comp, capability="wai.video.lcevc", companion_of=cap, companion_of_id=None), ref],
             [payloads[0], payloads[0], payloads[1], payloads[2]], 3),
        ]
        for tag, entries, pls, ri in cases:
            entries = [{k: v for k, v in e.items() if v is not None} for e in entries]
            e2 = wai2_of(entries, pls)
            h = SV.content_hash(e2)
            body = copy.deepcopy(base["body"])
            body["decoders"][0]["object"] = h
            body["decoders"][1].update(object=h, rendition=ri)
            objects = [{"content_hash": h, "path": "primary", "counted": CV.counted_of(e2), "transferred": len(e2), "presented": True}]
            restate(body, objects)
            ctx = dict(model["context"], objects=objects, envelopes=[{"envelope": e2.hex()}])
            ins.add(f"session_roles_{tag}_{cap}", CV.jcs(sign(dict(base, subject=h, body=body), VIEWER)), ctx)


def wai2_with(manifest_text, rest):
    """A WAI2 envelope of the manifest text (or bytes) given, with the rest of the session
    corpus's envelope: its rendition count, table and payloads."""
    t = manifest_text.encode() if isinstance(manifest_text, str) else manifest_text
    return b"WAI2" + len(t).to_bytes(4, "little") + t + rest


def model_envelope(ins, magic):
    """The model session's envelope of this magic: its base's WAI1, or its WAI2."""
    for e in ins.by_name["session_accept"]["context"]["envelopes"]:
        b = bytes.fromhex(e["envelope"])
        if b[:4] == magic:
            return b
    raise ValueError(f"the model session holds no {magic!r} envelope")


def sourced(ins, entry, env):
    """The model session with decoder entry `entry`'s object the envelope given, signed
    again: the session that decoded from it."""
    c = copy.deepcopy(ins.receipt("session_accept"))
    c["body"]["decoders"][entry]["object"] = SV.content_hash(env)
    return CV.jcs(sign(c, VIEWER))


def session_envelopes(ins):
    """§2.4.12 step 6, the envelopes: the model session against its WAI2 envelope with
    manifests every reader may read otherwise — members of the wrong type, pins that are
    not pins, a `security` that encrypts or is no security object, duplicate members,
    numbers past every range, nesting at and past the depth a reader allows, byte-order
    marks, lone surrogates — with the claim unchanged, so only the envelope decides."""
    env = model_envelope(ins, b"WAI2")
    ml = int.from_bytes(env[4:8], "little")
    text, rest = env[8:8 + ml].decode(), env[8 + ml:]
    idx = 1  # the classical rendition

    def mutate(f):
        m = json.loads(text)
        f(m)
        return json.dumps(m, separators=(",", ":"))

    def setr(k, val, i=idx):
        def f(m):
            m["renditions"][i][k] = val
        return f

    deep = lambda n: "[" * n + "]" * n
    cases = {
        "control": text,
        "id_int": mutate(setr("id", 5)),
        "sha256_int": mutate(setr("sha256", 5)),
        "companion_of_array": mutate(setr("companion_of", [])),
        "companion_of_id_object": mutate(setr("companion_of_id", {})),
        "prior_string": mutate(setr("prior", "x")),
        "prior_empty_object": mutate(setr("prior", {})),
        "prior_no_determinism": mutate(setr("prior", {"id": "a", "sha256": "b"})),
        "prior_id_int": mutate(setr("prior", {"id": 1, "sha256": "b", "determinism": "c"})),
        "prior_tier_int": mutate(setr("prior", {"id": "a", "sha256": "b", "determinism": "c", "tier": 5})),
        "prior_supersedes_string": mutate(setr("prior", {"id": "a", "sha256": "b", "determinism": "c", "supersedes": "x"})),
        "other_entry_id_int": mutate(setr("id", 7, 0)),
        "other_entry_prior_string": mutate(setr("prior", "x", 0)),
        "role_int": mutate(setr("role", 5)),
        "security_string": mutate(lambda m: m.__setitem__("security", "none")),
        "security_int": mutate(lambda m: m.__setitem__("security", 5)),
        "security_encrypted": mutate(lambda m: m.__setitem__("security", {"encryption": "aes-128-gcm", "kid": 1})),
        "security_none": mutate(lambda m: m.__setitem__("security", {"encryption": "none"})),
        "security_kid_negative": mutate(lambda m: m.__setitem__("security", {"encryption": "none", "kid": -1})),
        "duplicate_ids": mutate(lambda m: [r.__setitem__("id", "a") for r in m["renditions"]]),
        "missing_intent": mutate(lambda m: m.pop("intent")),
        "wai_number": mutate(lambda m: m.__setitem__("wai", 1.1)),
        "target_absent": mutate(lambda m: m.pop("target")),
        "target_1e400": text.replace('"target":null', '"target":1e400'),
        "target_neg_zero": text.replace('"target":null', '"target":-0'),
        "target_2pow64": text.replace('"target":null', '"target":18446744073709551616'),
        "target_depth_70": text.replace('"target":null', '"target":' + deep(70)),
        "target_depth_126": text.replace('"target":null', '"target":' + deep(126)),
        "target_depth_130": text.replace('"target":null', '"target":' + deep(130)),
        "duplicate_member_media": text.replace('"media":"video"', '"media":"video","media":"video"'),
        "duplicate_member_in_entry": text.replace('"kind":"second"', '"kind":"second","kind":"second"'),
        "trailing_garbage": text + " x",
        "trailing_space": text + "  ",
        "bom": "\ufeff" + text,
        "lone_surrogate_target": text.replace('"target":null', '"target":"\\ud800"'),
        "manifest_array": "[]",
        "kind_null": mutate(setr("kind", None)),
        "capability_int": mutate(setr("capability", 5)),
        "entry_not_object": mutate(lambda m: m["renditions"].__setitem__(0, "x")),
        "renditions_object": mutate(lambda m: m.__setitem__("renditions", {})),
        "envelope_manifest_int": mutate(setr("envelope_manifest", 5)),
        "refines_int": mutate(setr("refines", 5)),
        "id_null": mutate(setr("id", None)),
        "prior_null": mutate(setr("prior", None)),
        "role_companion": mutate(setr("role", "companion")),
        "role_unknown": mutate(setr("role", "rendition")),
        "media_audio": mutate(lambda m: m.__setitem__("media", "audio")),
    }
    for name, t in cases.items():
        e = wai2_with(t, rest)
        ins.add("session_envelope_" + name, sourced(ins, 1, e), {"envelopes": [{"envelope": e.hex()}]})
    # The table and payload framing around a sound manifest.
    t = text.encode()
    framing = {
        "count_short": b"WAI2" + len(t).to_bytes(4, "little") + t + (1).to_bytes(2, "little") + rest[2:],
        "truncated": wai2_with(t, rest)[:-1],
        "trailing_bytes": wai2_with(t, rest) + b"\x00",
        "manifest_length_past_end": b"WAI2" + (len(t) + len(rest) + 1).to_bytes(4, "little") + t + rest,
    }
    for name, e in framing.items():
        ins.add("session_envelope_" + name, sourced(ins, 1, e), {"envelopes": [{"envelope": e.hex()}]})


def session_wai1_envelopes(ins):
    """§2.4.12 step 6, the envelopes: the model session against its base's WAI1 envelope
    with manifests every reader may read otherwise, a capability or fallback the base's
    entry does or does not name, and framing short or long; the session signed again
    with the base's entry naming that envelope, so only the envelope decides."""
    env = model_envelope(ins, b"WAI1")
    multi = {"envelope": model_envelope(ins, b"WAI2").hex()}
    ml = int.from_bytes(env[4:8], "little")
    text, payload = env[8:8 + ml].decode(), env[12 + ml:]

    def mutate(f):
        m = json.loads(text)
        f(m)
        return json.dumps(m, separators=(",", ":"))

    def mr(k, val):
        return mutate(lambda m: m["model_requirement"].__setitem__(k, val))

    deep = lambda n: "[" * n + "]" * n
    cases = {
        "control": text,
        "capability_other": mr("capability", "wai.video.av1"),
        "fallback_is_entry": mutate(lambda m: m["model_requirement"].update(capability="wai.video.av1", fallback="wai.video.int_hyper")),
        "fallback_other": mr("fallback", "wai.video.hevc"),
        "capability_int": mr("capability", 5),
        "fallback_int": mr("fallback", 5),
        "min_sim_float": mr("min_sim", 0.5),
        "min_sim_string": mr("min_sim", "0.5"),
        "min_sim_huge": text.replace('"capability":"wai.video.int_hyper"', '"capability":"wai.video.int_hyper","min_sim":1e400'),
        "sim_metric_int": mr("sim_metric", 5),
        "model_requirement_missing": mutate(lambda m: m.pop("model_requirement")),
        "model_requirement_array": mutate(lambda m: m.__setitem__("model_requirement", [])),
        "conditioning_missing": mutate(lambda m: m.pop("conditioning")),
        "conditioning_kind_int": mutate(lambda m: m["conditioning"].__setitem__("kind", 5)),
        "prior_string": mutate(lambda m: m["conditioning"].__setitem__("prior", "x")),
        "security_none": mutate(lambda m: m.__setitem__("security", {"encryption": "none"})),
        "security_null": mutate(lambda m: m.__setitem__("security", None)),
        "security_encrypted": mutate(lambda m: m.__setitem__("security", {"encryption": "aes-128-gcm", "kid": 1})),
        "security_string": mutate(lambda m: m.__setitem__("security", "none")),
        "security_kid_negative": mutate(lambda m: m.__setitem__("security", {"encryption": "none", "kid": -1})),
        "security_kid_2pow64": mutate(lambda m: m.__setitem__("security", {"kid": 2 ** 64})),
        "missing_intent": mutate(lambda m: m.pop("intent")),
        "media_int": mutate(lambda m: m.__setitem__("media", 5)),
        "wai_number": mutate(lambda m: m.__setitem__("wai", 1.0)),
        "unknown_member": mutate(lambda m: m.__setitem__("x_unknown", 1)),
        "target_object": mutate(lambda m: m.__setitem__("target", {"a": 1})),
        "target_depth_130": mutate(lambda m: m.__setitem__("target", None)).replace('"target":null', '"target":' + deep(130)),
        "refines_int": mutate(lambda m: m.__setitem__("refines", 5)),
        "derived_from_string": mutate(lambda m: m.__setitem__("derived_from", "x")),
        "duplicate_member": text.replace('"media":"video"', '"media":"video","media":"video"'),
        "duplicate_in_requirement": text.replace('"capability":"wai.video.int_hyper"', '"capability":"wai.video.int_hyper","capability":"wai.video.int_hyper"'),
        "bom": "\ufeff" + text,
        "trailing_garbage": text + " x",
        "lone_surrogate": mutate(lambda m: m.__setitem__("target", None)).replace('"target":null', '"target":"\\ud800"'),
        "manifest_array": "[]",
    }
    for name, t in cases.items():
        tb = t.encode()
        e = SV.wai1(tb, payload)
        ins.add("session_wai1_" + name, sourced(ins, 0, e), {"envelopes": [{"envelope": e.hex()}, multi]})
    tb = text.encode()
    framing = {
        "payload_short": SV.wai1(tb, payload)[:-1],
        "trailing_bytes": SV.wai1(tb, payload) + b"\x00",
        "manifest_length_past_end": b"WAI1" + (len(tb) + 100).to_bytes(4, "little") + tb,
        "no_payload_length": b"WAI1" + len(tb).to_bytes(4, "little") + tb,
        "magic_other": b"WAI3" + SV.wai1(tb, payload)[4:],
        "short": b"WAI",
    }
    for name, e in framing.items():
        ins.add("session_wai1_" + name, sourced(ins, 0, e), {"envelopes": [{"envelope": e.hex()}, multi]})


def session_classical_sweep(ins):
    """§2.4.12's classical rendition: every registered capability, an unregistered string
    and the model's own, as the second rendition of the model session's WAI2 envelope —
    once in a manifest of the session's media and once in a manifest of the capability's
    own media — so that every rule that admits or excludes one is reached."""
    env = model_envelope(ins, b"WAI2")
    ml = int.from_bytes(env[4:8], "little")
    m0, rest = json.loads(env[8:8 + ml]), env[8 + ml:]
    for cap in sorted(CV.REGISTRY) + ["org.example.codec", "wai.video.int_hyper"]:
        medias = ["video"] + ([CV.REGISTRY[cap]["media"]] if cap in CV.REGISTRY and CV.REGISTRY[cap]["media"] not in (None, "video") else [])
        for media in medias:
            m = copy.deepcopy(m0)
            m["media"] = media
            m["renditions"][1]["capability"] = cap
            e = wai2_with(json.dumps(m, separators=(",", ":")), rest)
            ins.add(f"session_classical_{cap}_{media}", sourced(ins, 1, e), {"envelopes": [{"envelope": e.hex()}]})


def session_caps(ins):
    """§2.4.12 rules 7 and 8 at their caps, where nothing else decides: 300 units presented
    once and late, with the missed presentations and the switches written as many as their
    totals, as the cap, and one either side of each."""
    base = ins.receipt("session_accept")
    ps = [{"unit": u, "layers": 1, "decoder": 0, "out_digest": CV.blake3(b"differential: capped %d" % u).hex()}
          for u in range(300)]
    root = CV.out_mth([CV.out_leaf(p) for p in ps]).hex()
    decoder = dict(base["body"]["decoders"][0], leaves=300)
    for missed, written in [(300, 255), (300, 256), (300, 257), (255, 255), (256, 256), (257, 257), (256, 255), (255, 256)]:
        for total, switched in [(0, 0), (1100, 1023), (1100, 1024), (1100, 1025), (1023, 1023), (1024, 1024), (1025, 1024),
                                (1023, 1024)]:
            sw = [{"unit": u // 4, "from": "primary" if u % 2 == 0 else "fallback",
                   "to": "fallback" if u % 2 == 0 else "primary", "reason": "recovered"} for u in range(switched)]
            body = dict(base["body"], decoders=[decoder], units=300, leaf_count=300, output_root=root,
                        primary_units=300, fallback_units=0, switches=sw, switches_total=total,
                        deadline_us=40000, missed=missed, stage_sets=[],
                        missed_units=[{"unit": u, "layers": 1, "late_us": 1 + u} for u in range(written)])
            ins.add(f"session_caps_{missed}_{written}_{total}_{switched}", CV.jcs(sign(dict(base, body=body), VIEWER)),
                    {"outputs": copy.deepcopy(ps)})


def session_decode_paths(ins):
    """§2.4.12 rules 6, 13 and 14 where nothing else decides: one picture through one entry,
    for every capability below (classified at each tier, or not at all), decoded by a
    builtin or a module, on a path named by every term below (exact, float, a lowering a
    registration names, one it names for another capability, one no revision registers, or
    none), under a pin of either tier, naming the set read with no pin verified (the tier
    `none`), or naming no set."""
    base = ins.receipt("session_variant_float_accept")
    pin = CV.blake3(b"differential: a pin").hex()
    for cap in ["org.example.picture", "wai.image.jpegai", "wai.neural.int_hyper", "wai.audio.opus.dred.int", "wai.neural.dcvc_rt"]:
        for variant in [None, "exact", "float", "fp16", "dred-int/1", "whm1/1", "jpegai-int/1"]:
            for tier in [None, "none", "entropy-consistency", "decode-equivalence"]:
                for decoder, status in [("builtin:[email protected]", None), ("module:sha256:" + "5e" * 32, "pinned")]:
                    d = dict(base["body"]["decoders"][0], capability=cap, variant=variant, pin=pin if tier else None, tier=tier,
                             decoder=decoder, module_status=status)
                    body = dict(base["body"], decoders=[d])
                    ins.add(f"session_decode_path_{cap}_{variant}_{tier}_{status}", CV.jcs(sign(dict(base, body=body), VIEWER)))


def figures(ins, rng, n):
    oer_t = ins.by_name["energy_matches_oer_accept"]["context"]["oer"]["receipt"]
    claim_t = ins.receipt("energy_matches_oer_accept")
    keys = {OER_SIGNER: pk(METER)}
    cases = [(f, uj) for f in FIGURES + NUMERIC_FIGURES + RANGE_FIGURES for uj in (1, 12000, 12001, 11999, 2, 3)]
    for _ in range(n):
        uj = rng.randrange(1, 10 ** rng.randrange(1, 16))  # a claim's integers stop at 2^53 - 1
        frac = "".join(rng.choice("0123456789") for _ in range(rng.randrange(0, 50)))
        pick = rng.randrange(4)
        if pick == 0:  # exactly half-way, with trailing zeros
            f = f"{uj // 10**6}.{uj % 10**6:06d}5" + "0" * rng.randrange(0, 40)
        elif pick == 1:  # just either side of half-way
            f = f"{uj // 10**6}.{uj % 10**6:06d}{rng.choice(['49', '50', '51'])}{frac}"
        elif pick == 2:  # exponent form of the same digits
            s = f"{uj}5{frac}"
            f = f"{s}e-{len(s) - len(str(uj)) + 6}"
        else:  # a double near the figure
            f = (uj + rng.choice([0.5, 0.4999999, 0.5000001, 0.25])) / 1e6
        cases.append((f, rng.choice([uj, uj + 1, uj - 1 if uj > 1 else uj])))
    for f, uj in cases:
        r = seal_oer(oer_t, f)
        body = dict(claim_t["body"], joules_micro=uj, oer_receipt=r["receipt_id"])
        c = sign(dict(claim_t, body=body), METER)
        ins.add(f"figure_{str(f)[:40]}_uj{uj}", CV.jcs(c), {"oer": {"receipt": r, "keys": keys}})


def sums(ins, rng, n):
    """OER receipts whose breakdown or floor exercises §6's exact arithmetic: large terms
    that cancel, differences at and one unit past the 1e-9 tolerance, floors exactly on
    the 1% bound, and quantities at §4's limits. The claim names each receipt with the
    figure's microjoules, so the verdict turns on whether each OER verifier accepts the
    receipt."""
    oer_t = ins.by_name["energy_matches_oer_accept"]["context"]["oer"]["receipt"]
    claim_t = ins.receipt("energy_matches_oer_accept")
    keys = {OER_SIGNER: pk(METER)}
    e37 = "1" + "0" * 37
    cases = [
        ("cancel_1e24", "0.012", ["1000000000000000000000000.012", "-1000000000000000000000000", "0", "0", "0"], None),
        ("cancel_1e30", "0.012", ["1000000000000000000000000000000.012", "-1000000000000000000000000000000", "0", "0", "0"], None),
        ("cancel_1e37", "0.012", ["10000000000000000000000000000000000000.012", "-" + e37, "0", "0", "0"], None),
        ("wide_1e37_1e-38", e37, [e37, "0." + "0" * 37 + "1", "0", "0", "0"], None),
        ("at_tolerance", "9999999990000000000000000000000000000", [e37, "0", "0", "0", "0"], None),
        ("past_tolerance", "9999999989999999999999999999999999999", [e37, "0", "0", "0", "0"], None),
        ("at_tolerance_unit", "1.000000001", ["2", "-1", "0", "0", "0"], None),
        ("past_tolerance_unit", "1.000000001000000000000000000000000001", ["2", "-1", "0", "0", "0"], None),
        ("negative_total", "-0.012", ["-0.012", "0", "0", "0", "0"], None),
        ("floor_tiny_temperature", "0.012", None, (256, "1e-38", "0")),
        ("floor_huge_temperature", "0.012", None, (256, "9e37", "1.6e+17")),
        ("floor_zero_bits", "0.012", None, (0, "300", "0")),
        ("floor_zero_bits_tiny_floor_j", "0.012", None, (0, "300", "1e-38")),
        ("floor_bits_beyond_64", "0.012", None, (2 ** 64, "300", "1e-18")),
        ("floor_bits_float", "0.012", None, (256.0, "300", "1.838e-18")),
    ]
    # floor_j exactly on each 1% bound, and one unit of its last digit either side: §6's
    # check is exact, so a difference exactly at the tolerance passes and one past it does
    # not, in both verifiers or neither.
    with localcontext(OER.EXACT):
        for bits in (1, 256, 510, 4096, 2 ** 40):
            for temp in ("300", "3.000000e+02", "0.001", "77.5"):
                exact = Decimal(bits) * Decimal("1.380649e-23") * Decimal(temp) * Decimal("0.6931471805599453")
                for side in ("0.99", "1.01"):
                    on = exact * Decimal(side)
                    unit = Decimal((0, (1,), on.as_tuple().exponent))
                    for name, fj in (("on", on), ("below", on - unit), ("above", on + unit)):
                        cases.append((f"floor_{name}_bound_{bits}_{temp}_{side}", "0.012", None, (bits, temp, format(fj, "e"))))
    def term():
        whole = "".join(rng.choice("0123456789") for _ in range(rng.randrange(1, 20))).lstrip("0") or "0"
        frac = "".join(rng.choice("0123456789") for _ in range(rng.randrange(1, 25)))
        return ("-" if rng.random() < 0.3 else "") + whole + "." + frac

    for _ in range(n):
        # Five random terms, some negative, up to 43 digits (so some exceed §4's 34 and make
        # the receipt malformed), and a total exactly at, inside or outside the tolerance of
        # their sum.
        terms = [term() for _ in range(5)]
        with localcontext(OER.EXACT):
            total = sum(Decimal(t) for t in terms)
            tol = Decimal("1e-9") * max(Decimal(1), abs(total))
            jt = total + rng.choice([Decimal(0), tol, -tol, tol * 2, Decimal("1e-30")])
        cases.append((f"random_sum_{len(cases)}", format(jt, "f"), terms, None))
    for name, jt, terms, floor in cases:
        r = seal_oer(oer_t, jt, terms, floor)
        uj = Decimal(jt) * 1000000 if OER.DECIMAL.fullmatch(jt) else 1
        uj = max(1, min(int(abs(uj).to_integral_value()), 2 ** 53 - 1))
        body = dict(claim_t["body"], joules_micro=uj, oer_receipt=r["receipt_id"])
        ins.add("sum_" + name, CV.jcs(sign(dict(claim_t, body=body), METER)), {"oer": {"receipt": r, "keys": keys}})


def readings(ins):
    """oer-receipts §4 "Numbers": numbers a receipt body must not carry — `-0` however
    spelt, fractions, exponents, integers beyond I-JSON's range — beside the range's own
    bounds, in figures and in members no check reads. A body holding a forbidden number is
    `malformed` before its content address is computed, so the verdict is the same however
    a reader would have rounded it. And a delivered object's microjoules spelt as numbers
    the corpus does not give them as."""
    oer_t = ins.by_name["energy_matches_oer_accept"]["context"]["oer"]["receipt"]
    claim_t = ins.receipt("energy_matches_oer_accept")
    keys = {OER_SIGNER: pk(METER)}
    def k_floor(bits):  # temperature_k "300", computed as §6 fixes it
        with localcontext(OER.EXACT):
            return format(Decimal(bits) * OER.KBOLTZ * Decimal(300) * OER.LN2, "e")

    def at(path, spelling):
        def edit(r):
            *outer, last = path
            for p in outer:
                r = r[p]
            r[last] = raw(spelling)
        return edit

    def both(*edits):
        return lambda r: [e(r) for e in edits]

    zero_terms = ["0.012000", "0", "0", "0", "0"]
    cases = [  # (name, joules_total, breakdown, floor, edit, microjoules)
        ("term_neg_zero", "0.012000", zero_terms, None, at(("energy", "breakdown", "movement_j"), "-0"), 12000),
        ("term_neg_zero_float", "0.012000", zero_terms, None, at(("energy", "breakdown", "movement_j"), "-0.0"), 12000),
        ("term_neg_zero_exp", "0.012000", zero_terms, None, at(("energy", "breakdown", "movement_j"), "-0e7"), 12000),
        ("total_neg_zero", "0", ["0", "0", "0", "0", "0"], None,
         both(at(("energy", "joules_total"), "-0"), at(("energy", "breakdown", "compute_j"), "-0")), 1),
        ("latency_neg_zero", "0.012000", None, None, at(("work", "latency_s"), "-0"), 12000),
        ("status_neg_zero", "0.012000", None, None, at(("work", "status"), "-0"), 12000),
        ("status_i64_min", "0.012000", None, None, at(("work", "status"), "-9223372036854775808"), 12000),
        ("status_below_i64", "0.012000", None, None, at(("work", "status"), "-9223372036854775809"), 12000),
        ("status_u64_max", "0.012000", None, None, at(("work", "status"), "18446744073709551615"), 12000),
        ("status_i_json_max", "0.012000", None, None, at(("work", "status"), "9007199254740991"), 12000),
        ("status_i_json_min", "0.012000", None, None, at(("work", "status"), "-9007199254740991"), 12000),
        ("status_beyond_i_json", "0.012000", None, None, at(("work", "status"), "9007199254740992"), 12000),
        ("status_below_i_json", "0.012000", None, None, at(("work", "status"), "-9007199254740992"), 12000),
        ("status_one_point_zero", "0.012000", None, None, at(("work", "status"), "1.0"), 12000),
        ("status_exponent", "0.012000", None, None, at(("work", "status"), "1e0"), 12000),
        ("status_beyond_u64", "0.012000", None, None, at(("work", "status"), "18446744073709551616"), 12000),
        ("status_1e30_integer", "0.012000", None, None, at(("work", "status"), "1" + "0" * 30), 12000),
        ("status_308_digits", "0.012000", None, None, at(("work", "status"), "1" + "0" * 308), 12000),
        ("status_exponent_zeros", "0.012000", None, None, at(("work", "status"), "1e" + "0" * 400 + "5"), 12000),
        ("status_underflow", "0.012000", None, None, at(("work", "status"), "-1e-400"), 12000),
        ("bits_beyond_u64", "0.012000", None, (2 ** 64, "300", k_floor(float(2 ** 64))),
         at(("floor", "bits_committed"), "18446744073709551616"), 12000),
        ("bits_1e30_integer", "0.012000", None, (10 ** 30, "300", k_floor(1e30)),
         at(("floor", "bits_committed"), "1" + "0" * 30), 12000),
        ("bits_neg_zero", "0.012000", None, (0, "300", "0"), at(("floor", "bits_committed"), "-0"), 12000),
        ("total_beyond_u64_micro", "18446744073709.551616", ["18446744073709.551616", "0", "0", "0", "0"], None,
         at(("energy", "joules_total"), "18446744073709551616e-6"), 2 ** 53 - 1),
        ("total_number_spelt_long", "0.012000", zero_terms, None,
         at(("energy", "joules_total"), "0.01200000000000000000000000000000000000000000000001"), 12000),
    ]
    for name, jt, terms, floor, edit, uj in cases:
        r = seal_oer(oer_t, jt, terms, floor, edit)
        body = dict(claim_t["body"], joules_micro=uj, oer_receipt=r["receipt_id"])
        c = CV.jcs(sign(dict(claim_t, body=body), METER))
        ins.add(f"reading_{name}", c, {"oer": {"receipt": r, "keys": keys}})
    delivery = ins.by_name["energy_matches_delivery_accept"]
    for spelling in ["12000", "-0", "12000.0", "1.2e4", "18446744073709551616", "-12000"]:
        ctx = dict(delivery["context"], object=dict(delivery["context"]["object"], joules_micro=raw(spelling)))
        ins.add("delivered_joules_" + spelling[:24], delivery["receipt"], ctx)


# (name, spelling) of a value placed in an OER receipt's `work.outcome`, each in a file of
# its own: whether the file is read at all is the first thing compared.
DOCUMENTS = [
    ("control", '"encoded"'), ("max_double", "1.7976931348623157e308"), ("below_max_half_ulp", "1.7976931348623158e308"),
    ("past_max_half_ulp", "1.7976931348623159e308"), ("e400", "1e400"), ("neg_e400", "-1e400"), ("e308_8", "1.8e308"),
    ("digits_400", "1" + "0" * 400), ("neg_digits_400", "-1" + "0" * 400), ("zero_e_huge", "0e99999999999999999999"),
    ("one_e_huge", "1e99999999999999999999"), ("one_e_neg_huge", "1e-99999999999999999999"),
    ("nan", "NaN"), ("infinity", "Infinity"), ("neg_infinity", "-Infinity"),
    ("lone_high_surrogate", '"\\ud800"'), ("lone_low_surrogate", '"\\udc00"'), ("paired_surrogates", '"\\ud83d\\ude00"'),
    ("high_surrogate_then_letter", '"\\ud800\\u0041"'),
    # A repeated member name, in an object no check reads, and a repeated figure.
    ("repeated_member", '{"a":1,"a":2}'), ("repeated_member_same_value", '{"a":1,"a":1}'),
    ("distinct_members", '{"a":1,"b":2}'),
]
# work.outcome sits at level 7 of a corpus file ({vectors:[{context:{oer:{receipt:{work:
# ...}}}}]}), so arrays nested d deep inside it reach level 7 + d; §4 reads 64 levels.
DOCUMENTS += [(f"nesting_to_level_{7 + d}", "[" * d + "]" * d) for d in (55, 56, 57, 58, 121, 200, 1100, 5000)]


def documents(corpus):
    """One corpus-format file per DOCUMENTS entry: an energy claim naming an OER receipt that
    holds the value, sealed over §4's reading where the text is one §4 reads."""
    ins = Inputs(corpus)
    oer_t = ins.by_name["energy_matches_oer_accept"]["context"]["oer"]["receipt"]
    claim_t = ins.receipt("energy_matches_oer_accept")
    keys = {OER_SIGNER: pk(METER)}
    out = []
    for name, spelling in DOCUMENTS:
        def edit(r, spelling=spelling):
            r["work"]["outcome"] = raw(spelling)
        try:
            r = seal_oer(oer_t, "0.012000", None, None, edit)
        except (ValueError, RecursionError):
            # §4 does not read the receipt's own text. Sealed with a placeholder, which only
            # matters if a verifier reads the file anyway — and then the verdicts differ.
            r = seal_oer(oer_t, "0.012000")
            edit(r)
        body = dict(claim_t["body"], joules_micro=12000, oer_receipt=r["receipt_id"])
        one = Inputs(corpus)
        one.add("document_" + name, CV.jcs(sign(dict(claim_t, body=body), METER)), {"oer": {"receipt": r, "keys": keys}})
        out.append((name, text_of({"about": "one differential input", "vectors": one.out})))
    return out


def reports(ins, rng, n):
    """measurement and operating-point claims (§2.4.8-§2.4.9) around the edges of their
    rules: brackets at and one ppm below the least declaration, figures one off, ties at a
    half microjoule, the ratio thresholds, flat brackets and a held value, products past 128 bits,
    every pattern of liveness facts; operating points in every presence combination; and
    reports and energy claims against figure views and labelled objects."""
    m0 = ins.receipt("measurement_accept")
    fig0 = ins.by_name["measurement_accept"]["context"]["figure"]
    op0 = ins.receipt("operating_point_accept")
    hw = ["OnChipCounter", "CalibratedInstrument"]

    def summary(n_, mean, sd, live=None, run=None):
        live = n_ if live is None else live
        return {"n": n_, "mean_uw": mean, "sd_uw": sd, "live_intervals": live, "longest_live_run": n_ if run is None else run}

    def class_fields(label, ppm, window, figure):
        b = {"energy_provenance": label, "energy_uncertainty": None, "energy_counter_filtering": None,
             "energy_calibration_ref": None}
        if label == "OnChipCounter":
            b.update(energy_uncertainty={"relative_ppm": ppm, "window_us": window}, energy_counter_filtering="Unknown")
        elif label == "CalibratedInstrument":
            b.update(energy_uncertainty={"relative_ppm": ppm, "window_us": window},
                     energy_calibration_ref="https://calibration.example.org/certificates/0001")
        return dict(b, joules_micro=figure)

    def report(br, label, ppm, window, figure, name, ctx=None):
        body = dict(m0["body"], bracket=br, **class_fields(label, ppm, window, figure))
        ins.add(name, CV.jcs(sign(dict(m0, body=body), METER)), ctx)

    def no_reader(br, label, ppm, window, figure, name, ctx=None):
        body = dict(m0["body"], bracket=br, power_reader=None, **class_fields(label, ppm, window, figure))
        ins.add(name, CV.jcs(sign(dict(m0, body=body), METER)), ctx)

    def around(name, br, labels=("OnChipCounter", "CalibratedInstrument", "ModelBased")):
        f = CV.bracket_figure(br)
        for label in labels:
            if f is None:
                report(br, label, 300_000, br["window_us"], 1, f"{name}_nofig_{label}")
                continue
            p = CV.least_ppm(br, f) or 2 ** 32 - 1
            for d_ppm, d_fig, d_win in [(0, 0, 0), (-1, 0, 0), (1, 0, 0), (0, 1, 0), (0, -1, 0), (0, 0, 1)]:
                ppm = max(1, min(2 ** 32 - 1, p + d_ppm))
                fig = f + d_fig
                if fig <= 0 or fig > CV.MAX_INT:
                    continue
                report(br, label, ppm, br["window_us"] + d_win, fig, f"{name}_{label}_{d_ppm}_{d_fig}_{d_win}")

    def bracket(idle, busy, t, c, nn, opening):
        return {"idle": idle, "busy": busy, "window_us": t, "completions": c, "in_flight": nn, "opening": opening}

    # Fixed edges.
    around("half_tie", bracket(summary(10, 20_000_000, 1_000), summary(14, 21_000_000, 1_000), 2_500_000, 1_000_000, 1, "running"))
    around("flat_idle", bracket(summary(10, 20_000_000, 0), summary(14, 30_000_000, 115_471), 3_000_000, 60, 10, "from-rest"))
    around("flat_busy", bracket(summary(10, 20_000_000, 115_471), summary(14, 30_000_000, 0), 3_000_000, 60, 10, "from-rest"))
    around("held_value", bracket(summary(10, 5_000_000, 0, 1, 1), summary(14, 10_500_000, 2_980_708, 2, 1), 3_000_000, 60, 10, "from-rest"))
    # A report that names no power reader, with and without a bracket, under each class.
    for label in ("OnChipCounter", "CalibratedInstrument", "ModelBased", "Estimator"):
        unit_br = m0["body"]["bracket"]
        f = CV.bracket_figure(unit_br)
        p = CV.least_ppm(unit_br, f)
        for br in (unit_br, None):
            no_reader(br, label, p, unit_br["window_us"], f, f"no_reader_{label}_{br is None}", fig0)
    around("wide", bracket(summary(10, 20_000_000, 115_471), summary(14, 30_000_000, 115_471), 10 ** 11, 10 ** 13, 64, "from-rest"))
    for k in (10, 3, 1):
        for d in (0, -1):
            around(f"ratio_{k}_{d}", bracket(summary(10, 20_000_000, 100_000), summary(14, 20_000_000 + k * 100_000 + d, 100_000),
                                             3_000_000, 60, 10, "from-rest"), labels=("OnChipCounter", "ModelBased"))
    # Liveness facts: every pattern around the thresholds, in each bracket.
    for n_ in (4, 5, 6, 10, 14):
        for live in sorted({0, 1, 2, 4, 5, n_ * 4 // 5, n_ * 4 // 5 + 1, n_ - 1, n_, n_ + 1}):
            for run in sorted({0, 1, 2, 3, live, live + 1}):
                idle = summary(10, 20_000_000, 115_471)
                busy = summary(n_, 30_000_000, 115_471, live, run)
                br = bracket(idle, busy, 3_000_000, 60, 10, "from-rest")
                f = CV.bracket_figure(br)
                p = CV.least_ppm(br, f)
                report(br, "OnChipCounter", p, 3_000_000, f, f"live_busy_{n_}_{live}_{run}")
                br2 = bracket(summary(n_, 20_000_000, 115_471, live, run), summary(14, 30_000_000, 115_471), 3_000_000, 60, 10, "from-rest")
                report(br2, "OnChipCounter", CV.least_ppm(br2, f), 3_000_000, f, f"live_idle_{n_}_{live}_{run}")
    # Random brackets.
    for i in range(n):
        ni, nb = rng.choice([1, 2, 3, 5, 10, 14, 50]), rng.choice([1, 2, 3, 5, 10, 14, 50])
        idle_mean = rng.randrange(0, 50_000_000)
        busy_mean = idle_mean + rng.choice([-1, 0, 1, rng.randrange(1, 40_000_000)])
        sd = lambda: rng.choice([0, 1, rng.randrange(1, 5_000_000)])  # noqa: E731
        live = lambda m: rng.choice([m, m - 1, 0, 2, 5])  # noqa: E731
        idle = summary(ni, idle_mean, sd(), live(ni))
        busy = summary(nb, max(0, busy_mean), sd(), live(nb))
        t = rng.choice([0, 1, rng.randrange(1, 10 ** 7), rng.randrange(1, 10 ** 12)])
        c = rng.choice([0, 1, rng.randrange(1, 10 ** 6), rng.randrange(1, 10 ** 13)])
        nn = rng.choice([0, 1, rng.randrange(1, 65)])
        br = bracket(idle, busy, t, c, nn, rng.choice(["from-rest", "running"]))
        f = CV.bracket_figure(br)
        label = rng.choice(hw + ["ModelBased", "Estimator"])
        if f is None:
            report(br, label, 300_000, max(1, t), rng.randrange(1, 10 ** 6), f"random_{i}_nofig")
            continue
        p = CV.least_ppm(br, f)
        ppm = max(1, min(2 ** 32 - 1, (p or 2 ** 32 - 1) + rng.choice([0, 0, -1, 1, 1000])))
        if f * ppm < 288_676:
            ppm = 288_676 // f + 1
        report(br, label, ppm, rng.choice([max(1, t), t + 1]), f + rng.choice([0, 0, 0, 1, -1]) or 1, f"random_{i}")

    # Operating points: every presence combination, and points that are empty, not terms,
    # or not integers.
    pts = [None, {}, {"height": 1080}, {"height": 720}, {"frame rate": 1}, {"height": "1080"}, {"height": -1}]
    pred = op0["body"]["prediction"]
    for ri, r in enumerate(pts):
        for ci, c in enumerate(pts):
            for di, d in enumerate(pts):
                for pi, pr in enumerate([None, pred]):
                    body = dict(op0["body"], op_requested=r, op_committed=c, op_delivered=d, prediction=pr)
                    ins.add(f"op_{ri}{ci}{di}{pi}", CV.jcs(sign(dict(op0, body=body), METER)) if all(
                        not isinstance(x, dict) or all(isinstance(v, int) and v >= 0 for v in x.values()) for x in (r, c, d))
                        else text_of(sign_raw(dict(op0, body=body), METER)), {"figure": fig0})

    # Figure views that agree and disagree in each member.
    e = json.loads(fig0["claim"])
    eb = e["body"]
    base = {"profile": "moq-jwp:claim-link", "identity": CV.link_hash(e), "group": None,
            "figures": [{"index": None, "joules_micro": eb["joules_micro"], "energy_provenance": eb["energy_provenance"],
                         "energy_uncertainty": eb["energy_uncertainty"],
                         "energy_counter_filtering": eb["energy_counter_filtering"]}],
            "signer": pk(METER)}
    variants = [("same", {}), ("profile", {"profile": "wai:asset-receipt-id"}), ("identity", {"identity": "ab" * 32}),
                ("group", {"group": "cd" * 32}), ("signer", {"signer": pk(OPERATOR)}), ("no_signer", {"signer": None})]
    fig_variants = [("figure_plus", {"joules_micro": eb["joules_micro"] + 1}), ("unlabelled", {"energy_provenance": None,
                    "energy_uncertainty": None, "energy_counter_filtering": None}),
                    ("filtering_absent", {"energy_counter_filtering": None}), ("filtering_off", {"energy_counter_filtering": "Off"}),
                    ("ppm", {"energy_uncertainty": dict(eb["energy_uncertainty"], relative_ppm=eb["energy_uncertainty"]["relative_ppm"] + 1)}),
                    ("index", {"index": 0})]
    for name, edit in variants:
        ins.add("view_" + name, CV.jcs(m0), {"figure": {"view": dict(base, **edit)}})
        ins.add("view_op_" + name, CV.jcs(op0), {"figure": {"view": dict(base, **edit)}})
    for name, edit in fig_variants:
        fv = [{k: v for k, v in dict(base["figures"][0], **edit).items() if v is not None or k == "index"}]
        ins.add("view_fig_" + name, CV.jcs(m0), {"figure": {"view": dict(base, figures=fv)}})
    ins.add("view_none_given_claim_not_json", CV.jcs(m0), {"figure": {"claim": "not json"}})

    # Energy claims against labelled objects.
    ec = ins.receipt("energy_on_chip_counter_accept")
    b = ec["body"]
    lab = {"energy_provenance": "OnChipCounter", "energy_uncertainty": b["energy_uncertainty"], "energy_counter_filtering": "Unknown"}
    for name, j, extra in [("same", b["joules_micro"], lab), ("no_label", b["joules_micro"], {}),
                           ("filtering_absent", b["joules_micro"], {k: v for k, v in lab.items() if k != "energy_counter_filtering"}),
                           ("model", b["joules_micro"], {"energy_provenance": "ModelBased"}),
                           ("ppm", b["joules_micro"], dict(lab, energy_uncertainty={"relative_ppm": 20_001, "window_us": 1_000_000})),
                           ("figure", b["joules_micro"] + 1, lab)]:
        ins.add("labelled_object_" + name, CV.jcs(ec), {"object": dict({"content_hash": ec["subject"], "joules_micro": j}, **extra)})


def sign_raw(c, k):
    """`c` signed over its JCS where JCS can serialise it, with a member JCS refuses (a
    string or negative point value) left for the verifier to refuse at step 2. The
    signature covers a canonical form computed with that member as written."""
    c = copy.deepcopy(c)
    c.pop("sig", None)
    c["signer_pubkey"] = pk(k)
    def canon(v):
        if isinstance(v, int) and not isinstance(v, bool) and v < 0:
            return str(v)
        if isinstance(v, dict):
            keys = sorted(v, key=lambda x: x.encode("utf-16-be"))
            return "{" + ",".join(json.dumps(x, ensure_ascii=False) + ":" + canon(v[x]) for x in keys) + "}"
        if isinstance(v, list):
            return "[" + ",".join(canon(x) for x in v) + "]"
        return CV.jcs(v)
    c["sig"] = k.sign(b"moq-jwp:claim\x01" + canon(c).encode()).hex()
    return c


def contexts(ins):
    decl = ins.receipt("soft_binding_accept")
    record = ins.receipt("binding_recovery_accept")
    lawful = ins.receipt("lawful_basis_declaration_accept")
    decision = ins.receipt("lawful_basis_decision_accept")
    e0 = ins.receipt("energy_accept")
    e1 = ins.receipt("energy_chain_accept")
    ppk = pk(PUBLISHER)

    def recovery_case(name, declaration_wire, rec, algs=(WM_ALG,), publisher=ppk):
        ctx = {"declaration": declaration_wire, "publisher": publisher, "exact_match_algs": list(algs)}
        ins.add(name, CV.jcs(rec), ctx)

    def naming(rec_template, declaration, k, **body):
        b = dict(rec_template["body"], declaration=CV.link_hash(declaration), **body)
        return sign(dict(rec_template, body=b), k)

    # A resolver-signed declaration with its signer_pubkey rewritten to the publisher's
    # key, and a record naming that forgery: a declaration parsed but not verified.
    forged = dict(sign(decl, RESOLVER), signer_pubkey=ppk)
    recovery_case("recovery_forged_declaration", CV.jcs(forged), naming(record, forged, RESOLVER))
    forged_pub = dict(decl, seq=decl["seq"] + 1)  # the publisher's declaration, altered after signing
    recovery_case("recovery_altered_declaration", CV.jcs(forged_pub), naming(record, forged_pub, RESOLVER))
    recovery_case("recovery_declaration_not_json", "not json", record)
    recovery_case("recovery_declaration_not_a_claim", CV.jcs(dict(decl, kind="group")), record)
    recovery_case("recovery_declaration_lawful", CV.jcs(lawful), naming(record, lawful, RESOLVER))
    recovery_case("recovery_declaration_energy_forged", CV.jcs(dict(e0, seq=5)), naming(record, dict(e0, seq=5), RESOLVER))
    recovery_case("recovery_record_alg_differs", CV.jcs(decl), naming(record, decl, RESOLVER, alg="org.example.watermark.2"))
    recovery_case("recovery_record_type_differs", CV.jcs(decl), naming(record, decl, RESOLVER, binding_type="fingerprint"))
    recovery_case("recovery_record_no_matcher", CV.jcs(decl), record, algs=())
    recovery_case("recovery_record_ok", CV.jcs(decl), record)
    recovery_case("recovery_record_rogue_publisher", CV.jcs(decl), record, publisher=pk(RESOLVER))
    # A decision naming a declaration altered after the operator signed it (a purpose added).
    altered = copy.deepcopy(lawful)
    altered["body"]["purposes"].append({"purpose": "dpv:PersonalisedAdvertising", "basis": "dpv:Consent", "data": []})
    dec_b = dict(decision["body"], declaration=CV.link_hash(altered), purpose="dpv:PersonalisedAdvertising")
    ins.add("decision_altered_declaration", CV.jcs(sign(dict(decision, body=dec_b), VIEWER)), {"declaration": CV.jcs(altered)})
    ins.add("decision_declaration_not_json", CV.jcs(decision), {"declaration": "{"})
    ins.add("decision_declaration_is_decision", CV.jcs(sign(dict(decision, body=dict(decision["body"], declaration=CV.link_hash(decision))), VIEWER)),
            {"declaration": CV.jcs(decision)})
    ins.add("decision_declaration_is_soft_binding", CV.jcs(sign(dict(decision, body=dict(decision["body"], declaration=CV.link_hash(decl))), VIEWER)),
            {"declaration": CV.jcs(decl)})
    st = sign(dict(decision, subject_type="receipt"), VIEWER)
    ins.add("decision_subject_type_differs", CV.jcs(st), {"declaration": CV.jcs(lawful)})
    ins.add("declaration_given_to_a_declaration", CV.jcs(lawful), {"declaration": CV.jcs(lawful)})
    ins.add("declaration_given_to_energy", CV.jcs(e0), {"declaration": "not json"})
    # Step 5 against a previous claim that does not itself verify.
    bad_prev = dict(e0, seq=0, issued_at="2026-09-27T12:00:01Z")  # altered after signing
    e1_bad = sign(dict(e1, parent_receipt_hash=CV.link_hash(bad_prev)), METER)
    ins.add("previous_altered_named", CV.jcs(e1_bad), {"previous": CV.jcs(bad_prev)})
    ins.add("previous_altered_head", CV.jcs(sign(dict(e1, parent_receipt_hash=None), METER)), {"previous": CV.jcs(bad_prev)})
    ins.add("previous_not_json", CV.jcs(e1), {"previous": "[]"})
    ins.add("previous_ok", CV.jcs(e1), {"previous": CV.jcs(e0)})
    # §3.2 with the same declarations.
    for name, d in [("forged", forged), ("altered", forged_pub), ("ok", decl)]:
        ins.add("recover_" + name, None, {"declaration": CV.jcs(d), "publisher": ppk, "observed_value": decl["body"]["value"],
                                          "observed_hash": None, "exact_match_algs": [WM_ALG]}, procedure="recover")


def spans(ins, rng, n):
    """jwp-receipts §2.6.4 and §3.2 through a span: `n` span declarations with random ranges
    near one another — media time, groups and objects each present or not, in order or
    inverted, halves missing — each signed by the publisher and verified on its own, and,
    for each, recoveries through it with delivery receipts on and around the edges of its
    ranges, on another track, under another key, and of the anchor or another object, and a
    recovery record naming it checked with the same receipts."""
    decl = ins.receipt("soft_binding_span_groups_accept")
    record = ins.receipt("binding_recovery_span_accept")
    ppk, value, observed = pk(PUBLISHER), decl["body"]["value"], record["body"]["observed"]
    anchor, other = decl["subject"], "ab" * 32  # the anchor, and an object of the track that is not it
    near = lambda: rng.choice([0, 1, 2, 3, 40, 41, 42, 2 ** 53 - 1])
    for i in range(n):
        g = rng.choice([None, (near(), near()), (near(), None), (None, near())])
        o = rng.choice([None, (near(), near()), (near(), None)])
        t0 = rng.choice([0, 1, 120000, 179999])
        t1 = rng.choice([t0, t0 + 1, max(t0 - 1, 0), 179999])
        sp = {"track_alias": 7, "start_ms": t0, "end_ms": t1,
              "first_group": g[0] if g else None, "last_group": g[1] if g else None,
              "first_object": o[0] if o else None, "last_object": o[1] if o else None}
        d = sign(dict(decl, body=dict(decl["body"], span=sp)), PUBLISHER)
        ins.add(f"span_{i}", CV.jcs(d))
        rec = sign(dict(record, body=dict(record["body"], declaration=CV.link_hash(d))), RESOLVER)
        for j in range(6):
            r = {"signer": ppk if rng.random() < 0.9 else pk(RESOLVER), "track_alias": 7 if rng.random() < 0.9 else 8,
                 "group_id": near(), "object_id": near(), "content_hash": rng.choice([anchor, other])}
            hand = rng.choice([None, observed, r["content_hash"], anchor])
            ins.add(f"span_{i}_recover_{j}", None, {"declaration": CV.jcs(d), "receipt": r, "observed_value": value,
                                                   "observed_hash": hand, "exact_match_algs": [WM_ALG]}, procedure="recover")
            ins.add(f"span_{i}_record_{j}", CV.jcs(rec), {"declaration": CV.jcs(d), "receipt": r, "exact_match_algs": [WM_ALG]})


# ------------------------------------------------------------------ run and compare

def run_raw(cmd):
    env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONDONTWRITEBYTECODE="1")
    return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", env=env)


def run(cmd):
    p = run_raw(cmd)
    if p.returncode != 0:
        sys.exit(f"{' '.join(cmd)} exited {p.returncode}: {p.stderr.strip()[:400]}")
    return [line.split("\t", 1) for line in p.stdout.splitlines()]


def document_outcome(p, python):
    """`refused`, the verdict lines, or `crash: ...`. The reference refuses a file with exit
    status 1, no output and `not read:` on stderr; anything else that exits non-zero is a
    crash. wai-rs exits 1 with its reader's error on stderr."""
    if p.returncode == 0:
        return "\n".join(p.stdout.splitlines())
    if p.stdout or p.returncode != 1 or (python and "not read:" not in p.stderr):
        return "crash: exit %d: %s" % (p.returncode, p.stderr.strip().splitlines()[-1:] or "")
    return "refused"


def main(argv):
    ap = argparse.ArgumentParser()
    ap.add_argument("rust_bin")
    ap.add_argument("--random", type=int, default=200)
    ap.add_argument("--seed", type=int, default=1)
    ap.add_argument("--keep")
    a = ap.parse_args(argv[1:])
    rng = random.Random(a.seed)
    corpus = OER.load(open(os.path.join(HERE, "vectors.json"), encoding="utf-8"))
    ins = Inputs(corpus)
    numbers(ins)
    strings(ins)
    nesting(ins)
    dates(ins, rng, a.random)
    endpoints(ins, rng, a.random)
    figures(ins, rng, a.random)
    sums(ins, rng, a.random)
    classes(ins, random.Random(f"classes {a.seed}"), a.random)  # its own stream: other families' draws are unchanged
    readings(ins)
    contexts(ins)
    bodies(ins)
    reports(ins, random.Random(f"reports {a.seed}"), a.random)  # its own stream, as classes()
    stages(ins, random.Random(f"stages {a.seed}"), a.random)  # its own stream, as classes()
    sessions(ins, random.Random(f"sessions {a.seed}"), a.random)  # its own stream, as classes()
    session_caps(ins)
    session_decode_paths(ins)
    session_stage_bounds(ins, random.Random(f"session bounds {a.seed}"), a.random)  # its own stream, as classes()
    session_stage_epochs(ins, random.Random(f"session epochs {a.seed}"), a.random)  # its own stream, as classes()
    session_held(ins, random.Random(f"session held {a.seed}"), a.random)  # its own stream, as classes()
    session_envelopes(ins)
    session_wai1_envelopes(ins)
    session_classical_sweep(ins)
    session_roles_sweep(ins)
    session_encrypted(ins, random.Random(f"session encrypted {a.seed}"), a.random)  # its own stream, as classes()
    spans(ins, random.Random(f"spans {a.seed}"), a.random)  # its own stream, as classes()
    doc = text_of({"about": "differential inputs; no outcome is declared", "vectors": ins.out})
    with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8") as f:
        f.write(doc)
        path = f.name
    try:
        rust = run([a.rust_bin, "verdicts", path])
        py = run([sys.executable, os.path.join(HERE, "claim_verify.py"), "verdicts", path])
    finally:
        if a.keep:
            os.replace(path, a.keep)
        else:
            os.unlink(path)
    if len(rust) != len(ins.out) or len(py) != len(ins.out):
        sys.exit(f"verdict counts differ: {len(ins.out)} inputs, wai-rs {len(rust)}, python {len(py)}")
    differ = [(n, r, p) for (n, r), (_, p) in zip(rust, py) if r != p]
    crashes = [(n, v) for n, v in rust + py if v.startswith("crash")]
    tally = {}
    for _, v in rust:
        k = v.split(" link_hash=")[0].split(" ancestor=")[0]
        tally[k] = tally.get(k, 0) + 1
    oer = "reject step=6 reason=energy_oer_invalid"
    for n, r, p in differ:
        # Marked, never set apart: the verdicts differ either way.
        mark = "  (one side's OER verifier refused the receipt)" if oer in (r, p) else ""
        print(f"DIFFER {n}{mark}\n    wai-rs: {r}\n    python: {p}")
    for n, v in crashes:
        print(f"CRASH {n}: {v}")
    print(f"{len(ins.out)} inputs; verdicts: " + ", ".join(f"{k} x{c}" for k, c in sorted(tally.items())))
    # Files of their own: read by both or refused by both, and the same verdicts if read.
    doc_differ, doc_crash, doc_tally = 0, 0, {}
    with tempfile.TemporaryDirectory() as d:
        for name, text in documents(corpus):
            path = os.path.join(d, name + ".json")
            with open(path, "w", encoding="utf-8") as f:
                f.write(text)
            r = document_outcome(run_raw([a.rust_bin, "verdicts", path]), False)
            p = document_outcome(run_raw([sys.executable, os.path.join(HERE, "claim_verify.py"), "verdicts", path]), True)
            key = r.split("\t", 1)[-1].split(" link_hash=")[0]
            doc_tally[key] = doc_tally.get(key, 0) + 1
            if r.startswith("crash") or p.startswith("crash"):
                doc_crash += 1
                print(f"CRASH document {name}\n    wai-rs: {r}\n    python: {p}")
            elif r != p:
                doc_differ += 1
                print(f"DIFFER document {name}\n    wai-rs: {r}\n    python: {p}")
    print(f"{len(DOCUMENTS)} documents; wai-rs: " + ", ".join(f"{k} x{c}" for k, c in sorted(doc_tally.items())))
    print(f"{len(differ)} differ, {len(crashes)} crash; documents: {doc_differ} differ, {doc_crash} crash")
    return 1 if differ or crashes or doc_differ or doc_crash else 0


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