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

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

Written from the extension's text and the sections of energy-measurement, jwp-receipts and
live-metering it cites. It shares no code with wai-rs, which generates the corpus, and never
runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's strict
rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library, and reads each
per-hop record `WMH1` with the live-metering corpus's own second implementation,
live_metering_verify.py (live-metering §3.6, §3.5). Everything else is here, from the text:

- RFC 8785 canonical JSON over the values a record may hold, and a JSON reader that keeps
  duplicate members and non-integer numbers visible, so that §7 step 1 can refuse them;
- the three records (§3): their shape (§7 step 1), signature (step 2), classes (step 3)
  and identities;
- the decode table (§4, step 4): the readings recomputed into rounds (§4.3), each round's
  figure and least declaration under energy-measurement §6.2 from rest, the row's status,
  median, range, verdict and the declaration floor `max(p, q)` (§4.5);
- the segment receipt (§5, step 5): its method, the bounded bracket (`Wn = Wd = 1`,
  `N = 0`), the table formula against the held table, the estimator
  `estimator-cpu-time/1` (energy-measurement §9.4), and the chain (§5.3);
- the session receipt (§6, step 6): its hop entries, the display estimator
  `estimator-display/1` (energy-measurement §9.5), a hop's figure under the sole-user
  rule against its `WMH1` record, the streams against the segment receipts, and the sink-decode part's figure,
  class and declaration ⌈Σ(Fᵢ·pᵢ)/ΣFᵢ⌉ (§6.3);

all with Python's integers and no floating point.

For every case of vectors.json it requires the outcome the corpus states, and for an
accepted case its identity; of the corpus, that every reason of §8 refuses a case and that
§8 lists exactly the reasons this verifier raises (and its prose exactly the row and
unmetered reasons it knows); for every measured table under tables/<sink>/, that it is
accepted with its readings held, and that its readings and procedure are the files its
digests name. Then it mutates copies of the corpus, one defect each (re-signing a mutated
record with a key of its own where the defect must reach a rule past the signature), and
requires every copy to fail, most of them with the reason the text gives the defect.

    python3 -I sink_energy_verify.py <live-conformance/sink-energy>
    python3 -I sink_energy_verify.py --table <dir>
    python3 -I sink_energy_verify.py --self-test

`--table <dir>` verifies one measured table directory, <dir>/table.json, readings.json and
procedure.txt, exactly as each tables/<sink>/ of the corpus is verified: the table accepted
(§7 steps 1–4) with its readings held, `readings_sha256` the SHA-256 of readings.json's
bytes and `procedure_sha256` that of procedure.txt's, and its rows summarised. The hardware
jobs of continuous integration run it on a table the procedure has just written (§9).

Exit status 0 only if every check holds and every mutated copy fails (with `--table`, only
if that table holds). Python 3.8 or later;
standard library only.
"""
import copy
import hashlib
import json
import math
import os
import re
import sys

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

TABLE, SEGMENT, SESSION = "table", "segment", "session"
# §3: first member, signing domain, identity domain.
KINDS = {
    TABLE: ("wai_sink_table", b"wai:sink-table\x01", b"wai:sink-table-id\x01"),
    SEGMENT: ("wai_sink_segment", b"wai:sink-segment\x01", b"wai:sink-segment-id\x01"),
    SESSION: ("wai_sink_session", b"wai:sink-session\x01", b"wai:sink-session-id\x01"),
}
MAX = 2 ** 53 - 1
U32 = 2 ** 32 - 1
FLOOR = 288676  # energy-measurement §2.3
LIVE_SPAN_US = 300000  # energy-measurement §3.1, sink-energy §4.5 rule 2
# §4.2 and energy-measurement §9.2, §9.3: the counter profiles and their minimum windows.
MINIMUM_US = {"macos-task": 1000000, "macos-gpu": 1000000, "linux-rapl": 1000000, "linux-nvml": 5000000}
CPU_EST = "estimator-cpu-time/1"
DISPLAY_EST = "estimator-display/1"
PROFILES = frozenset(MINIMUM_US) | {CPU_EST, DISPLAY_EST}  # §2 *Profile*
# §4.2, in energy-measurement §9.1's fabrics.
FABRICS = {"macos-task": ["cpu"], "linux-rapl": ["cpu"], CPU_EST: ["cpu"], "macos-gpu": ["gpu"], "linux-nvml": ["gpu"]}
# energy-measurement §2, strongest first.
CLASSES = ("CalibratedInstrument", "OnChipCounter", "HwShunt", "ModelBased", "Estimator")
RANK = {c: i for i, c in enumerate(CLASSES)}
HARDWARE = ("CalibratedInstrument", "OnChipCounter")
FILTERING = ("On", "Off", "Unknown")
COUNTER_STATUS = ("live", "frozen", "needs-privilege", "no-counter", "not-implemented", "implausible")
ABSENT_REASONS = ("no_decoder", "no_work_item", "no_hardware_path", "no_counter")
ROUND_RULES = ("too_few_samples", "rail_not_live", "window_short", "empty_burst", "no_marginal",
               "below_attribution_floor", "figure_zero", "not_declarable")
ROW_REASONS = ABSENT_REASONS + ROUND_RULES + ("noisy",)  # §8 *row reasons*
UNMETERED_REASONS = ("no_counter", "rail_not_live", "window_short", "no_marginal", "below_attribution_floor",
                     "figure_zero", "not_declarable", "not_reported")  # §8 *unmetered reasons*
ORIGIN, RELAY, SINK_DECODE, DISPLAY = 1, 2, 3, 4  # §6.2's roles
ROLE_NAME = {ORIGIN: "origin", RELAY: "relay", SINK_DECODE: "sink decode", DISPLAY: "display"}
MARKER_REASONS = ("not_reported", "hop_unmetered", "hop_shared", "unmetered", "display_undeclared")
PART_REASONS = {"origin": ("not_reported", "hop_unmetered", "hop_shared"),
                "relay": ("not_reported", "hop_unmetered", "hop_shared"),
                "sink_decode": ("unmetered",), "display": ("display_undeclared", "unmetered")}
PART_COVERAGE = {"origin": "Partial", "relay": "Partial", "display": "Whole"}  # §6.2
ROLE_REASONS = {ORIGIN: PART_REASONS["origin"], RELAY: PART_REASONS["relay"], SINK_DECODE: PART_REASONS["sink_decode"],
                DISPLAY: PART_REASONS["display"]}
ROLE_COVERAGE = {ORIGIN: "Partial", RELAY: "Partial", DISPLAY: "Whole"}
FILTER_BYTE = {0: "Unknown", 1: "Off", 2: "On"}  # energy-measurement §2.7
# Unicode White_Space, for a calibration reference "not empty once trimmed" (jwp-receipts §2.4.5).
WHITE_SPACE = frozenset([0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x20, 0x85, 0xA0, 0x1680, 0x2028, 0x2029, 0x202F, 0x205F,
                         0x3000] + list(range(0x2000, 0x200B)))

REASONS = [
    "record_malformed", "signature_invalid", "hwshunt_forbidden", "energy_class_invalid", "unmetered_as_zero",
    "table_rounds_invalid", "table_row_order", "table_row_status", "table_row_figure", "table_row_declaration",
    "table_readings_mismatch", "segment_method_class", "segment_rail_not_live", "segment_window_short",
    "segment_below_floor", "segment_figure_mismatch", "segment_declaration_below", "segment_table_mismatch",
    "segment_estimator_mismatch", "segment_chain_broken", "split_part_invalid", "split_display_mismatch",
    "split_hop_mismatch", "split_decode_mismatch", "session_streams_mismatch",
]


class Refused(Exception):
    """A refusal: the reason of §8, and where and why, for a reader of the output."""

    def __init__(self, reason, why=""):
        super().__init__(reason)
        self.reason, self.why = reason, why


def need(ok, reason, why):
    if not ok:
        raise Refused(reason, why)


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


# ---------------------------------------------------------------- JSON: reading and RFC 8785

class JObj(dict):
    """A JSON object in its wire order; `dup` is set when a member name repeats (§7 step 1)."""
    dup = False


class JNum:
    """A number no record admits: a fraction, an exponent, a sign or a constant
    (jwp-receipts §2.4.3's integers, which §2 adopts)."""

    def __init__(self, text):
        self.text = text

    def __repr__(self):
        return "JNum(%s)" % self.text

    def __deepcopy__(self, memo):
        return self


def _pairs(ps):
    o = JObj()
    for k, v in ps:
        if k in o:
            o.dup = True
        o[k] = v
    return o


def _int(text):
    return JNum(text) if text.startswith("-") else int(text)


def loads(text):
    return json.loads(text, object_pairs_hook=_pairs, parse_float=JNum, parse_int=_int, parse_constant=JNum)


_ESC = {'"': '\\"', "\\": "\\\\", "\b": "\\b", "\f": "\\f", "\n": "\\n", "\r": "\\r", "\t": "\\t"}


def _jstr(s):
    out = ['"']
    for ch in s:
        o = ord(ch)
        if ch in _ESC:
            out.append(_ESC[ch])
        elif o < 0x20:
            out.append("\\u%04x" % o)
        elif 0xD800 <= o <= 0xDFFF:
            raise ValueError("a lone surrogate is not I-JSON")
        else:
            out.append(ch)
    out.append('"')
    return "".join(out)


def jcs(v):
    """RFC 8785 over strings, integers from 0 to 2⁵³ − 1, null, arrays and objects: members
    sorted by their UTF-16 code units, no whitespace, integers in plain decimal, strings
    escaped as §3.2.2.2 writes them. Anything else raises ValueError."""
    out = []

    def enc(x):
        if x is None:
            out.append("null")
        elif type(x) is int:
            if not 0 <= x <= MAX:
                raise ValueError("an integer outside 0 to 2^53 - 1")
            out.append(str(x))
        elif type(x) is str:
            out.append(_jstr(x))
        elif type(x) is list:
            out.append("[")
            for i, y in enumerate(x):
                if i:
                    out.append(",")
                enc(y)
            out.append("]")
        elif isinstance(x, dict):
            if getattr(x, "dup", False):
                raise ValueError("a duplicate member")
            out.append("{")
            for i, k in enumerate(sorted(x, key=lambda k: k.encode("utf-16-be"))):
                if i:
                    out.append(",")
                out.append(_jstr(k))
                out.append(":")
                enc(x[k])
            out.append("}")
        else:
            raise ValueError("a value JCS is not used for here: %r" % (x,))
    enc(v)
    return "".join(out).encode("utf-8")


# ---------------------------------------------------------------- §7 step 1: shape

INT, STR, HEX32, HEX64, DECODER, DATE, FABRIC_LIST = "int", "str", "hex32", "hex64", "decoder", "date", "fabrics"
FABRIC_TERMS = ("cpu", "gpu", "media", "neural")  # energy-measurement §9.1


def nul(s):
    return ("null", s)


def enum(*v):
    return ("enum", frozenset(v))


def const(v):
    return ("const", v)


def arr(s):
    return ("arr", s)


def pair(s):
    return ("pair", s)


def obj(members):
    return ("obj", members)


_DEC_BUILTIN = re.compile(r"builtin:[a-z0-9_-]{1,64}@[0-9][A-Za-z0-9.+-]{0,63}")
_DEC_PLATFORM = re.compile(r"platform:[a-z0-9-]{1,64}")
_DATE = re.compile(r"([0-9]{4})-([0-9]{2})-([0-9]{2})")


def decoder_ok(v):
    """§4.2: `builtin:<crate>@<version>` as jwp-receipts §2.4.12 writes one, or
    `platform:<interface>`."""
    return _DEC_BUILTIN.fullmatch(v) is not None or _DEC_PLATFORM.fullmatch(v) is not None


def date_ok(v):
    """`YYYY-MM-DD`, a day its month has in its year (proleptic Gregorian)."""
    m = _DATE.fullmatch(v)
    if not m:
        return False
    y, mo, d = (int(g) for g in m.groups())
    if not 1 <= mo <= 12:
        return False
    leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
    days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][mo - 1]
    return 1 <= d <= days


