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

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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. It takes from the second implementations of the texts this extension builds
on what they define: group and object receipts from the JWP corpus's
(`../jwp-conformance/jwp_verify.py`, jwp-receipts §2.2, §3); a claim's steps 1 to 5 from the
claim corpus's (`../claim-conformance/claim_verify.py`, jwp-receipts §3.1), whose step 4 holds
§4.2's rules; a transmission claim's step 6, a schedule's, and a receiver's passes from the
carousel corpus's (`carousel_verify.py`: bearer-receipts §4.3, carousel §4.3 and §7); a holdings
record's form, signature and holdings from the holdings corpus's (`holdings_verify.py`, holdings
§4.4, §5.4); and a WAI1 envelope's manifest from the staged corpus's reader (SPEC §2, §3).
Everything else this extension adds is written here:

- §4.3: a timeliness claim against its group, the object receipts and envelopes held, and each
  rung's group and object receipt;
- §5: an entry's value at an instant, and its end;
- §6: the intents that stand, the records that stand and the held-at statements they give
  (`held_from_records`, with §2's `before_group`), each intent's lines (`lines`), and a
  statement offered;
- §7: the plan's inputs, each source's arrival, and the choice, the ladder's rung included;
- §7.5: a sink's run of its plans against its clock (`run`), each plan reading the claims and records
  held by its instant.

For every case it requires the outcome the corpus states, field for field; that every reason
of §9's table, every outcome of §6.3's and every action §7.5.3 states appears in the corpus; then
it mutates copies of the
corpus, one defect each, and requires every copy to fail; and it drops each of its own checks in
turn and requires the corpus to catch the weakened verifier. A bound no vector carries is checked
directly.

    python3 timeliness_verify.py <live-conformance/timeliness>

Exit status 0 only if every case verifies, every mutated copy fails, every weakened verifier is
caught and every direct check holds. Python 3.8 or later; the `cryptography` package, which the
claim verifier imports.
"""

import copy
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
for d in ("claim-conformance", "jwp-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", d))
import claim_verify as CV  # noqa: E402
import jwp_verify as JV  # noqa: E402
import carousel_verify as KV  # noqa: E402
import holdings_verify as HV  # noqa: E402

SV = CV.SV  # the staged corpus's strict WAI1 reader, which the claim verifier loads

EXTENSION = os.path.join(HERE, "..", "extensions", "timeliness.md")
SPEC = os.path.join(HERE, "..", "SPEC.md")
JWP_RECEIPTS = os.path.join(HERE, "..", "extensions", "jwp-receipts.md")
TELEMETRY = os.path.join(HERE, "..", "extensions", "delivery-telemetry.md")
FULL = 1_000_000                       # §2: a value in parts per million of the full value
LAST_INSTANT = 253_402_300_799_999     # §2: 9999-12-31T23:59:59.999Z
MAX_INT = (1 << 53) - 1
ACTIONS = ("fetch", "abandon", "present", "passed", "ended")  # §7.5.3
T0_PLUS_LATER = 1_793_664_000_000 + 3_600_000  # An instant past every plan of the corpus's runs.

# A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything.
DROPPED = None


def drops(name):
    return DROPPED == name


def text_of(path):
    with open(path, encoding="utf-8") as f:
        return f.read()


def table_rows(text, heading):
    """The first backticked cell of each row of the table under `heading`, up to the next heading
    of its level or above."""
    at = text.index("\n" + heading)
    level = heading.split(" ")[0]
    ends = [text.find("\n" + h + " ", at + 1) for h in ("#" * n for n in range(2, len(level) + 1))]
    ends = [e for e in ends if e >= 0]
    rows = text[at:min(ends) if ends else len(text)].splitlines()
    return [r.split("`")[1] for r in rows if r.startswith("| `")]


def reasons_table():
    """§9's codes, in order, each with the section the table says raises it."""
    text = text_of(EXTENSION)
    at = text.index("\n## 9. Reasons")
    end = text.find("\n## ", at + 1)
    rows = text[at:end].splitlines()
    return [(r.split("`")[1], r.split("|")[2].strip()) for r in rows if r.startswith("| `")]


def outcomes_table():
    """§6.3's outcomes, in order."""
    return table_rows(text_of(EXTENSION), "### 6.3 Lines")


def refinement_only():
    """SPEC §5, *Refinement capabilities*: the capabilities that are refinement-only."""
    return set(table_rows(text_of(SPEC), "### Refinement capabilities (staged delivery)"))


REFINEMENT_ONLY = refinement_only()


def need(ok, reason):
    """jwp-receipts §3.1 step 6."""
    if not ok:
        raise CV.Reject(6, reason)


# ------------------------------------------------------------------ what the verifier holds

# Steps 1 to 4 of a claim, a group's steps 1 and 2 with its objects', and a record's reading are
# pure in their bytes: each is computed once and kept, which nothing a weakened run drops reads.
_CLAIMS, _GROUPS, _RECORDS = {}, {}, {}


def claim_steps(wire):
    """jwp-receipts §3.1 steps 1 to 4 (claim_verify.py). Returns the claim; raises CV.Reject."""
    if wire not in _CLAIMS:
        try:
            _CLAIMS[wire] = CV.verify_claim(wire)
        except CV.Reject as r:
            _CLAIMS[wire] = r
    v = _CLAIMS[wire]
    if isinstance(v, CV.Reject):
        raise v
    return v


def read_record(b):
    """holdings §4.4 (holdings_verify.py). Returns the record; raises HV.Refused."""
    if b not in _RECORDS:
        try:
            _RECORDS[b] = HV.read(b)
        except HV.Refused as r:
            _RECORDS[b] = r
    v = _RECORDS[b]
    if isinstance(v, HV.Refused):
        raise v
    return v


class Group:
    """A group given, after jwp-receipts §3 steps 1 and 2, with its object receipts and envelopes."""

    def __init__(self, item):
        self.receipt = JV.read_group(item["group"])
        JV.verify_group(self.receipt)
        self.link = JV.link(self.receipt).hex()
        self.count = self.receipt["leaf_count"]
        self.signer = self.receipt["signer_pubkey"]
        self.start = JV.seconds(self.receipt["group_start"])
        self.verdicts = []   # bearer-receipts §6.2 step 2, one for each object receipt given
        self.pairs = []      # [(object receipt, envelope or None)] for each that reads
        self.included = {}   # position -> content hash of each object receipt that passes step 5
        self.envelopes = {}  # content hash -> envelope bytes held
        for o in item["objects"]:
            try:
                ob = JV.read_object(o["receipt"])
            except JV.Refuse as r:
                self.verdicts.append("reject reason=" + r.reason)
                continue
            env = None if o["envelope"] is None else bytes.fromhex(o["envelope"])
            self.pairs.append((ob, env))
            try:
                if env is None:
                    JV.inclusion(ob, self.receipt)
                    self.verdicts.append("inclusion-verified")
                else:
                    JV.accept(ob, self.receipt, env)
                    self.verdicts.append("delivered-verified")
            except JV.Refuse as r:
                self.verdicts.append("reject reason=" + r.reason)
            try:
                JV.inclusion(ob, self.receipt)
                self.included[ob["object_id"]] = ob["content_hash"].hex()
            except JV.Refuse:
                pass
            if env is not None:
                self.envelopes[JV.blake3(JV.OBJECT + env).hex()] = env


def read_groups(items):
    """§6.2 step 1: each group under jwp-receipts §3 steps 1 and 2, and each of its object receipts
    as bearer-receipts §6.2 step 2 checks it. Raises JV.Refuse for a group that fails."""
    out = []
    for item in items:
        key = json.dumps(item, sort_keys=True)
        if key not in _GROUPS:
            try:
                _GROUPS[key] = Group(item)
            except JV.Refuse as r:
                _GROUPS[key] = r
        g = _GROUPS[key]
        if isinstance(g, JV.Refuse):
            raise g
        out.append(g)
    return out


def manifest(env):
    """A WAI1 envelope's manifest (SPEC §2, §3), or None for bytes that are not one."""
    try:
        m, _, _ = SV.read_wai1(env)
    except SV.NotWai1:
        return None
    return m


def rungs(entry):
    """§4.1: an entry's ladder, in its order: none for `null`, the one object, or the array's."""
    fb = entry["fallback"]
    return [] if fb is None else fb if isinstance(fb, list) else [fb]


def checked_rungs(entry, check):
    """The rungs §4.3 reads, every one; a weakened run reads only the first."""
    return rungs(entry)[:1] if drops(check) else rungs(entry)


def is_layer(env):
    """§4.3 step 6: a WAI1 whose capability is refinement-only (SPEC §5)."""
    m = manifest(env)
    return m is not None and m["model_requirement"]["capability"] in REFINEMENT_ONLY


def derived_from(env, motif):
    """§4.3 step 8: a WAI1 whose manifest carries `derived_from` (SPEC §7.1, *Derived renditions*:
    its four members in their forms) with `motif` the primary's content hash. A WAI2 is none."""
    m = manifest(env)
    df = None if m is None else m.get("derived_from")
    return (isinstance(df, dict) and all(isinstance(df.get(k), str) and df[k] for k in
                                         ("motif", "motif_capability", "renderer", "settings"))
            and CV.is_hex(df["motif"], 32) and CV.is_hex(df["settings"], 32) and df["motif"] == motif)


# ------------------------------------------------------------------ §2: the held-at statement

def before_group(word_s, start):
    """§2: a held-at statement is before its group when the whole seconds of the time its word
    states, any fraction dropped (`word_s`), are below the seconds jwp-receipts §2.4.3 reads from
    the group's `group_start` (`start`, None where it is not a date-time, which nothing is before).
    §4.3 step 4 reads an entry's `from_ms` by the same rule."""
    if start is None:
        return False
    return word_s <= start if drops("same_second") else word_s < start


# ------------------------------------------------------------------ §4.3: step 6

def check_timeliness(c, g, groups):
    """§4.3 against the group `g` the claim is checked against, which passed jwp-receipts §3 steps
    1 and 2, and every group given (`groups`, by link hash) with the object receipts and envelopes
    held under it."""
    need(g.start is not None, "context_invalid")
    entries = c["body"]["objects"]
    envelopes = {}
    for x in groups.values():
        envelopes.update(x.envelopes)
    need(drops("subject") or g.link == c["subject"], "timeliness_subject_mismatch")
    need(drops("signer") or g.signer == c["signer_pubkey"], "timeliness_signer_mismatch")
    need(drops("range") or all(e["leaf_index"] < g.count for e in entries), "timeliness_leaf_out_of_range")
    need(drops("before") or not any(before_group(e["from_ms"] // 1000, g.start) for e in entries),
         "timeliness_before_group")
    need(drops("content") or all(g.included.get(e["leaf_index"], e["content_hash"]) == e["content_hash"]
                                 for e in entries), "timeliness_content_mismatch")
    named = [e["content_hash"] for e in entries] + [f["content_hash"] for e in entries for f in checked_rungs(e, "rung_layer")]
    need(drops("layer") or not any(h in envelopes and is_layer(envelopes[h]) for h in named), "timeliness_on_layer")
    for f in (f for e in entries for f in checked_rungs(e, "rung_position")):
        fg = groups.get(f["group"])
        if fg is not None and not drops("fallback_position"):
            need(f["leaf_index"] < fg.count and fg.included.get(f["leaf_index"], f["content_hash"]) == f["content_hash"],
                 "timeliness_fallback_mismatch")
    for e in entries:
        for f in checked_rungs(e, "rung_derived"):
            if f["content_hash"] in envelopes and not drops("derived"):
                need(derived_from(envelopes[f["content_hash"]], e["content_hash"]), "timeliness_fallback_not_derived")


# ------------------------------------------------------------------ §5: value over time

def value(entry, h):
    """§5.1: the entry's value at the instant h, floored."""
    by = entry["by_ms"]
    if h <= by:
        return FULL
    pts = [(by, FULL)] + [(p["at_ms"], p["value_ppm"]) for p in entry["decay"]]
    for (ta, va), (tb, vb) in zip(pts, pts[1:]):
        if ta < h <= tb:
            num, den = va * (tb - h) + vb * (h - ta), tb - ta
            return (2 * num + den) // (2 * den) if drops("floor") else num // den
    return 0


def end(entry):
    """§5.2: the `at_ms` of its last point."""
    return entry["decay"][-1]["at_ms"]


# ------------------------------------------------------------------ §6: judging delivery

def claims_step(wires, groups, plan, chained=True):
    """§6.2 step 2, and §7.2's (`plan`): each claim under jwp-receipts §3.1 steps 1 to 5, each
    signer's previous claim the last of its claims held before it, then step 6 against the group
    whose link hash is its subject, or the first group. Returns each claim's label, and each
    claim accepted (None for one refused). `chained` is False only in a weakened run."""
    by_link = {}
    for g in groups:
        by_link.setdefault(g.link, g)
    held, labels, accepted = {}, [], []
    for wire in wires:
        try:
            c = claim_steps(wire)
        except CV.Reject as r:
            labels.append(f"reject step={r.step} reason={r.reason}")
            accepted.append(None)
            continue
        previous = held.get(c["signer_pubkey"])
        held[c["signer_pubkey"]] = c
        try:
            if chained:
                CV.check_chain(c, previous)
            g = by_link.get(c["subject"], groups[0] if groups else None)
            if c["class"] == "timeliness":
                need(g is not None, "timeliness_subject_mismatch")
                check_timeliness(c, g, by_link)
            elif plan and c["class"] == "transmission" and g is not None:
                KV.check_transmission(c, g.receipt)
            elif plan and c["class"] == "schedule" and g is not None:
                KV.check_schedule(c, g.receipt, g.pairs)
            elif not drops("context"):
                raise CV.Reject(6, "context_mismatch")
        except CV.Reject as r:
            labels.append(f"reject step={r.step} reason={r.reason}")
            accepted.append(None)
            continue
        labels.append("accepted")
        accepted.append(c)
    return labels, accepted


def intents_step(accepted, groups):
    """§6.2 step 3: the entries of the accepted claims by group and `leaf_index`, in the order of
    the groups given and then of `leaf_index`: [(group, leaf_index, entry, or None when set
    aside)]."""
    stated = {}
    for c in accepted:
        if c is not None and c["class"] == "timeliness":
            for e in c["body"]["objects"]:
                stated.setdefault((c["subject"], e["leaf_index"]), []).append(e)
    out, seen = [], set()
    for g in groups:
        if g.link in seen:
            continue
        seen.add(g.link)
        for i in sorted(i for link, i in stated if link == g.link):
            es = stated[(g.link, i)]
            stands = drops("intent_contradiction") or all(e == es[0] for e in es)
            out.append((g.link, i, es[0] if stands else None))
    return out


def records_step(records, groups):
    """§6.2 step 4: each record as holdings §4.4 reads and verifies it, refused whole when its
    entry for a group given misstates the leaf count or is before the group, two of one holder at
    one `as_of_ms` that differ set aside. Returns each record's label and the records that are
    not refused (index -> record)."""
    by_link = {g.link: g for g in groups}
    labels, kept = [], {}
    for k, hexed in enumerate(records):
        try:
            rec = read_record(bytes.fromhex(hexed))
            mine = [(by_link[g.hex()], n) for g, n, _ in rec[3] if g.hex() in by_link]
            if not drops("leaf_count") and any(g.count != n for g, n in mine):
                raise HV.Refused("holdings_leaf_count_mismatch")
            if not drops("held_before") and any(before_group(rec[1] // 1000, g.start) for g, _ in mine):
                raise HV.Refused("held_before_group")
            kept[k] = rec
            labels.append("stands")
        except HV.Refused as r:
            labels.append("reject reason=" + r.reason)
    at = {}
    for k, rec in kept.items():
        at.setdefault((rec[0], rec[1]), []).append(k)
    for ks in at.values():
        if len({HV.digest(kept[k]) for k in ks}) > 1 and not drops("record_contradiction"):
            for k in ks:
                labels[k] = "set-aside reason=holdings_contradicted"
    return labels, kept


def held_from_records(standing, group, index):
    """§6.2 step 5: the held-at statements the records that stand give for object `index` of
    `group` (a link hash in hexadecimal): (holder, group, index, instant) for each record that holds
    it (holdings §5.4), its instant the record's `as_of_ms`."""
    g = bytes.fromhex(group)
    return [(rec[0].hex(), group, index, rec[1]) for rec in standing if HV.holds(rec, g, index) is None]


def has_entry(rec, group):
    return any(g.hex() == group for g, _, _ in rec[3])


def line_holders(group, entry, standing):
    """§6.3: the holders with a record that stands and has an entry for the intent's group or a
    rung's group, in the order of their first such record."""
    rs = checked_rungs(entry, "rung_holders")
    out = []
    for rec in standing:
        h = rec[0].hex()
        if h not in out and (drops("line_holders") or has_entry(rec, group)
                             or any(has_entry(rec, r["group"]) for r in rs)):
            out.append(h)
    return out


def grade(entry, p, fs):
    """§6.3's table, the first outcome that holds, for a holder whose least instants for the
    primary and for each rung of the ladder are p and fs (None where none)."""
    by = entry["by_ms"]
    if p is not None and (p < by if drops("deadline_inclusive") else p <= by):
        return "on_time", FULL
    # The first rung, in the ladder's order, held by the deadline.
    in_time = [r for r, f in zip(rungs(entry), fs) if f is not None and (f <= by or drops("fallback_deadline"))]
    if in_time:
        r = in_time[-1] if drops("rung_order") else in_time[0]
        # §5.3: what arrived after the deadline grades nothing, so a later primary of more value
        # does not displace a rung held by then.
        if not (drops("sink_rule") and p is not None and value(entry, p) > r["value_ppm"]):
            return "replaced", r["value_ppm"]
    if p is not None:
        return "after_deadline", value(entry, p)
    return "not_shown", 0


def lines(entry, holders, primary, fallback):
    """§6.3: an intent's line for each of `holders` (keys in hexadecimal), in increasing order of
    key bytes, from that holder's own held-at statements for its primary and for the rungs of its
    fallback, each statement naming its object. *f* is one instant for a ladder of one rung, and
    the list of them for an array."""
    def least(statements, holder, latest=False):
        instants = [s[3] for s in statements if drops("own_statements") or s[0] == holder]
        if not instants:
            return None
        return max(instants) if latest or drops("least") else min(instants)

    out = []
    for h in (holders if drops("key_order") else sorted(holders, key=bytes.fromhex)):
        p = least(primary, h)
        fs = [least([s for s in fallback if (s[1], s[2]) == (r["group"], r["leaf_index"])], h, drops("least_fallback"))
              for r in rungs(entry)]
        outcome, v = grade(entry, p, fs)
        f = fs if isinstance(entry["fallback"], list) else (fs[0] if fs else None)
        out.append({"holder": h, "held_ms": p, "fallback_ms": f, "outcome": outcome, "value_ppm": v})
    return out


def intent_lines(link, entry, standing):
    primary = held_from_records(standing, link, entry["leaf_index"])
    fallback = [s for r in rungs(entry) for s in held_from_records(standing, r["group"], r["leaf_index"])]
    return lines(entry, line_holders(link, entry, standing), primary, fallback)


def verdict(case):
    """§6: the verdict, or `refuse reason=R` for a group that fails."""
    try:
        groups = read_groups(case["groups"])
    except JV.Refuse as r:
        return "refuse reason=" + r.reason
    labels, accepted = claims_step(case["claims"], groups, plan=False)
    intents = intents_step(accepted, groups)
    rlabels, kept = records_step(case["records"], groups)
    standing = [kept[k] for k in sorted(kept) if rlabels[k] == "stands"]
    out, found = [], {}
    for link, i, entry in intents:
        ls = [] if entry is None else intent_lines(link, entry, standing)
        found[(link, i)] = (entry, ls)
        out.append({"group": link, "leaf_index": i,
                    "intent": "stands" if entry is not None else "set-aside reason=timeliness_contradicted",
                    "end_ms": None if entry is None else end(entry), "lines": ls})
    statement = None
    s = case["statement"]
    if s is not None:
        entry, ls = found.get((s["group"], s["index"]), (None, []))
        if entry is None and not drops("no_intent"):
            statement = "reject reason=deadline_not_stated"
        else:
            line = next((x for x in ls if x["holder"] == s["holder"]), None)
            statement = ({"outcome": "not_shown", "value_ppm": 0} if line is None
                         else {"outcome": line["outcome"], "value_ppm": line["value_ppm"]})
    return {"objects": [g.verdicts for g in groups], "claims": labels, "records": rlabels, "intents": out,
            "statement": statement}


# ------------------------------------------------------------------ §7: planning

def sources_valid(sources, loss, h):
    """§7.1: the sources, each exactly a bearer instance, a holder or the origin."""
    if not isinstance(sources, list) or (not sources and not drops("sources_empty")):
        return False
    bearers, holders, origins = set(), set(), 0
    for s in sources:
        if not isinstance(s, dict):
            return False
        kind = s.get("kind")
        fetch = s.get("fetch_ms")
        if kind != "bearer" and not (drops("sources_fetch") or (CV.is_int(fetch) and fetch >= 1)):
            return False
        if kind == "bearer":
            if set(s) != {"kind", "bearer_id"} or not (drops("sources_uri") or CV.is_uri(s["bearer_id"])):
                return False
            if s["bearer_id"] in bearers and not drops("sources_twice"):
                return False
            bearers.add(s["bearer_id"])
        elif kind == "holder":
            if set(s) != {"kind", "holder", "fetch_ms"} or not (drops("sources_key") or CV.is_hex(s["holder"], 32)):
                return False
            if s["holder"] in holders and not drops("sources_twice"):
                return False
            holders.add(s["holder"])
        elif kind == "origin":
            if set(s) != {"kind", "fetch_ms"}:
                return False
            origins += 1
        elif not drops("sources_kind"):
            return False
    if origins > 1 and not drops("sources_twice"):
        return False
    return not (bearers and (loss is None or h is None)) or drops("sources_loss")


def plan_inputs(loss, h, t, sources):
    """§7.1's checks, in their order: the reason the plan is refused, or None."""
    if loss is not None and not drops("loss_bound") and not (CV.is_int(loss) and loss <= 999_999):
        return "loss_bound_invalid"
    if h is not None and not drops("overhead") and not (CV.is_int(h) and h <= 2):
        return "overhead_invalid"
    if t != "from_ms" and not drops("instant") and not (CV.is_int(t) and t <= LAST_INSTANT):
        return "instant_invalid"
    if not sources_valid(sources, loss, h):
        return "sources_invalid"
    return None


def hash_and_position(v, kind):
    """§7.1: the `group` of a wanted object or a statement is 64 lowercase hexadecimal digits, and its
    `index` an integer from 0 to 2^53 - 1."""
    if not (drops(f"{kind}_group") or CV.is_hex(v.get("group"), 32)):
        return False
    index = v.get("index")
    if drops(f"{kind}_index"):
        return True
    return CV.is_int(index) and not (drops(f"{kind}_greatest") and index == MAX_INT)


def wanted_formed(wanted):
    """§7.1: a list, each entry an object with a `group` and an `index`."""
    if not isinstance(wanted, list):
        return drops("wanted_list")
    return all(hash_and_position(w, "wanted") if isinstance(w, dict) else drops("wanted_object") for w in wanted)


def wanted_list(wanted):
    """What a verifier without §7.1's rule for the list reads of one: nothing wanted."""
    return wanted if isinstance(wanted, list) else []


def named(w):
    """What a wanted entry names. A verifier without §7.1's rule reads an entry that is no object as
    naming none, so as an object with no intent."""
    return (w.get("group"), w.get("index")) if isinstance(w, dict) else (None, None)


def statement_bearer(y, kind):
    """§7.1: a `bearer_id` that is a string and an absolute URI."""
    bearer = y.get("bearer_id")
    if not isinstance(bearer, str):
        return drops(f"{kind}_bearer_string")
    return drops(f"{kind}_uri") or CV.is_uri(bearer)


def symbols_formed(symbols):
    """§7.1: a list, each statement an object with a `bearer_id` that is an absolute URI, a `group`, an
    `index` and a list of at least one number, each an integer from 0 to 2^53 - 1; and no two of one
    object on one instance."""
    if not isinstance(symbols, list):
        return drops("symbols_list")
    seen = set()
    for y in symbols:
        if not isinstance(y, dict):
            if drops("symbols_statement"):
                continue
            return False
        if not statement_bearer(y, "symbols") or not hash_and_position(y, "symbols"):
            return False
        if not statement_numbers(y.get("held"), "symbols"):
            return False
        key = (y.get("bearer_id"), y.get("group"), y.get("index"))
        if key in seen and not drops("symbols_twice"):
            return False
        seen.add(key)
    return True


def statement_numbers(held, kind):
    """§7.1: a list of at least one number, each an integer from 0 to 2^53 - 1."""
    if not isinstance(held, list):
        return drops(f"{kind}_held_list")
    if not held and not drops(f"{kind}_empty"):
        return False
    if not drops(f"{kind}_integer") and any(type(n) is not int for n in held):
        return False
    return drops(f"{kind}_number") or all(CV.is_int(n) for n in held if type(n) is int)


def entry_on(schedules, bearer_id, group, index):
    """§7.2: the entry for an object in the schedule that stands for its group on an instance, or None."""
    body = schedules.get((group, bearer_id))
    return None if body is None else next((x for x in body["objects"] if x["leaf_index"] == index), None)


def holding_fits(e, held):
    """§7.2, by carousel §7's own check: one number for each source block, and no more of a `no-code`
    block than it has. Whatever else the plan's loss bound is does not enter into it."""
    p = KV.plan({"entry": e, "loss_ppm": 0, "overhead": 0, "at_ms": 0, "held": held})
    return p.get("reason") != "holding_invalid"


def schedules_step(labels, accepted):
    """§7.2: the declarations (bearer-receipts §6.2 step 4) and the schedules (carousel §6.2 step
    4), each of its group. Relabels the claims; returns the schedules that stand, by (group,
    bearer_id)."""
    def key(c):
        return c["body"]["bearer_id"] if drops("per_group") else (c["subject"], c["body"]["bearer_id"])

    def by_instance(cls):
        out = {}
        for i, c in enumerate(accepted):
            if c is not None and c["class"] == cls:
                out.setdefault(key(c), []).append(i)
        return out

    declared, contradicted = {}, set()
    for k, idx in by_instance("transmission").items():
        bodies = [accepted[i]["body"] for i in idx]
        same = all(b == bodies[0] for b in bodies)
        if same:
            declared[k] = bodies[0]["bearer"]
        else:
            contradicted.add(k)
        for i in idx:
            labels[i] = "declared" if same else "set-aside reason=bearer_contradicted"
    schedules = {}
    for k, idx in by_instance("schedule").items():
        bodies = [accepted[i]["body"] for i in idx]
        if not KV.monotone(bodies) and not drops("schedule_contradicted"):
            aside = "schedule_contradicted"
        elif k in contradicted:
            aside = "bearer_contradicted"
        elif k not in declared and not drops("undeclared"):
            aside = "bearer_not_declared"
        elif declared.get(k) == "bidirectional" and not drops("bidirectional"):
            aside = "schedule_on_bidirectional"
        else:
            aside = None
        # The last of an instance's schedules is the one that stands (carousel §6.2 step 4).
        last = bodies[0] if drops("schedule_last") else bodies[-1]
        for i in idx:
            if aside is not None:
                labels[i] = "set-aside reason=" + aside
            else:
                labels[i] = "stands" if accepted[i]["body"] == last else "extended"
                schedules[(accepted[i]["subject"], accepted[i]["body"]["bearer_id"])] = last
    return schedules


def standing_at(kept, rlabels, s):
    """§7.2: the records that stand at s, by holdings §5.1 to §5.3 over those §6.2 step 4 did not
    refuse, and not set aside there."""
    live = [rec for k, rec in kept.items() if drops("record_at_s") or rec[1] <= s < rec[2]]
    out = []
    for holder in {rec[0] for rec in live}:
        mine = [rec for rec in live if rec[0] == holder]
        latest = max(rec[1] for rec in mine)
        top = [rec for rec in mine if rec[1] == latest]
        if len({HV.digest(rec) for rec in top}) == 1:
            out.append(top[0])
    aside = {HV.digest(kept[k]) for k in kept if rlabels[k] != "stands"}
    return [rec for rec in out if HV.digest(rec) not in aside]


def arrival(src, group, index, s, ctx):
    """§7.2: a source's arrival of object `index` of `group` planned from s, or None."""
    kind = src["kind"]
    if kind == "origin":
        return s + src["fetch_ms"]
    if kind == "holder":
        recs = standing_at(ctx["kept"], ctx["rlabels"], s)
        held = any((drops("holder_record") or rec[0].hex() == src["holder"])
                   and HV.holds(rec, bytes.fromhex(group), index) is None for rec in recs)
        return s + src["fetch_ms"] if held else None
    if kind != "bearer":
        return None
    e = entry_on(ctx["schedules"], src["bearer_id"], group, index)
    if e is None:
        return None
    blocks = sum(n for _, n in CV.schedule_blocks(e))
    held = [0] * blocks
    for y in ctx["symbols"]:
        same_instance = y["bearer_id"] == src["bearer_id"] or drops("symbols_instance")
        if same_instance and y["group"] == group and y["index"] == index and not drops("symbols_ignored"):
            held = y["held"]
            break
    p = KV.plan({"entry": e, "loss_ppm": ctx["loss"], "overhead": ctx["h"], "at_ms": s, "held": held})
    return p["complete_ms"]


def choose(entry, primary, ladder):
    """§7.3 for an object whose intent stands, from each source's arrivals of the primary and,
    for each rung of the ladder in its order, of that rung."""
    by = entry["by_ms"]

    def choice(outcome, obj, r, i, arrived, v):
        return {"outcome": outcome, "object": obj, "rung": r, "source": i, "arrival_ms": arrived, "value_ppm": v}

    timely = [i for i, a in enumerate(primary) if a is not None and a <= by]
    if timely:
        i = min(timely, key=lambda j: (primary[j], j)) if drops("plan_order") else timely[0]
        return choice("on_time", "primary", None, i, primary[i], FULL)
    late = [(value(entry, a), i) for i, a in enumerate(primary) if a is not None]
    best = max((v for v, _ in late), default=0)
    # Rule 2: the first rung of at least the best late value that arrives by the deadline, from the
    # first such source in the asker's order.
    order = list(enumerate(rungs(entry)))
    if drops("plan_rung_first"):
        order = order[:1]
    if drops("plan_rung_order"):
        order.reverse()
    for r, fb in order:
        at_least = fb["value_ppm"] > best if drops("fallback_tie") else fb["value_ppm"] >= best
        if not (at_least or drops("fallback_value")):
            continue
        arrivals = ladder[r]
        in_time = [i for i, a in enumerate(arrivals)
                   if a is not None and (a < by if drops("plan_fallback_deadline") else a <= by)]
        if in_time and drops("fallback_order"):
            in_time.sort(key=lambda j: (arrivals[j], j))
        if in_time:
            return choice("replaced", "fallback", r, in_time[0], arrivals[in_time[0]], fb["value_ppm"])
    if late and (best >= 1 or drops("late_zero")):
        ties = [i for v, i in late if v == best]
        i = ties[-1] if drops("late_tie") else ties[0]
        return choice("after_deadline", "primary", None, i, primary[i], best)
    return choice("no_source", None, None, None, None, 0)


def plan(case):
    """§7: the plan, or `refuse reason=R`."""
    loss, h, t, sources = case["loss_ppm"], case["overhead"], case["at_ms"], case["sources"]
    bad = plan_inputs(loss, h, t, sources)
    if drops("wanted_before_sources") and not wanted_formed(case["wanted"]):
        return "refuse reason=wanted_invalid"
    if bad:
        return "refuse reason=" + bad
    if drops("wanted_before_symbols") and not wanted_formed(case["wanted"]):
        return "refuse reason=wanted_invalid"
    symbols = case.get("symbols", [])
    if not symbols_formed(symbols):
        return "refuse reason=symbols_invalid"
    if not wanted_formed(case["wanted"]):
        return "refuse reason=wanted_invalid"
    # A verifier without those rules reads what it cannot count as nothing held; a formed list stays whole.
    symbols = [y for y in symbols if isinstance(y, dict) and all(k in y for k in ("bearer_id", "group", "index"))
               and isinstance(y.get("held"), list) and not any(isinstance(n, str) for n in y["held"])] \
        if isinstance(symbols, list) else []
    try:
        groups = read_groups(case["groups"])
    except JV.Refuse as r:
        return "refuse reason=" + r.reason
    labels, accepted = claims_step(case["claims"], groups, plan=True, chained=not case.get("unchained"))
    intents = {(link, i): entry for link, i, entry in intents_step(accepted, groups)}
    rlabels, kept = records_step(case["records"], groups)
    schedules = schedules_step(labels, accepted)
    # §7.2: a statement against the entry that stands for its object on its instance; one with no such
    # entry is compared with nothing.
    if not drops("symbols_holding"):
        for y in symbols:
            e = entry_on(schedules, y["bearer_id"], y["group"], y["index"])
            if e is not None and not holding_fits(e, y["held"]):
                return "refuse reason=holding_invalid"
    ctx = {"kept": kept, "rlabels": rlabels, "schedules": schedules, "loss": loss, "h": h, "symbols": symbols}
    choices = []
    for w in wanted_list(case["wanted"]):
        entry = intents.get(named(w))
        if entry is None:
            choices.append("reject reason=deadline_not_stated")
            continue
        s = entry["from_ms"] if t == "from_ms" else t
        primary = [arrival(src, w["group"], w["index"], s, ctx) for src in sources]
        ladder = [[arrival(src, r["group"], r["leaf_index"], s, ctx) for src in sources] for r in rungs(entry)]
        choices.append(choose(entry, primary, ladder))
    return {"claims": labels, "records": rlabels, "choices": choices}


# ------------------------------------------------------------------ §7.5: running a plan

def run_inputs(case):
    """§7.5.1's checks, in their order: the reason the run is refused, or None."""
    plans = case["plans"]
    if not plans and not drops("run_plans_none"):
        return "plans_invalid"
    if any(p["at_ms"] == "from_ms" for p in plans) and not drops("run_plans_from_ms"):
        return "plans_invalid"
    # §7.5.1 check 1: one that is not an integer reads as above every other, and §7.1 refuses it.
    instants = [p["at_ms"] if type(p["at_ms"]) is int and p["at_ms"] >= 0 else float("inf") for p in plans]
    if not drops("run_plans_order") and any(a >= b for a, b in zip(instants, instants[1:])):
        return "plans_invalid"
    if drops("run_wanted_before_twice") and not all(wanted_formed(p["wanted"]) for p in plans):
        return "wanted_invalid"
    if not drops("run_plans_twice"):
        for p in plans:
            wanted = wanted_list(p["wanted"])
            # An entry §7.1 refuses names no object: that is check 2's, not this one's.
            objects = [json.dumps(w, sort_keys=True) for w in wanted] if drops("run_wanted_malformed_twice") \
                else [named(w) for w in wanted if wanted_formed([w])]
            if len(set(objects)) != len(objects):
                return "plans_invalid"
    for p in plans:
        if drops("run_wanted_before_sources") and not wanted_formed(p["wanted"]):
            return "wanted_invalid"
        bad = plan_inputs(case["loss_ppm"], case["overhead"], p["at_ms"], p["sources"])
        if bad:
            return bad
        if not drops("run_wanted_late") and not wanted_formed(p["wanted"]):
            return "wanted_invalid"
    arrivals = case["arrivals"]
    if not drops("run_arrivals_last") and any(not CV.is_int(a["at_ms"]) or a["at_ms"] > LAST_INSTANT for a in arrivals):
        return "arrivals_invalid"
    if not drops("run_arrivals_order") and any(a["at_ms"] > b["at_ms"] for a, b in zip(arrivals, arrivals[1:])):
        return "arrivals_invalid"
    if not drops("run_arrivals_twice") and len({(a["group"], a["index"]) for a in arrivals}) != len(arrivals):
        return "arrivals_invalid"
    def formed(kind, items, key, seconds):
        if key not in case or (case[key] is None and drops(f"run_{kind}_null")):
            return True
        return dated_formed(kind, items, case[key], seconds)

    checks = [("claims_invalid", formed("claims", case["claims"], "claims_at_ms", claim_seconds)),
              ("records_invalid", formed("records", case["records"], "records_at_ms", record_seconds)),
              ("groups_invalid", formed("groups", [json.dumps(g["group"], sort_keys=True) for g in case["groups"]],
                                        "groups_at_ms", group_seconds)
               and ("receipts_at_ms" not in case or receipts_formed(case["groups"], case["receipts_at_ms"]))),
              ("symbols_invalid", symbols_trace_formed(case.get("symbols", [])))]
    # Checks 4 to 7, and each weakening that swaps two neighbours of them.
    for k, name in enumerate(("run_dated_order", "run_order_records", "run_order_groups")):
        if drops(name):
            checks[k], checks[k + 1] = checks[k + 1], checks[k]
    for reason, ok in checks:
        if not ok:
            return reason
    return None


def group_seconds(key):
    """§7.5.1 check 6: the whole seconds jwp-receipts §2.4.3 reads from the `group_start` of a group that
    passes jwp-receipts §3 steps 1 and 2, or None: one that fails them starts at no instant it can be
    held to."""
    item = json.loads(key)
    try:
        r = JV.read_group(item)
        if not drops("run_groups_unverified"):
            JV.verify_group(r)
        return JV.seconds(r["group_start"])
    except JV.Refuse:
        return None


def receipts_formed(groups, at):
    """§7.5.1 check 6: a list for each group, an entry for each of its object receipts, none past the
    last instant."""
    if not isinstance(at, list):
        return drops("run_receipts_not_list")
    if len(at) != len(groups) and not drops("run_receipts_groups"):
        return False
    for g, a in zip(groups, at):
        if not isinstance(a, list):
            if drops("run_receipts_member_not_list"):
                continue
            return False
        if len(a) != len(g["objects"]) and not drops("run_receipts_objects"):
            return False
        if not drops("run_receipts_last") and any(t is not None and (not CV.is_int(t) or t > LAST_INSTANT) for t in a):
            return False
    return True


def symbols_trace_formed(trace):
    """§7.5.1 check 7: each instant an integer to the last instant, none below the one before it; each
    statement as §7.1 forms it; of one object on one instance, each statement at a later instant than
    the one before it, of as many source blocks, and with no number below that one's."""
    if not isinstance(trace, list):
        return drops("run_symbols_list") or (trace is None and drops("run_symbols_null"))
    last, previous = {}, 0
    for s in trace:
        if not isinstance(s, dict):
            if drops("run_symbols_statement"):
                continue
            return False
        at, held = s.get("at_ms"), s.get("held")
        if not drops("run_symbols_last") and (not CV.is_int(at) or at > LAST_INSTANT):
            return False
        if at < previous and not drops("run_symbols_order"):
            return False
        previous = at
        if not statement_bearer(s, "run_symbols") or not hash_and_position(s, "run_symbols"):
            return False
        if not statement_numbers(held, "run_symbols"):
            return False
        key = (s.get("bearer_id"), s.get("group"), s.get("index"))
        if key in last:
            was_at, was = last[key]
            if was_at == at and not drops("run_symbols_twice"):
                return False
            if len(was) != len(held) and not drops("run_symbols_blocks"):
                return False
            if any(b < a for a, b in zip(was, held)) and not drops("run_symbols_fall"):
                return False
        last[key] = (at, held)
    return True


def claim_seconds(wire):
    """§7.5.1 check 4: the whole seconds jwp-receipts §2.4.3 reads from the `issued_at` of a claim that
    passes steps 1 to 4, or None: one that fails them is not known to be its signer's."""
    try:
        return JV.seconds(claim_steps(wire)["issued_at"])
    except CV.Reject:
        return JV.seconds(json.loads(wire)["issued_at"]) if drops("run_claims_unverified") else None


def record_seconds(hexed):
    """§7.5.1 check 5: the whole seconds of a record's `as_of_ms`, where holdings §4.4 reads and
    verifies it, or None."""
    try:
        return read_record(bytes.fromhex(hexed))[1] // 1000
    except HV.Refused:
        return None


def dated_formed(kind, items, at, seconds):
    """§7.5.1 checks 4 to 6: one entry for each; none past the last instant; no statement given twice;
    none held in a whole second before its own."""
    if not isinstance(at, list):
        return drops(f"run_{kind}_not_list")
    if len(at) != len(items) and not drops(f"run_{kind}_length"):
        return False
    if not drops(f"run_{kind}_last") and any(a is not None and (not CV.is_int(a) or a > LAST_INSTANT) for a in at):
        return False
    if not drops(f"run_{kind}_twice") and len(set(items)) != len(items):
        return False
    if drops(f"run_{kind}_signed"):
        return True
    for item, a in zip(items, at):
        since = seconds(item)
        if a is None or since is None:
            continue
        # A statement's own second admits an instant anywhere in it.
        if (a < since * 1000 + 999 if drops(f"run_{kind}_signed_ms") else a // 1000 < since):
            return False
    return True


def read_by(at, i, t, kind):
    """§7.5.1: what is held by t: its instant is at most t, or it has none."""
    if not isinstance(at, list) or i >= len(at) or at[i] is None or drops(f"run_{kind}_undated"):
        return True
    return at[i] < t if drops(f"run_{kind}_strict") else at[i] <= t


def read_at(case, t):
    """§7.5.1: a run's inputs for its plan at t: the claims and records held by then, in the order given,
    and their positions."""
    claims = [i for i, _ in enumerate(case["claims"]) if read_by(case.get("claims_at_ms"), i, t, "claims")]
    records = [i for i, _ in enumerate(case["records"]) if read_by(case.get("records_at_ms"), i, t, "records")]
    return claims, records


def envelope_held(case, g, o, t):
    """§7.5.1: an envelope is held from its object's arrival; one whose object no arrival names, from
    before the first plan. A group or receipt that does not read names no object."""
    if drops("run_envelopes_undated"):
        return True
    try:
        link, index = JV.link(JV.read_group(g["group"])).hex(), JV.read_object(o["receipt"])["object_id"]
    except JV.Refuse:
        return True
    a = next((a for a in case["arrivals"] if a["group"] == link and a["index"] == index), None)
    if a is None:
        return not drops("run_envelopes_no_arrival")
    return a["at_ms"] < t if drops("run_envelopes_strict") else a["at_ms"] <= t


def groups_read(case, t):
    """§7.5.1: the groups a plan at t reads. A run that dates none reads each whole, envelopes and all;
    one that dates either reads what the sink held by t."""
    at_groups, at_receipts = case.get("groups_at_ms"), case.get("receipts_at_ms")
    if at_groups is None and at_receipts is None and not drops("run_envelopes_opt_in"):
        return case["groups"]
    out = []
    for i, g in enumerate(case["groups"]):
        if not read_by(at_groups, i, t, "groups"):
            continue
        at = at_receipts[i] if isinstance(at_receipts, list) and i < len(at_receipts) else None
        objects = [dict(o, envelope=o["envelope"] if envelope_held(case, g, o, t) else None)
                   for k, o in enumerate(g["objects"]) if read_by(at, k, t, "receipts")]
        out.append({"group": g["group"], "objects": objects})
    return out


def symbols_read(case, t):
    """§7.5.1: what the sink holds at t: of each object on each instance, the last statement by t, as
    §7.1 gives it."""
    read = {}
    trace = case.get("symbols", [])
    for s in [s for s in trace if isinstance(s, dict)] if isinstance(trace, list) else []:
        if not (drops("run_symbols_undated") or (s["at_ms"] < t if drops("run_symbols_strict") else s["at_ms"] <= t)):
            continue
        key = (s.get("bearer_id"), s.get("group"), s.get("index"))
        if key not in read or not drops("run_symbols_latest"):
            read[key] = {k: s[k] for k in ("bearer_id", "group", "index", "held") if k in s}
    return list(read.values())


def not_held(n, read, labels):
    """§7.5.3: each of n statements' labels, `not-held` for one the plan did not read."""
    out = ["not-held"] * n
    for i, label in zip(read, labels):
        out[i] = label
    return out


def same_source(a, b):
    """§7.5.2: both name one bearer instance or one holder's key, or both are the origin, whatever
    their `fetch_ms`."""
    if a["kind"] != b["kind"]:
        return False
    if drops("run_same_fetch_ms") and a.get("fetch_ms") != b.get("fetch_ms"):
        return False
    return a.get("bearer_id") == b.get("bearer_id") and a.get("holder") == b.get("holder")


class Fetch:
    """§7.5.2: the object a fetch takes, its source, and the plan it serves."""

    def __init__(self, obj, rung, takes, plan, index, source):
        self.obj, self.rung, self.takes, self.plan, self.index, self.source = obj, rung, takes, plan, index, source


def takes(key, entry, choice):
    """The object a choice takes: the primary, or the rung it names."""
    if choice["object"] == "primary":
        return key
    r = rungs(entry)[choice["rung"]]
    return (r["group"], r["leaf_index"])


def run(case):
    """§7.5: the run of a sink's plans against its arrivals, or `refuse reason=R`."""
    bad = run_inputs(case)
    if bad:
        return "refuse reason=" + bad
    # Each plan derived by §7.2 and §7.3 from what the sink held by its instant, with its own
    # sources and wanted objects; a group that fails refuses the run as it refuses a plan.
    plans, read, intents_of = [], [], []
    dated = ("claims_at_ms" in case or "records_at_ms" in case
             or ("groups_at_ms" in case and not drops("run_groups_read"))
             or ("receipts_at_ms" in case and not drops("run_receipts_read")))
    unchained = dated and drops("run_claims_chain")
    for p in case["plans"]:
        cs, rs = read_at(case, p["at_ms"])
        claims, records = [case["claims"][i] for i in cs], [case["records"][i] for i in rs]
        held_groups = groups_read(case, p["at_ms"])
        derived = plan(dict(case, at_ms=p["at_ms"], sources=p["sources"], wanted=p["wanted"], claims=claims,
                            records=records, groups=held_groups, symbols=symbols_read(case, p["at_ms"]),
                            unchained=unchained))
        if isinstance(derived, str):
            return derived
        plans.append(derived["choices"])
        read.append({"claims": not_held(len(case["claims"]), cs, derived["claims"]),
                     "records": not_held(len(case["records"]), rs, derived["records"])})
        groups = read_groups(held_groups)
        _, accepted = claims_step(claims, groups, plan=True, chained=not unchained)
        intents_of.append({(link, i): entry for link, i, entry in intents_step(accepted, groups)})

    def objects_run(k):
        """§7.5.2: what plan k wants whose choice is not refused, in its wanted order, each with its
        choice and the entry that plan read."""
        return [((w["group"], w["index"]), c, intents_of[k][(w["group"], w["index"])])
                for w, c in zip(wanted_list(case["plans"][k]["wanted"]), plans[k]) if isinstance(c, dict)]

    instants = {p["at_ms"] for p in case["plans"]} | {a["at_ms"] for a in case["arrivals"]}
    for k in range(len(plans)):
        for _, _, e in objects_run(k):
            instants |= {e["by_ms"], end(e)}

    held, settled, passed, fetches, force, out = {}, set(), set(), {}, None, []

    def decide(n, k, action, key, obj=None, rung=None, source=None, v=None):
        out.append({"at_ms": n, "plan": k, "action": action, "group": key[0], "index": key[1],
                    "object": obj, "rung": rung, "source": source, "value_ppm": v})

    def abandon(n, key):
        f = fetches.pop(key)
        decide(n, f.plan, "abandon", key, f.obj, f.rung, f.index)

    for n in sorted(instants):
        # Step 1.
        for a in case["arrivals"]:
            if a["at_ms"] == n:
                arrived = (a["group"], a["index"])
                held[arrived] = n
                if not drops("run_arrival_completes"):
                    for key in [key for key, f in fetches.items() if f.takes == arrived]:
                        del fetches[key]
        # Step 2.
        new = next((k for k, p in enumerate(case["plans"]) if p["at_ms"] == n), None)
        if new is not None:
            sources = case["plans"][new]["sources"]
            choices = {key: (c, e) for key, c, e in objects_run(new)}

            def continues(key, f):
                c, e = choices.get(key, (None, None))
                if c is None or c["object"] is None or drops("run_continue"):
                    return False
                return takes(key, e, c) == f.takes and same_source(sources[c["source"]], f.source)

            if force is not None:
                for key, _, _ in objects_run(force):
                    if key not in settled and key in fetches and not continues(key, fetches[key]):
                        if drops("run_drop_abandon") and key not in choices:
                            continue
                        abandon(n, key)
            force = new
            for key, c, e in objects_run(new):
                if (key in settled and not drops("run_settled_refetch")) or c["object"] is None:
                    continue
                if drops("run_symbols_hold") and c["arrival_ms"] == n and sources[c["source"]]["kind"] == "bearer":
                    # §7.5.1: symbols are not an arrival, however many are stated.
                    held[takes(key, e, c)] = n
                if takes(key, e, c) in held and not drops("run_held_fetch"):
                    continue
                f = fetches.get(key)
                if f is not None and continues(key, f):
                    if not drops("run_continued_plan"):
                        f.plan, f.index = new, c["source"]
                    continue
                fetches[key] = Fetch(c["object"], c["rung"], takes(key, e, c), new, c["source"], sources[c["source"]])
                decide(n, new, "fetch", key, c["object"], c["rung"], c["source"])
        if force is None:
            continue
        current = objects_run(force)
        if drops("run_passed_keeps_running"):
            # §7.5.4: an object a later plan has no intent for is no longer run, passed or not.
            run_now = {key for key, _, _ in current}
            current = current + [x for k in range(force) for x in objects_run(k)
                                 if x[0] in passed and x[0] not in run_now]
        # Step 3, by what the sink holds by `by_ms`.
        for key, _, e in current:
            if key in settled or key in passed or e["by_ms"] > n:
                continue
            limit = n if drops("run_by_deadline") else e["by_ms"]
            holds = [(r, f) for r, f in enumerate(rungs(e))
                     if held.get((f["group"], f["leaf_index"]), limit + 1) <= limit]
            if held.get(key, limit + 1) <= limit:
                settled.add(key)
                decide(n, force, "present", key, "primary", v=FULL)
            elif holds:
                r, f = holds[-1] if drops("run_rung_order") else holds[0]
                settled.add(key)
                decide(n, force, "present", key, "fallback", r, v=f["value_ppm"])
            else:
                passed.add(key)
                decide(n, force, "passed", key)
            if key in fetches:
                after_present = key in settled and not drops("run_present_abandon")
                after_passed = key in passed and fetches[key].obj == "fallback" and not drops("run_passed_abandon")
                if after_present or after_passed:
                    abandon(n, key)
        # Step 4, at the primary's value at its arrival.
        for key, _, e in current:
            if key in passed and key not in settled and key in held:
                v = value(e, n if drops("run_late_value") else held[key])
                if v >= 1 or drops("run_late_zero"):
                    settled.add(key)
                    decide(n, force, "present", key, "primary", v=v)
        # Step 5.
        for key, _, e in current:
            if key in passed and key not in settled and end(e) <= n:
                if key in fetches and not drops("run_end_abandon"):
                    abandon(n, key)
                settled.add(key)
                if not drops("run_ended"):
                    decide(n, force, "ended", key)
    return {"plans": plans, "read": read, "decisions": out} if dated else {"plans": plans, "decisions": out}


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

def claim_outcome(c):
    try:
        claim_steps(c["claim"])
        return "accept"
    except CV.Reject as r:
        return f"reject step={r.step} reason={r.reason}"


def value_outcome(c):
    """§5 for an entry read alone, as §3.1 steps 2 and 4 read an entry."""
    e = c["entry"]
    CV.timeliness_entry_body(e)
    CV.timeliness_entry_rules(e, None)
    return {"value_ppm": value(e, c["at_ms"]), "end_ms": end(e)}


RUNS = (("claims", claim_outcome), ("values", value_outcome), ("verdicts", verdict), ("plans", plan), ("runs", run))
KINDS = tuple(k for k, _ in RUNS)


def strings(v, out):
    if isinstance(v, str):
        out.append(v)
    elif isinstance(v, list):
        for x in v:
            strings(x, out)
    elif isinstance(v, dict):
        for x in v.values():
            strings(x, out)


def check(doc, quiet=False):
    """The failures of a corpus document: each case's outcome, every reason of §9, every outcome of
    §6.3 and every action of §7.5.3."""
    fails = []
    for kind, run in RUNS:
        for c in doc[kind]:
            try:
                got = run(c)
            except CV.Reject as r:
                got = {"error": f"reject step={r.step} reason={r.reason}"}
            except (KeyError, TypeError, ValueError, IndexError, AttributeError, StopIteration,
                    ZeroDivisionError) as e:
                got = {"error": repr(e)}
            if got != c["outcome"]:
                fails.append(c["name"])
                if not quiet:
                    print(f"FAIL {c['name']}\n  stated {json.dumps(c['outcome'])}\n  got    {json.dumps(got)}")
    stated = []
    for kind in KINDS:
        for c in doc[kind]:
            strings(c["outcome"], stated)
    for code, _ in reasons_table():
        if not any(s == code or s.endswith("reason=" + code) for s in stated):
            fails.append("no case gives " + code)
    for outcome in outcomes_table():
        if outcome not in stated:
            fails.append("no case gives the outcome " + outcome)
    for action in ACTIONS:
        if action not in stated:
            fails.append("no case gives the action " + action)
    return fails


def mutations(doc):
    """Copies of the corpus, one defect each, every one of which must fail."""
    def case(d, name):
        return next(c for k in KINDS for c in d[k] if c["name"] == name)

    def m_instant(d):
        # A millisecond before the end is 75; at the end, 0.
        case(d, "example_last_ms")["at_ms"] += 1

    def m_end(d):
        case(d, "hard_one_ms")["outcome"]["end_ms"] -= 1

    def m_floor(d):
        case(d, "step_mid")["outcome"]["value_ppm"] += 1

    def m_accept(d):
        case(d, "decay_near_collinear")["outcome"] = "reject step=4 reason=timeliness_decay_invalid"

    def m_line(d):
        case(d, "holder_one_ms_after")["outcome"]["intents"][0]["lines"][0]["value_ppm"] = 999_801

    def m_earliest(d):
        c = case(d, "earliest_record_counts")
        c["records"] = c["records"][1:]

    def m_statement(d):
        case(d, "statement_line")["statement"]["holder"] = d["keys"]["holder_c"]

    def m_envelope(d):
        c = case(d, "on_layer")
        c["groups"][0]["objects"][3]["envelope"] = None

    def m_order(d):
        c = case(d, "order_over_earliest")
        c["sources"] = list(reversed(c["sources"]))

    def m_deadline(d):
        case(d, "on_time_at_deadline")["at_ms"] += 1

    def m_fetch(d):
        # Both late sources at 300,000: the first in order, the origin.
        case(d, "late_choice_by_value")["sources"][1]["fetch_ms"] = 66_000

    def m_loss(d):
        # 70%: 3 symbols a pass of the 12 needed, a fourth pass.
        case(d, "every_pass_from_first")["loss_ppm"] = 700_000

    def m_from_ms(d):
        case(d, "publisher_check_fails")["at_ms"] = d["plans"][0]["at_ms"]

    def m_object(d):
        case(d, "fallback_from_carousel")["outcome"]["choices"][0]["object"] = "primary"

    def m_reason_gone(d):
        d["verdicts"].remove(case(d, "record_before_group"))

    def m_extended(d):
        case(d, "plan_bearer_extended")["outcome"]["claims"][1] = "stands"

    def m_withdrawn(d):
        # The withdrawn pass's schedule given first and last alike: one body, which stands.
        c = case(d, "plan_bearer_withdrawn")
        c["claims"][2] = c["claims"][1]

    def m_rung_extended(d):
        # The rung above the one the extended schedule reaches.
        case(d, "ladder_rung_on_extended_schedule")["outcome"]["choices"][0]["rung"] = 0

    def m_rung_value(d):
        # The lower rung presented at the higher's value.
        case(d, "ladder_lower_rung_alone")["outcome"]["intents"][0]["lines"][0]["value_ppm"] = 700_000

    def m_rung_instant(d):
        # The lower rung a millisecond late: nothing held by the deadline.
        r = case(d, "ladder_lower_rung_alone")
        r["outcome"]["intents"][0]["lines"][0]["fallback_ms"][1] += 1

    def m_rung_form(d):
        # A ladder's line reports its list.
        r = case(d, "ladder_first_rung_replaces")
        r["outcome"]["intents"][0]["lines"][0]["fallback_ms"] = r["outcome"]["intents"][0]["lines"][0]["fallback_ms"][0]

    def m_plan_rung(d):
        case(d, "ladder_lower_rung")["outcome"]["choices"][0]["rung"] = 0

    def m_arrival_late(d):
        # The trace a millisecond later is a late object at another value.
        case(d, "run_late_object")["arrivals"][0]["at_ms"] += 1

    def m_arrival_gone(d):
        case(d, "run_on_time")["arrivals"] = []

    def m_decision_value(d):
        case(d, "run_late_object")["outcome"]["decisions"][-1]["value_ppm"] += 1

    def m_decision_action(d):
        case(d, "run_every_rung_passed_no_primary")["outcome"]["decisions"][1]["action"] = "ended"

    def m_plan_dropped(d):
        c = case(d, "run_replaced_changes_source")
        c["plans"] = c["plans"][:1]

    def m_plan_instant(d):
        case(d, "run_replaced_at_deadline")["plans"][1]["at_ms"] += 1

    def m_decision_order(d):
        c = case(d, "run_in_wanted_order")
        c["outcome"]["decisions"][0], c["outcome"]["decisions"][1] = c["outcome"]["decisions"][1], c["outcome"]["decisions"][0]

    def m_dates_dropped(d):
        # The schedules given undated: every plan reads the extension, and the fetches differ.
        c = case(d, "run_extension_reaches_mid_run")
        del c["claims_at_ms"]

    def m_date_later(d):
        # Past the plan that read it.
        case(d, "run_extension_reaches_mid_run")["claims_at_ms"][3] = T0_PLUS_LATER

    def m_read_label(d):
        case(d, "run_extension_reaches_mid_run")["outcome"]["read"][1]["claims"][3] = "not-held"

    def m_dated_fetch(d):
        # The rung fetched at T0, as the undated run does.
        first = case(d, "run_extension_reaches_mid_run")["outcome"]["decisions"][0]
        first["object"], first["rung"], first["source"] = "fallback", 1, 1

    def m_withdrawn_read(d):
        # The withdrawal never read: every schedule stands in the second plan.
        c = case(d, "run_contradiction_mid_run")
        c["claims_at_ms"][3] = T0_PLUS_LATER

    def m_parent_first(d):
        # The parent held with its child: nothing is refused.
        c = case(d, "run_claim_parent_not_held_yet")
        c["claims_at_ms"][0] = c["claims_at_ms"][1]

    def m_before_signed(d):
        # Within the claim's own second: no refusal.
        case(d, "claims_held_before_signed")["claims_at_ms"][0] += 1

    def m_record_date(d):
        del case(d, "run_record_reaches_mid_run")["records_at_ms"]

    def m_claims_reason_gone(d):
        d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=claims_invalid"]

    def m_records_reason_gone(d):
        d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=records_invalid"]

    def m_reason_gone_run(d):
        d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=arrivals_invalid"]

    def m_symbols_gone(d):
        # Holding nothing, no rung is in reach: the rung the case states is not chosen.
        del case(d, "run_symbols_reach_a_rung")["symbols"]

    def m_symbols_later(d):
        # Past the only plan.
        case(d, "run_symbols_reach_a_rung")["symbols"][0]["at_ms"] = T0_PLUS_LATER

    def m_symbols_fewer(d):
        # 6 of 12 takes two passes, so the rung comes a pass later.
        case(d, "run_symbols_reach_a_rung")["symbols"][0]["held"] = [6]

    def m_symbols_instance(d):
        case(d, "run_symbols_reach_a_rung")["symbols"][0]["bearer_id"] = "urn:example:bearer:b3"

    def m_symbols_arrival_stated(d):
        case(d, "run_symbols_reach_a_rung")["outcome"]["plans"][0][0]["arrival_ms"] += 1

    def m_symbols_fetch_stated(d):
        d_ = case(d, "run_symbols_reach_a_rung")["outcome"]["decisions"]
        d_[:] = d_[1:]

    def m_plan_symbols_gone(d):
        del case(d, "plan_symbols_reach_a_rung")["symbols"]

    def m_symbols_reason_gone(d):
        for k in ("plans", "runs"):
            d[k] = [c for c in d[k] if c["outcome"] != "refuse reason=symbols_invalid"]

    def m_group_date_dropped(d):
        del case(d, "run_group_reaches_mid_run")["groups_at_ms"]

    def m_group_date_later(d):
        case(d, "run_group_reaches_mid_run")["groups_at_ms"][0] = T0_PLUS_LATER

    def m_envelope_read_early(d):
        # With no list the envelope is read throughout, so the schedule is refused from T0.
        del case(d, "run_envelope_read_from_its_arrival")["groups_at_ms"]

    def m_envelope_arrival_gone(d):
        # An envelope whose object no arrival names was held from before the first plan.
        case(d, "run_envelope_read_from_its_arrival")["arrivals"] = []

    def m_receipt_date_dropped(d):
        del case(d, "run_object_receipt_reaches_mid_run")["receipts_at_ms"]

    def m_groups_reason_gone(d):
        d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=groups_invalid"]

    def m_passed_then_presented(d):
        # A primary arriving after an entry is set aside is not presented.
        case(d, "run_entry_contradicted_after_passed")["outcome"]["decisions"].append(
            case(d, "run_late_object")["outcome"]["decisions"][-1])

    def m_contradiction_early(d):
        # Read from before the first plan, the entry is set aside before anything is fetched.
        case(d, "run_entry_contradicted_after_passed")["claims_at_ms"][1] = None

    def m_enough_then_arrives(d):
        # With an arrival the sink holds the object, and presents it at the deadline.
        case(d, "run_symbols_enough_no_arrival")["arrivals"] = [{"group": d["groups"]["F"], "index": 0, "at_ms": 1_793_664_021_000}]

    def restored(name, pointer, value):
        """A copy in which one case written over holds its well-formed member: its refusal goes."""
        def m(d):
            target = case(d, name)
            for key in pointer[:-1]:
                target = target[key]
            target[pointer[-1]] = value
        m.__name__ = "m_restored_" + name
        return m

    bearer_b2 = "urn:example:bearer:b2"

    def statement(d):
        return [{"bearer_id": bearer_b2, "group": d["groups"]["F"], "index": 0, "at_ms": 1_793_664_020_000, "held": [10]}]

    def symbols_restored(name):
        def m(d):
            case(d, name)["symbols"] = statement(d)
        m.__name__ = "m_restored_" + name
        return m

    def plan_symbols_restored(name):
        def m(d):
            case(d, name)["symbols"] = [{k: v for k, v in statement(d)[0].items() if k != "at_ms"}]
        m.__name__ = "m_restored_" + name
        return m

    def wanted_restored(name, pointer, as_list=False):
        """A copy in which one case written over names a wanted object again."""
        def m(d):
            target = case(d, name)
            for key in pointer[:-1]:
                target = target[key]
            formed = {"group": d["groups"]["G"], "index": 0}
            target[pointer[-1]] = [formed] if as_list else formed
        m.__name__ = "m_restored_" + name
        return m

    def statement_restored(name, instant):
        """A copy in which one case written over holds a formed statement again."""
        def m(d):
            formed = {"bearer_id": "urn:example:bearer:b3", "group": d["groups"]["F"], "index": 0, "held": [10]}
            case(d, name)["symbols"][0] = dict(formed, at_ms=1_793_664_020_000) if instant else formed
        m.__name__ = "m_restored_" + name
        return m

    def m_wanted_reason_gone(d):
        for k in ("plans", "runs"):
            d[k] = [c for c in d[k] if c["outcome"] != "refuse reason=wanted_invalid"]

    def m_wanted_twice_reason(d):
        # One object wanted twice stays the plans' reason.
        case(d, "run_wanted_twice_and_malformed")["outcome"] = "refuse reason=wanted_invalid"

    def m_not_an_instant_restored(d):
        for name, key in (("claims_at_not_an_instant", "claims_at_ms"), ("records_at_not_an_instant", "records_at_ms"),
                          ("groups_at_not_an_instant", "groups_at_ms")):
            case(d, name)[key][0] = 1_793_664_002_000

    return [m_instant, m_end, m_floor, m_accept, m_line, m_earliest, m_statement, m_envelope, m_order, m_deadline,
            m_fetch, m_loss, m_from_ms, m_object, m_reason_gone, m_extended, m_withdrawn, m_rung_extended, m_rung_value,
            m_rung_instant, m_rung_form, m_plan_rung,
            m_arrival_late, m_arrival_gone, m_decision_value, m_decision_action, m_plan_dropped, m_plan_instant,
            m_decision_order, m_reason_gone_run, m_dates_dropped, m_date_later, m_read_label, m_dated_fetch,
            m_withdrawn_read, m_parent_first, m_before_signed, m_record_date, m_claims_reason_gone,
            m_records_reason_gone, m_symbols_gone, m_symbols_later, m_symbols_fewer, m_symbols_instance,
            m_symbols_arrival_stated, m_symbols_fetch_stated, m_plan_symbols_gone, m_symbols_reason_gone,
            m_group_date_dropped, m_group_date_later, m_envelope_read_early, m_envelope_arrival_gone,
            m_receipt_date_dropped, m_groups_reason_gone, m_passed_then_presented, m_contradiction_early,
            m_not_an_instant_restored, m_enough_then_arrives,
            restored("claims_at_not_a_list", ["claims_at_ms"], [None]),
            restored("claims_at_null", ["claims_at_ms"], [None]),
            restored("records_at_not_a_list", ["records_at_ms"], [None]),
            restored("groups_at_not_a_list", ["groups_at_ms"], [None]),
            restored("receipts_at_not_a_list", ["receipts_at_ms"], [[None] * 4]),
            restored("receipts_at_one_not_a_list", ["receipts_at_ms", 0], [None] * 4),
            restored("symbols_held_a_string", ["symbols", 0, "held"], [10]),
            restored("symbols_held_a_fraction", ["symbols", 0, "held"], [10]),
            restored("symbols_held_not_a_list", ["symbols", 0, "held"], [10]),
            symbols_restored("symbols_not_a_list"),
            symbols_restored("symbols_null"),
            restored("plan_symbols_held_a_string", ["symbols", 0, "held"], [10]),
            restored("plan_symbols_held_a_fraction", ["symbols", 0, "held"], [10]),
            restored("plan_symbols_held_not_a_list", ["symbols", 0, "held"], [10]),
            plan_symbols_restored("plan_symbols_not_a_list"),
            statement_restored("plan_symbols_statement_not_an_object", False),
            statement_restored("plan_symbols_bearer_id_not_a_string", False),
            statement_restored("plan_symbols_group_uppercase", False),
            statement_restored("plan_symbols_index_a_string", False),
            statement_restored("symbols_statement_not_an_object", True),
            statement_restored("symbols_bearer_id_not_a_string", True),
            statement_restored("symbols_group_uppercase", True),
            statement_restored("symbols_index_a_string", True),
            wanted_restored("plan_wanted_not_an_object", ["wanted", 0]),
            wanted_restored("plan_wanted_group_uppercase", ["wanted", 0]),
            wanted_restored("plan_wanted_index_a_string", ["wanted", 0]),
            wanted_restored("plan_wanted_not_a_list", ["wanted"], as_list=True),
            wanted_restored("wanted_not_an_object", ["plans", 0, "wanted", 0]),
            wanted_restored("wanted_group_uppercase", ["plans", 0, "wanted", 0]),
            wanted_restored("wanted_index_a_string", ["plans", 0, "wanted", 0]),
            wanted_restored("wanted_not_a_list", ["plans", 0, "wanted"], as_list=True),
            m_wanted_reason_gone, m_wanted_twice_reason]


WEAKENINGS = ["subject", "signer", "range", "before", "same_second", "content", "layer", "fallback_position", "derived",
              "context", "intent_contradiction", "leaf_count", "held_before", "record_contradiction", "line_holders",
              "least", "least_fallback", "own_statements", "key_order", "deadline_inclusive", "fallback_deadline",
              "sink_rule", "floor", "no_intent", "loss_bound",
              "overhead", "instant", "sources_empty", "sources_kind", "sources_uri", "sources_key", "sources_fetch",
              "sources_twice", "sources_loss", "per_group", "schedule_contradicted", "schedule_last", "undeclared",
              "bidirectional", "record_at_s",
              "holder_record", "plan_order", "fallback_value", "fallback_tie", "fallback_order",
              "plan_fallback_deadline", "late_zero", "late_tie",
              "rung_layer", "rung_position", "rung_derived", "rung_holders", "rung_order",
              "plan_rung_first", "plan_rung_order",
              "run_plans_none", "run_plans_from_ms", "run_plans_order", "run_plans_twice", "run_arrivals_last",
              "run_arrivals_order", "run_arrivals_twice", "run_same_fetch_ms", "run_arrival_completes", "run_continue",
              "run_drop_abandon", "run_settled_refetch", "run_held_fetch", "run_continued_plan", "run_by_deadline",
              "run_rung_order", "run_present_abandon", "run_passed_abandon", "run_late_value", "run_late_zero",
              "run_end_abandon", "run_ended",
              "run_claims_undated", "run_claims_strict", "run_claims_length", "run_claims_last", "run_claims_twice",
              "run_claims_signed", "run_claims_signed_ms", "run_claims_unverified", "run_claims_chain",
              "run_records_undated", "run_records_strict", "run_records_length", "run_records_last",
              "run_records_twice", "run_records_signed", "run_records_signed_ms", "run_dated_order",
              "symbols_uri", "symbols_empty", "symbols_number", "symbols_twice", "symbols_holding", "symbols_ignored",
              "symbols_instance", "run_symbols_last", "run_symbols_order", "run_symbols_uri", "run_symbols_empty",
              "run_symbols_number", "run_symbols_twice", "run_symbols_blocks", "run_symbols_fall",
              "run_symbols_undated", "run_symbols_strict", "run_symbols_latest",
              "run_groups_undated", "run_groups_strict", "run_groups_length", "run_groups_last", "run_groups_twice",
              "run_groups_signed", "run_groups_signed_ms", "run_groups_unverified", "run_groups_read",
              "run_receipts_undated", "run_receipts_strict", "run_receipts_groups", "run_receipts_objects",
              "run_receipts_last", "run_receipts_read", "run_envelopes_undated", "run_envelopes_strict",
              "symbols_statement", "symbols_bearer_string", "symbols_group", "symbols_index", "symbols_greatest",
              "run_symbols_statement", "run_symbols_bearer_string", "run_symbols_group", "run_symbols_index",
              "run_symbols_greatest", "wanted_list", "wanted_object", "wanted_group", "wanted_index", "wanted_greatest",
              "wanted_before_sources", "wanted_before_symbols", "run_wanted_before_twice", "run_wanted_malformed_twice",
              "run_wanted_before_sources", "run_wanted_late",
              "symbols_list", "symbols_held_list", "symbols_integer", "run_symbols_list", "run_symbols_null",
              "run_symbols_held_list", "run_symbols_integer", "run_symbols_hold", "run_claims_not_list", "run_claims_null",
              "run_records_not_list", "run_groups_not_list", "run_receipts_not_list", "run_receipts_member_not_list",
              "run_envelopes_opt_in", "run_envelopes_no_arrival", "run_order_records", "run_order_groups", "run_passed_keeps_running"]


def direct_checks():
    """What no vector carries, and what this text states of the texts beside it."""
    fails = []
    # §7.1: the greatest fetch estimate is admitted; the corpus carries only the one past it.
    if not sources_valid([{"kind": "origin", "fetch_ms": MAX_INT}], None, None):
        fails.append("a fetch_ms of 2^53 - 1 is refused")
    # §4.2 rule 3.2 where the products pass 2^64: three points on one line are refused, and a
    # part per million off it is not. A float of 64 bits cannot tell them apart.
    half = (LAST_INSTANT - 1) // 2
    for v1, ok in ((500_000, False), (500_001, True)):
        e = {"leaf_index": 0, "content_hash": "00" * 32, "from_ms": 0, "by_ms": 0, "fallback": None,
             "decay": [{"at_ms": half, "value_ppm": v1}, {"at_ms": 2 * half, "value_ppm": 0}]}
        try:
            CV.timeliness_entry_rules(e, None)
            got = True
        except CV.Reject:
            got = False
        if got != ok:
            fails.append(f"a point of {v1} at {half} is {'refused' if ok else 'accepted'}")
    # §7.5.3 states each action a run decides.
    decisions = text_of(EXTENSION)
    decisions = decisions[decisions.index("\n#### 7.5.3 Decisions"):decisions.index("\n#### 7.5.4 ")]
    for action in ACTIONS:
        if f"`{action}`" not in decisions:
            fails.append(f"§7.5.3 does not state {action}")
    # jwp-receipts §3.1 lists §4.2's codes at step 4 and §4.3's at step 6; delivery-telemetry §5.1
    # lists §8.6's switch reason.
    jwp = text_of(JWP_RECEIPTS)
    for code, where in reasons_table():
        step = {"§4.2": 4, "§4.3": 6}.get(where.split(" ")[0])
        if step is not None and f"| {step} | `{code}` |" not in jwp:
            fails.append(f"jwp-receipts §3.1 does not list {code} at step {step}")
    if "- `not-held-by-deadline`:" not in text_of(TELEMETRY):
        fails.append("delivery-telemetry §5.1 does not list not-held-by-deadline")
    if REFINEMENT_ONLY != {"wai.image.int_refine", "wai.video.int_refine", "wai.audio.int_refine",
                           "wai.splat.int_refine"}:
        fails.append(f"SPEC §5's refinement capabilities read as {sorted(REFINEMENT_ONLY)}")
    return fails


def main(argv):
    global DROPPED
    if len(argv) != 2:
        print(__doc__)
        return 2
    doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8"))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    fails = check(doc)
    if extra:
        fails.append(f"files the corpus does not hold: {extra}")
    fails += direct_checks()
    print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): "
          f"{len(fails)} failure(s)")
    caught = 0
    for m in mutations(doc):
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append(f"mutation {m.__name__} was not caught")
    print(f"{caught} of {len(mutations(doc))} mutated copies fail")
    weak = 0
    for w in WEAKENINGS:
        DROPPED = w
        if check(doc, quiet=True):
            weak += 1
        else:
            fails.append(f"a verifier without its {w} check passes the corpus")
    DROPPED = None
    print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught")
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


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