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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library, and
computes BLAKE3 with ../claim-conformance/_blake3.py, written from its specification. Each room
is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py
(live-rooms §9), written from its text alone. Everything else is here:

- the ad policy (§3): its layout, rules and signature, and the decider grant;
- the break annotation and the schedule (§4), with its refusals;
- the HLS interstitial, closing and preload tags (§5) and the DASH events (§6), byte for byte;
- the decision and the asset lists (§7), byte for byte;
- the break form of the wai.video.ssai receipt (§8): its JSON carriage, hashes, signing
  payload, identity and rules;
- verification, from the room to the spend (§9).

A policy of version 2 is read here with its presentation fields (§3.1, ad-presentation §8). A
room whose policy requires presentation claims is verified with ad_presentation_verify.py,
which gives accept() the judge of those claims (ad-presentation §9).

For every room it requires the outcome, reason and place, and everything else the corpus
states; every HLS tag file, byte for byte; of the corpus, that every reason of §10 appears in
it and that no other file is there. Then it mutates copies of the corpus, one defect each, and
requires every copy to fail.

    python3 live_ads_verify.py <live-conformance/ads>

Exit status 0 only if every case verifies and every mutated copy fails. Python 3.8 or later;
standard library only.
"""

import copy
import hashlib
import json
import os
import struct
import sys

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

Refuse = LEV.Refuse
EXTENSION = os.path.join(HERE, "..", "extensions", "live-ads.md")

POLICY_MAGIC, ANNOTATION_MAGIC = b"WAP1", b"WAB1"
POLICY_DOMAIN = b"wai:live-ads-policy\x01"                                       # §3.2
BREAK_DOMAIN = b"wai:video-ssai-break-receipt\x01"                               # §8.4
ID_DOMAIN = b"wai:video-ssai-receipt-id\x01"
POD_DOMAIN, IMP_DOMAIN = b"wai:video-ssai-pod\x01", b"wai:video-ssai-imp\x01"    # §8.3
LEAF, NODE = b"moq-jwp:merkle-leaf\x01", b"moq-jwp:merkle-node\x01"              # jwp-receipts §2.2
INTERSTITIAL = "com.apple.hls.interstitial"  # allow-name: the interstitial date-range class, a protocol identifier
PRELOAD = "com.apple.hls.preload"  # allow-name: the preload date-range class, a protocol identifier
REPLACE = "urn:mpeg:dash:event:alternativeMPD:replace:2025"                      # §6.1
INSERT = "urn:mpeg:dash:event:alternativeMPD:insert:2025"
OUT, IN, CHAPTER, BREAK = 1, 2, 3, 4
META, CUE = 0x01, 0x02
MID, POST = 1, 2
HLS, DASH = 1, 2
NO_SKIP = 65535
NO_CEILING = 2 ** 64 - 1
U64 = 2 ** 64 - 1
FLOOR = 288676                                                                    # §8.5 rule 5
# Unicode White_Space, the code points a calibration reference must hold more than (§8.5).
WHITE_SPACE = set(map(chr, [0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x20, 0x85, 0xA0, 0x1680] + list(range(0x2000, 0x200B))
                      + [0x2028, 0x2029, 0x202F, 0x205F, 0x3000]))


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


# ---------------------------------------------------------------- the extension's own table

def reasons():
    """§10's codes, from the text."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 10. Reasons")
    end = text.index("\n## ", at + 1)
    return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]


REASONS = reasons()


# ---------------------------------------------------------------- §3 the ad policy

def is_base(b):
    """§3.1."""
    if not b.startswith(b"https://"):
        return False
    rest = b[8:]
    return (1 <= len(rest) <= 492 and rest[-1:] == b"/"
            and all(0x21 <= c <= 0x7E and c not in b"\"#&'<>?\\" for c in rest))