def is_hex(v, n):
    return type(v) is str and len(v) == 2 * n and all(c in "0123456789abcdef" for c in v)


def utf8_ok(s):
    try:
        s.encode("utf-8")
        return True
    except UnicodeEncodeError:
        return False


def fits(v, s):
    if s == INT:
        return type(v) is int and 0 <= v <= MAX
    if s == STR:
        return type(v) is str and utf8_ok(v)
    if s == HEX32:
        return is_hex(v, 32)
    if s == HEX64:
        return is_hex(v, 64)
    if s == DECODER:
        return type(v) is str and decoder_ok(v)
    if s == DATE:
        return type(v) is str and date_ok(v)
    if s == FABRIC_LIST:
        # §7 step 1: every member one of energy-measurement §9.1's fabrics, none twice.
        return type(v) is list and all(type(x) is str and x in FABRIC_TERMS for x in v) and len(set(v)) == len(v)
    tag = s[0]
    if tag == "null":
        return v is None or fits(v, s[1])
    if tag == "enum":
        return type(v) is str and v in s[1]
    if tag == "const":
        return type(v) is type(s[1]) and v == s[1]
    if tag == "arr":
        return type(v) is list and all(fits(x, s[1]) for x in v)
    if tag == "pair":
        return type(v) is list and len(v) == 2 and all(fits(x, s[1]) for x in v)
    if tag == "obj":
        return (isinstance(v, dict) and not getattr(v, "dup", False) and set(v) == set(s[1])
                and all(fits(v[k], t) for k, t in s[1].items()))
    raise AssertionError(s)


SUMMARY = obj({"n": INT, "mean_uw": INT, "sd_uw": INT, "live_intervals": INT, "longest_live_run": INT})
CLASS_MEMBERS = {
    "joules_micro": INT,
    "energy_provenance": nul(STR),
    "energy_uncertainty": nul(obj({"relative_ppm": INT, "window_us": INT})),
    "energy_counter_filtering": nul(STR),
    "energy_calibration_ref": nul(STR),
}
COVERAGE = nul(enum("Whole", "Partial"))
UNIT = enum("image", "frame", "content-ms", "byte")
PATH = enum("software", "hardware")
ROW = obj(dict({
    "capability": STR, "path": PATH, "decoder": DECODER, "profile": nul(enum(*MINIMUM_US)), "fabrics": FABRIC_LIST,
    "coverage": COVERAGE, "unit": UNIT, "units": INT, "width": nul(INT), "height": nul(INT), "workers": INT,
    "status": enum("figure", "relative", "unmetered", "absent"), "reason": nul(enum(*ROW_REASONS)),
    "rounds": arr(obj({"idle": SUMMARY, "busy": SUMMARY, "window_us": INT, "completions": INT})),
    "median": nul(INT), "range": nul(pair(INT)),
}, **CLASS_MEMBERS))
SCHEMA = {
    TABLE: obj({
        "wai_sink_table": const(1), "signer": HEX32, "procedure": const("wai-sink-table/1"),
        "procedure_sha256": HEX32, "readings_sha256": HEX32,
        "machine": obj({"os": STR, "arch": STR, "processor": STR, "cores": INT, "memory_gib": INT}),
        "build": obj({"profile": enum("release", "debug"), "toolchain": nul(STR), "target": nul(STR)}),
        "measured": DATE, "rounds": INT, "sample_us": INT,
        "counters": arr(obj({"profile": enum(*MINIMUM_US), "status": enum(*COUNTER_STATUS), "detail": STR})),
        "rows": arr(ROW), "sig": HEX64,
    }),
    SEGMENT: obj(dict({
        "wai_sink_segment": const(1), "signer": HEX32, "session": HEX32, "stream": HEX32, "segment": INT,
        "prev": nul(HEX32), "object": HEX32, "capability": STR, "path": PATH, "decoder": DECODER, "unit": UNIT,
        "units": INT, "width": nul(INT), "height": nul(INT), "media_us": INT,
        "method": enum("measured", "table", "estimator", "unmetered"), "profile": nul(enum(*PROFILES)),
        "fabrics": FABRIC_LIST, "coverage": COVERAGE,
        "bracket": nul(obj({"idle": SUMMARY, "busy": SUMMARY, "window_us": INT, "sample_us": INT})),
        "table": nul(obj({"id": HEX32, "row": INT, "per_completion_uj": INT, "units_per_completion": INT})),
        "estimator": nul(obj({"method": const(CPU_EST), "cpu_us": INT, "active_mw": INT, "basis": STR,
                              "relative_ppm": INT})),
        "reason": nul(enum(*UNMETERED_REASONS)), "sig": HEX64,
    }, **CLASS_MEMBERS)),
    SESSION: obj({
        "wai_sink_session": const(1), "signer": HEX32, "session": HEX32,
        "streams": arr(obj({"stream": HEX32, "segments": INT, "last": HEX32})),
        "media_us": INT, "presented_us": INT,
        "display": nul(obj({"method": const(DISPLAY_EST), "panel_class": enum("backlit", "emissive"),
                            "on_mode_mw": INT, "reference_nits": INT, "nits": INT, "basis": STR,
                            "relative_ppm": INT})),
        "hops": arr(obj(dict({
            "role": INT,
            "unmetered": nul(enum(*MARKER_REASONS)),
            "coverage": COVERAGE,
            "source": nul(obj({"kind": enum("live-metering-hop", "other"), "record": HEX32, "hop_joules": INT,
                               "hop_class": STR, "share_num": INT, "share_den": INT})),
        }, **CLASS_MEMBERS))),
        "sig": HEX64,
    }),
}

_REGISTRY = None


def registry():
    """§4.2: `capability` is a capability the registry lists (registry/capabilities.json)."""
    global _REGISTRY
    if _REGISTRY is None:
        with open(os.path.join(HERE, "..", "registry", "capabilities.json"), encoding="utf-8") as f:
            _REGISTRY = frozenset(c["capability"] for c in json.load(f)["capabilities"])
    return _REGISTRY


def unit_ok(x):
    """§7 step 1: `width` and `height` at least 1 for `image` and `frame`, `null` otherwise;
    `units` at least 1."""
    if x["unit"] in ("image", "frame"):
        wh = type(x["width"]) is int and x["width"] >= 1 and type(x["height"]) is int and x["height"] >= 1
    else:
        wh = x["width"] is None and x["height"] is None
    return wh and x["units"] >= 1


def shape(rec, kind):
    """§7 step 1 (`record_malformed`)."""
    first = KINDS[kind][0]
    need(isinstance(rec, dict), "record_malformed", "the record is not a JSON object")
    need(not getattr(rec, "dup", False), "record_malformed", "a member name repeats")
    need(first in rec, "record_malformed", "the record has no `%s`, its first member (§3)" % first)
    need(fits(rec, SCHEMA[kind]), "record_malformed", "a member is missing, extra, or not of its type (§7 step 1)")
    if kind == TABLE:
        for i, row in enumerate(rec["rows"]):
            need(unit_ok(row), "record_malformed", "row %d: width, height or units (§7 step 1)" % i)
            need(row["capability"] in registry(), "record_malformed",
                 "row %d: `%s` is not a capability the registry lists (§4.2)" % (i, row["capability"]))
        keys = [(c["profile"].encode("utf-8"), c["detail"].encode("utf-8")) for c in rec["counters"]]
        need(all(a <= b for a, b in zip(keys, keys[1:])), "record_malformed",
             "`counters` is not in ascending order of (profile, detail) (§4.1)")
    elif kind == SEGMENT:
        need(unit_ok(rec), "record_malformed", "width, height or units (§7 step 1)")
        need(rec["capability"] in registry(), "record_malformed",
             "`%s` is not a capability the registry lists (§4.2, §5.1)" % rec["capability"])
        need(rec["media_us"] >= 1, "record_malformed", "`media_us` is below 1 (§5.1)")
        need((rec["prev"] is None) == (rec["segment"] == 0), "record_malformed",
             "`prev` is not null exactly when `segment` is 0 (§5.1, §7 step 1)")
    else:
        keys = [s["stream"] for s in rec["streams"]]
        need(len(keys) >= 1 and all(a < b for a, b in zip(keys, keys[1:])), "record_malformed",
             "`streams` is empty or not in strictly ascending order of `stream` (§6.1)")
        roles = [h["role"] for h in rec["hops"]]
        need(all(1 <= r <= 4 for r in roles) and all(a < b for a, b in zip(roles, roles[1:])), "record_malformed",
             "a hop's `role` is not 1 to 4, or the entries are not in strictly ascending order of `role` (§6.1, §7 step 1)")


# ---------------------------------------------------------------- §3: signature and identity

def body_of(rec):
    return {k: v for k, v in rec.items() if k != "sig"}


_SIG_CACHE = {}


def sig_ok(pk, msg, sig):
    key = (pk, sha(msg), sig)
    if key not in _SIG_CACHE:
        _SIG_CACHE[key] = ED.verify(pk, msg, sig) is None
    return _SIG_CACHE[key]


def check_signature(rec, kind):
    """§7 step 2: `sig` over `domain ‖ JCS(body)` under `signer`."""
    msg = KINDS[kind][1] + jcs(body_of(rec))
    need(sig_ok(bytes.fromhex(rec["signer"]), msg, bytes.fromhex(rec["sig"])), "signature_invalid",
         "the signature does not verify under `signer` (§3, SPEC §7)")


def identity(rec, kind):
    """§3: SHA-256(id-domain ‖ JCS(body) ‖ signature bytes)."""
    return sha(KINDS[kind][2], jcs(body_of(rec)), bytes.fromhex(rec["sig"]))


