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

Written from the extension's text and the texts it cites, alone. It shares no code with the
reference implementation in Rust, which generates the corpus, and never runs it, nor the
generator. 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 computes BLAKE3 with ../claim-conformance/_blake3.py;
SHA-256 (FIPS 180-4) is the standard library's. Everything else is here:

- the room descriptor (live-rooms §3.1, §3.2): its reader (§3.4 steps 1 to 3), its digest,
  the room id, and its signature under its `authority`;
- the room grants (live-rooms §4.1, §4.2, §4.4): each grant's reader (§4.3 step 1), its
  signature under the descriptor's authority (step 2), its room (step 4) and its
  attributes (the `attrs` rule of step 5). The other checks of live-rooms §4.3 (step 3's
  exchange linkage, step 5's other field rules, steps 6 to 8) and the descriptor's field
  rules (§3.4, `descriptor_invalid`) are the live-rooms verifier's, live_rooms_verify.py,
  with its own corpus, and are not repeated here;
- the world statement (live-worlds §3): its reader and rules, in §3.4's order;
- the world tracks (live-worlds §4, §9.2): the grant attribute `world` (§5.1), the group
  sequence, the receipts track's object (jwp-receipts §2.7, binary form `WJR1`) and its
  verification by jwp-receipts §3 (key, group signature over the legacy or the labelled
  payload of §2.3, the chain kept per track, the content hash of §2.1 over the bytes the
  receipt covers, the class label, the proof's position and the Merkle walk of §2.2, and
  the group total), the op batch (interactive-worlds §5.1), each op's signature, the
  actor, the action body of `wai.world.model`, redelivery, equivocation and the Lamport
  rule (§4.2);
- private worlds (live-worlds §8, §9.1, §9.2): the receipts cover `payload`, the ops are
  the batch `opened` from it, and a track's first object may be at any group; above group
  0 it starts its chain whatever parent it names (jwp-receipts §3 step 3, *Joining*), and
  at group 0 it starts it as any first group does. Opening is confidentiality §6's, so the
  relation between `payload` and `opened` is not checked here;
- the session hash (live-worlds §9.3) over the admitted ops in canonical order
  (interactive-worlds §5.2), and the outcome (§9.4).

Of the acquisition classes (energy-measurement §2.7) it reads every tag, so that a class
that does not read is `malformed` as jwp-receipts §2.7.4 says; it holds tags 0 to 3 to
jwp-receipts §2.4.1 and §2.4.2. A refined hardware class (tags 4 and 5) needs the
resolution floor of energy-measurement §2.3, which this verifier does not implement: it
refuses such a receipt as `class_not_supported_here`, a verifier-local outcome that no
conforming corpus vector states.

For every vector it requires the vector's `expect`: for an accepted world its session
hash, the number of admitted ops and each actor's count; for a refused one its reason and
where (`"descriptor"`, `"statement"`, `{"grant": i}` or `{"track": i, "group": g}`).

    python3 -I live_worlds_verify.py <live-conformance/worlds>

Exit status 0 only if every vector agrees. Python 3.8 or later; standard library only.
"""

import hashlib
import json
import os
import sys

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

# live-rooms §3.1, §3.2, §4.1, §4.2
DESCRIPTOR_MAGIC = b"WRD1"
GRANT_MAGIC = b"WRG1"
DESCRIPTOR_DOMAIN = b"wai:live-room\x01"           # 14 bytes
GRANT_DOMAIN = b"wai:live-room-grant\x01"          # 20 bytes
# live-worlds §3
STATEMENT_MAGIC = b"WLW1"
STATEMENT_DOMAIN = b"wai:live-world\x01"           # 15 bytes
CAPABILITY = b"wai.world.model"
WORLD_LEN = 52                                     # §3.2: id 32, seed 8, width, height, rate 4 each
MIN_OP, MAX_OP = 136, 65536
MIN_OPS, MAX_OPS = 1, 4096
# live-worlds §4.3, §9.3
WORLD_RECEIPT_ALIAS = 4
SESSION_DOMAIN = b"wai:world-model-session\x01"   # 24 bytes
# interactive-worlds §5.1
OP_DOMAIN = b"wai:world-op\x01"
OP_HEAD = 32 + 8 + 8 + 2                           # actor, lamport, effect_tick, n_deps
OP_DEP = 32 + 8
OP_MID = 16 + 2 + 4                                # target_node, mutation_kind, mut_len
OP_SIG = 64
ACTION_KIND = 1                                    # wai.world.model: an action
ACTION_HEAD = 16                                   # dx i32, dy i32, buttons u32, payload_len u32
# jwp-receipts §2.1, §2.2, §2.3, §2.7, §3
RECEIPTS_MAGIC = b"WJR1"
SIGN_LEGACY = b"moq-jwp:group-sign\x01"            # 19 bytes
SIGN_LABELLED = b"moq-jwp:group-sign\x02"
OBJECT_DOMAIN = b"moq-jwp:object\x01"
LEAF_LEGACY = b"moq-jwp:merkle-leaf\x01"
LEAF_LABELLED = b"moq-jwp:merkle-leaf\x02"
NODE_DOMAIN = b"moq-jwp:merkle-node\x01"
LINK_DOMAIN = b"moq-jwp:receipt-link\x01"
ZERO32 = bytes(32)
# energy-measurement §2.7: the tags, and §2's names for those this verifier compares
UNMETERED, HW_SHUNT, MODEL_BASED, ESTIMATOR, ON_CHIP, CALIBRATED = 0, 1, 2, 3, 4, 5

# The reasons the corpus is to refuse with (live-worlds §11 and the codes it names from
# live-events §9 and jwp-receipts §3.3). `world_incomplete` is a participant's report about
# its own copy (§7), not a refusal; `bad_key` at the statement is not listed, since a
# statement is signed by the descriptor's authority, whose key the descriptor's own
# signature has already held to SPEC §7's steps 1 and 2.
CORPUS_REASONS = [
    "truncated", "malformed", "unknown_version", "world_capability_unknown",
    "world_statement_invalid", "wrong_room", "bad_signature",
    "world_not_granted", "world_gap", "moqt_mismatch", "signature_verify_failed",
    "content_hash_mismatch", "chain_break", "parent_not_held",
    "op_malformed", "batch_too_large", "op_too_large", "bad_op_signature",
    "actor_not_owner", "op_not_action", "equivocation", "lamport_regress",
]


class Refuse(Exception):
    def __init__(self, reason, at):
        Exception.__init__(self, reason)
        self.reason, self.at = reason, at


class Short(Exception):
    """The bytes end before a field the layout requires."""


class Reader:
    """Reads a record field by field, in either byte order."""

    def __init__(self, b, order):
        self.b, self.i, self.order = b, 0, order

    def take(self, n):
        if self.i + n > len(self.b):
            raise Short()
        out = self.b[self.i:self.i + n]
        self.i += n
        return out

    def uint(self, n):
        return int.from_bytes(self.take(n), self.order)

    def u8(self):
        return self.uint(1)

    def short(self):           # rooms: a u8 length and that many bytes
        return self.take(self.u8())

    def long(self):            # rooms: a u16 (little-endian) length and that many bytes
        return self.take(self.uint(2))

    def string(self):          # receipts: u32_be(len) and UTF-8
        return self.take(self.uint(4))

    def left(self):
        return len(self.b) - self.i


def sha256(b):
    return hashlib.sha256(b).digest()


def is_utf8(b):
    try:
        b.decode("utf-8")
        return True
    except UnicodeDecodeError:
        return False


def is_slot_name(b):           # live-rooms §2: 1 to 32 bytes of a-z, 0-9 and _
    return 1 <= len(b) <= 32 and all(c in b"abcdefghijklmnopqrstuvwxyz0123456789_" for c in b)


def read_head(b, magic):
    """live-rooms §3.4 step 1, and live-worlds §3.4 step 1: the magic and the version."""
    if len(b) < 4:
        raise Refuse("truncated", None)
    if b[:4] != magic:
        raise Refuse("malformed", None)
    if len(b) == 4:
        raise Refuse("truncated", None)
    if b[4] != 1:
        raise Refuse("unknown_version", None)


# --- the room descriptor (live-rooms §3) ---------------------------------------------------

def read_descriptor(b):
    """§3.4 steps 1 to 3 and §3.2's signature; the field rules are live-rooms's verifier's."""
    at = "descriptor"
    try:
        read_head(b, DESCRIPTOR_MAGIC)
        r = Reader(b, "little")
        r.take(5)
        authority, _clock = r.take(32), r.take(32)
        r.take(8 + 8)                          # t0, tick
        r.long()                               # origin
        r.take(4 + 2 + 4)                      # max_present, max_publishers, reval
        r.take(65)                             # ceiling (WEP1)
        for _ in range(r.u8()):                # roles
            r.short()
            r.take(1 + 1 + 1 + 1 + 32)
        for _ in range(r.u8()):                # slots
            r.short()
            r.take(1 + 1 + 4)
        for _ in range(r.u8()):                # publish profiles (§8.1)
            r.short()
            r.take(1)                          # media
            for _ in range(r.u8()):
                r.short()                      # codecs
            r.take(1)                          # layering
            r.short()                          # mode
            r.take(14 * r.u8())                # layers
            r.take(4 + 1 + 1)                  # sample_rate, channels, processing
        r.take(37)                             # review
        r.take(1 + 2 + 4 + 4)                  # recording
        r.take(2 * r.u8())                     # regions
        body_end = r.i
        sig = r.take(64)
    except Short:
        raise Refuse("truncated", at)
    except Refuse as e:
        raise Refuse(e.reason, at)
    if r.left():
        raise Refuse("malformed", at)
    body = b[:body_end]
    bad = ED.verify(authority, DESCRIPTOR_DOMAIN + body, sig)
    if bad:
        raise Refuse(bad, at)                  # bad_key or bad_signature
    return {"authority": authority, "room": sha256(body)}


# --- the grants (live-rooms §4) -----------------------------------------------------------

def read_attrs(b):
    """§4.4: keys (short, a slot name) in strictly ascending order, values (long) UTF-8,
    at most 1024 bytes. None when the field breaks the rule."""
    if len(b) > 1024:
        return None
    r, out, last = Reader(b, "little"), {}, None
    try:
        while r.left():
            k, v = r.short(), r.long()
            if not is_slot_name(k) or (last is not None and k <= last) or not is_utf8(v):
                return None
            out[k.decode()] = v.decode("utf-8")
            last = k
    except Short:
        return None
    return out


def read_grant(b, d, index):
    at = {"grant": index}
    try:
        read_head(b, GRANT_MAGIC)
        r = Reader(b, "little")
        r.take(5)
        room, principal = r.take(32), r.take(32)
        r.take(32 + 8 + 8 + 8)                 # prev, effective, nbf, exp
        r.take(1 + 4 + 1)                      # role, publish..powers, precedence
        r.take(16 + 16 + 65)                   # once, viewer, policy
        r.take(8 + 8 + 3)                      # joules_ceiling, funds_ceiling, currency
        r.take(r.u8())                         # profiles
        for _ in range(r.u8()):
            r.short()                          # origins
        attrs = r.long()
        body_end = r.i
        sig = r.take(64)
    except Short:
        raise Refuse("truncated", at)
    except Refuse as e:
        raise Refuse(e.reason, at)
    if r.left():
        raise Refuse("malformed", at)
    if ED.verify(d["authority"], GRANT_DOMAIN + b[:body_end], sig):
        raise Refuse("bad_signature", at)      # §4.3 step 2
    if room != d["room"]:
        raise Refuse("wrong_room", at)         # §4.3 step 4
    a = read_attrs(attrs)
    if a is None:
        raise Refuse("grant_invalid", at)      # §4.3 step 5, the attrs rule
    return {"principal": principal, "attrs": a}


# --- the world statement (live-worlds §3) -------------------------------------------------

def read_statement(b, d):
    at = "statement"
    try:
        read_head(b, STATEMENT_MAGIC)
        r = Reader(b, "little")
        r.take(5)
        room = r.take(32)
        capability = r.short()
        world = r.long()
        max_op_bytes = r.uint(4)
        max_ops = r.uint(2)
        body_end = r.i
        sig = r.take(64)
    except Short:
        raise Refuse("truncated", at)
    except Refuse as e:
        raise Refuse(e.reason, at)
    if r.left():
        raise Refuse("malformed", at)
    if capability != CAPABILITY:
        raise Refuse("world_capability_unknown", at)
    if (len(world) != WORLD_LEN or any(int.from_bytes(world[i:i + 4], "little") == 0 for i in (40, 44, 48))
            or not MIN_OP <= max_op_bytes <= MAX_OP or not MIN_OPS <= max_ops <= MAX_OPS):
        raise Refuse("world_statement_invalid", at)
    if room != d["room"]:
        raise Refuse("wrong_room", at)
    bad = ED.verify(d["authority"], STATEMENT_DOMAIN + b[:body_end], sig)
    if bad:
        raise Refuse(bad, at)                  # bad_key or bad_signature
    return {"world": world, "max_op_bytes": max_op_bytes, "max_ops": max_ops}


# --- receipts (jwp-receipts §2.7, §3) -----------------------------------------------------

def read_class(r):
    """energy-measurement §2.7: a tag byte and its evidence. Returns (tag, raw bytes)."""
    start = r.i
    tag = r.u8()
    if tag in (UNMETERED, HW_SHUNT, MODEL_BASED, ESTIMATOR):
        pass
    elif tag == ON_CHIP:
        r.take(4 + 8)
        if r.u8() > 2:
            raise ValueError("filtering")
    elif tag == CALIBRATED:
        r.take(4 + 8)
        if not is_utf8(r.string()):
            raise ValueError("calibration_ref")
    else:
        raise ValueError("tag")
    return tag, r.b[start:r.i]


def read_flag_parent(r):
    flag, parent = r.u8(), r.take(32)
    if flag not in (0, 1) or (flag == 0 and parent != ZERO32):
        raise ValueError("parent flag")
    return parent if flag else None


def read_receipts(b):
    """§2.7.3 read by §2.7.4: any failure is `malformed`."""
    try:
        r = Reader(b, "big")
        if r.take(4) != RECEIPTS_MAGIC:
            raise ValueError("magic")
        start = r.i
        domain = r.take(19)
        if domain not in (SIGN_LEGACY, SIGN_LABELLED):
            raise ValueError("domain")
        g = {"labelled": domain == SIGN_LABELLED}
        g["track_alias"], g["group_id"] = r.uint(8), r.uint(8)
        g["root"], g["leaf_count"], g["total"] = r.take(32), r.uint(8), r.uint(8)
        for k in ("group_start", "group_end"):
            if not is_utf8(r.string()):
                raise ValueError(k)
        g["pubkey"] = r.take(32)
        if not is_utf8(r.string()):
            raise ValueError("signer_id")
        if g["labelled"]:
            g["class"] = read_class(r)
            g["coverage"] = r.u8()
            if g["coverage"] not in (0, 1):
                raise ValueError("coverage")
            g["parent"] = read_flag_parent(r)
        payload_end = r.i
        g["sig"] = r.take(64)
        g["payload"] = b[start:payload_end]
        g["canonical"] = b[start:r.i]
        if not g["labelled"]:
            g["parent"] = read_flag_parent(r)
        objects = []
        for _ in range(r.uint(4)):
            o = {"object_id": r.uint(8), "content_hash": r.take(32), "joules": r.uint(8)}
            o["origin"] = r.string()
            if not is_utf8(o["origin"]):
                raise ValueError("origin")
            o["siblings"] = [r.take(32) for _ in range(r.u8())]
            o["direction_bits"], o["leaf_index"] = r.uint(8), r.uint(8)
            o["leaf_version"] = r.u8()
            if o["leaf_version"] not in (1, 2):
                raise ValueError("leaf_version")
            o["class"] = read_class(r) if o["leaf_version"] == 2 else None
            objects.append(o)
        if r.left():
            raise ValueError("trailing bytes")
    except (Short, ValueError):
        return None
    return g, objects


def merkle_leaf(o):
    """§2.2: the labelled leaf where the receipt carries a label, the legacy one otherwise."""
    if o["class"] is None:
        return blake3(LEAF_LEGACY + o["content_hash"])
    return blake3(LEAF_LABELLED + o["content_hash"] + o["joules"].to_bytes(8, "big") + o["class"][1]
                  + len(o["origin"]).to_bytes(4, "big") + o["origin"])


def check_receipts(rb, covered, owner, group, chain, joins, at):
    """live-worlds §9.2 step 3. `chain` is the track's chain state: the head's link hash and
    the link hashes of the earlier groups the verifier holds, and the head's group_id.
    `joins` is True for a private world's first opened object at a group above 0."""
    read = read_receipts(rb)
    if read is None:
        raise Refuse("malformed", at)
    g, objects = read
    if len(objects) != 1:
        raise Refuse("malformed", at)
    o = objects[0]
    # moq-streaming-format §4.1: before any signature is checked.
    if g["pubkey"] != owner or g["track_alias"] != WORLD_RECEIPT_ALIAS or g["group_id"] != group or o["object_id"] != 0:
        raise Refuse("moqt_mismatch", at)
    # jwp-receipts §3 step 1.
    if g["labelled"] and g["class"][0] == UNMETERED and g["coverage"] != 0:
        raise Refuse("group_label_invalid", at)        # §2.7.4: a coverage beside the marker
    if not ED.strict_key(g["pubkey"]):
        raise Refuse("bad_signer_key", at)
    # Step 2: the group signature, then the class with the group alone (§2.4.1).
    if ED.verify(g["pubkey"], g["payload"], g["sig"]):
        raise Refuse("signature_verify_failed", at)
    if g["labelled"]:
        tag = g["class"][0]
        if tag == HW_SHUNT or (tag == UNMETERED) != (g["total"] == 0):
            raise Refuse("group_label_invalid", at)
        if tag in (ON_CHIP, CALIBRATED):
            raise Refuse("class_not_supported_here", at)
    # Step 3: the chain of the track (live-worlds §9.2 step 3).
    link = blake3(LINK_DOMAIN + g["canonical"])
    parent = g["parent"]
    if chain["head"] is None:
        if not joins and parent is not None:
            raise Refuse("parent_not_held", at)
    else:
        if not g["labelled"] and g["group_id"] <= chain["head_group"]:
            raise Refuse("group_id_not_increasing", at)
        if parent != chain["head"]:
            if parent is None or parent in chain["earlier"]:
                raise Refuse("chain_break", at)
            raise Refuse("parent_not_held", at)
    # Step 4: the content hash over the bytes the receipt covers (§2.1).
    if blake3(OBJECT_DOMAIN + covered) != o["content_hash"]:
        raise Refuse("content_hash_mismatch", at)
    # Step 5: the label (§2.4.2), the position, the walk to the verified root.
    if o["class"] is not None:
        tag = o["class"][0]
        if tag == HW_SHUNT or (tag == UNMETERED) != (o["joules"] == 0):
            raise Refuse("object_label_invalid", at)
        if tag in (ON_CHIP, CALIBRATED):
            raise Refuse("class_not_supported_here", at)
    n = g["leaf_count"]
    depth = (n - 1).bit_length() if n else -1
    if (o["leaf_index"] >= n or len(o["siblings"]) != depth or o["direction_bits"] != o["leaf_index"]
            or o["object_id"] != o["leaf_index"]):
        raise Refuse("proof_position_mismatch", at)
    cur = merkle_leaf(o)
    for i, sib in enumerate(o["siblings"]):
        cur = blake3(NODE_DOMAIN + (sib + cur if (o["direction_bits"] >> i) & 1 else cur + sib))
    if cur != g["root"]:
        raise Refuse("inclusion_proof_failed", at)
    # The group total: with one leaf, the verifier holds an accepted receipt for every index.
    if n == 1:
        if o["joules"] != g["total"]:
            raise Refuse("group_total_mismatch", at)
        if g["labelled"]:
            if o["joules"] != 0 and o["class"] is None:
                raise Refuse("group_class_disagrees", at)   # its objects compose to no class
            want = (UNMETERED, bytes([UNMETERED])) if o["joules"] == 0 else o["class"]
            if g["class"] != want:
                raise Refuse("group_class_disagrees", at)
            if g["coverage"] != 0:
                raise Refuse("group_coverage_disagrees", at)
    # Accepted: the group heads the track's chain.
    if chain["head"] is not None:
        chain["earlier"].add(chain["head"])
    chain["head"], chain["head_group"] = link, g["group_id"]


# --- ops (interactive-worlds §5.1) --------------------------------------------------------

def read_batch(b):
    """One or more ops, each whole, concatenated with nothing between them. None when the
    batch is not that."""
    ops, i = [], 0
    while i < len(b):
        start = i
        if len(b) - i < OP_HEAD:
            return None
        actor = b[i:i + 32]
        lamport = int.from_bytes(b[i + 32:i + 40], "little")
        tick = int.from_bytes(b[i + 40:i + 48], "little")
        n_deps = int.from_bytes(b[i + 48:i + 50], "little")
        i += OP_HEAD + OP_DEP * n_deps
        if len(b) < i + OP_MID:
            return None
        kind = int.from_bytes(b[i + 16:i + 18], "little")
        mut_len = int.from_bytes(b[i + 18:i + 22], "little")
        i += OP_MID
        mutation = b[i:i + mut_len]
        i += mut_len
        if len(b) < i + OP_SIG:
            return None
        i += OP_SIG
        raw = b[start:i]
        ops.append({"raw": raw, "actor": actor, "lamport": lamport, "tick": tick, "n_deps": n_deps,
                    "kind": kind, "mutation": mutation, "signed": raw[:-OP_SIG], "sig": raw[-OP_SIG:]})
    return ops if ops else None


def is_action(op):
    """`wai.world.model`'s action: mutation kind 1, no dependency, and a body that reads:
    dx i32, dy i32, buttons u32, payload_len u32 (little-endian), then the payload."""
    m = op["mutation"]
    if op["kind"] != ACTION_KIND or op["n_deps"] != 0 or len(m) < ACTION_HEAD:
        return False
    return len(m) == ACTION_HEAD + int.from_bytes(m[12:16], "little")


# --- a world (live-worlds §9) -------------------------------------------------------------

def verify_world(v):
    d = read_descriptor(bytes.fromhex(v["descriptor"]))
    grants = [read_grant(bytes.fromhex(g), d, i) for i, g in enumerate(v["grants"])]
    st = read_statement(bytes.fromhex(v["statement"]), d)
    private = v["private"]
    if private not in (True, False):
        raise ValueError("private is not a boolean")
    actors = {g["principal"] for g in grants if g["attrs"].get("world") == "act"}
    admitted = {}                                      # op id -> op bytes
    for ti, track in enumerate(v["tracks"]):
        owner = bytes.fromhex(track["owner"])
        chain = {"head": None, "head_group": None, "earlier": set()}
        last_group, owner_lamports = None, []
        for obj in track["objects"]:
            group = obj["group"]
            at = {"track": ti, "group": group}
            # Step 1.
            if owner not in actors:
                raise Refuse("world_not_granted", at)
            # Step 2.
            if last_group is None:
                if not private and group != 0:
                    raise Refuse("world_gap", at)
            elif group != last_group + 1:
                raise Refuse("world_gap", at)
            joins = private and last_group is None and group > 0
            last_group = group
            # Step 3: the receipts cover `payload`; in a private world the ops are `opened`.
            covered = bytes.fromhex(obj["payload"])
            if private:
                if obj.get("opened") is None:
                    raise ValueError("a private world's object without its opened batch")
                batch = bytes.fromhex(obj["opened"])
            else:
                if obj.get("opened") is not None:
                    raise ValueError("an opened batch in a world that is not private")
                batch = covered
            check_receipts(bytes.fromhex(obj["receipts"]), covered, owner, group, chain, joins, at)
            # Step 4.
            ops = read_batch(batch)
            if ops is None:
                raise Refuse("op_malformed", at)
            if len(ops) > st["max_ops"]:
                raise Refuse("batch_too_large", at)
            if any(len(op["raw"]) > st["max_op_bytes"] for op in ops):
                raise Refuse("op_too_large", at)
            # Step 5, each op in batch order.
            for op in ops:
                if ED.verify(op["actor"], OP_DOMAIN + op["signed"], op["sig"]):
                    raise Refuse("bad_op_signature", at)
                if not private and op["actor"] != owner:
                    raise Refuse("actor_not_owner", at)
                if not is_action(op):
                    raise Refuse("op_not_action", at)
                oid = (op["actor"], op["lamport"])
                held = admitted.get(oid)
                if held is not None and held != op["raw"]:
                    raise Refuse("equivocation", at)
                redelivery = held is not None
                if op["actor"] == owner:
                    if not redelivery and owner_lamports and op["lamport"] <= max(owner_lamports):
                        raise Refuse("lamport_regress", at)
                    owner_lamports.append(op["lamport"])
                if not redelivery:
                    admitted[oid] = op["raw"]
    return session(st["world"], admitted)


def session(world, admitted):
    """§9.3 over the admitted ops in canonical order, and §9.4's outcome."""
    order = sorted(admitted.items(), key=lambda kv: (int.from_bytes(kv[1][40:48], "little"), kv[0][1], kv[0][0]))
    h = blake3(SESSION_DOMAIN + world)
    counts = {}
    for (actor, _lamport), raw in order:
        h = blake3(SESSION_DOMAIN + h + raw)
        counts[actor.hex()] = counts.get(actor.hex(), 0) + 1
    return {"accept": True, "session_hash": h.hex(), "ops": len(order), "actors": counts}


def outcome(v):
    try:
        return verify_world(v)
    except Refuse as e:
        return {"accept": False, "reason": e.reason, "at": e.at}


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    if st:
        fails.append("ed25519 self-test")
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    reasons = set()
    for v in doc["vectors"]:
        name, want = v["name"], v["expect"]
        try:
            got = outcome(v)
        except (ValueError, KeyError, TypeError) as e:
            got = {"accept": None, "unreadable": repr(e)}
        ok = got == want
        if not want.get("accept"):
            reasons.add(want.get("reason"))
        if got.get("accept"):
            what = "accept %s, %d ops" % (got["session_hash"][:16], got["ops"])
        elif got.get("accept") is False:
            what = "refuse %s at %s" % (got["reason"], json.dumps(got["at"]))
        else:
            what = "unreadable: %s" % got["unreadable"]
        print(("PASS" if ok else "FAIL") + " vector %s: %s" % (name, what))
        if not ok:
            fails.append("%s: verifier reached %s, the corpus states %s" % (name, json.dumps(got), json.dumps(want)))
    missing = [r for r in CORPUS_REASONS if r not in reasons]
    accepted = [v for v in doc["vectors"] if v["expect"].get("accept")]
    kinds = {v["private"] for v in accepted}
    if missing:
        fails.append("no vector refused with %s" % ", ".join(missing))
    if kinds != {True, False}:
        fails.append("the corpus does not accept both a public and a private world")
    print("%s every refusal of §3.4 and §9.2 refuses a vector, and a public and a private world are accepted"
          % ("PASS" if not missing and kinds == {True, False} else "FAIL"))
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " live-worlds corpus: %d vectors" % len(doc["vectors"]))
    return 1 if fails else 0


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: live_worlds_verify.py <live-conformance/worlds>", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))