def read_policy(b):
    """§3.3: the policy, or Refuse."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != POLICY_MAGIC:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] not in (1, 2):
        raise Refuse("unknown_version")
    r = LEV.Reader(b, "truncated")
    r.take(5)
    p = {"room": r.take(32), "decider_grant": r.take(32), "slots": r.u8(), "open_avail_ms": r.u32(),
         "preload_ms": r.u32(), "restrict": r.u8(), "skip_after_s": r.u16(), "snap": r.u8()}
    p["asset_base"] = r.take(r.u16())
    p["creative_base"] = r.take(r.u16())
    p["presentation"] = None
    if b[4] == 2:                                                       # ad-presentation §8
        p["presentation"] = {"require": r.u8(), "rule": r.u8(), "token_type": r.u16(), "issuer": r.take(r.u16())}
    p["body"] = b[:r.p]
    p["sig"] = r.take(64)
    if not r.done():
        raise Refuse("malformed")
    if (not 1 <= p["slots"] <= 16 or p["restrict"] > 3 or p["snap"] > 3
            or not is_base(p["asset_base"]) or not is_base(p["creative_base"])
            or (p["presentation"] is not None and not presentation_ok(p["presentation"]))):
        raise Refuse("ad_policy_malformed")
    p["asset_base"], p["creative_base"] = p["asset_base"].decode(), p["creative_base"].decode()
    return p


def presentation_ok(q):
    """ad-presentation §8's rules for a policy of version 2."""
    req, claim, attest = q["require"], q["require"] & 1, q["require"] & 2
    if req > 7 or (not claim and req != 0) or q["rule"] > 2 or bool(claim) != (q["rule"] != 0):
        return False
    if attest:
        return q["token_type"] in (1, 2) and len(q["issuer"]) > 0
    return q["token_type"] == 0 and q["issuer"] == b""


def requires_claims(p):
    return p["presentation"] is not None and p["presentation"]["require"] & 1 == 1


# ---------------------------------------------------------------- §4 breaks

def read_annotation(b):
    """§4.2."""
    if len(b) != 38 or b[:4] != ANNOTATION_MAGIC or b[4] != 1 or b[37] not in (MID, POST):
        raise Refuse("annotation_malformed")
    return {"cue": b[5:37], "placement": b[37]}


def cue_of(ev):
    return LEV.read_payload(CUE, ev["payload"], 65535)


def schedule(held, p):
    """§4.3 over the held events in the merged order: (breaks, chapters, written) or Stop."""
    breaks, chapters, written = [], [], []
    open_, last_end, closed = None, None, False
    for k, (c, i, ev) in enumerate(held):
        where = ["record", c, i]
        if ev["kind"] == CUE:
            q = cue_of(ev)
            if closed:
                raise Stop("cue_after_close", where)
            t = ev["t"]
            overlaps = open_ is not None or (last_end is not None and t < last_end)
            if q["type"] == OUT:
                if overlaps:
                    raise Stop("break_overlap", where)
                avail = q["duration_ms"] or p["open_avail_ms"]
                if avail == 0:
                    raise Stop("break_duration_unknown", where)
                open_ = len(breaks)
                written.append(("open", len(breaks)))
                breaks.append({"digest": ev["digest"], "author": ev["principal"], "cue_id": q["cue_id"], "t": t,
                               "type": OUT, "placement": MID, "planned_ms": q["duration_ms"], "avail_ms": avail,
                               "duration_ms": None, "closed_by": None, "end": None})
            elif q["type"] == IN:
                b = breaks[open_] if open_ is not None else None
                if b is None or b["cue_id"] != q["cue_id"] or b["author"] != ev["principal"]:
                    raise Stop("cue_in_unmatched", where)
                duration = (t - b["t"]) // 1000
                if duration == 0:
                    raise Stop("break_zero_duration", where)
                if q["duration_ms"] != 0 and q["duration_ms"] != duration:
                    raise Stop("cue_duration_mismatch", where)
                b.update(duration_ms=duration, closed_by=ev["digest"], end=t)
                written.append(("close", open_))
                open_, last_end = None, t
            elif q["type"] == CHAPTER:
                chapters.append({"digest": ev["digest"], "t": t, "duration_ms": q["duration_ms"]})
            else:
                if q["duration_ms"] == 0:
                    raise Stop("break_zero_duration", where)
                if overlaps:
                    raise Stop("break_overlap", where)
                last_end = t + q["duration_ms"] * 1000
                written.append(("open", len(breaks)))
                breaks.append({"digest": ev["digest"], "author": ev["principal"], "cue_id": q["cue_id"], "t": t,
                               "type": BREAK, "placement": MID, "planned_ms": q["duration_ms"],
                               "avail_ms": q["duration_ms"], "duration_ms": q["duration_ms"], "closed_by": None,
                               "end": last_end})
        elif ev["kind"] == META and ev["payload"][:4] == ANNOTATION_MAGIC:
            try:
                a = read_annotation(ev["payload"])
            except Refuse as e:
                raise Stop(e.reason, where)
            before = next((x for _, _, x in reversed(held[:k])
                           if x["principal"] == ev["principal"] and x["seq"] == ev["seq"] - 1), None)
            if (before is None or before["digest"] != a["cue"] or before["kind"] != CUE or before["t"] != ev["t"]
                    or cue_of(before)["type"] != BREAK):
                raise Stop("annotation_misplaced", where)
            if a["placement"] == POST:
                for b in breaks:
                    if b["digest"] == a["cue"]:
                        b["placement"] = POST
                closed = True
    return breaks, chapters, written