def try_identity(rec, kind):
    try:
        if isinstance(rec, dict) and is_hex(rec.get("sig"), 64):
            return identity(rec, kind).hex()
    except (ValueError, TypeError, AttributeError):
        pass
    return None


# ---------------------------------------------------------------- §7 step 3: classes

def _trimmed_empty(s):
    return all(ord(c) in WHITE_SPACE for c in s)


def check_class(x, where):
    """One class beside its figure, as §2 writes it (jwp-receipts §2.4.5)."""
    prov, unc = x["energy_provenance"], x["energy_uncertainty"]
    filt, ref, j = x["energy_counter_filtering"], x["energy_calibration_ref"], x["joules_micro"]
    need(prov != "HwShunt", "hwshunt_forbidden", "%s: the class is HwShunt (energy-measurement §2.6)" % where)
    need((prov is None) == (j == 0), "unmetered_as_zero",
         "%s: %s beside a figure of %d (energy-measurement §2)" % (where, prov or "no class", j))
    if prov is None:
        need(unc is None and filt is None and ref is None, "energy_class_invalid",
             "%s: evidence without a class (energy-measurement §2.7)" % where)
    else:
        need(prov in CLASSES, "energy_class_invalid", "%s: `%s` is not a class of energy-measurement §2" % (where, prov))
        if prov == "OnChipCounter":
            ok = unc is not None and filt in FILTERING and ref is None
        elif prov == "CalibratedInstrument":
            ok = unc is not None and filt is None and type(ref) is str and not _trimmed_empty(ref)
        else:
            ok = unc is None and filt is None and ref is None
        need(ok, "energy_class_invalid", "%s: the evidence is not exactly %s's (energy-measurement §2.7)" % (where, prov))
        if unc is not None:
            need(unc["relative_ppm"] >= 1 and unc["window_us"] >= 1, "energy_class_invalid",
                 "%s: an `energy_uncertainty` member of 0 (energy-measurement §2.3)" % where)
            need(unc["relative_ppm"] <= U32, "energy_class_invalid",
                 "%s: `relative_ppm` wider than its u32 encoding (energy-measurement §2.7, jwp-receipts §2.4.5)" % where)
            need(j * unc["relative_ppm"] >= FLOOR, "energy_class_invalid",
                 "%s: %d x %d is below the resolution floor 288 676 (energy-measurement §2.3)"
                 % (where, j, unc["relative_ppm"]))


def check_hop_class(label, where):
    need(label != "HwShunt", "hwshunt_forbidden", "%s: the hop's class is HwShunt (§6.4)" % where)
    need(label in CLASSES, "energy_class_invalid", "%s: `hop_class` %r is not a label of a class (§7 step 3)" % (where, label))


# ---------------------------------------------------------------- the integer forms