# ---------------------------------------------------------------- §4.4, §5, §6, §7

def break_id(d):
    return "wai-ad-" + d.hex()


def event_id(d):
    return int.from_bytes(d[:8], "big") & ((1 << 53) - 1)


def secs(ms):
    return "%d.%03d" % (ms // 1000, ms % 1000)


def hls_uri(p, d):
    i = break_id(d)
    return "%s%s.json?_HLS_interstitial_id=%s" % (p["asset_base"], i, i)


def dash_uri(p, d):
    return "%s%s.mpd" % (p["asset_base"], break_id(d))


def date(t0, t, less=0):
    return LEV.iso((t0 + t) // 1000 - less)


def bits(v, names):
    return ",".join(n for k, n in enumerate(names) if v & (1 << k))


def interstitial(p, t0, b):
    """§5.1."""
    post = b["placement"] == POST
    a = ['ID="%s"' % break_id(b["digest"]), 'CLASS="%s"' % INTERSTITIAL, 'START-DATE="%s"' % date(t0, b["t"]),
         'CUE="%s"' % ("POST,ONCE" if post else "ONCE")]
    known = b["planned_ms"] if b["type"] == BREAK else None
    if known is not None:
        a.append("DURATION=" + secs(known))
    elif b["planned_ms"]:
        a.append("PLANNED-DURATION=" + secs(b["planned_ms"]))
    a.append('X-ASSET-LIST="%s"' % hls_uri(p, b["digest"]))
    if known is not None:
        if not post:
            a.append("X-RESUME-OFFSET=" + secs(known))
        a.append("X-PLAYOUT-LIMIT=" + secs(known))
    if p["snap"]:
        a.append('X-SNAP="%s"' % bits(p["snap"], ["OUT", "IN"]))
    if p["restrict"]:
        a.append('X-RESTRICT="%s"' % bits(p["restrict"], ["SKIP", "JUMP"]))
    if p["skip_after_s"] != NO_SKIP:
        a.append("X-SKIP-CONTROL-OFFSET=%d" % p["skip_after_s"])
    return "#EXT-X-DATERANGE:" + ",".join(a)


def closing(b):
    """§5.2."""
    s = secs(b["duration_ms"])
    return '#EXT-X-DATERANGE:ID="%s",CLASS="%s",DURATION=%s,X-RESUME-OFFSET=%s,X-PLAYOUT-LIMIT=%s' % (
        break_id(b["digest"]), INTERSTITIAL, s, s, s)


def preload(p, t0, b):
    """§5.3."""
    i = break_id(b["digest"])
    return ('#EXT-X-DATERANGE:ID="%s-preload",CLASS="%s",START-DATE="%s",DURATION=%s,X-URI="%s",X-TARGET-ID="%s",'
            'X-TARGET-CLASS="%s"' % (i, PRELOAD, date(t0, b["t"], p["preload_ms"]), secs(p["preload_ms"]),
                                     hls_uri(p, b["digest"]), i, INTERSTITIAL))


def hls_tags(p, t0, breaks, written):
    """§5.4: in the order a packager writes them."""
    out = []
    for what, i in written:
        if what == "open":
            if p["preload_ms"]:
                out.append(preload(p, t0, breaks[i]))
            out.append(interstitial(p, t0, breaks[i]))
        else:
            out.append(closing(breaks[i]))
    return out


def dash_event(p, pt0, b, closed):
    """§6.2."""
    known = b["duration_ms"] if closed else None
    post = b["placement"] == POST
    e = '<Event presentationTime="%d"' % (pt0 + b["t"] // 1000)
    if known is not None:
        e += ' duration="%d"' % known
    e += ' id="%d">' % event_id(b["digest"])
    child = "InsertPresentation" if post else "ReplacePresentation"
    e += '<%s uri="%s"' % (child, dash_uri(p, b["digest"]))
    if p["preload_ms"]:
        e += ' earliestResolutionTimeOffset="%d"' % p["preload_ms"]
    if known is not None:
        e += ' maxDuration="%d"' % known
    e += ' executeOnce="true"'
    skip = p["skip_after_s"] if p["skip_after_s"] != NO_SKIP else -(-b["avail_ms"] // 1000)
    e += ' skipAfter="PT%dS"' % skip
    if known is not None and not post:
        e += ' returnOffset="%d"' % known
    return e + "/></Event>"


def dash_streams(p, pt0, breaks):
    """§6.3."""
    out = ""
    for scheme, post in ((REPLACE, False), (INSERT, True)):
        evs = [dash_event(p, pt0, b, True) for b in breaks if (b["placement"] == POST) == post]
        if evs:
            out += '<EventStream schemeIdUri="%s" timescale="1000">%s</EventStream>' % (scheme, "".join(evs))
    return out


def decide(pod, slots, avail, funds):
    """§7.1: the fill, each (slot, ad_id, creative_hash, duration_ms, price_micros)."""
    fill = []
    for c in sorted(pod, key=lambda c: (-c["price_micros"], c["ad_id"].encode())):
        if len(fill) >= slots:
            break
        if c["duration_ms"] <= avail and c["price_micros"] <= funds:
            fill.append((len(fill), c["ad_id"], c["creative_hash"], c["duration_ms"], c["price_micros"]))
            avail -= c["duration_ms"]
            funds -= c["price_micros"]
    return fill


def asset_list(p, fmt, fill):
    """§7.2, §7.3: the exact bytes."""
    if fmt == HLS:
        assets = ['{"URI":"%s%s.m3u8","DURATION":%s}' % (p["creative_base"], h.hex(), secs(d)) for _, _, h, d, _ in fill]
        return ('{"ASSETS":[%s]}' % ",".join(assets)).encode()
    if not fill:
        return b""
    periods = "".join('<Period id="%d" duration="PT%sS"><ImportedMPD>%s%s.mpd</ImportedMPD></Period>'
                      % (s, secs(d), p["creative_base"], h.hex()) for s, _, h, d, _ in fill)
    return ('<?xml version="1.0" encoding="UTF-8"?>\n<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" '
            'profiles="urn:mpeg:dash:profile:list:2024" type="list" minBufferTime="PT1S">%s</MPD>' % periods).encode()


# ---------------------------------------------------------------- §8 the break form

MEMBERS = ["form", "room", "cue", "session", "start_offset_ms", "format", "asset_list_sha256", "avail_ms", "slots",
           "funds_remaining_micros", "pod_hash", "impressions", "merkle_root", "budget_spent_micros",
           "work_decide_ops", "joules_micro", "energy_provenance", "grant", "signer_pubkey", "signer_id", "sig"]
EVIDENCE = {"OnChipCounter": ["energy_uncertainty", "energy_counter_filtering"],
            "CalibratedInstrument": ["energy_uncertainty", "energy_calibration_ref"],
            "ModelBased": [], "Estimator": []}
CLASS_TAG = {"ModelBased": 2, "Estimator": 3, "OnChipCounter": 4, "CalibratedInstrument": 5}
FILTERING = {"Unknown": 0, "Off": 1, "On": 2}


def u(v, top=U64):
    if not isinstance(v, int) or isinstance(v, bool) or not 0 <= v <= top:
        raise Refuse("receipt_malformed")
    return v


def hexn(v, n):
    if not isinstance(v, str) or len(v) != 2 * n or any(c not in "0123456789abcdef" for c in v):
        raise Refuse("receipt_malformed")
    return bytes.fromhex(v)


def text(v):
    if not isinstance(v, str):
        raise Refuse("receipt_malformed")
    return v


def read_receipt(o):
    """§8.2: the receipt, or Refuse receipt_malformed."""
    if not isinstance(o, dict):
        raise Refuse("receipt_malformed")
    label = o.get("energy_provenance")
    if label is not None and label not in EVIDENCE:
        raise Refuse("receipt_malformed")
    want = set(MEMBERS) | set(EVIDENCE.get(label, []))
    if set(o) != want or u(o["form"]) != 2:
        raise Refuse("receipt_malformed")
    r = {"room": hexn(o["room"], 32), "cue": hexn(o["cue"], 32), "session": hexn(o["session"], 16),
         "start_offset_ms": u(o["start_offset_ms"]), "format": u(o["format"], 255),
         "asset_list_sha256": hexn(o["asset_list_sha256"], 32), "avail_ms": u(o["avail_ms"]), "slots": u(o["slots"]),
         "funds_remaining_micros": u(o["funds_remaining_micros"]), "pod_hash": hexn(o["pod_hash"], 32),
         "merkle_root": hexn(o["merkle_root"], 32), "budget_spent_micros": u(o["budget_spent_micros"]),
         "work_decide_ops": u(o["work_decide_ops"]), "joules_micro": u(o["joules_micro"]),
         "signer_pubkey": hexn(o["signer_pubkey"], 32), "signer_id": text(o["signer_id"]), "sig": hexn(o["sig"], 64)}
    if not isinstance(o["impressions"], list):
        raise Refuse("receipt_malformed")
    r["impressions"] = []
    for i in o["impressions"]:
        if not isinstance(i, dict) or set(i) != {"slot", "ad_id", "creative_hash", "duration_ms", "price_micros",
                                                 "quartiles_reached"}:
            raise Refuse("receipt_malformed")
        r["impressions"].append({"slot": u(i["slot"]), "ad_id": text(i["ad_id"]),
                                 "creative_hash": hexn(i["creative_hash"], 32), "duration_ms": u(i["duration_ms"]),
                                 "price_micros": u(i["price_micros"]), "quartiles_reached": u(i["quartiles_reached"], 255)})
    g = o["grant"]
    if not isinstance(g, dict) or set(g) != {"grant_hash", "capability", "joule_ceiling_micro", "funds_ceiling"}:
        raise Refuse("receipt_malformed")
    r["grant"] = {"grant_hash": hexn(g["grant_hash"], 32), "capability": text(g["capability"]),
                  "joule_ceiling_micro": u(g["joule_ceiling_micro"]),
                  "funds_ceiling": None if g["funds_ceiling"] is None else u(g["funds_ceiling"])}
    r["class"] = None
    if label is not None:
        c = {"label": label}
        if label in ("OnChipCounter", "CalibratedInstrument"):
            unc = o["energy_uncertainty"]
            if not isinstance(unc, dict) or set(unc) != {"relative_ppm", "window_us"}:
                raise Refuse("receipt_malformed")
            c["ppm"], c["window"] = u(unc["relative_ppm"], 2 ** 32 - 1), u(unc["window_us"])
        if label == "OnChipCounter":
            if o["energy_counter_filtering"] not in FILTERING:
                raise Refuse("receipt_malformed")
            c["filtering"] = FILTERING[o["energy_counter_filtering"]]
        if label == "CalibratedInstrument":
            c["ref"] = text(o["energy_calibration_ref"])
        r["class"] = c
    return r


def b3(*parts):
    return B3.blake3(b"".join(parts))


def be64(v):
    return struct.pack(">Q", v)


def pod_hash(pod):
    """§8.3."""
    out = [POD_DOMAIN, be64(len(pod))]
    for c in pod:
        a = c["ad_id"].encode()
        out += [struct.pack(">I", len(a)), a, c["creative_hash"], be64(c["duration_ms"]), be64(c["price_micros"])]
    return b3(*out)


def leaf(i):
    a = i["ad_id"].encode()
    return b3(IMP_DOMAIN, be64(i["slot"]), struct.pack(">I", len(a)), a, i["creative_hash"], be64(i["duration_ms"]),
              be64(i["price_micros"]), bytes([i["quartiles_reached"]]))


def merkle(hashes):
    """jwp-receipts §2.2, each leaf of its first form."""
    if not hashes:
        return bytes(32)
    level = [b3(LEAF, h) for h in hashes]
    while len(level) > 1:
        if len(level) % 2:
            level.append(level[-1])
        level = [b3(NODE, level[k], level[k + 1]) for k in range(0, len(level), 2)]
    return level[0]


def class_bytes(c):
    """energy-measurement §2.7's signed encoding."""
    if c is None:
        return b"\x00"
    out = bytes([CLASS_TAG[c["label"]]])
    if c["label"] == "OnChipCounter":
        out += struct.pack(">IQ", c["ppm"], c["window"]) + bytes([c["filtering"]])
    elif c["label"] == "CalibratedInstrument":
        ref = c["ref"].encode()
        out += struct.pack(">IQ", c["ppm"], c["window"]) + struct.pack(">I", len(ref)) + ref
    return out


def payload(r):
    """§8.4."""
    g = r["grant"]
    cap, sid = g["capability"].encode(), r["signer_id"].encode()
    out = [BREAK_DOMAIN, r["room"], r["cue"], r["session"], be64(r["start_offset_ms"]), bytes([r["format"]]),
           r["asset_list_sha256"], be64(r["avail_ms"]), be64(r["slots"]), be64(r["funds_remaining_micros"]),
           r["pod_hash"], r["merkle_root"], be64(r["budget_spent_micros"]), be64(r["work_decide_ops"]),
           be64(r["joules_micro"]), class_bytes(r["class"]), g["grant_hash"], struct.pack(">I", len(cap)), cap,
           be64(g["joule_ceiling_micro"])]
    out.append(b"\x00" if g["funds_ceiling"] is None else b"\x01" + be64(g["funds_ceiling"]))
    out.append(be64(len(r["impressions"])))
    out += [leaf(i) for i in r["impressions"]]
    out += [r["signer_pubkey"], struct.pack(">I", len(sid)), sid]
    return b"".join(out)


def own_rules(r):
    """§8.5: None, or the refusal."""
    if r["format"] not in (HLS, DASH):
        return "receipt_invalid"
    bad = ED.verify(r["signer_pubkey"], payload(r), r["sig"])
    if bad:
        return bad
    imps = r["impressions"]
    spend = sum(i["price_micros"] for i in imps if i["quartiles_reached"] >= 1)
    if merkle([leaf(i) for i in imps]) != r["merkle_root"] or spend != r["budget_spent_micros"]:
        return "receipt_invalid"
    f = r["grant"]["funds_ceiling"]
    if f is not None and spend > f * 10 ** 6:
        return "receipt_invalid"
    if any(i["slot"] != k or i["quartiles_reached"] > 5 for k, i in enumerate(imps)) or r["work_decide_ops"] < len(imps):
        return "receipt_invalid"
    c, j = r["class"], r["joules_micro"]
    if c is None:
        return None if j == 0 else "receipt_invalid"
    if j == 0:
        return "receipt_invalid"
    if c["label"] in ("OnChipCounter", "CalibratedInstrument"):
        if c["ppm"] == 0 or c["window"] == 0 or j * c["ppm"] < FLOOR:
            return "receipt_invalid"
    if c["label"] == "CalibratedInstrument" and all(ch in WHITE_SPACE for ch in c["ref"]):
        return "receipt_invalid"
    return None


def identity(r):
    return b3(ID_DOMAIN, payload(r), r["sig"])


# ---------------------------------------------------------------- §9 verification

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


def ads(v, judge=None):
    """§9: a dict of what the corpus states for vector v, and the run's HLS tags. judge, when
    given, makes the judge of a policy's presentation claims (ad-presentation §9)."""
    descriptor = bytes.fromhex(v["descriptor"])
    grants = [bytes.fromhex(g) for g in v["grants"]]
    chains = [[bytes.fromhex(h) for h in c] for c in v["chains"]]
    room = LRV.verify_room(descriptor, grants, v["key_log"], chains)                   # step 1
    if room["outcome"] == "refuse":
        return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}, None
    if room["outcome"] == "gap":
        return {"outcome": "gap", "gaps": room["gaps"]}, None
    try:
        return accept(v, descriptor, grants, chains, room, judge)
    except Stop as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}, None


def accept(v, descriptor, grants, chains, room, judge=None):
    d = LRV.read_descriptor(descriptor)
    held_grants, _ = LRV.check_grants(d, grants)
    try:                                                                               # step 2
        p = read_policy(bytes.fromhex(v["policy"]))
        if p["room"] != d["room"]:
            raise Refuse("wrong_room")
        bad = ED.verify(d["authority"], POLICY_DOMAIN + p["body"], p["sig"])
        if bad:
            raise Refuse(bad)
        decider = next((g for g in held_grants if g["digest"] == p["decider_grant"]), None)
        if decider is None:
            raise Refuse("decider_grant_unknown")
        if LRV.revoked(decider, LRV.check_key_log(d, v["key_log"])):
            raise Refuse("grant_revoked")
    except Refuse as e:
        raise Stop(e.reason, ["policy"])
    held, seen = [], set()                                                             # step 3
    for c, i, _ in room["order"]:
        ev = LEV.read_record(chains[c][i])
        if ev["digest"] not in seen:
            seen.add(ev["digest"])
            held.append((c, i, ev))
    breaks, chapters, written = schedule(held, p)
    t0, pt0 = d["t0"], v["pt0_ms"]                                                     # step 4
    tags = hls_tags(p, t0, breaks, written)
    funds = None if decider["funds"] == NO_CEILING else decider["funds"]
    claims = None
    if requires_claims(p):                                     # ad-presentation §9.2 steps 1, 2
        if judge is None:
            raise ValueError("a policy that requires presentation claims: verify it with ad_presentation_verify.py")
        claims = judge(v, d, p)
    stated, sessions, total = [], set(), 0                                             # step 5
    for k, (o, pod_json) in enumerate(zip(v["receipts"], v["pods"] + [None] * len(v["receipts"]))):
        at = ["receipt", k]
        try:
            r = read_receipt(o)
        except Refuse as e:
            raise Stop(e.reason, at)
        if r["room"] != d["room"]:
            raise Stop("wrong_room", at)
        bad = own_rules(r)
        if bad:
            raise Stop(bad, at)
        if r["signer_pubkey"] != decider["principal"]:
            raise Stop("receipt_not_decider", at)
        g = r["grant"]
        if (g["grant_hash"] != decider["digest"] or g["capability"] != "wai.live.room"
                or g["joule_ceiling_micro"] != decider["joules"] or g["funds_ceiling"] != funds):
            raise Stop("receipt_grant_mismatch", at)
        b = next((x for x in breaks if x["digest"] == r["cue"]), None)
        if b is None:
            raise Stop("receipt_break_unknown", at)
        if not LRV.good_at(decider, d, b["t"]):
            raise Stop("outside_grant_window", at)
        pod = [{"ad_id": c["ad_id"], "creative_hash": bytes.fromhex(c["creative_hash"]), "duration_ms": c["duration_ms"],
                "price_micros": c["price_micros"]} for c in (pod_json or [])]
        if r["pod_hash"] != pod_hash(pod):
            raise Stop("pod_mismatch", at)
        fill = decide(pod, p["slots"], max(b["avail_ms"] - r["start_offset_ms"], 0), r["funds_remaining_micros"])
        got = [(i["slot"], i["ad_id"], i["creative_hash"], i["duration_ms"], i["price_micros"]) for i in r["impressions"]]
        if r["avail_ms"] != b["avail_ms"] or r["slots"] != p["slots"] or got != fill:
            raise Stop("decision_mismatch", at)
        listed = asset_list(p, r["format"], fill)
        if r["asset_list_sha256"] != sha(listed):
            raise Stop("asset_list_mismatch", at)
        if (r["cue"], r["session"]) in sessions:
            raise Stop("receipt_duplicate", at)
        sessions.add((r["cue"], r["session"]))
        counted, extra = r["budget_spent_micros"], {}                                  # step 12
        if claims is not None:
            counted, extra = claims.receipt(k, r)
        total += counted
        if funds is not None and (r["funds_remaining_micros"] > funds * 10 ** 6 or total > funds * 10 ** 6):
            raise Stop("funds_exceeded", at)
        fact = {"identity": identity(r).hex(), "asset_list": listed.decode(), "spend_micros": r["budget_spent_micros"]}
        fact.update(extra)
        stated.append(fact)
    out = {
        "outcome": "accept",
        "room": d["room"].hex(),
        "breaks": [{"digest": b["digest"].hex(), "id": break_id(b["digest"]), "event_id": event_id(b["digest"]),
                    "author": b["author"].hex(), "cue_id": b["cue_id"], "t": b["t"],
                    "type": "out" if b["type"] == OUT else "break", "placement": "post" if b["placement"] == POST else "mid",
                    "planned_ms": b["planned_ms"], "avail_ms": b["avail_ms"], "duration_ms": b["duration_ms"],
                    "closed_by": b["closed_by"].hex() if b["closed_by"] else None, "end": b["end"],
                    "hls_list_uri": hls_uri(p, b["digest"]), "dash_list_uri": dash_uri(p, b["digest"])} for b in breaks],
        "chapters": [{"digest": c["digest"].hex(), "t": c["t"], "duration_ms": c["duration_ms"]} for c in chapters],
        "hls": v["name"] + ".m3u8",
        "dash": dash_streams(p, pt0, breaks),
        "dash_open": [{"break": k, "event": dash_event(p, pt0, b, False)} for k, b in enumerate(breaks) if b["type"] == OUT],
        "receipt_facts": stated,
        "spend_micros": min(total, U64),
    }
    if claims is not None:
        out["claim_facts"] = claims.finish()
    return out, "".join(t + "\n" for t in tags).encode()


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

def check(doc, files, quiet=False):
    """Every room against its statement, and every tag file: the failures, and the files written."""
    fails, written = [], {}
    for v in doc.get("vectors", []):
        try:
            got, tags = ads(v)
        except Exception as e:  # a crash on a mutated copy is a failure to read it
            got, tags = None, None
            bad = ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
        else:
            bad = [k for k in got if v.get(k) != got[k]]
            bad += [k for k in ("outcome", "reason", "at", "gaps", "breaks", "hls", "dash", "receipt_facts")
                    if k in v and k not in got]
        if tags is not None:
            name = v["name"] + ".m3u8"
            written[name] = tags
            if files.get(name) != tags:
                bad.append("its tag file %s" % name)
        if not quiet:
            what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
            print(("PASS" if not bad else "FAIL") + " room %s: %s" % (v.get("name"), what))
        fails += ["room %s: %s differs from the corpus" % (v.get("name"), k) for k in bad]
    return fails, written


def flip(h, at):
    b = bytearray(bytes.fromhex(h))
    b[at] ^= 0x01
    return b.hex()


def mutations(doc, files):
    """Copies of the corpus, each with one defect a conforming verifier must catch."""
    vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
    live, mid, post, chap, dup = vn["live_break"], vn["mid_roll"], vn["post_roll"], vn["chapters"], vn["receipt_duplicate"]

    def mutate(label, f):
        d, fs = copy.deepcopy(doc), dict(files)
        f(d["vectors"], fs)
        return label, d, fs

    def drop_last(vs, chain_of):
        """The last record of the chain holding a producer's cues."""
        ch = vs[chain_of]["chains"]
        longest = max(range(len(ch)), key=lambda c: len(ch[c]))
        ch[longest].pop()

    def swap_pod(vs, fs):
        pod = vs[mid]["pods"][0]
        pod[1], pod[3] = pod[3], pod[1]

    def edit_file(name, a, b):
        return lambda vs, fs: fs.update({name: fs[name].replace(a, b, 1)})

    return [
        mutate("a byte of the live break's policy signature", lambda vs, fs: vs[live].update(policy=flip(vs[live]["policy"], -2))),
        mutate("the live break's in withheld from the end of its chain", lambda vs, fs: drop_last(vs, live)),
        mutate("the post-roll's annotation withheld", lambda vs, fs: drop_last(vs, post)),
        mutate("a stated asset list with one ad's duration changed",
               lambda vs, fs: vs[live]["receipt_facts"][0].update(asset_list=vs[live]["receipt_facts"][0]["asset_list"].replace("15.000", "14.000", 1))),
        mutate("the DASH events stated with executeOnce false",
               lambda vs, fs: vs[live].update(dash=vs[live]["dash"].replace('executeOnce="true"', 'executeOnce="false"'))),
        mutate("the open out's DASH event stated with its duration",
               lambda vs, fs: vs[live]["dash_open"][0].update(event=vs[live]["dash"].split(">", 1)[1].rsplit("</EventStream>", 1)[0])),
        mutate("a byte of the mid-roll's tag file", edit_file("mid_roll.m3u8", b"X-RESUME-OFFSET=30.000", b"X-RESUME-OFFSET=31.000")),
        mutate("the post-roll's tag stated without POST", edit_file("post_roll.m3u8", b'CUE="POST,ONCE"', b'CUE="ONCE"')),
        mutate("the mid-roll pod given in another order", swap_pod),
        mutate("the mid-roll's presentation time of t0 changed", lambda vs, fs: vs[mid].update(pt0_ms=vs[mid]["pt0_ms"] + 1)),
        mutate("a live break receipt's quartile raised",
               lambda vs, fs: vs[live]["receipts"][0]["impressions"][1].update(quartiles_reached=4)),
        mutate("a live break receipt's session changed",
               lambda vs, fs: vs[live]["receipts"][1].update(session="00" * 16)),
        mutate("the stated spend", lambda vs, fs: vs[live].update(spend_micros=vs[live]["spend_micros"] + 1)),
        mutate("a chapter's stated length", lambda vs, fs: vs[chap]["chapters"][0].update(duration_ms=59999)),
        mutate("the duplicated receipt given once", lambda vs, fs: vs[dup]["receipts"].pop()),
        mutate("the funds control's second receipt given once",
               lambda vs, fs: vs[vn["funds_exceeded"]]["receipts"].pop()),
        mutate("a tag file removed", lambda vs, fs: fs.pop("chapters.m3u8")),
    ]


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    fails += ["ed25519 self-test"] if st else []
    ok = len(REASONS) == 20
    print(("PASS" if ok else "FAIL") + " §10 read from the extension: %d reasons" % len(REASONS))
    fails += [] if ok else ["§10 does not read"]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    files = {}
    for name in sorted(os.listdir(corpus_dir)):
        if name.endswith(".m3u8"):
            with open(os.path.join(corpus_dir, name), "rb") as f:
                files[name] = f.read()
    got, written = check(doc, files)
    fails += got
    extra = sorted(set(os.listdir(corpus_dir)) - set(written) - {"vectors.json", "README.md"})
    print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and the tag files the rooms write")
    fails += ["an extra file: %s" % n for n in extra]
    seen = {v.get("reason") for v in doc["vectors"]}
    missing = [r for r in REASONS if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of §10 appears in the corpus")
    fails += ["no case gives %s" % r for r in missing]
    for label, d, fs in mutations(doc, files):
        caught, _ = check(d, fs, quiet=True)
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " live-ads corpus: %d rooms, %d tag files" % (len(doc["vectors"]), len(files)))
    return 1 if fails else 0


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