def least_sq(k, p):
    """The least integer x ≥ 0 with x² · k ≥ p, or None when there is none (k = 0 < p)."""
    if p <= 0:
        return 0
    if k <= 0:
        return None
    x = math.isqrt(p // k)
    while x * x * k < p:
        x += 1
    return x


def ceil_div(a, b):
    return -(-a // b)


def figure_of(m, t, wn, wd):
    """energy-measurement §6.2: F = ⌊(2·M·T·Wd + 10⁶·Wn) / (2·10⁶·Wn)⌋."""
    return (2 * m * t * wd + 10 ** 6 * wn) // (2 * 10 ** 6 * wn)


def least_p(ni, si, nb, sb, m, t, wn, wd, n, f):
    """energy-measurement §6.2: the least p with p² · K ≥ P."""
    k = 12 * ni * nb * wn ** 4 * f * f
    p = (12 * wn * wn * t * t * wd * wd * (si * si * nb + sb * sb * ni)
         + 4 * ni * nb * m * m * t * t * wd * wd * n * n
         + 10 ** 12 * ni * nb * wn ** 4)
    return least_sq(k, p)


def live(s, span):
    """§4.5 rule 2 (energy-measurement §3.1, §6.2): `sd_uw` above 0, at least 5 live
    intervals, a run of at least 2, `5 · live ≥ 4 · n`, a span of at least 300 ms."""
    return (s["sd_uw"] > 0 and s["live_intervals"] >= 5 and s["longest_live_run"] >= 2
            and 5 * s["live_intervals"] >= 4 * s["n"] and span >= LIVE_SPAN_US)


def summary_of(intervals):
    """§4.3: a bracket's summary from its intervals [d nJ, t µs]."""
    n = len(intervals)
    ps = [(2000 * d + t) // (2 * t) for d, t in intervals]
    s1, s2 = sum(ps), sum(p * p for p in ps)
    best = run = 0
    for d, _ in intervals:
        run = run + 1 if d > 0 else 0
        best = max(best, run)
    return {"n": n, "mean_uw": (2 * s1 + n) // (2 * n) if n else 0,
            "sd_uw": least_sq(n * (n - 1), n * s2 - s1 * s1) if n >= 2 else 0,
            "live_intervals": sum(1 for d, _ in intervals if d > 0), "longest_live_run": best}


def estimate_cpu(cpu_us, active_mw):
    """energy-measurement §9.4."""
    return (2 * cpu_us * active_mw + 1000) // 2000


def estimate_display(on_mode_mw, nits, presented_us, reference_nits):
    """energy-measurement §9.5."""
    return (2 * on_mode_mw * nits * presented_us + 1000 * reference_nits) // (2000 * reference_nits)


def share_of(hop_joules, num, den):
    """§6.4."""
    return (2 * hop_joules * num + den) // (2 * den)


def table_figure(per_completion, units, per):
    """§5.2 `table`."""
    return (2 * per_completion * units + per) // (2 * per)


# ---------------------------------------------------------------- §4.5: a round and a row

def round_figure(rd, workers, minimum, sample_us):
    """§4.5: the first rule a round breaks, or (F, p, M, s_idle). The bracket opened from
    rest: Wn = 2C + N, Wd = 2, N the row's workers."""
    i, b = rd["idle"], rd["busy"]
    t, c = rd["window_us"], rd["completions"]
    if i["n"] < 2 or b["n"] < 2:
        return "too_few_samples"
    if not (live(i, i["n"] * sample_us) and live(b, t)):
        return "rail_not_live"
    if t < minimum:
        return "window_short"
    if c < 1:
        return "empty_burst"
    if b["mean_uw"] <= i["mean_uw"]:
        return "no_marginal"
    m = b["mean_uw"] - i["mean_uw"]
    if m < i["sd_uw"]:
        return "below_attribution_floor"
    wn = 2 * c + workers
    f = figure_of(m, t, wn, 2)
    if f < 1:
        return "figure_zero"
    if f > MAX:
        return "not_declarable"
    p = least_p(i["n"], i["sd_uw"], b["n"], b["sd_uw"], m, t, wn, 2, workers, f)
    if p is None or p > U32:
        return "not_declarable"
    return (f, p, m, i["sd_uw"])


def row_outcome(row, sample_us):
    """§4.5 *The row*: its status and reason, and for a figure its median, range, F, p, q
    and window."""
    minimum = MINIMUM_US[row["profile"]]
    figs = []
    for r, rd in enumerate(row["rounds"]):
        got = round_figure(rd, row["workers"], minimum, sample_us)
        if isinstance(got, str):
            return {"status": "unmetered", "reason": got, "round": r}
        figs.append((got[0], r, got))
    order = sorted(figs, key=lambda x: (x[0], x[1]))
    _, med, (f, p, m, si) = order[len(order) // 2]
    if m >= 10 * si:
        status, reason = "figure", None
    elif m >= 3 * si:
        status, reason = "relative", None
    else:
        return {"status": "unmetered", "reason": "noisy", "round": med}
    lo, hi = min(x[0] for x in figs), max(x[0] for x in figs)
    q = least_sq(12 * f * f, 10 ** 12 * (hi - lo) ** 2)
    if max(p, q) > U32:
        return {"status": "unmetered", "reason": "not_declarable", "round": med}
    return {"status": status, "reason": reason, "median": med, "range": [lo, hi], "F": f, "p": p, "q": q,
            "window": row["rounds"][med]["window_us"]}


def power_of_two(n):
    return n >= 1 and n & (n - 1) == 0


def marker(x):
    """The unmetered marker: `joules_micro` 0 and the four class members `null` (§2)."""
    return (x["joules_micro"] == 0 and x["energy_provenance"] is None and x["energy_uncertainty"] is None
            and x["energy_counter_filtering"] is None and x["energy_calibration_ref"] is None)


def row_key(row):
    """§4.1: (capability, path, decoder, profile), strings by their bytes, a null profile first."""
    return (row["capability"].encode("utf-8"), row["path"].encode("utf-8"), row["decoder"].encode("utf-8"),
            (0, b"") if row["profile"] is None else (1, row["profile"].encode("utf-8")))


# ---------------------------------------------------------------- §7 step 4: the decode table

def readings_bytes_of(readings):
    """§4.3: the readings file is the JCS form of its object, so one held inline as an
    object is digested as its JCS bytes."""
    try:
        return jcs(readings)
    except (ValueError, TypeError, AttributeError):
        return None


def _intervals_ok(v):
    return type(v) is list and all(
        type(x) is list and len(x) == 2 and type(x[0]) is int and type(x[1]) is int
        and 0 <= x[0] <= MAX and 1 <= x[1] <= MAX for x in v)


def check_readings(rec, readings, raw):
    """§4.3, §7 step 4's last bullet (`table_readings_mismatch`)."""
    why = "the readings file (§4.3)"
    need(raw is not None and sha(raw).hex() == rec["readings_sha256"], "table_readings_mismatch",
         "the readings file's SHA-256 is not `readings_sha256` (§4.1)")
    need(isinstance(readings, dict) and not getattr(readings, "dup", False)
         and set(readings) == {"wai_sink_readings", "rows"} and type(readings["wai_sink_readings"]) is int
         and readings["wai_sink_readings"] == 1 and type(readings["rows"]) is list,
         "table_readings_mismatch", why + " does not read")
    need(len(readings["rows"]) == len(rec["rows"]), "table_readings_mismatch",
         "the readings file has %d entries for %d rows" % (len(readings["rows"]), len(rec["rows"])))
    for i, (e, row) in enumerate(zip(readings["rows"], rec["rows"])):
        need(isinstance(e, dict) and not getattr(e, "dup", False) and set(e) == {"capability", "path", "decoder", "profile", "rounds"}
             and type(e["rounds"]) is list, "table_readings_mismatch", "entry %d does not read" % i)
        need(e["capability"] == row["capability"] and e["path"] == row["path"] and e["decoder"] == row["decoder"]
             and e["profile"] == row["profile"]
             and (e["profile"] is None) == (row["profile"] is None),
             "table_readings_mismatch", "entry %d does not name row %d" % (i, i))
        need(len(e["rounds"]) == len(row["rounds"]), "table_readings_mismatch",
             "entry %d has %d rounds, row %d %d" % (i, len(e["rounds"]), i, len(row["rounds"])))
        for r, (er, rr) in enumerate(zip(e["rounds"], row["rounds"])):
            need(isinstance(er, dict) and not getattr(er, "dup", False) and set(er) == {"idle", "busy", "completions"}
                 and _intervals_ok(er["idle"]) and _intervals_ok(er["busy"]) and type(er["completions"]) is int,
                 "table_readings_mismatch", "row %d round %d's readings do not read" % (i, r))
            got = {"idle": summary_of(er["idle"]), "busy": summary_of(er["busy"]),
                   "window_us": sum(t for _, t in er["busy"]), "completions": er["completions"]}
            for k in ("idle", "busy", "window_us", "completions"):
                need(got[k] == rr[k], "table_readings_mismatch",
                     "row %d round %d: %s is %r from the readings, the table states %r (§4.3)" % (i, r, k, got[k], rr[k]))


def verify_table(rec, readings=None, raw=None, held_readings=False):
    """§7 steps 1–4 for a decode table; its identity. `held_readings` says the verifier holds
    the readings file (`readings`, digested as `raw`)."""
    shape(rec, TABLE)
    check_signature(rec, TABLE)
    for i, row in enumerate(rec["rows"]):
        check_class(row, "row %d" % i)
    n = rec["rounds"]
    need(n % 2 == 1 and n >= 7, "table_rounds_invalid", "`rounds` is %d: odd and at least 7 (§4.1)" % n)
    for i, row in enumerate(rec["rows"]):
        need(row["status"] == "absent" or len(row["rounds"]) == n, "table_rounds_invalid",
             "row %d has %d rounds, not %d" % (i, len(row["rounds"]), n))
    keys = [row_key(r) for r in rec["rows"]]
    for i in range(1, len(keys)):
        need(keys[i - 1] < keys[i], "table_row_order", "row %d is not above row %d (§4.1)" % (i, i - 1))
    live_profiles = {c["profile"] for c in rec["counters"] if c["status"] == "live"}
    for i, row in enumerate(rec["rows"]):
        where = "row %d (%s %s %s)" % (i, row["capability"], row["path"], row["profile"])
        # its status and reason
        out = None
        if row["status"] == "absent":
            need(row["reason"] in ABSENT_REASONS and row["rounds"] == [] and row["workers"] == 0
                 and (row["profile"] is None) == (row["reason"] == "no_decoder") and row["fabrics"] == []
                 and row["coverage"] is None and row["median"] is None and row["range"] is None,
                 "table_row_status", "%s: not an absent row (§4.5, §7 step 4)" % where)
        else:
            need(row["profile"] is not None, "table_row_status", "%s: no profile, and not absent for no_decoder" % where)
            need(power_of_two(row["workers"]), "table_row_status", "%s: workers %d not a power of two" % (where, row["workers"]))
            out = row_outcome(row, rec["sample_us"])
            need((row["status"], row["reason"]) == (out["status"], out["reason"]), "table_row_status",
                 "%s: states %s/%s, its rounds give %s/%s (round %s)"
                 % (where, row["status"], row["reason"], out["status"], out["reason"], out.get("round")))
        if row["status"] in ("absent", "unmetered"):
            need(marker(row), "table_row_status", "%s: an %s row without the unmetered marker (§7 step 4)" % (where, row["status"]))
        if row["profile"] is not None:
            is_live = row["profile"] in live_profiles
            need(is_live == (not (row["status"] == "absent" and row["reason"] == "no_counter")), "table_row_status",
                 "%s: its profile is %slive in `counters`" % (where, "" if is_live else "not "))
        # its figure
        if row["status"] != "absent":
            need(row["fabrics"] == FABRICS[row["profile"]], "table_row_figure",
                 "%s: fabrics %r, not its profile's %r (§4.2)" % (where, row["fabrics"], FABRICS[row["profile"]]))
            need(row["coverage"] == "Partial", "table_row_figure", "%s: coverage %r, not Partial (§4.5)" % (where, row["coverage"]))
            if row["status"] in ("figure", "relative"):
                need(row["median"] == out["median"], "table_row_figure",
                     "%s: median %r, its rounds give %d" % (where, row["median"], out["median"]))
                need(row["range"] == out["range"], "table_row_figure",
                     "%s: range %r, its rounds give %r" % (where, row["range"], out["range"]))
                need(row["joules_micro"] == out["F"], "table_row_figure",
                     "%s: joules_micro %d, the median's figure is %d" % (where, row["joules_micro"], out["F"]))
                need(row["energy_provenance"] == "CalibratedInstrument" or (row["energy_provenance"] == "OnChipCounter"
                     and row["energy_counter_filtering"] == "Unknown"), "table_row_figure",
                     "%s: class %s/%s, not OnChipCounter with filtering Unknown or CalibratedInstrument"
                     % (where, row["energy_provenance"], row["energy_counter_filtering"]))
                need(row["energy_uncertainty"]["window_us"] == out["window"], "table_row_figure",
                     "%s: window_us %d, the median round's is %d" % (where, row["energy_uncertainty"]["window_us"], out["window"]))
            else:
                need(row["median"] is None and row["range"] is None, "table_row_figure",
                     "%s: an unmetered row with a median or range" % where)
        # its declaration
        if row["status"] in ("figure", "relative"):
            least = max(out["p"], out["q"])
            need(row["energy_uncertainty"]["relative_ppm"] >= least, "table_row_declaration",
                 "%s: relative_ppm %d below max(p %d, q %d)" % (where, row["energy_uncertainty"]["relative_ppm"], out["p"], out["q"]))
    if held_readings:  # §7 step 4: once every row has passed
        check_readings(rec, readings, raw)
    return identity(rec, TABLE)


# ---------------------------------------------------------------- §7 step 5: the segment receipt

def held_tables(tables):
    out = {}
    for t in tables or []:
        h = try_identity(t, TABLE)
        if h is not None:
            out.setdefault(h, t)
    return out


def verify_segment(rec, tables):
    """§7 steps 1–5 for one segment receipt (not its chain); its identity. `tables` is the
    list of decode tables the verifier holds, found by identity."""
    shape(rec, SEGMENT)
    check_signature(rec, SEGMENT)
    check_class(rec, "the segment")
    m, prov, prof = rec["method"], rec["energy_provenance"], rec["profile"]
    has = tuple(rec[k] is not None for k in ("bracket", "table", "estimator"))
    if m == "measured":
        ok = has == (True, False, False) and rec["reason"] is None and prof in MINIMUM_US and prov in HARDWARE
    elif m == "table":
        ok = has == (False, True, False) and rec["reason"] is None and prof in MINIMUM_US and prov in ("ModelBased", "Estimator")
    elif m == "estimator":
        ok = has == (False, False, True) and rec["reason"] is None and prof == CPU_EST and prov == "Estimator"
    else:
        ok = (has == (False, False, False) and prof is None and rec["coverage"] is None and rec["fabrics"] == []
              and marker(rec) and rec["reason"] in UNMETERED_REASONS)
    if ok and m != "unmetered":
        ok = rec["fabrics"] == FABRICS[prof] and rec["coverage"] == "Partial" and rec["reason"] is None
    need(ok, "segment_method_class", "its members or class do not fit the method `%s` (§5.2, §7 step 5)" % m)
    if m == "measured":
        b = rec["bracket"]
        i, bu, t = b["idle"], b["busy"], b["window_us"]
        need(live(i, i["n"] * b["sample_us"]) and live(bu, t), "segment_rail_not_live",
             "a bracket is not live (§4.5 rule 2, §5.2)")
        need(t >= MINIMUM_US[prof], "segment_window_short", "window %d below %s's minimum" % (t, prof))
        mm = bu["mean_uw"] - i["mean_uw"]
        need(mm >= i["sd_uw"], "segment_below_floor", "M %d below the idle sd %d" % (mm, i["sd_uw"]))
        f = figure_of(mm, t, 1, 1)
        need(rec["joules_micro"] == f and f >= 1 and rec["energy_uncertainty"]["window_us"] == t,
             "segment_figure_mismatch", "joules_micro %d window_us %d; the bracket gives F %d over T %d"
             % (rec["joules_micro"], rec["energy_uncertainty"]["window_us"], f, t))
        p = least_p(i["n"], i["sd_uw"], bu["n"], bu["sd_uw"], mm, t, 1, 1, 0, f)
        need(p is not None and rec["energy_uncertainty"]["relative_ppm"] >= p, "segment_declaration_below",
             "relative_ppm %d below p %s" % (rec["energy_uncertainty"]["relative_ppm"], p))
    elif m == "table":
        tb = rec["table"]
        per = tb["units_per_completion"]
        need(per >= 1 and rec["joules_micro"] == table_figure(tb["per_completion_uj"], rec["units"], per)
             and rec["joules_micro"] >= 1, "segment_table_mismatch", "the figure is not the formula's (§5.2)")
        held = held_tables(tables).get(tb["id"])
        if held is not None:
            try:
                verify_table(held)
            except Refused as e:
                raise Refused("segment_table_mismatch", "the held table %s does not verify (%s)" % (tb["id"], e.reason))
            need(prov == ("ModelBased" if held["signer"] == rec["signer"] else "Estimator"), "segment_table_mismatch",
                 "class %s; the table's signer is %sthe receipt's (§5.2)" % (prov, "" if held["signer"] == rec["signer"] else "not "))
            rows = held["rows"]
            need(tb["row"] < len(rows), "segment_table_mismatch", "row %d is not in the table" % tb["row"])
            row = rows[tb["row"]]
            need(row["status"] == "figure", "segment_table_mismatch", "row %d's status is %s" % (tb["row"], row["status"]))
            for k in ("capability", "path", "decoder", "profile", "unit", "width", "height", "fabrics", "coverage"):
                need(row[k] == rec[k], "segment_table_mismatch", "`%s` is not row %d's" % (k, tb["row"]))
            need(tb["per_completion_uj"] == row["joules_micro"] and per == row["units"], "segment_table_mismatch",
                 "per_completion_uj or units_per_completion is not row %d's" % tb["row"])
    elif m == "estimator":
        e = rec["estimator"]
        est = estimate_cpu(e["cpu_us"], e["active_mw"])
        need(rec["joules_micro"] == est and est >= 1 and len(e["basis"]) > 0 and e["relative_ppm"] >= 1,
             "segment_estimator_mismatch", "joules_micro %d; E is %d (energy-measurement §9.4)" % (rec["joules_micro"], est))
    return identity(rec, SEGMENT)


def chain_ok(receipts, ids):
    """§5.3: one session, signer and stream throughout; segments 0 to k in order; each
    `prev` the identity of the receipt before it, and `null` for segment 0."""
    if not receipts:
        return "no receipt"
    r0 = receipts[0]
    for k, r in enumerate(receipts):
        if (r["session"], r["signer"], r["stream"]) != (r0["session"], r0["signer"], r0["stream"]):
            return "receipt %d names another session, signer or stream" % k
        if r["segment"] != k:
            return "receipt %d is segment %d" % (k, r["segment"])
        if r["prev"] != (None if k == 0 else ids[k - 1]):
            return "receipt %d's prev is not the identity of the receipt before it" % k
    return None


def verify_chain(receipts, tables):
    """A `chain` case: each receipt by steps 1–5, then §5.3; the last receipt's identity."""
    ids = [verify_segment(r, tables).hex() for r in receipts]
    bad = chain_ok(receipts, ids)
    need(bad is None, "segment_chain_broken", bad)
    return bytes.fromhex(ids[-1])


# ---------------------------------------------------------------- §7 step 6: the session receipt

def expected_decode(segments):
    """§6.3 *sink_decode* from the segment receipts."""
    metered = [s for s in segments if s["method"] != "unmetered"]
    if not metered:
        return {"status": "absent", "reason": "unmetered"}
    total = sum(s["joules_micro"] for s in metered)
    weakest = max((s["energy_provenance"] for s in metered), key=lambda c: RANK[c])
    whole = len(metered) == len(segments) and all(s["coverage"] == "Whole" for s in metered)
    out = {"status": "figure", "reason": None, "joules_micro": total, "coverage": "Whole" if whole else "Partial",
           "energy_counter_filtering": None, "energy_calibration_ref": None, "window_us": None, "bound": None}
    if weakest not in HARDWARE:
        out["energy_provenance"] = weakest
        return out
    refs = {s["energy_calibration_ref"] for s in metered}
    if all(s["energy_provenance"] == "CalibratedInstrument" for s in metered) and len(refs) == 1:
        out["energy_provenance"], out["energy_calibration_ref"] = "CalibratedInstrument", refs.pop()
    else:
        filt = [s["energy_counter_filtering"] for s in metered]
        out["energy_provenance"] = "OnChipCounter"
        every_off = all(s["energy_provenance"] == "OnChipCounter" and s["energy_counter_filtering"] == "Off" for s in metered)
        out["energy_counter_filtering"] = "On" if "On" in filt else "Off" if every_off else "Unknown"
    out["window_us"] = max(s["energy_uncertainty"]["window_us"] for s in metered)
    out["bound"] = ceil_div(sum(s["joules_micro"] * s["energy_uncertainty"]["relative_ppm"] for s in metered), total)
    return out


def check_hop(name, part, held_hops):
    """§6.4 for the `origin` or `relay` part (`split_hop_mismatch`): the sole-user rule, the
    figure, the class and evidence, and, holding the record, the record."""
    if part["energy_provenance"] is None:  # the marker: no figure to check
        return
    src = part["source"]
    hj, num, den = src["hop_joules"], src["share_num"], src["share_den"]
    need(hj >= 1 and 1 <= num <= den, "split_hop_mismatch", "%s: hop_joules or the share is out of range" % name)
    need(100 * num >= 99 * den, "split_hop_mismatch",
         "%s: %d of %d is below the sole user's 99%%; a shared hop is absent for hop_shared (§6.4)" % (name, num, den))
    j = share_of(hj, num, den)
    need(j >= 1 and part["joules_micro"] == j, "split_hop_mismatch",
         "%s: joules_micro %d, §6.4 gives %d" % (name, part["joules_micro"], j))
    need(part["energy_provenance"] == src["hop_class"], "split_hop_mismatch",
         "%s: class %s, the hop's is %s (§6.4)" % (name, part["energy_provenance"], src["hop_class"]))
    if src["kind"] != "live-metering-hop" or src["record"] not in held_hops:
        return
    b = held_hops[src["record"]]
    try:
        x = LMV.read_hop(b, b[5:37])
    except LMV.Refuse as e:
        raise Refused("split_hop_mismatch", "%s: the hop record does not read (%s, live-metering §3.6)" % (name, e.reason))
    need(x["digest"].hex() == src["record"], "split_hop_mismatch", "%s: the record's digest is not `record`" % name)
    need(x["joules"] == hj, "split_hop_mismatch", "%s: the record's joules %d, hop_joules %d" % (name, x["joules"], hj))
    label = LMV.LABELS.get(x["class"][0]) if x["class"] not in (None, LMV.INVALID) else None
    need(label == src["hop_class"], "split_hop_mismatch", "%s: the record's class is %s" % (name, label))
    c = x["class"]
    if c[0] == 4:
        ev = ({"relative_ppm": c[1], "window_us": c[2]}, FILTER_BYTE[c[3]], None)
    elif c[0] == 5:
        ev = ({"relative_ppm": c[1], "window_us": c[2]}, None, c[3])
    else:
        ev = (None, None, None)
    got = (part["energy_uncertainty"], part["energy_counter_filtering"], part["energy_calibration_ref"])
    need(got == ev, "split_hop_mismatch", "%s: the part's evidence is not the record's" % name)
    need(sum(x["sub"]) == den, "split_hop_mismatch",
         "%s: share_den %d, the record's subscriber-microseconds sum to %d" % (name, den, sum(x["sub"])))


def verify_session(rec, segments, tables, held_hops):
    """§7 steps 1–3 and 6 for a session receipt; its identity."""
    shape(rec, SESSION)
    check_signature(rec, SESSION)
    for h in rec["hops"]:
        where = "the %s entry" % ROLE_NAME[h["role"]]
        check_class(h, where)
        if h["source"] is not None:
            check_hop_class(h["source"]["hop_class"], where + "'s source")
    by_role = {h["role"]: h for h in rec["hops"]}
    need(SINK_DECODE in by_role and ORIGIN in by_role, "split_part_invalid",
         "the sink decode and the origin always have an entry (§6.2)")
    need((DISPLAY in by_role) == (rec["presented_us"] > 0), "split_part_invalid",
         "the display has an entry exactly when presented_us is above 0 (§6.2); presented_us %d" % rec["presented_us"])
    for h in rec["hops"]:
        role = h["role"]
        if h["energy_provenance"] is not None:  # a figure (step 3: a class beside a figure of at least 1)
            cov = h["coverage"] in ("Whole", "Partial") if role == SINK_DECODE else h["coverage"] == ROLE_COVERAGE[role]
            ok = h["unmetered"] is None and h["joules_micro"] >= 1 and cov and (h["source"] is not None) == (role in (ORIGIN, RELAY))
        else:
            ok = marker(h) and h["coverage"] is None and h["source"] is None and h["unmetered"] in ROLE_REASONS[role]
        need(ok, "split_part_invalid", "the %s entry is neither a figure nor a marker with its reason (§6.2)" % ROLE_NAME[role])
    d = rec["display"]
    if d is not None:
        need(d["on_mode_mw"] >= 1 and d["reference_nits"] >= 1 and d["relative_ppm"] >= 1 and len(d["basis"]) > 0,
             "split_display_mismatch", "the display's declaration is out of range (§6.3)")
    dp = by_role.get(DISPLAY)
    if dp is not None:
        if d is None:
            need(dp["energy_provenance"] is None and dp["unmetered"] == "display_undeclared", "split_display_mismatch",
                 "no display declared, and the entry is not the marker display_undeclared (§6.3)")
        else:
            e = estimate_display(d["on_mode_mw"], d["nits"], rec["presented_us"], d["reference_nits"])
            if e >= 1:
                need(dp["energy_provenance"] == "Estimator" and dp["joules_micro"] == e, "split_display_mismatch",
                     "the display entry is not the Estimator figure %d (energy-measurement §9.5)" % e)
            else:
                need(dp["energy_provenance"] is None and dp["unmetered"] == "unmetered", "split_display_mismatch",
                     "E is 0, and the display entry is not the marker unmetered (§6.3)")
    for role in (ORIGIN, RELAY):
        if role in by_role:
            check_hop(ROLE_NAME[role], by_role[role], held_hops)
    if segments:
        ids = []
        for k, s in enumerate(segments):
            try:
                ids.append(verify_segment(s, tables).hex())
            except Refused as e:
                raise Refused("session_streams_mismatch", "segment receipt %d is refused %s (%s)" % (k, e.reason, e.why))
            need(s["session"] == rec["session"] and s["signer"] == rec["signer"], "session_streams_mismatch",
                 "segment receipt %d names another session or signer" % k)
        by_stream = {}
        for s, h in zip(segments, ids):
            by_stream.setdefault(s["stream"], []).append((s, h))
        listed = {st["stream"]: st for st in rec["streams"]}
        need(set(by_stream) == set(listed), "session_streams_mismatch", "the streams are not the receipts' streams")
        for stream, items in by_stream.items():
            bad = chain_ok([s for s, _ in items], [h for _, h in items])
            need(bad is None, "session_streams_mismatch", "stream %s: %s" % (stream, bad))
            need(listed[stream]["segments"] == len(items) and listed[stream]["last"] == items[-1][1],
                 "session_streams_mismatch", "stream %s: its count or last identity" % stream)
        total = sum(s["media_us"] for s in segments)
        need(rec["media_us"] == total, "session_streams_mismatch", "media_us %d, the receipts sum to %d" % (rec["media_us"], total))
        want, got = expected_decode(segments), by_role[SINK_DECODE]
        if want["status"] == "absent":
            ok = got["energy_provenance"] is None and got["unmetered"] == "unmetered"
        else:
            ok = (got["energy_provenance"] is not None and got["joules_micro"] == want["joules_micro"]
                  and got["energy_provenance"] == want["energy_provenance"]
                  and got["energy_counter_filtering"] == want["energy_counter_filtering"]
                  and got["energy_calibration_ref"] == want["energy_calibration_ref"]
                  and got["coverage"] == want["coverage"])
            if ok and want["window_us"] is not None:
                ok = (got["energy_uncertainty"]["window_us"] == want["window_us"]
                      and got["energy_uncertainty"]["relative_ppm"] >= want["bound"])
        need(ok, "split_decode_mismatch", "the sink-decode entry is not what §6.3 gives: %r" % (want,))
    return identity(rec, SESSION)


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

def held_hops_of(case, doc):
    out = {}
    for d in case.get("hops") or []:
        if d in (doc.get("hops") or {}):
            try:
                out[d] = bytes.fromhex(doc["hops"][d])
            except (ValueError, TypeError):
                pass
    return out


def run_case(case, doc):
    """The outcome a case's inputs give: (outcome, identity hex or None, why)."""
    subj, tables = case["subject"], case.get("tables") or []
    try:
        if subj == "table":
            held = case.get("readings") is not None
            ident = verify_table(case["record"], case.get("readings"),
                                 readings_bytes_of(case["readings"]) if held else None, held)
        elif subj == "segment":
            ident = verify_segment(case["record"], tables)
        elif subj == "chain":
            ident = verify_chain(case["segments"], tables)
        elif subj == "session":
            ident = verify_session(case["record"], case.get("segments") or [], tables, held_hops_of(case, doc))
        else:
            return ("<subject %r>" % subj, None, "")
        return ("accept", ident.hex(), "")
    except Refused as e:
        return (e.reason, None, e.why)
    except (KeyError, TypeError, ValueError, AttributeError, IndexError) as e:
        return ("<verifier error %s: %s>" % (type(e).__name__, e), None, "")


def check(doc, quiet=False, notes=None):
    """Every case against its stated outcome; the reasons refused. (fails, outcomes). What
    may be deliberate in a vector (a hop record carried under a digest that is not its own,
    which §6.4's digest rule refuses) goes to `notes`, not to the failures."""
    fails, outcomes, seen = [], {}, set()
    notes = [] if notes is None else notes
    for k in ("wai_sink_energy_corpus", "keys", "hops", "cases"):
        if k not in doc:
            fails.append("the corpus has no `%s`" % k)
    names = [c.get("name") for c in doc.get("cases", [])]
    if len(set(names)) != len(names):
        fails.append("two cases share a name")
    for d in (doc.get("hops") or {}):
        try:
            b = bytes.fromhex(doc["hops"][d])
        except (ValueError, TypeError):
            fails.append("the hop record under %s is not hexadecimal" % d[:16])
            continue
        if LMV.given_digest(b).hex() != d:
            notes.append("the hop record under %s has the digest %s" % (d[:16], LMV.given_digest(b).hex()[:16]))
    for case in doc.get("cases", []):
        name = case.get("name")
        for d in case.get("hops") or []:
            if d not in (doc.get("hops") or {}):
                notes.append("%s: holds hop %s, which the corpus does not carry (read as not held)" % (name, d[:16]))
        got, ident, why = run_case(case, doc)
        outcomes[name] = (got, ident, why)
        if got != case.get("expect"):
            fails.append("%s: expected %s, got %s%s" % (name, case.get("expect"), got, (" — " + why) if why else ""))
        elif got == "accept" and ident != case.get("identity"):
            fails.append("%s: accepted, identity %s, the corpus states %s" % (name, ident, case.get("identity")))
        elif got != "accept" and case.get("identity") is not None:
            fails.append("%s: refused, and the corpus states an identity" % name)
        if got in REASONS:
            seen.add(got)
    for r in REASONS:
        if r not in seen:
            fails.append("nothing in the corpus is refused %s" % r)
    if not quiet:
        for f in fails:
            print("FAIL " + f)
    return fails, outcomes


def section(name):
    path = os.path.join(HERE, "..", "extensions", "sink-energy.md")
    with open(path, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## " + name)
    end = text.find("\n## ", at + 1)
    return text[at:end if end >= 0 else len(text)]


def reasons_listed():
    """§8's table, in order."""
    return [line.split("`")[1] for line in section("8. Reasons").splitlines() if line.startswith("| `")]


def prose_reasons():
    """§8's prose: the row reasons and the unmetered reasons."""
    body = " ".join(section("8. Reasons").split())
    a = body.index("The **row reasons**")
    b = body.index("The **unmetered reasons**")
    return re.findall(r"`([a-z_]+)`", body[a:b]), re.findall(r"`([a-z_]+)`", body[b:])


# ---------------------------------------------------------------- the measured tables

def load_table_dir(d):
    """One measured table's three files, as bytes (None where a file is missing)."""
    files = {}
    for n in ("table.json", "readings.json", "procedure.txt"):
        p = os.path.join(d, n)
        if os.path.isfile(p):
            with open(p, "rb") as f:
                files[n] = f.read()
        else:
            files[n] = None
    return files


def load_tables(corpus_dir):
    out = []
    root = os.path.join(corpus_dir, "tables")
    if not os.path.isdir(root):
        return out
    for sink in sorted(os.listdir(root)):
        d = os.path.join(root, sink)
        if os.path.isdir(d):
            out.append((sink, load_table_dir(d)))
    return out


# §9 has the corpus hold the reference sinks' measured tables. While they are being taken, a
# corpus without them is a NOTE, not a failure; set this to True once they are committed.
REQUIRE_TABLES = True


def check_tables(tables, quiet=False, prefix="tables/"):
    fails, lines = [], []
    if not tables:
        msg = "no measured table under tables/ (§9: the corpus holds the reference sinks' tables)"
        if REQUIRE_TABLES:
            fails.append(msg)
        else:
            lines.append("NOTE " + msg)
    for sink, files in tables:
        if any(files[n] is None for n in files):
            fails.append("%s: missing %s" % (prefix + sink, ", ".join(n for n in files if files[n] is None)))
            continue
        try:
            rec = loads(files["table.json"].decode("utf-8"))
            readings = loads(files["readings.json"].decode("utf-8"))
        except (ValueError, UnicodeDecodeError) as e:
            fails.append("%s: does not read as JSON (%s)" % (prefix + sink, e))
            continue
        try:
            ident = verify_table(rec, readings, files["readings.json"], True)
        except Refused as e:
            fails.append("%s: refused %s — %s" % (prefix + sink, e.reason, e.why))
            continue
        except (KeyError, TypeError, ValueError, AttributeError, IndexError) as e:
            fails.append("%s: verifier error %s: %s" % (prefix + sink, type(e).__name__, e))
            continue
        try:
            if not_jcs(readings, files["readings.json"]):
                lines.append("NOTE %s: readings.json is not the JCS form of its object (§4.3)" % (prefix + sink))
        except ValueError:
            pass
        if sha(files["procedure.txt"]).hex() != rec["procedure_sha256"]:
            fails.append("%s: procedure.txt's SHA-256 is not `procedure_sha256`" % (prefix + sink))
            continue
        m = rec["machine"]
        lines.append("%s: %s %s, %s, %d cores, %d GiB; measured %s; %d rounds at %d us; identity %s"
                     % (prefix + sink, m["os"], m["arch"], m["processor"], m["cores"], m["memory_gib"], rec["measured"],
                        rec["rounds"], rec["sample_us"], ident.hex()))
        for row in rec["rows"]:
            lines.append("    %-28s %-8s %-11s %-10s %-12s %s" % (
                row["capability"], row["path"], row["profile"], row["status"] + ("/" + row["reason"] if row["reason"] else ""),
                row["joules_micro"] if row["status"] in ("figure", "relative") else "-",
                row["range"] if row["range"] is not None else "-"))
    if not quiet:
        for f in fails:
            print("FAIL " + f)
    return fails, lines


def not_jcs(obj_, raw):
    return jcs(obj_) != raw


# ---------------------------------------------------------------- RFC 8032 signing, for mutations

def _mul(s, p):
    return ED._double_mul(s, p, 0, ED.IDENTITY)


def keypair(seed):
    """RFC 8032 §5.1.5."""
    h = hashlib.sha512(seed).digest()
    a = int.from_bytes(h[:32], "little")
    a &= (1 << 254) - 8
    a |= 1 << 254
    return (a, h[32:], ED.encode(_mul(a, ED.B)))


def sign(kp, msg):
    """RFC 8032 §5.1.6."""
    a, prefix, pk = kp
    r = int.from_bytes(hashlib.sha512(prefix + msg).digest(), "little") % ED.L
    rr = ED.encode(_mul(r, ED.B))
    k = int.from_bytes(hashlib.sha512(rr + pk + msg).digest(), "little") % ED.L
    return rr + ((r + k * a) % ED.L).to_bytes(32, "little")


MUT_KEY = keypair(sha(b"sink_energy_verify: the mutation self-test's own key"))


def resign(rec, kind):
    rec["signer"] = MUT_KEY[2].hex()
    rec["sig"] = sign(MUT_KEY, KINDS[kind][1] + jcs(body_of(rec))).hex()


# ---------------------------------------------------------------- mutations

def rekey(case):
    """A copy of an accepted case whose records signed by the case's sink are re-signed by
    MUT_KEY, with every link (table ids, `prev`, `last`) carried across, so that a defect
    put in afterwards is met by the rule it breaks rather than by the signature."""
    c = copy.deepcopy(case)
    if c["subject"] == "table":
        resign(c["record"], TABLE)
        return c
    main = c["record"] if c["record"] is not None else c["segments"][0]
    old_signer = main["signer"]
    ids = {}
    for t in c.get("tables") or []:
        if t["signer"] == old_signer:
            old = identity(t, TABLE).hex()
            resign(t, TABLE)
            ids[old] = identity(t, TABLE).hex()
    segs = list(c.get("segments") or []) + ([c["record"]] if c["subject"] == "segment" else [])
    for s in segs:
        if s["signer"] != old_signer:
            continue
        old = identity(s, SEGMENT).hex()
        if s["table"] is not None and s["table"]["id"] in ids:
            s["table"]["id"] = ids[s["table"]["id"]]
        if s["prev"] in ids:
            s["prev"] = ids[s["prev"]]
        resign(s, SEGMENT)
        ids[old] = identity(s, SEGMENT).hex()
    if c["subject"] == "session":
        for st in c["record"]["streams"]:
            st["last"] = ids.get(st["last"], st["last"])
        resign(c["record"], SESSION)
    return c


def flip_hex(h, at, bit=1):
    b = bytearray(bytes.fromhex(h))
    b[at] ^= bit
    return b.hex()


def mutations(doc, tables):
    """(name, doc, tables, case name or None, the outcome the text gives the defect or None)."""
    out, notes = [], []
    cases = doc.get("cases", [])

    def accepted(c):
        return c.get("expect") == "accept"

    def find(pred, rekeyed=False):
        """The first accepted case meeting `pred` (and, if `rekeyed`, still accepted once
        re-signed: the positive control)."""
        for c in cases:
            try:
                if not accepted(c) or not pred(c):
                    continue
            except (KeyError, TypeError, IndexError, AttributeError):
                continue
            if not rekeyed:
                return c
            r = rekey(c)
            got = run_case(r, doc)
            if got[0] == "accept":
                return r
            notes.append("the re-signed copy of %s gives %s, not accept (%s)" % (c["name"], got[0], got[2]))
        return None

    def put(name, case, f, want, rekeyed_case=None):
        if case is None:
            notes.append("no case to mutate for '%s'" % name)
            return
        d = copy.deepcopy(doc)
        i = next(k for k, c in enumerate(d["cases"]) if c["name"] == case["name"])
        if rekeyed_case is not None:
            d["cases"][i] = copy.deepcopy(rekeyed_case)
        w = f(d["cases"][i], d)
        out.append((name, d, tables, case["name"], w if isinstance(w, str) else want))

    def figure_row(c):
        rows = c["record"]["rows"]
        return next((k for k, r in enumerate(rows) if r["status"] == "figure"), None)

    is_table = lambda c: c["subject"] == "table"  # noqa: E731

    # 1. a table's signature
    c = find(is_table)
    put("a table's signature byte", c, lambda x, d: x["record"].update(sig=flip_hex(x["record"]["sig"], 0)), "signature_invalid")

    # 2. a row's figure, changed by one
    r = find(lambda c: is_table(c) and figure_row(c) is not None, True)

    def figure_plus(x, d):
        row = x["record"]["rows"][figure_row(x)]
        row["joules_micro"] += 1
        resign(x["record"], TABLE)
    put("a row's figure, by one", r, figure_plus, "table_row_figure", r)

    # 3. a round dropped
    def drop_round(x, d):
        x["record"]["rows"][figure_row(x)]["rounds"].pop()
        resign(x["record"], TABLE)
    put("a round dropped", r, drop_round, "table_rounds_invalid", r)

    # 4. two rows swapped
    r4 = find(lambda c: is_table(c) and len(c["record"]["rows"]) >= 2, True)

    def swap_rows(x, d):
        rows = x["record"]["rows"]
        rows[0], rows[1] = rows[1], rows[0]
        resign(x["record"], TABLE)
    put("two rows swapped", r4, swap_rows, "table_row_order", r4)

    # 5. a reading's interval, the digest and signature carried across
    def with_readings(c):
        return is_table(c) and c.get("readings") is not None and any(r["rounds"] for r in c["record"]["rows"])
    r5 = find(with_readings, True)

    def interval(x, d):
        k = next(k for k, row in enumerate(x["record"]["rows"]) if row["rounds"])
        iv = x["readings"]["rows"][k]["rounds"][0]["busy"][0]
        iv[0] = 0 if iv[0] > 0 else 10 ** 9
        x["record"]["readings_sha256"] = sha(readings_bytes_of(x["readings"])).hex()
        resign(x["record"], TABLE)
    put("a reading's interval", r5, interval, "table_readings_mismatch", r5)

    # 6. the readings changed under their digest
    c6 = find(with_readings)

    def readings_only(x, d):
        k = next(k for k, row in enumerate(x["record"]["rows"]) if row["rounds"])
        x["readings"]["rows"][k]["rounds"][0]["completions"] += 1
    put("the readings, under their digest", c6, readings_only, "table_readings_mismatch")

    # 7. a row's status
    def status_flip(x, d):
        row = x["record"]["rows"][figure_row(x)]
        row["status"] = "relative"
        resign(x["record"], TABLE)
    put("a row's status", r, status_flip, "table_row_status", r)

    # 8. a row's declaration, one below its floor
    def decl_minus(x, d):
        rec = x["record"]
        row = rec["rows"][figure_row(x)]
        o = row_outcome(row, rec["sample_us"])
        row["energy_uncertainty"]["relative_ppm"] = max(o["p"], o["q"]) - 1
        resign(rec, TABLE)
        return ("energy_class_invalid" if row["joules_micro"] * row["energy_uncertainty"]["relative_ppm"] < FLOOR
                or row["energy_uncertainty"]["relative_ppm"] < 1 else "table_row_declaration")
    put("a row's declaration", r, decl_minus, None, r)

    # 9. a counter's status
    def counter_frozen(x, d):
        rec = x["record"]
        prof = rec["rows"][figure_row(x)]["profile"]
        for cn in rec["counters"]:
            if cn["profile"] == prof:
                cn["status"] = "frozen"
        resign(rec, TABLE)
    put("a counter's status", r, counter_frozen, "table_row_status", r)

    # 10. an extra member
    put("an extra member", c, lambda x, d: x["record"].update(note="x"), "record_malformed")

    # 11. a segment's signature
    seg = lambda m: (lambda c: c["subject"] == "segment" and c["record"]["method"] == m)  # noqa: E731
    c11 = find(lambda c: c["subject"] == "segment")
    put("a segment's signature byte", c11, lambda x, d: x["record"].update(sig=flip_hex(x["record"]["sig"], 63, 0x10)),
        "signature_invalid")

    # 12. a measured segment's figure
    r12 = find(seg("measured"), True)

    def seg_plus(x, d):
        x["record"]["joules_micro"] += 1
        resign(x["record"], SEGMENT)
    put("a measured segment's figure", r12, seg_plus, "segment_figure_mismatch", r12)

    # 13. a measured segment's declaration
    def seg_decl(x, d):
        s = x["record"]
        b = s["bracket"]
        mm = b["busy"]["mean_uw"] - b["idle"]["mean_uw"]
        f = figure_of(mm, b["window_us"], 1, 1)
        p = least_p(b["idle"]["n"], b["idle"]["sd_uw"], b["busy"]["n"], b["busy"]["sd_uw"], mm, b["window_us"], 1, 1, 0, f)
        s["energy_uncertainty"]["relative_ppm"] = p - 1
        resign(s, SEGMENT)
        return "energy_class_invalid" if f * (p - 1) < FLOOR or p - 1 < 1 else "segment_declaration_below"
    put("a measured segment's declaration", r12, seg_decl, None, r12)

    # 14. a table segment's class
    r14 = find(lambda c: seg("table")(c) and c["record"]["table"]["id"] in held_tables(c.get("tables")), True)

    def seg_class(x, d):
        s = x["record"]
        s["energy_provenance"] = "Estimator" if s["energy_provenance"] == "ModelBased" else "ModelBased"
        resign(s, SEGMENT)
    put("a table segment's class", r14, seg_class, "segment_table_mismatch", r14)

    # 15. an estimated segment's figure
    r15 = find(seg("estimator"), True)
    put("an estimated segment's figure", r15, seg_plus, "segment_estimator_mismatch", r15)

    # 16–18. a chain
    chain2 = lambda c: c["subject"] == "chain" and len(c["segments"]) >= 2  # noqa: E731
    c16 = find(chain2)

    def chain_swap(x, d):
        s = x["segments"]
        s[0], s[1] = s[1], s[0]
    put("a chain's receipts swapped", c16, chain_swap, "segment_chain_broken")

    def chain_drop(x, d):
        s = x["segments"]
        s.pop(len(s) // 2 if len(s) >= 3 else 0)
    put("a chain's receipt dropped", c16, chain_drop, "segment_chain_broken")
    r18 = find(chain2, True)

    def chain_prev(x, d):
        s = x["segments"][1]
        s["prev"] = sha(b"not the receipt before it").hex()
        resign(s, SEGMENT)
    put("a chain's prev", r18, chain_prev, "segment_chain_broken", r18)

    # 19–20. a hop's share, a hop's record
    def entry(c, role):
        return next((h for h in c["record"]["hops"] if h["role"] == role), None)

    def hop_part(x):
        """The origin's or relay's figure entry whose live-metering record the case holds."""
        return next((h for h in x["record"]["hops"] if h["role"] in (ORIGIN, RELAY) and h["source"] is not None
                     and h["source"]["kind"] == "live-metering-hop" and h["source"]["record"] in (x.get("hops") or [])), None)

    def held_hop(c):
        return c["subject"] == "session" and hop_part(c) is not None
    r19 = find(held_hop, True)

    def share_den(x, d):
        hop_part(x)["source"]["share_den"] += 1
        resign(x["record"], SESSION)
    put("a hop's share", r19, share_den, "split_hop_mismatch", r19)
    c20 = find(held_hop)

    def hop_byte(x, d):
        dg = hop_part(x)["source"]["record"]
        d["hops"][dg] = flip_hex(d["hops"][dg], 86)
    put("a hop record's byte", c20, hop_byte, "split_hop_mismatch")

    # 21. the display's figure
    r21 = find(lambda c: c["subject"] == "session" and entry(c, DISPLAY) is not None
               and entry(c, DISPLAY)["energy_provenance"] is not None, True)

    def display_plus(x, d):
        entry(x, DISPLAY)["joules_micro"] += 1
        resign(x["record"], SESSION)
    put("the display's figure", r21, display_plus, "split_display_mismatch", r21)

    # 22–24. a session against its segment receipts
    holding = lambda c: c["subject"] == "session" and len(c.get("segments") or []) >= 1  # noqa: E731
    r22 = find(holding, True)

    def media_plus(x, d):
        x["record"]["media_us"] += 1
        resign(x["record"], SESSION)
    put("a session's media_us", r22, media_plus, "session_streams_mismatch", r22)
    c23 = find(holding)
    put("a session's segment left out", c23, lambda x, d: x["segments"].pop(), "session_streams_mismatch")
    r24 = find(lambda c: holding(c) and entry(c, SINK_DECODE) is not None
               and entry(c, SINK_DECODE)["energy_provenance"] is not None, True)

    def decode_plus(x, d):
        entry(x, SINK_DECODE)["joules_micro"] += 1
        resign(x["record"], SESSION)
    put("the sink-decode entry's figure", r24, decode_plus, "split_decode_mismatch", r24)

    # 24b. the display entry left out of a session that presented content
    r24b = find(lambda c: c["subject"] == "session" and entry(c, DISPLAY) is not None, True)

    def display_out(x, d):
        x["record"]["hops"] = [h for h in x["record"]["hops"] if h["role"] != DISPLAY]
        resign(x["record"], SESSION)
    put("the display entry left out", r24b, display_out, "split_part_invalid", r24b)

    # 25–27. the corpus's own statements
    c25 = find(lambda c: is_hex(c.get("identity"), 32))

    def ident(x, d):
        x["identity"] = flip_hex(x["identity"], 0)
    if c25 is not None:
        d = copy.deepcopy(doc)
        ident(next(x for x in d["cases"] if x["name"] == c25["name"]), d)
        out.append(("a stated identity", d, tables, None, None))
    refused = next((x for x in cases if x.get("expect") in REASONS), None)
    if refused is not None:
        d = copy.deepcopy(doc)
        x = next(y for y in d["cases"] if y["name"] == refused["name"])
        x["expect"] = next(r for r in REASONS if r != refused["expect"])
        out.append(("a stated reason", d, tables, None, None))
    counts = {}
    for x in cases:
        counts[x.get("expect")] = counts.get(x.get("expect"), 0) + 1
    only = next((x for x in cases if x.get("expect") in REASONS and counts[x["expect"]] == 1), None)
    if only is not None:
        d = copy.deepcopy(doc)
        d["cases"] = [y for y in d["cases"] if y["name"] != only["name"]]
        out.append(("the only case refused %s" % only["expect"], d, tables, None, None))
    else:
        notes.append("no reason is refused by exactly one case (the coverage mutation needs one)")

    # 28–29. the measured tables
    # on the first measured table that holds, alone, so that the defect is what fails
    good = next((t for t in tables if not check_tables([t], quiet=True)[0]), None)
    if good is not None:
        sink, files = good
        t2 = [(sink, dict(files))]
        t2[0][1]["readings.json"] = files["readings.json"].replace(b"]]", b"] ]", 1) if b"]]" in files["readings.json"] \
            else files["readings.json"] + b"\n"
        out.append(("a measured table's readings", doc, t2, None, None))
        t3 = [(sink, dict(files))]
        t3[0][1]["procedure.txt"] = bytes([files["procedure.txt"][0] ^ 1]) + files["procedure.txt"][1:]
        out.append(("a measured table's procedure", doc, t3, None, None))
    elif tables:
        notes.append("no measured table holds, so the two table mutations have nothing to mutate")
    return out, notes


# ---------------------------------------------------------------- the self-test

def self_test():
    """Hand cases for JCS, the integer forms, the signer and the §8 parse; a list of the
    checks that fail."""
    fails = []

    def eq(name, got, want):
        if got != want:
            fails.append("%s: %r, not %r" % (name, got, want))
    # RFC 8785 §3.2.3's sorting example (strings only), and escapes.
    o = loads('{"\\u20ac":"Euro Sign","\\r":"Carriage Return","\\ufb33":"Hebrew Letter Dalet With Dagesh",'
              '"1":"One","\\ud83d\\ude00":"Emoji: Grinning Face","\\u0080":"Control",'
              '"\\u00f6":"Latin Small Letter O With Diaeresis"}')
    eq("jcs sorting", jcs(o), ('{"\\r":"Carriage Return","1":"One","\u0080":"Control",'
                               '"ö":"Latin Small Letter O With Diaeresis","€":"Euro Sign",'
                               '"\U0001F600":"Emoji: Grinning Face","דּ":"Hebrew Letter Dalet With Dagesh"}').encode("utf-8"))
    eq("jcs escapes", jcs({"b": [3, None, "\u000f\"\\\n\u007f
"], "a": 9007199254740991}),
       b'{"a":9007199254740991,"b":[3,null,"\\u000f\\"\\\\\\n\x7f\xe2\x80\xa8"]}')
    for bad in (True, 1.0, -1, 2 ** 53, "\ud800"):
        try:
            jcs([bad])
            fails.append("jcs takes %r" % (bad,))
        except ValueError:
            pass
    x = loads('{"a":1,"a":2,"b":1.0,"c":-0,"d":1e3}')
    eq("reader dup", x.dup, True)
    eq("reader numbers", [type(x[k]).__name__ for k in "bcd"], ["JNum"] * 3)
    eq("reader dup kept by copy", copy.deepcopy(x).dup, True)
    # §4.3: powers, mean half up, sd rounded up, liveness.
    eq("p half up", summary_of([[1, 2000], [1, 2000]])["mean_uw"], 1)
    eq("p 1 mW", (2000 * 200000 + 200000) // (2 * 200000), 1000)
    s = summary_of([[2, 2000], [4, 2000], [6, 2000], [8, 2000]])  # p = 1, 2, 3, 4
    eq("summary 1..4", s, {"n": 4, "mean_uw": 3, "sd_uw": 2, "live_intervals": 4, "longest_live_run": 4})
    # p = 500, 500, 0, 500, 500, 500: n·Σp² − (Σp)² = 1 250 000 over n(n − 1) = 30 is
    # 41 666.7, whose root 204.1 rounds up to 205.
    s = summary_of([[5, 10], [5, 10], [0, 10], [5, 10], [5, 10], [5, 10]])
    eq("liveness", (s["live_intervals"], s["longest_live_run"], s["sd_uw"]), (5, 3, 205))
    eq("sd one interval", summary_of([[7, 3]])["sd_uw"], 0)
    eq("least_sq", [least_sq(12, 0), least_sq(12, 12), least_sq(12, 13), least_sq(0, 1), least_sq(1, 10 ** 30)],
       [0, 1, 2, None, 10 ** 15])
    # energy-measurement §6.2 from rest, against its real-valued §2.3 form.
    m, t, c, n = 500000, 2000000, 100, 4
    wn = 2 * c + n
    f = figure_of(m, t, wn, 2)
    eq("F from rest", f, 9804)
    p = least_p(10, 10000, 10, 20000, m, t, wn, 2, n, f)
    e = m * t * 2 / (1e6 * wn)
    u_meas = math.sqrt((math.sqrt(10000 ** 2 / 10 + 20000 ** 2 / 10) / m) ** 2 + (n / (math.sqrt(3) * wn)) ** 2)
    u_rel = math.sqrt((e * u_meas) ** 2 + 1 / 12) / f
    if not 0 <= p - u_rel * 1e6 < 1.001:
        fails.append("p from rest: %d against %.3f" % (p, u_rel * 1e6))
    if f * p < FLOOR:
        fails.append("p from rest below the floor")
    # bounded (§5.2): Wn = Wd = 1, N = 0.
    f = figure_of(1000, 1000000, 1, 1)
    eq("F bounded", f, 1000)
    p = least_p(10, 100, 20, 200, 1000, 1000000, 1, 1, 0, f)
    u = math.sqrt((1e6 * math.sqrt(100 ** 2 / 10 + 200 ** 2 / 20) / 1e6) ** 2 + 1 / 12) / f
    if not 0 <= p - u * 1e6 < 1.001:
        fails.append("p bounded: %d against %.3f" % (p, u * 1e6))
    eq("F rounds half up", figure_of(1, 500000, 1, 1), 1)
    eq("F rounds down", figure_of(1, 499999, 1, 1), 0)
    # the estimators and the shares.
    eq("estimator-cpu-time/1", estimate_cpu(1000000, 1500), 1500000)
    eq("estimator-cpu-time/1 half up", [estimate_cpu(1, 500), estimate_cpu(1, 499)], [1, 0])
    eq("estimator-display/1", estimate_display(30000, 125, 1000000, 250), 15000000)
    eq("estimator-display/1 nits 0", estimate_display(30000, 0, 1000000, 250), 0)
    eq("shares", [share_of(1000, 1, 3), share_of(1000, 2, 3), share_of(1, 1, 2), share_of(1, 1, 3)], [333, 667, 1, 0])
    eq("table formula", [table_figure(1000, 3, 2), table_figure(1, 1, 2), table_figure(1, 1, 3)], [1500, 1, 0])
    eq("q", least_sq(12 * 1000 * 1000, 10 ** 12 * 10 ** 2), 2887)
    eq("ceil", ceil_div(7, 2), 4)
    # the median by (F, r)
    figs = sorted([(5, 0), (3, 1), (5, 2), (3, 3), (4, 4), (9, 5), (1, 6)])
    eq("median", figs[len(figs) // 2], (4, 4))
    eq("decoder", [decoder_ok(v) for v in ("builtin:[email protected]", "builtin:wai@v1", "platform:videotoolbox",
                                            "platform:Video", "builtin:a@1+b.c-d")], [True, False, True, False, True])
    eq("dates", [date_ok(v) for v in ("2026-02-29", "2028-02-29", "2026-13-01", "2026-10-22", "26-10-22")],
       [False, True, False, True, False])
    # RFC 8032 §7.1 TEST 1 through the mutation signer.
    kp = keypair(bytes.fromhex("9d61b19deffd5a60ba844af492ec2cc44449c5697b326919703bac031cae7f60"))
    eq("signer public key", kp[2].hex(), "d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a")
    eq("signer TEST 1", sign(kp, b"").hex(), "e5564300c360ac729086e2cc806e828a84877f1eb8e5d974d873e06522490155"
                                                "5fb8821590a33bacc61e39701cf9b46bd25bf5f0595bbe24655141438e7a100b")
    eq("signer verifies", ED.verify(MUT_KEY[2], b"m", sign(MUT_KEY, b"m")), None)
    eq("§8 row reasons", tuple(prose_reasons()[0]), ROW_REASONS)
    eq("§8 unmetered reasons", tuple(prose_reasons()[1]), UNMETERED_REASONS)
    return fails


def main(argv):
    fail = 0
    st = self_test()
    for f in st:
        print("FAIL self-test " + f)
    print("%s the self-test: RFC 8785, the reader, the integer forms, the estimators, RFC 8032 signing" % ("PASS" if not st else "FAIL"))
    fail += bool(st)
    listed = reasons_listed()
    ok = listed == REASONS
    print("%s §8 lists exactly the reasons this verifier raises%s" % ("PASS" if ok else "FAIL", "" if ok else ": %r" % listed))
    fail += not ok
    if argv == ["--self-test"]:
        return 1 if fail else 0
    if argv[0] == "--table":
        d = argv[1]
        if not os.path.isdir(d):
            print("FAIL %s is not a directory" % d)
            return 1
        tfails, lines = check_tables([(d, load_table_dir(d))], prefix="")
        for line in lines:
            print(line)
        print("%s the measured table in %s is accepted with its readings and procedure" % ("PASS" if not tfails else "FAIL", d))
        return 1 if fail or tfails else 0
    corpus_dir = argv[0]
    path = os.path.join(corpus_dir, "vectors.json")
    tables = load_tables(corpus_dir)
    if not os.path.exists(path):
        print("FAIL %s does not exist" % path)
        return 1
    with open(path, encoding="utf-8") as f:
        doc = loads(f.read())
    notes = []
    fails, outs = check(doc, notes=notes)
    for n in notes:
        print("NOTE " + n)
    n_acc = sum(1 for o in outs.values() if o[0] == "accept")
    print("%s %d cases verify from the text (%d accepted, %d refused)"
          % ("PASS" if not fails else "FAIL", len(doc.get("cases", [])), n_acc, len(outs) - n_acc))
    fail += bool(fails)
    tfails, lines = check_tables(tables)
    for line in lines:
        print(line)
    n_bad = sum(1 for t in tables if check_tables([t], quiet=True)[0])
    print("%s %d of %d measured tables accepted with their readings and procedures"
          % ("PASS" if not tfails else "FAIL", len(tables) - n_bad, len(tables)))
    fail += bool(tfails)
    muts, notes = mutations(doc, tables)
    for n in notes:
        print("NOTE " + n)
    caught = 0
    for name, d, tb, case, want in muts:
        f1, outs = check(d, quiet=True)
        f2, _ = check_tables(tb, quiet=True) if tb is not tables else ([], [])
        got = outs.get(case, (None,))[0] if case is not None else None
        if not (f1 or f2):
            print("FAIL the mutation '%s' goes unnoticed" % name)
            fail += 1
        elif want is not None and got != want:
            print("FAIL the mutation '%s' of %s gives %s, the text gives %s (%s)" % (name, case, got, want, outs[case][2]))
            fail += 1
        else:
            caught += 1
    kinds_ok = len(muts) >= 12
    print("%s %d of %d mutated copies of the corpus fail%s" % ("PASS" if caught == len(muts) and kinds_ok else "FAIL", caught,
                                                                len(muts), "" if kinds_ok else " (fewer than 12 kinds)"))
    fail += not kinds_ok
    return 1 if fail else 0


if __name__ == "__main__":
    args = sys.argv[1:]
    if not (len(args) == 1 and args[0] != "--table" or len(args) == 2 and args[0] == "--table"):
        print("usage: sink_energy_verify.py <live-conformance/sink-energy> | --table <dir> | --self-test", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1:]))