#!/usr/bin/env python3
"""Second verifier for http-conformance/ — HTTP delivery (extensions/http-delivery.md).

Written from the extension text and SPEC §7.1. It shares no code with wai-rs, which
generates the corpus: the head (§2, §5.1), covered_len, the head proof (§6), the
response checks (§7.1), the plans and classes of the reference client (§7, §10.1), the
byte-range procedure with its whole-object fallback (§7), the base presented as it
verifies and the refinement deadline on the caller's clock (§10.1, §10.2), SPEC §7.1's selection with
companions and refinements, the codecs grammar, union, floor and support (§4), all
written here. BLAKE3 is ../claim-conformance/_blake3.py (written from its specification,
shared with the other Python verifiers) and SHA-256 is hashlib. The registry's kinds,
roles and classes are read from ../registry/capabilities.json, the published export.
Only the standard library is needed, except for the receipt and key cases of §8.1-§8.5,
which http_receipts.py checks through the corpora's own receipt, claim and key-discovery
verifiers and so with the `cryptography` package; without it those cases fail.

Pins (SPEC §3.1) are read as far as the corpus needs: a pin applies to a capability that
decodes against a parameter set (registry class neural-int-exact, neural-entropy-exact,
neural-float or generative-exact) or that the registry does not hold; a form-less pin
resolves when the sink holds a file whose SHA-256 is the pin's; a pin in the set form
resolves to nothing here, since the corpus's sinks hold no set listing. Any other shape
stops the verifier, never passes.

    python3 http_verify.py [dir]                       verify every case and the self-test
    python3 http_verify.py [dir] --mutate NAME         the same with a deliberate defect:
        left-half | no-root | counter-zero | floor-as-codecs | blake3-iv | base-at-done |
        tick-noop | refinement-unclipped | nonconforming-offered | conforming-withheld |
        refines-read-whole | … | rcpt-name-unchecked | legacy-admitted |
        walk-skipped | walk-unclamped | memo-off | budget-unchecked | bytes-unchecked |
        caps-unchecked | nothing-as-delivered | groups-from-listed | step6-skipped |
        decision-unchecked | recovery-record-unchecked | declaration-unnamed |
        first-publisher | matchers-need-object | operating-point-unchecked |
        session-accepted | bytes-read-whole | key-log-uncapped | ancestor-any |
        anchor-follows-redirect |
        cached-rollback-unchecked | grants-module-unchecked | set-grants-ignored |
        set-by-name (falsify/run.sh runs every one)
    python3 http_verify.py cases FILE [dir]
        give every case of a case file (wai_http_vectors fuzz) its stated outcome
    python3 http_verify.py diff FILE [--flip N --seed S]
        verify every head proof `wai_http_vectors diff` wrote; with --flip, change one
        chaining-value byte in N of them, chosen by S, and require exactly those refused

Exit status 0 only when every case gives its committed outcome (or, for diff, every
proof verifies, or exactly the flipped ones fail).
"""
import hashlib, json, os, random, re, sys, time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
sys.path.insert(0, os.path.join(HERE, "..", "staged-conformance"))
# Its own directory too (http_receipts.py, imported where receipts are checked): under
# `python3 -I` the script's directory is not on sys.path.
sys.path.insert(0, HERE)
import _blake3 as B3  # noqa: E402
import staged_verify as SV  # noqa: E402  (strict JSON, the manifest members, refines, explode)

MUTATE = None
DOMAIN = b"moq-jwp:object\x01"
READABLE = ["wai.prior.set"]


class Refusal(Exception):
    def __init__(self, code, skipped=None):
        super().__init__(code)
        self.code, self.skipped = code, skipped


def sha(b):
    return hashlib.sha256(b).hexdigest()


def blake3(b):
    return B3.blake3(b)


def content_hash(env):
    return blake3(DOMAIN + env)


# ------------------------------------------------------------------ the registry

REG = {}


def load_registry():
    d = json.load(open(os.path.join(HERE, "..", "registry", "capabilities.json")))
    for c in d["capabilities"]:
        REG[c["capability"]] = c


def kind(cap):
    return REG[cap]["kind"] if cap in REG else None


def candidate(cap):
    return cap in REG and REG[cap]["status"] == "candidate"


def dispatchable(cap):
    k = kind(cap)
    return k is None or (k == "payload" and not candidate(cap))


def enhancement(cap):
    return cap in REG and REG[cap]["role"] == "enhancement"


def pin_applies(cap):
    return cap not in REG or REG[cap]["replicate"] in ("neural-int-exact", "neural-entropy-exact", "neural-float", "generative-exact")


# ---------------------------------------------------------------------- the head

def u32(b, at):
    return int.from_bytes(b[at:at + 4], "little")


def read_manifest(b, v2):
    """SPEC §2, §3, §7.1: the manifest read strictly, with the members SPEC requires and
    their types (staged_verify's reader); `head_malformed` otherwise."""
    try:
        # SPEC §2, *Reader limits* (--mutate manifest-values-uncounted drops the value limit).
        m = SV.strict_json(b, None if MUTATE == "manifest-values-uncounted" else SV.MANIFEST_MAX_VALUES)
        (SV.check_manifest_v2 if v2 else SV.check_manifest_v1)(m)
    except (SV.NotJSON, SV.Schema):
        raise Refusal("head_malformed")
    return m


def parse_head(prefix, n):
    """§2, §5.1: ("need", k) or a head dict; Refusal on a malformed or mis-sized object."""
    if len(prefix) < 8:
        return ("need", 8)
    magic = prefix[:4]
    if magic not in (b"WAI1", b"WAI2"):
        raise Refusal("head_malformed")
    ml = u32(prefix, 4)
    # SPEC §2, *Reader limits*: refused before it is asked for (--mutate
    # manifest-limit-dropped asks for it).
    if ml > SV.MANIFEST_MAX and MUTATE != "manifest-limit-dropped":
        raise Refusal("head_malformed")
    if magic == b"WAI1":
        head_len = 12 + ml
        if len(prefix) < head_len:
            return ("need", head_len)
        comps = [(head_len, u32(prefix, 8 + ml))]
    else:
        if len(prefix) < 10 + ml:
            return ("need", 10 + ml)
        k = int.from_bytes(prefix[8 + ml:10 + ml], "little")
        head_len = 10 + ml + 8 * k
        if len(prefix) < head_len:
            return ("need", head_len)
        comps, expected = [], 0
        for i in range(k):
            off, ln = u32(prefix, 10 + ml + 8 * i), u32(prefix, 14 + ml + 8 * i)
            if off != expected:
                raise Refusal("head_malformed")
            comps.append((head_len + off, ln))
            expected += ln
    if head_len + sum(c[1] for c in comps) != n:
        raise Refusal("object_length_mismatch")
    m = read_manifest(prefix[8:8 + ml], magic == b"WAI2")
    if magic == b"WAI2" and len(m["renditions"]) != len(comps):
        raise Refusal("head_malformed")
    return {"version": magic.decode(), "head_len": head_len, "envelope_len": n, "components": comps, "manifest": m}


def is_layer(h):
    """§10.4: a WAI1 whose capability is refinement-only (staged delivery §3). A payload
    capability that carries `refines` is an ordinary object: SPEC §4 dispatches it and
    `refines` is not read."""
    if h["version"] != "WAI1":
        return False
    return kind(h["manifest"]["model_requirement"]["capability"]) == "refinement"


def whole_head(env):
    """The head of a whole envelope; one shorter than its head is malformed."""
    h = parse_head(env, len(env))
    if isinstance(h, tuple):
        raise Refusal("head_malformed")
    return h


def covered_len(head_len, n):
    return min(n, 1024 * -(-(15 + head_len) // 1024) - 15)


# ------------------------------------------------------------------ the head proof

def left(n):
    if MUTATE == "left-half":
        return max(1, (n // 2) // 1024) * 1024
    return 1024 * 2 ** ((((n - 1) // 1024)).bit_length() - 1)


def merge(l, r, root):
    if MUTATE == "no-root":
        root = False
    return B3.parent_cv(l, r, root)


def subtree(x, start):
    return B3.subtree_cv(x, 0 if MUTATE == "counter-zero" else start)


def read_proof(b):
    if len(b) < 53 or b[:4] != b"WHP1":
        raise Refusal("head_proof_malformed")
    k = b[52]
    if k > 64 or len(b) != 53 + 32 * k:
        raise Refusal("head_proof_malformed")
    n, c = int.from_bytes(b[36:44], "little"), int.from_bytes(b[44:52], "little")
    if n > 2 ** 64 - 1 - 15 or c > n:
        raise Refusal("head_proof_malformed")
    if not (c == n or (c < n and (15 + c) % 1024 == 0)) or (c == n and k):
        raise Refusal("head_proof_malformed")
    return {"cid": b[4:36], "n": n, "covered": c, "cvs": [b[53 + 32 * i:85 + 32 * i] for i in range(k)]}


def verify_proof(p, cid, prefix):
    """§6.3, checks 1-3, in order."""
    if p["cid"] != cid or len(prefix) < p["covered"]:
        raise Refusal("head_proof_malformed")
    covered = prefix[:p["covered"]]
    if p["covered"] == p["n"]:
        if content_hash(covered) != cid:
            raise Refusal("head_proof_mismatch")
    else:
        x, L, c = DOMAIN + covered, 15 + p["n"], 15 + p["covered"]
        it = iter(p["cvs"])

        def cv(start, ln):
            if start >= c:
                v = next(it, None)
                if v is None:
                    raise Refusal("head_proof_mismatch")
                return v
            if start + ln <= c:
                return subtree(x[start:start + ln], start)
            l = left(ln)
            return merge(cv(start, l), cv(start + l, ln - l), False)

        l = left(L)
        root = merge(cv(0, l), cv(l, L - l), True)
        if next(it, None) is not None or root != cid:
            raise Refusal("head_proof_mismatch")
    try:
        h = parse_head(covered, p["n"])
    except Refusal as e:
        if e.code == "object_length_mismatch":
            raise Refusal("head_proof_malformed")
        raise
    if isinstance(h, tuple) or covered_len(h["head_len"], p["n"]) != p["covered"]:
        raise Refusal("head_proof_malformed")


# ------------------------------------------------------------- responses (§7.1)

DIGITS = re.compile(r"[0-9]+", re.ASCII)


def digits(s):
    """RFC 9110's 1*DIGIT: ASCII digits only, no sign, separator, space or other script."""
    return int(s) if DIGITS.fullmatch(s) else None


def check_response(req, status, cr, enc, body_len, n):
    coded = enc is not None and enc.strip() and enc.strip().lower() != "identity"
    if status == 206:
        if req is None:
            raise Refusal("http_status_unexpected")
        if coded:
            raise Refusal("content_coding_present")
        field = (cr or "").strip(" \t")
        if not field.startswith("bytes ") or "/" not in field:
            raise Refusal("content_range_mismatch")
        span, complete = field[len("bytes "):].split("/", 1)
        if "-" not in span:
            raise Refusal("content_range_mismatch")
        a, b = span.split("-", 1)
        a, b = digits(a), digits(b)
        if a is None or b is None or a != req[0] or b != min(req[1], max(n - 1, 0)) or b < a or body_len != b - a + 1:
            raise Refusal("content_range_mismatch")
        if digits(complete) != n:
            raise Refusal("complete_length_mismatch")
        return (a, b)
    if status == 200:
        if coded:
            raise Refusal("content_coding_present")
        if body_len != n or n == 0:
            raise Refusal("complete_length_mismatch")
        return (0, n - 1)
    raise Refusal("http_status_unexpected")


def origin_serve(n, rng):
    """An origin's answer to a Range header on n bytes (RFC 9110 §14), as the corpus's
    origin gives it: (status, Content-Range, first, end)."""
    if rng is None or not rng.startswith("bytes=") or "," in rng:
        return 200, None, 0, n
    a, b = rng[len("bytes="):].split("-")
    if a == "":
        s = int(b)
        return 206, "bytes %d-%d/%d" % (max(0, n - s), n - 1, n), max(0, n - s), n
    first = int(a)
    last = n - 1 if b == "" else min(int(b), n - 1)
    if first >= n:
        return 416, "bytes */%d" % n, 0, 0
    return 206, "bytes %d-%d/%d" % (first, last, n), first, last + 1


# ------------------------------------------------------------------ selection

def selectable(e):
    return e.get("role") not in ("companion", "refinement") and dispatchable(e["capability"])


def resolve_pin(cap, prior, held):
    if not isinstance(prior, dict) or "sha256" not in prior:
        raise SystemExit("http_verify: a pin shape this verifier does not read")
    form = prior.get("form")
    if form is None:
        return "ok" if any(sha(h) == prior["sha256"] for h in held) else "prior_not_held"
    if form in READABLE:
        return "prior_not_held"
    raise SystemExit("http_verify: a pin form this verifier does not read")


def plaintext_v2(m):
    """SPEC §6.4 step 2 for a WAI2: `security` absent or null, or an object whose
    `encryption` is absent, null or "none"; a member that is not a security object (a
    string, `true`, a mistyped member) means the payloads cannot be read here."""
    sec = m.get("security")
    if sec is None:
        return True
    if not isinstance(sec, dict):
        return False
    enc, kid = sec.get("encryption"), sec.get("kid")
    if enc is not None and not isinstance(enc, str):
        return False
    if kid is not None and not SV.a_u64(kid):
        return False
    return enc in (None, "none")


def select(m, order, adv, held, digest_ok):
    """SPEC §7.1 step 2 in `order`: (pick, skipped) or Refusal('inert', skipped)."""
    entries, skipped = m["renditions"], []
    if not plaintext_v2(m):
        raise Refusal("inert", [[i, "unsupported_encryption"] for i in range(len(entries))])
    for i in order:
        if i >= len(entries):
            continue
        e = entries[i]
        if not selectable(e):
            skipped.append([i, "not_selectable"]); continue
        if e["capability"] not in adv:
            skipped.append([i, "missing_capability"]); continue
        if not digest_ok(i, e):
            skipped.append([i, "component_digest_mismatch"]); continue
        if e.get("prior") is not None and pin_applies(e["capability"]):
            r = resolve_pin(e["capability"], e["prior"], held)
            if r != "ok":
                skipped.append([i, r]); continue
        return i, skipped
    raise Refusal("inert", skipped)


def dispatch_wai1(m, adv, held):
    """SPEC §4 with §6.4 step 2 first: an encrypted payload is inert at a sink that
    opens no scheme."""
    sec = m.get("security")
    if sec is not None and sec.get("encryption") not in (None, "none"):
        raise Refusal("inert", [[0, "unsupported_encryption"]])
    mr = m["model_requirement"]
    cap, fb = mr["capability"], mr.get("fallback")
    if kind(cap) in ("companion", "refinement"):
        raise Refusal("inert", [[0, "missing_capability"]])
    first = None
    for c in [cap, fb]:
        if c is None or not dispatchable(c) or c not in adv:
            continue
        p = m.get("conditioning", {}).get("prior")
        if p is not None and pin_applies(c):
            r = resolve_pin(c, p, held)
            if r == "ok":
                return 0
            first = first or r
            continue
        return 0
    raise Refusal("inert", [[0, first or "missing_capability"]])


def companions_of(entries, i):
    base = entries[i]
    out = []
    for j, e in enumerate(entries):
        if e.get("role") != "companion":
            continue
        cid = e.get("companion_of_id")
        if (cid is not None and base.get("id") == cid) or (cid is None and e.get("companion_of") == base["capability"]):
            out.append(j)
    return out


def refinements_of(entries, i):
    """The refinement components whose well-formed `refines` (staged-delivery §3, ids in
    place of hashes) names entry i's id as its base."""
    bid = entries[i].get("id")
    if bid is None:
        return []
    out = []
    for j, e in enumerate(entries):
        if e.get("role") != "refinement":
            continue
        try:
            if SV.parse_refines_ids(e.get("refines"))["base"] == bid:
                out.append(j)
        except (SV.Malformed, KeyError, TypeError):
            pass
    return out


def refinements_naming(entries, i):
    """The refinement components whose `refines.base`, read on its own, is entry i's id:
    refinements_of without the rest of `refines` read. Of a conforming envelope the two
    agree; of one that does not, these are the refinements a plan drops and reports
    (components_nonconforming, http-delivery §7 step 4), a malformed `refines` naming the
    pick among them."""
    if MUTATE == "refines-read-whole":
        return refinements_of(entries, i)
    bid = entries[i].get("id")
    if bid is None:
        return []
    return [j for j, e in enumerate(entries) if e.get("role") == "refinement"
            and isinstance(e.get("refines"), dict) and isinstance(e["refines"].get("base"), str)
            and e["refines"]["base"] == bid]


def payload_ok(e, payload):
    return e.get("sha256") is None or e["sha256"] == sha(payload)


# ------------------------------------------------------------------- the plan

def plan(head, wanted, held, permille):
    """§7 steps 5-6 and the threshold, over base-class bytes only:
    ("rest", (first, last), why) or ("ranges", [...], dropped)."""
    n, cov = head["envelope_len"], covered_len(head["head_len"], head["envelope_len"])
    if cov == n:
        return ("ranges", [], [])
    rest = (cov, n - 1)
    entries = head["manifest"]["renditions"]
    for i, cl in wanted:
        if cl == "base" and entries[i].get("sha256") is None:
            return ("rest", rest, "component_digest_absent")
    pieces, dropped = [], []
    for i, cl in wanted:
        if entries[i].get("sha256") is None:
            dropped.append([i, "component_digest_absent"])
            continue
        s, ln = head["components"][i]
        segs = [(s, s + ln)]
        for (hf, hl) in held:
            nxt = []
            for a, b in segs:
                hs, he = hf, hl + 1
                if he <= a or hs >= b:
                    nxt.append((a, b)); continue
                if a < hs:
                    nxt.append((a, hs))
                if he < b:
                    nxt.append((he, b))
            segs = nxt
        pieces += [(0 if cl == "base" else 1, a, b) for a, b in segs]
    pieces.sort()
    merged = []
    for c, a, b in pieces:
        if merged and merged[-1][0] == c and merged[-1][2] >= a:
            merged[-1][2] = max(merged[-1][2], b)
        else:
            merged.append([c, a, b])
    base = sum(b - a for c, a, b in merged if c == 0)
    if base > 0 and base * 1000 >= permille * (n - cov):
        return ("rest", rest, "range_threshold")
    return ("ranges", [(a, b - 1, "base" if c == 0 else "refinement") for c, a, b in merged if b > a], dropped)


# ---------------------------------------------------------------- the codecs

def cap_ok(cap):
    parts = cap.split(".")
    lc = lambda s, extra="": s and all(ch in "abcdefghijklmnopqrstuvwxyz0123456789" + extra for ch in s)
    return len(parts) >= 2 and lc(parts[0]) and all(lc(p, "_") for p in parts[1:])


def check_cap(c, first):
    if not cap_ok(c):
        raise Refusal("codecs_malformed")
    full = "wai." + c
    k = kind(full)
    if k in ("derivation", "record"):
        raise Refusal("codecs_not_media")
    if candidate(full):
        raise Refusal("codecs_candidate")
    if first and k in ("companion", "refinement"):
        raise Refusal("codecs_malformed")
    if not first and k == "payload":
        raise Refusal("codecs_malformed")


def check_alt(alt):
    for i, c in enumerate(alt):
        check_cap(c, i == 0)


def parse_element(s):
    if "." not in s:
        raise Refusal("codecs_malformed")
    prefix, rest = s.split(".", 1)
    if prefix not in ("wai1", "wai2"):
        raise Refusal("codecs_malformed")
    alts = []
    for a in rest.split("|"):
        alt = a.split("+")
        check_alt(alt)
        if len(set(alt)) != len(alt) or any(x[0] == alt[0] for x in alts):
            raise Refusal("codecs_malformed")
        alts.append(alt)
    return (prefix, alts)


def render(e):
    return e[0] + "." + "|".join("+".join(a) for a in e[1])


def parse_codecs(value):
    out = []
    for e in (x.strip(" \t") for x in value.split(",")):
        if e == "":
            raise Refusal("codecs_malformed")
        if e.encode("utf-8")[:3].lower() == b"wai":
            out.append(render(parse_element(e)))
        else:
            out.append(None)
    return out


def short(c):
    return c[4:] if c.startswith("wai.") else c


def element_of(head):
    if is_layer(head):
        raise Refusal("codecs_malformed")
    m = head["manifest"]
    if head["version"] == "WAI1":
        mr = m["model_requirement"]
        alts, prefix = [], "wai1"
        for c in [mr["capability"], mr.get("fallback")]:
            if c is not None and dispatchable(c) and short(c) not in [a[0] for a in alts]:
                alts.append([short(c)])
    else:
        e, alts, prefix = m["renditions"], [], "wai2"
        for i, r in enumerate(e):
            if not selectable(r):
                continue
            cap = short(r["capability"])
            a = next((x for x in alts if x[0] == cap), None)
            if a is None:
                a = [cap]; alts.append(a)
            for j in sorted(companions_of(e, i) + refinements_of(e, i)):
                p = short(e[j]["capability"])
                if p not in a[1:]:
                    a.append(p)
        pos = lambda p: next(k for k, r in enumerate(e) if short(r["capability"]) == p)
        alts = [[a[0]] + sorted(a[1:], key=pos) for a in alts]
    if not alts:
        raise Refusal("codecs_not_media")
    for a in alts:
        check_alt(a)
    return (prefix, alts)


def seq_object(env):
    h = whole_head(env)
    comps = []
    if h["version"] == "WAI2":
        try:
            comps = [content_hash(x).hex() if x is not None else None for x in SV.explode_each(env)]
        except (SV.NotWai2, SV.NotJSON, SV.Schema, ValueError, KeyError):
            comps = []
    return {"cid": content_hash(env).hex(), "head": h, "components": comps}


def base_capability(b, base):
    """The capability through which object b is the base whose content hash is `base`: a
    WAI1's capability when its cid is `base`, or the capability of the selectable component
    of a WAI2 whose exploded envelope has that hash; None otherwise (§4.2)."""
    if is_layer(b["head"]):
        return None
    if b["cid"] == base and b["head"]["version"] == "WAI1":
        return short(b["head"]["manifest"]["model_requirement"]["capability"])
    if base in b["components"]:
        r = b["head"]["manifest"]["renditions"][b["components"].index(base)]
        return short(r["capability"]) if selectable(r) else None
    return None


def described(objects):
    """§4.2: each object that is not a layer as an element, each layer's capability after
    `+` in the alt of every object of the sequence that is its base. An object listed more
    than once is one object, described once with every layer it has."""
    seen, uniq = set(), []
    for o in objects:
        if o["cid"] not in seen:
            seen.add(o["cid"])
            uniq.append(o)
    objects = uniq
    els = [None if is_layer(o["head"]) else element_of(o["head"]) for o in objects]
    for o in objects:
        if not is_layer(o["head"]):
            continue
        m = o["head"]["manifest"]
        cap = short(m["model_requirement"]["capability"])
        check_cap(cap, False)
        try:
            base = SV.parse_refines(m.get("refines"))["base"]
        except SV.Malformed:
            raise Refusal("codecs_malformed")
        targets = [(k, bc) for k, b in enumerate(objects) for bc in [base_capability(b, base)] if bc is not None]
        if not targets:
            raise Refusal("codecs_layer_unbound")
        for k, bc in targets:
            alt = next((a for a in els[k][1] if a[0] == bc), None)
            if alt is None:
                raise Refusal("codecs_layer_unbound")
            if cap not in alt[1:]:
                alt.append(cap)
    return [e for e in els if e is not None]


def union_of(objects):
    if MUTATE == "floor-as-codecs":
        return floor_of(objects)
    each = described(objects)
    prefix = "wai2" if any(o["head"]["version"] == "WAI2" for o in objects) else "wai1"
    alts = []
    for p, al in each:
        for a in al:
            x = next((x for x in alts if x[0] == a[0]), None)
            if x is None:
                alts.append(list(a))
            else:
                x += [c for c in a[1:] if c not in x[1:]]
    if not alts:
        raise Refusal("codecs_no_common_capability")
    return (prefix, alts)


def floor_of(objects):
    each = described(objects)
    prefix = "wai2" if any(o["head"]["version"] == "WAI2" for o in objects) else "wai1"
    seen = []
    for e in each:
        for a in e[1]:
            x = next((s for s in seen if s[0] == a[0]), None)
            if x is None:
                seen.append(list(a))
            else:
                x += [c for c in a[1:] if c not in x[1:]]
    out = []
    for a in seen:
        found = [next((x for x in e[1] if x[0] == a[0]), None) for e in each]
        if any(f is None for f in found):
            continue
        out.append([a[0]] + [c for c in a[1:] if all(c in f[1:] for f in found)])
    if not out:
        raise Refusal("codecs_no_common_capability")
    return (prefix, out)


def check_declared(head, union, base):
    """§4.2: a layer is declared only after `+` in the alt of its own base's capability,
    the base (`base`, a sequence object) being the one its `refines.base` names."""
    if is_layer(head):
        m = head["manifest"]
        cap = short(m["model_requirement"]["capability"])
        try:
            named = SV.parse_refines(m.get("refines"))["base"]
        except SV.Malformed:
            raise Refusal("codecs_undeclared")
        bc = base_capability(base, named) if base is not None else None
        if bc is None or not any(a[0] == bc and cap in a[1:] for a in union[1]):
            raise Refusal("codecs_undeclared")
        return
    p, alts = element_of(head)
    if p > union[0]:
        raise Refusal("codecs_undeclared")
    for a in alts:
        u = next((x for x in union[1] if x[0] == a[0]), None)
        if u is None or any(c not in u[1:] for c in a[1:]):
            raise Refusal("codecs_undeclared")


def supports(floor, containers, caps):
    reads = "WAI1" in containers and (floor[0] == "wai1" or "WAI2" in containers)
    return reads and any(("wai." + a[0]) in caps for a in floor[1])


def parse_layer_tag(line):
    """§10.4's #WAI-LAYER line, as an HLS attribute list."""
    bad = Refusal("layer_tag_malformed")
    if not line.startswith("#WAI-LAYER:"):
        raise bad
    rest = line[len("#WAI-LAYER:"):].encode("utf-8")
    n, uri, names = None, None, []
    while True:
        eq = rest.find(b"=")
        if eq < 0:
            raise bad
        name = rest[:eq]
        if not name or not re.fullmatch(rb"[A-Z0-9-]+", name) or name in names:
            raise bad
        names.append(name)
        rest = rest[eq + 1:]
        if rest[:1] == b'"':
            close = rest.find(b'"', 1)
            if close < 0:
                raise bad
            value, quoted, after = rest[1:close], True, rest[close + 1:]
            if b"\r" in value or b"\n" in value:
                raise bad
        else:
            end = rest.find(b",")
            end = len(rest) if end < 0 else end
            value, quoted, after = rest[:end], False, rest[end:]
            if not value or b'"' in value or re.search(rb"[ \t\n\r\x0c]", value):
                raise bad
        if name == b"N":
            if quoted or len(value) > 3 or not re.fullmatch(rb"[0-9]+", value) or not 1 <= int(value) <= 255:
                raise bad
            n = int(value)
        elif name == b"URI":
            if not quoted:
                raise bad
            try:
                uri = value.decode("utf-8")
            except UnicodeDecodeError:
                raise bad
        if not after:
            break
        if after[:1] != b",":
            raise bad
        rest = after[1:]
    if n is None or uri is None:
        raise bad
    return {"n": n, "uri": uri}


# ------------------------------------------------------------- the session (§7)

class Session:
    """The reference client of §7, step by step as the requests are answered: it says
    which requests to make, checks each response, and decides as each arrives."""

    def __init__(self, cid, frag, permille, sink, layer=False, deadline=None):
        self.cid, self.frag, self.permille, self.sink, self.deadline = cid, frag, permille, sink, deadline
        self.next_id, self.queued, self.issued, self.early = 0, [], {}, []
        self.proof, self.held, self.head, self.stage = None, [], None, ("head",)
        self.abandoned, self.dropped, self.body, self.requests, self.result = [], [], 0, 0, None
        # §10.1, §10.2: the caller's clock as last given; the base once it verified, with
        # the time of the response that verified it; each verified component, frozen.
        self.now, self.base, self.base_at, self.verified = 0, None, None, {}
        if layer:
            self.stage = ("layer",)
            self.queue(("whole", "refinement"))
            return
        n = frag.get("n")
        if n is not None and n < 1009:
            self.stage = ("rest", "below_one_chunk", "base")
            self.queue(("whole", "base"))
            return
        e = frag.get("h") or 16369
        if n is not None:
            e = min(e, n)
        self.queue(("range", (0, e - 1), "base"))
        self.queue(("proof",))

    @staticmethod
    def cls(w):
        return "base" if w[0] == "proof" else w[-1]

    def queue(self, w):
        self.queued.append((self.next_id, w)); self.next_id += 1

    def wants(self):
        out, self.queued = self.queued, []
        for i, w in out:
            self.issued[i] = w
            self.requests += 1
        return out

    def finish(self, r):
        self.stage, self.queued, self.result = ("done",), [], r

    def abandon(self, w):
        st = self.stage[0]
        if st == "layer" or (st == "rest" and self.stage[2] == "refinement"):
            if 0 not in self.abandoned:
                self.abandoned.append(0)
        elif st == "components" and w[0] == "range" and w[2] == "refinement":
            (a, b) = w[1]
            for i, cl in self.stage[3]:
                s, ln = self.head["components"][i]
                if cl == "refinement" and s <= b and a < s + ln and i not in self.abandoned:
                    self.abandoned.append(i)

    def on_deadline(self, i):
        if self.result is not None:
            return
        w = self.issued.get(i)
        if w is not None and self.cls(w) == "refinement":
            del self.issued[i]
            self.abandon(w)
            self.advance()

    def base_ready(self):
        if self.base is not None:
            return self.base
        return self.result if self.result is not None and "error" not in self.result else None

    def tick(self, now):
        """§10.2: the deadline runs from the response that verified the base; the sum
        saturates at the end of u64, and a clock that steps back has not passed it."""
        self.now = now
        if self.base_at is None or self.deadline is None or MUTATE == "tick-noop":
            return []
        if self.result is not None or now < min(self.base_at + self.deadline, 2 ** 64 - 1):
            return []
        ids = sorted(i for i, w in self.issued.items() if self.cls(w) == "refinement")
        for i in ids:
            self.on_deadline(i)
        return ids

    def on_response(self, i, status, cr, enc, body, now):
        self.now = now
        if self.result is not None and "error" in self.result:
            raise Refusal(self.result["error"], self.result.get("skipped"))
        w = self.issued.pop(i, None)
        if w is None or self.result is not None:
            return
        self.body += len(body)
        try:
            self.take(w, status, cr, enc, body)
            self.advance()
        except Refusal as e:
            self.finish(refusal(e))
            raise

    def n(self):
        return self.proof["n"] if self.proof else None

    def take(self, w, status, cr, enc, body):
        if w[0] == "proof":
            if status != 200:
                raise Refusal("http_status_unexpected")
            if enc is not None and enc.strip() and enc.strip().lower() != "identity":
                raise Refusal("content_coding_present")
            p = read_proof(body)
            if p["cid"] != self.cid:
                raise Refusal("head_proof_malformed")
            self.proof = p
            early, self.early = self.early, []
            for x in early:
                self.take(*x)
            return
        if w[0] == "whole":
            try:
                check_response(None, status, cr, enc, len(body), len(body))
            except Refusal:
                if w[1] == "refinement":
                    self.abandon(w); return
                raise
            if content_hash(body) != self.cid:
                raise Refusal("object_digest_mismatch")
            if body:
                self.held.append((0, len(body) - 1, body))
            return
        if self.proof is None:
            self.early.append((w, status, cr, enc, body)); return
        try:
            a, b = check_response(w[1], status, cr, enc, len(body), self.n())
        except Refusal:
            if w[2] == "refinement":
                self.abandon(w); return
            raise
        if w[2] == "refinement" and MUTATE != "refinement-unclipped":
            # A refinement's response adds the bytes it asked for and no others (§10.1).
            first, last = max(a, w[1][0]), min(b, w[1][1])
            if first <= last:
                self.held.append((first, last, body[first - a:last + 1 - a]))
            return
        self.held.append((a, b, body))

    def have(self, first, end):
        out, at = b"", first
        while at < end:
            best = max((h for h in self.held if h[0] <= at <= h[1]), key=lambda h: h[1], default=None)
            if best is None:
                return None
            take = min(best[1] + 1, end)
            out += best[2][at - best[0]:take - best[0]]
            at = take
        return out

    def drop_from(self, frm):
        kept = []
        for a, b, body in self.held:
            if a >= frm:
                continue
            kept.append((a, b, body) if b < frm else (a, frm - 1, body[:frm - a]))
        self.held = kept

    def prefix_len(self):
        at = 0
        while True:
            e = max((h[1] + 1 for h in self.held if h[0] <= at <= h[1]), default=None)
            if e is None:
                return at
            at = e

    def advance(self):
        if self.result is not None:
            return
        st = self.stage[0]
        if st == "head":
            self.advance_head()
        elif st == "components":
            self.advance_components()
        elif st in ("rest", "layer"):
            self.advance_rest()

    def to_rest(self, why, cl):
        self.stage = ("rest", why, cl)
        n, cov = self.head["envelope_len"], covered_len(self.head["head_len"], self.head["envelope_len"])
        if cov < n and self.have(cov, n) is None:
            self.queue(("rest", (cov, n - 1), cl))

    def nonconforming(self, entries, wanted):
        """§7 step 4: the refinement components among `wanted` of an envelope whose
        components do not all conform (SPEC §7.1) — every one of them where it does not,
        none where it does."""
        conforms = SV.component_roles(entries) is None and MUTATE != "conforming-withheld"
        if conforms or MUTATE == "nonconforming-offered":
            return []
        return [i for i, _ in wanted if entries[i].get("role") == "refinement"]

    def applied(self, entries, pick):
        adv, applies = self.sink[0], self.sink[1]
        out = [(j, "refinement" if enhancement(entries[j]["capability"]) else "base")
               for j in companions_of(entries, pick) if entries[j]["capability"] in applies]
        out += [(j, "refinement") for j in refinements_naming(entries, pick) if entries[j]["capability"] in applies]
        return out

    def advance_head(self):
        if self.proof is None:
            return
        cov = self.proof["covered"]
        h0 = self.prefix_len()
        if h0 < cov:
            pending = [w for w in list(self.issued.values()) + [w for _, w in self.queued] if w[0] in ("range", "rest") and w[-1] == "base"]
            if not pending:
                self.queue(("range", (h0, cov - 1), "base"))
            return
        prefix = self.have(0, cov)
        verify_proof(self.proof, self.cid, prefix)
        head = parse_head(prefix, self.proof["n"])
        if isinstance(head, tuple):
            raise Refusal("head_malformed")
        adv, applies, held_sets, order, declared, base = self.sink
        if declared is not None:
            check_declared(head, declared, base)
        self.head = head
        if is_layer(head):
            self.to_rest("layer", "refinement")
        elif head["version"] == "WAI1":
            dispatch_wai1(head["manifest"], adv, held_sets)
            self.to_rest("single_component", "base")
        else:
            entries = head["manifest"]["renditions"]
            ordr = order(head) if callable(order) else (order if order is not None else list(range(len(entries))))
            pick, skipped = select(head["manifest"], ordr, adv, held_sets, lambda i, e: True)
            wanted = [(pick, "base")] + self.applied(entries, pick)
            nc = self.nonconforming(entries, wanted)
            planned = [(i, c) for i, c in wanted if i not in nc or entries[i].get("sha256") is None]
            pl = plan(head, planned, [(a, b) for a, b, _ in self.held], self.permille)
            if pl[0] == "rest":
                self.to_rest(pl[2], "base")
            else:
                for (a, b, cl) in pl[1]:
                    self.queue(("range", (a, b), cl))
                self.dropped = pl[2] + [[i, "components_nonconforming"] for i in nc]
                gone = [i for i, _ in pl[2]]
                self.stage = ("components", pick, skipped, [(i, c) for i, c in wanted if i not in gone])
        if self.stage[0] == "components":
            self.advance_components()
        elif self.stage[0] == "rest":
            self.advance_rest()

    def advance_components(self):
        _, pick, skipped, wanted = self.stage
        entries = self.head["manifest"]["renditions"]
        parts, pending, base_pending = {}, False, False
        for i, cl in wanted:
            if i in self.abandoned or any(d == i for d, _ in self.dropped):
                continue
            if i in self.verified:
                parts[i] = self.verified[i]
                continue
            s, ln = self.head["components"][i]
            bts = self.have(s, s + ln)
            if bts is None:
                pending = True
                base_pending = base_pending or cl == "base"
            elif entries[i].get("sha256") is not None and sha(bts) == entries[i]["sha256"]:
                parts[i] = self.verified[i] = bts
            elif cl == "refinement":
                self.dropped.append([i, "component_digest_mismatch"])
            else:
                self.issued, self.queued, self.abandoned, self.dropped, self.verified = {}, [], [], [], {}
                self.drop_from(covered_len(self.head["head_len"], self.head["envelope_len"]))
                self.to_rest("component_digest_mismatch", "base")
                self.advance_rest()
                return
        classes = dict(wanted)

        def assemble(only_base):
            js = [j for j, _ in wanted[1:] if j in parts and not (only_base and classes[j] == "refinement")]
            comps = [j for j in js if entries[j].get("role") == "companion"]
            refs = [j for j in js if entries[j].get("role") != "companion"]
            return self.fetched("base", pick, parts[pick], comps, refs, self.abandoned, self.dropped, skipped, "head-proof", None)
        if not (pending if MUTATE == "base-at-done" else base_pending) and self.base is None:
            self.base, self.base_at = assemble(True), self.now
        if pending:
            return
        self.finish(assemble(False))

    def fetched(self, role, pick, comp, comps, refs, abandoned, dropped, skipped, by, why):
        return {"role": role, "pick": pick, "verified_by": by, "why_whole": why, "body_bytes": self.body, "requests": self.requests,
                "companions": comps, "refinements": refs, "abandoned": list(abandoned), "dropped": [list(x) for x in dropped],
                "skipped": skipped, "component_sha256": sha(comp)}

    def advance_rest(self):
        layer_listed = self.stage[0] == "layer"
        why = "layer" if layer_listed else self.stage[1]
        if 0 in self.abandoned and (layer_listed or why == "layer"):
            self.finish(self.fetched("layer", 0, b"", [], [], [0], [], [], "whole-object", "layer"))
            return
        if self.proof is not None:
            n = self.proof["n"]
        elif self.held and self.held[0][0] == 0:
            n = self.held[0][1] + 1
        else:
            return
        envb = self.have(0, n)
        if envb is None:
            return
        if content_hash(envb) != self.cid:
            raise Refusal("object_digest_mismatch")
        head = parse_head(envb, n)
        adv, applies, held_sets, order, declared, base = self.sink
        if declared is not None:
            check_declared(head, declared, base)
        if is_layer(head):
            s, ln = head["components"][0]
            self.finish(self.fetched("layer", 0, envb[s:s + ln], [], [], [], [], [], "whole-object", "layer"))
            return
        if layer_listed:
            raise Refusal("not_a_layer")
        m = head["manifest"]
        if head["version"] == "WAI1":
            dispatch_wai1(m, adv, held_sets)
            s, ln = head["components"][0]
            self.finish(self.fetched("base", 0, envb[s:s + ln], [], [], [], [], [], "whole-object", why))
            return
        entries = m["renditions"]
        pay = [envb[s:s + ln] for s, ln in head["components"]]
        ordr = order(head) if callable(order) else (order if order is not None else list(range(len(entries))))
        pick, skipped = select(m, ordr, adv, held_sets, lambda i, e: payload_ok(e, pay[i]))
        comps, refs, dropped = [], [], []
        applied = self.applied(entries, pick)
        nc = self.nonconforming(entries, applied)
        for j, cl in applied:
            # §7 step 4: not offered from an envelope that does not conform; one stating no
            # digest is dropped for that too, as on the range path.
            if j in nc:
                if entries[j].get("sha256") is None:
                    dropped.append([j, "component_digest_absent"])
                continue
            # A refinement stating no digest is dropped here as on the range path (§7 step 5).
            if cl == "refinement" and entries[j].get("sha256") is None:
                dropped.append([j, "component_digest_absent"])
                continue
            if not payload_ok(entries[j], pay[j]):
                if cl == "refinement":
                    dropped.append([j, "component_digest_mismatch"])
                continue
            (comps if entries[j].get("role") == "companion" else refs).append(j)
        dropped += [[j, "components_nonconforming"] for j in nc]
        self.finish(self.fetched("base", pick, pay[pick], comps, refs, [], dropped, skipped, "whole-object", why))


def drive(s, answer):
    """Run a session: `answer(want)` gives (status, Content-Range, Content-Encoding,
    body) or None for a request that never arrives."""
    while True:
        if s.result is not None:
            if "error" in s.result:
                raise Refusal(s.result["error"], s.result.get("skipped"))
            return s.result
        ws = s.wants()
        if not ws:
            for i in [i for i, w in s.issued.items() if Session.cls(w) == "refinement"]:
                s.on_deadline(i)
            if s.result is not None:
                continue
            if not s.queued:
                if s.issued:
                    raise Refusal("base_unanswered")
                raise SystemExit("http_verify: the session stalled")
            continue
        for i, w in ws:
            r = answer(w)
            if r is not None:
                s.on_response(i, *r, s.now)


def fetch(env, served, frag, sink, permille, tamper, listed_as_layer=False):
    """The reference client against an origin serving `served` under the name of `env`,
    with the corpus's tampers."""
    proof_bytes = make_proof_file(env) if not listed_as_layer else b""
    n_real = len(served)
    kind_t, ref_t, first_flip = tamper.get("kind"), tamper.get("refinement"), tamper.get("first_range_flip")
    state = {"first": True}

    def answer(w):
        cl = Session.cls(w)
        if kind_t == "no_refinement" and cl == "refinement":
            return None
        if w[0] == "proof":
            return None if kind_t == "no_proof" else (200, None, None, proof_bytes)
        rng = None if w[0] == "whole" else ("bytes=%d-%d" % w[1] if w[0] == "range" else "bytes=%d-" % w[1][0])
        status, cr, a, b = origin_serve(n_real, rng)
        body = served[a:b]
        enc = "gzip" if kind_t == "gzip" else None
        if first_flip is not None and w[0] == "range" and state["first"]:
            state["first"] = False
            if a <= first_flip < b:
                body = bytearray(body); body[first_flip - a] ^= 1; body = bytes(body)
        if ref_t is not None and cl == "refinement" and w[0] == "range":
            (ra, rb) = w[1]
            if ref_t == "gzip":
                enc = "gzip"
            elif ref_t == "bad_range":
                cr = "bytes %d-%d/%d" % (ra + 1, rb + 1, n_real)
            elif ref_t == "416":
                return (416, "bytes */%d" % n_real, None, b"")
            elif ref_t == "complete_length":
                cr = "bytes %d-%d/1" % (ra, rb)
            elif ref_t == "plus_sign":
                cr = cr.replace("bytes ", "bytes +", 1)
            elif ref_t == "short_200":
                return (200, None, None, body)
        return (status, cr, enc, body)

    s = Session(content_hash(env), frag, permille, sink, layer=listed_as_layer)
    return drive(s, answer)


def fetch_timed(env, frag, sink, permille, deadline, script, flip):
    """The reference client on a clock (§10.1, §10.2), driven by the corpus's script."""
    proof_bytes = make_proof_file(env)
    n = len(env)

    def answer(w):
        if w[0] == "proof":
            return (200, None, None, proof_bytes)
        rng = None if w[0] == "whole" else ("bytes=%d-%d" % w[1] if w[0] == "range" else "bytes=%d-" % w[1][0])
        status, cr, a, b = origin_serve(n, rng)
        return (status, cr, None, env[a:b])

    def summary(f):
        return {"pick": f["pick"], "verified_by": f["verified_by"], "component_sha256": f["component_sha256"],
                "companions": f["companions"], "refinements": f["refinements"]}

    s = Session(content_hash(env), frag, permille, sink, deadline=deadline)
    pending, ready_step, first, ticks = [], None, None, []
    stop = False
    for k, step in enumerate(script):
        kind, t = step[0], step[1]
        if kind in ("base", "base_once", "refinement"):
            cl = "refinement" if kind == "refinement" else "base"
            while True:
                pending += s.wants()
                now = [(i, w) for i, w in pending if Session.cls(w) == cl]
                if not now:
                    break
                pending = [(i, w) for i, w in pending if Session.cls(w) != cl]
                for i, w in now:
                    if len(step) > 2 and step[2] == "whole_200":
                        b = bytearray(env)
                        if flip is not None:
                            b[flip] ^= 1
                        r = (200, None, None, bytes(b))
                    else:
                        r = answer(w)
                    try:
                        s.on_response(i, *r, t)
                    except Refusal:
                        stop = True
                        break
                if stop or kind != "base":
                    break
        elif kind == "tick":
            ids = s.tick(t)
            pending = [(i, w) for i, w in pending if i not in ids]
            ticks.append([t, ids])
        else:
            raise SystemExit("unknown step " + kind)
        if ready_step is None and s.base_ready() is not None:
            ready_step, first = k, summary(s.base_ready())
        if stop:
            break
    end = s.base_ready()
    return {"ready_step": ready_step, "ready_at": s.base_at, "base": first, "base_end": summary(end) if end is not None else None,
            "ticks": ticks, "outcome": s.result}


def make_proof_file(env):
    """A producer's head proof (§6.2), written from the text: the chaining values of the
    subtrees at or after c, in the order §6.3 consumes them."""
    h = whole_head(env)
    cov = covered_len(h["head_len"], len(env))
    cvs = []
    if cov < len(env):
        x, L, c = DOMAIN + env, 15 + len(env), 15 + cov

        def walk(start, ln):
            if start >= c:
                cvs.append(B3.subtree_cv(x[start:start + ln], start))
            elif start + ln > c:
                l = 1024 * 2 ** ((((ln - 1) // 1024)).bit_length() - 1)
                walk(start, l); walk(start + l, ln - l)

        l = 1024 * 2 ** ((((L - 1) // 1024)).bit_length() - 1)
        walk(0, l); walk(l, L - l)
    return b"WHP1" + content_hash(env) + len(env).to_bytes(8, "little") + cov.to_bytes(8, "little") + bytes([len(cvs)]) + b"".join(cvs)


# ----------------------------------------------------------------- the cases

def apply_patch(b, patch):
    b = bytearray(b)
    for off, hx in patch or []:
        bs = bytes.fromhex(hx)
        b[off:off + len(bs)] = bs
    return bytes(b)


def frag_of(s):
    out = {}
    if s and "#" in s:
        for kv in s.split("#", 1)[1].split("&"):
            if "=" in kv:
                k, v = kv.split("=", 1)
                if k in ("n", "h") and v.isdigit():
                    out[k] = int(v)
    return out


def sink_of(d, inp, objs):
    """The sink a case describes. Its declared union is parsed, and the base a layer is
    declared against read, before anything is fetched: either refuses the case."""
    held = [open(os.path.join(d, p), "rb").read() for p in inp.get("held") or []]
    declared = parse_element(inp["declared"]) if isinstance(inp.get("declared"), str) else None
    base = None
    if inp.get("layer_base") is not None or inp.get("layer_base_hex") is not None:
        b = bytes.fromhex(inp["layer_base_hex"]) if isinstance(inp.get("layer_base_hex"), str) else objs[inp["layer_base"]]
        base = seq_object(b)
    order = inp.get("order")
    if inp.get("abr") is not None:
        adv, pol, st = inp.get("advertises") or [], inp["abr"]["policy"], abr_state(inp["abr"]["state"])
        check_policy(pol)
        check_state(st)

        def order(head):
            try:
                return order_entries(head, adv, pol, st)
            except Refusal:
                return list(range(len(head["manifest"]["renditions"])))
    return (inp.get("advertises") or [], inp.get("applies") or [], held, order, declared, base)


# ------------------------------------------------- adaptation (Appendix F)

LOG2_TABLE = [0, 5732, 11136, 16248, 21098, 25711, 30109, 34312, 38336, 42196, 45904, 49472, 52911, 56229, 59434, 62534, 65536]


def ilog2_q16(a, b):
    """log2(a/b) in Q16: the ratio's exponent, and the table interpolated on the
    mantissa's top four fractional bits with the next twelve as the weight."""
    b = max(b, 1)
    if a <= b:
        return 0
    x = (a << 16) // b
    e = x.bit_length() - 1 - 16
    f = (x >> e) - 65536
    i, r = f >> 12, f & 4095
    if MUTATE == "ilog2-no-interp":
        r = 0
    return e * 65536 + LOG2_TABLE[i] + (((LOG2_TABLE[i + 1] - LOG2_TABLE[i]) * r) >> 12)


def throughput_bps(samples, n):
    k = min(max(n, 1), len(samples))
    tail = samples[len(samples) - k:]
    t = sum(max(s[2], 1) for s in tail)
    return sum(s[1] for s in tail) * 8000 // t if t else 0


def option_bps(nbytes, seg):
    return -(-(nbytes * 8000) // max(seg, 1))


def highest_under(st, budget):
    for k in range(len(st["option_bytes"]) - 1, -1, -1):
        if option_bps(st["option_bytes"][k], st["segment_ms"]) <= budget:
            return k
    return 0


def utilities(st):
    s0 = max(st["option_bytes"][0] if st["option_bytes"] else 1, 1)
    return [ilog2_q16(max(b, 1), s0) for b in st["option_bytes"]]


def choose(p, st):
    opts, seg, k = st["option_bytes"], st["segment_ms"], len(st["option_bytes"])
    if k == 0:
        return 0
    if p["id"] == "throughput":
        return highest_under(st, throughput_bps(st["samples"], p["last_n"]) * p["safety_permille"] // 1000)
    if p["id"] == "buffer_based":
        b, r, c = st["buffer_ms"], p["reservoir_ms"], p["cushion_ms"]
        if b <= r:
            return 0
        if c == 0 or b >= r + c:
            return k - 1
        lo, hi = option_bps(opts[0], seg), option_bps(opts[-1], seg)
        return highest_under(st, lo + max(hi - lo, 0) * (b - r) // c)
    if p["id"] == "bola_like":
        sg = max(seg, 1)
        q, qmax = st["buffer_ms"] * 65536 // sg, p["max_buffer_ms"] * 65536 // sg
        u = utilities(st)
        vn, vd = max(qmax - 65536, 0), max(u) + 65536
        score = [vn * p["v_milli"] * (x + 65536) // (vd * 1000) - q for x in u]
        size = [max(b, 1) for b in opts]
        best = None
        if any(x > 0 for x in score):
            for i in range(k):
                if score[i] > 0 and (best is None or score[i] * size[best] > score[best] * size[i]):
                    best = i
        elif MUTATE == "bola-waits":
            return 0
        else:
            for i in range(k):
                if best is None or score[i] > score[best]:
                    best = i
        return best
    if p["id"] == "mpc":
        pred = throughput_bps(st["samples"], 5) * p["safety_permille"] // 1000
        if pred == 0:
            return 0
        h = min(max(p["horizon"], 1), 5)
        while h > 1 and k ** h > 65536:
            h -= 1
        u = utilities(st)
        u_top = max(u)
        best = None
        for code in range(k ** h):
            seq, c = [0] * h, code
            for t in range(h - 1, -1, -1):
                seq[t], c = c % k, c // k
            b, rebuf, qoe = st["buffer_ms"], 0, 0
            last = st["last"]
            prev = u[last] if last is not None and last < k else None
            for o in seq:
                dl = -(-(opts[o] * 8000) // pred)
                if dl > b:
                    rebuf, b = rebuf + dl - b, 0
                else:
                    b -= dl
                b += seg
                qoe += 1000 * u[o]
                if prev is not None:
                    qoe -= 1000 * abs(u[o] - prev)
                prev = u[o]
            qoe -= rebuf * (u_top + 65536)
            if best is None or qoe > best[0]:
                best = (qoe, seq[0])
        return best[1]
    raise SystemExit("http_verify: an unknown policy " + str(p["id"]))


def clamp_to_budget(st, choice):
    """Appendix F's budget: it never raises the choice; the highest option at or below it
    whose labelled figure is within it, else option 0 unmet; not applied only when no
    option has a labelled figure."""
    budget = st["joule_budget_micro"]
    if budget is None:
        return choice, "not_set"
    figs = st["option_joules_micro"]
    n = max(len(st["option_bytes"]), choice + 1)
    fig = lambda j: figs[j] if j < len(figs) and j < n else None
    if MUTATE == "unlabelled-as-zero":
        fig = lambda j: (figs[j] if j < len(figs) and j < n else None) or 0
    labelled = [(j, fig(j)) for j in range(n) if fig(j) is not None]
    if not labelled or (MUTATE == "unlabelled-escapes" and fig(choice) is None):
        return choice, "not_applied"
    if fig(choice) is not None and fig(choice) <= budget:
        return choice, "within"
    below = [j for j, f in labelled if j < choice and f <= budget]
    if below:
        return below[-1], "lowered"
    if MUTATE == "budget-raises":
        above = [j for j, f in labelled if j > choice and f <= budget]
        if above:
            return above[0], "unmet"
    return 0, "unmet"


MAX_OPTIONS = 256
OUT_OF_BOUNDS = "abr_state_out_of_bounds"


def bounded(v, top):
    """An unsigned integer of at most top; out of bounds for any other number."""
    if isinstance(v, bool) or not isinstance(v, (int, float)):
        raise SystemExit("http_verify: an adaptation value is not a number")
    if not isinstance(v, int) or not 0 <= v <= top:
        raise Refusal(OUT_OF_BOUNDS)
    return v


def check_policy(p):
    """Appendix F: each parameter within its type."""
    tops = {"throughput": {"last_n": 255, "safety_permille": 65535}, "buffer_based": {"reservoir_ms": U32, "cushion_ms": U32},
            "bola_like": {"v_milli": U32, "max_buffer_ms": U32}, "mpc": {"horizon": 255, "safety_permille": 65535}}
    if p.get("id") not in tops:
        raise SystemExit("http_verify: an unknown policy " + str(p.get("id")))
    for k, top in tops[p["id"]].items():
        bounded(p[k], top)


def check_state(st):
    if (len(st["option_bytes"]) > MAX_OPTIONS or len(st["option_joules_micro"]) > MAX_OPTIONS) and MUTATE != "abr-unbounded":
        raise Refusal(OUT_OF_BOUNDS)


def decide(p, st):
    check_policy(p)
    check_state(st)
    return clamp_to_budget(st, choose(p, st))


def abr_state(v):
    """A state from its JSON form, each value within its field's type (Appendix F)."""
    opt = lambda x, top: None if x is None else bounded(x, top)
    return {"buffer_ms": bounded(v["buffer_ms"], U32), "samples": [tuple(bounded(x, U64) for x in s[:3]) for s in v.get("samples") or []],
            "last": opt(v.get("last"), U64), "option_bytes": [bounded(x, U64) for x in v.get("option_bytes") or []],
            "option_joules_micro": [opt(x, U64) for x in v.get("option_joules_micro") or []],
            "joule_budget_micro": opt(v.get("joule_budget_micro"), U64), "segment_ms": bounded(v["segment_ms"], U32)}


def depth(layers_n, p, st):
    check_policy(p)
    check_state(st)
    if len(layers_n) > MAX_OPTIONS:
        raise Refusal(OUT_OF_BOUNDS)
    st = dict(st)
    acc, cum, jcum, fsum = 0, [], [], 0
    figs = st["option_joules_micro"]
    for k, n in enumerate(layers_n):
        acc += n
        if acc > U64 and MUTATE != "abr-unbounded":
            raise Refusal(OUT_OF_BOUNDS)
        cum.append(acc)
        f = figs[k] if k < len(figs) else None
        fsum = None if fsum is None or f is None else fsum + f
        if fsum is not None and fsum > U64 and MUTATE != "abr-unbounded":
            raise Refusal(OUT_OF_BOUNDS)
        jcum.append(fsum)
    st["option_bytes"], st["option_joules_micro"] = cum, jcum
    if not layers_n:
        return 0, clamp_to_budget(st, 0)[1]
    return clamp_to_budget(st, choose(p, st))


def order_entries(head, adv, p, st):
    """SPEC §7.1 step 2's order under a policy: each advertised selectable capability's
    entries by component length, the choice first, then those below it nearest first,
    then those above; every other entry after, in table order."""
    check_policy(p)
    check_state(st)
    if head["version"] != "WAI2":
        return [0]
    entries = head["manifest"]["renditions"]
    ln = lambda i: head["components"][i][1]
    groups = []
    for i, e in enumerate(entries):
        if not selectable(e) or e["capability"] not in adv:
            continue
        g = next((g for g in groups if g[0] == e["capability"]), None)
        if g is None:
            groups.append((e["capability"], [i]))
        else:
            g[1].append(i)
    order = []
    for _, g in groups:
        g = sorted(g, key=lambda i: (ln(i), i))
        s2 = dict(st)
        figs = st["option_joules_micro"]
        s2["option_bytes"] = [ln(i) for i in g]
        s2["option_joules_micro"] = [figs[i] if i < len(figs) else None for i in g]
        c = min(decide(p, s2)[0], len(g) - 1)
        order += [g[c]] + list(reversed(g[:c])) + g[c + 1:]
    return order + [i for i in range(len(entries)) if i not in order]


# ------------------------------------------------------ playlists (§10.4, §11)

U64 = 2 ** 64 - 1
U32 = 2 ** 32 - 1
DIGITS3 = re.compile(r"[0-9]+(\.[0-9]+)?", re.ASCII)


def decimal_ms(s):
    """A decimal number of seconds, exactly, in milliseconds rounded half up."""
    if not DIGITS3.fullmatch(s):
        return None
    whole, _, frac = s.partition(".")
    f = (frac + "0000")[:4]
    ms = int(whole) * 1000 + int(f[:3]) + (1 if int(f[3]) >= (6 if MUTATE == "round-half-down" else 5) else 0)
    return ms if ms <= U64 else None


def decimal_int(s):
    if not re.fullmatch(r"[0-9]+", s, re.ASCII):
        return None
    v = int(s)
    return v if v <= U64 else None


def attributes(s):
    """An HLS attribute list (RFC 8216 §4.2) as [(name, quoted, value)]."""
    bad = Refusal("playlist_malformed")
    out, rest = [], s
    while True:
        if "=" not in rest:
            raise bad
        name, after = rest.split("=", 1)
        if not re.fullmatch(r"[A-Z0-9-]+", name, re.ASCII) or any(n == name for n, _, _ in out):
            raise bad
        if after.startswith('"'):
            close = after.find('"', 1)
            if close < 0:
                raise bad
            v = after[1:close]
            if "\r" in v or "\n" in v:
                raise bad
            out.append((name, True, v))
            after = after[close + 1:]
        else:
            end = after.find(",")
            end = len(after) if end < 0 else end
            v = after[:end]
            if not v or '"' in v or any(ch in " \t\n\r\x0c" for ch in v):
                raise bad
            out.append((name, False, v))
            after = after[end:]
        if not after:
            return out
        if not after.startswith(","):
            raise bad
        rest = after[1:]


def attr(a, name):
    return next(((q, v) for n, q, v in a if n == name), None)


def attr_ms(a, name):
    x = attr(a, name)
    if x is None:
        return None
    if x[0] or decimal_ms(x[1]) is None:
        raise Refusal("playlist_malformed")
    return decimal_ms(x[1])


def attr_quoted(a, name):
    x = attr(a, name)
    if x is None:
        return None
    if not x[0]:
        raise Refusal("playlist_malformed")
    return x[1]


def attr_yes(a, name):
    x = attr(a, name)
    if x is None:
        return False
    if x[0] or x[1] not in ("YES", "NO"):
        raise Refusal("playlist_malformed")
    return x[1] == "YES"


def new_extra():
    return {"discontinuity": False, "gap": False, "pdt": None, "parts": [], "layers": [], "hint": None}


def push_layer(layers, l):
    if layers and layers[-1]["n"] >= l["n"]:
        raise Refusal("playlist_malformed")
    layers.append(l)


def parse_media_playlist(text):
    """A media playlist, classical or WAI, read as §11 needs it."""
    bad = Refusal("playlist_malformed")
    lines = [l[:-1] if l.endswith("\r") else l for l in text.split("\n")]
    lines = [l for l in lines if l]
    if not lines or lines[0] != "#EXTM3U":
        raise bad
    p = {"version": None, "target": None, "seq": 0, "dseq": 0, "part_target": None, "hold_back": None, "block": False,
         "segments": [], "open": None, "hint": None, "end": False, "ptype": None, "indep": False, "map": None}
    pending, extinf, anchor, started = new_extra(), None, None, False
    for line in lines[1:]:
        if line.startswith("#WAI-LAYER:"):
            l = parse_layer_tag(line)
            if anchor == "extinf":
                push_layer(pending["layers"], l)
            elif anchor == "part":
                push_layer(pending["parts"][-1]["layers"], l)
            else:
                raise bad
            continue
        anchor = None
        if not line.startswith("#"):
            if extinf is None:
                raise bad
            p["segments"].append({"ms": extinf, "uri": line, "extra": pending})
            pending, extinf, started = new_extra(), None, True
            continue
        if not line.startswith("#EXT"):
            continue
        tag, _, value = line.partition(":")
        if tag == "#EXTM3U":
            raise bad
        elif tag == "#EXT-X-VERSION":
            v = decimal_int(value)
            if v is None or v > 255:
                raise bad
            p["version"] = v
        elif tag == "#EXT-X-TARGETDURATION":
            v = decimal_int(value)
            if v is None:
                raise bad
            p["target"] = v
        elif tag in ("#EXT-X-MEDIA-SEQUENCE", "#EXT-X-DISCONTINUITY-SEQUENCE"):
            if started or extinf is not None or pending["parts"]:
                raise bad
            v = decimal_int(value)
            if v is None:
                raise bad
            p["seq" if tag == "#EXT-X-MEDIA-SEQUENCE" else "dseq"] = v
        elif tag == "#EXT-X-PART-INF":
            a = attributes(value)
            pt = attr_ms(a, "PART-TARGET")
            if pt is None:
                raise bad
            p["part_target"] = pt
        elif tag == "#EXT-X-SERVER-CONTROL":
            a = attributes(value)
            p["block"] = attr_yes(a, "CAN-BLOCK-RELOAD")
            p["hold_back"] = attr_ms(a, "PART-HOLD-BACK")
        elif tag == "#EXT-X-DISCONTINUITY":
            pending["discontinuity"] = True
        elif tag == "#EXT-X-GAP":
            pending["gap"] = True
        elif tag == "#EXT-X-INDEPENDENT-SEGMENTS":
            p["indep"] = True
        elif tag == "#EXT-X-PLAYLIST-TYPE":
            if value not in ("VOD", "EVENT"):
                raise bad
            p["ptype"] = value
        elif tag == "#EXT-X-MAP":
            uri = attr_quoted(attributes(value), "URI")
            if uri is None:
                raise bad
            if not started and extinf is None and pending == new_extra() and p["map"] is None:
                p["map"] = uri
        elif tag == "#EXT-X-PROGRAM-DATE-TIME":
            pending["pdt"] = value
        elif tag == "#EXT-X-PART":
            a = attributes(value)
            d, uri = attr_ms(a, "DURATION"), attr_quoted(a, "URI")
            if d is None or uri is None:
                raise bad
            pending["parts"].append({"ms": d, "independent": attr_yes(a, "INDEPENDENT"), "uri": uri, "layers": []})
            anchor = "part"
        elif tag == "#EXTINF":
            if extinf is not None:
                raise bad
            extinf = decimal_ms(value.split(",", 1)[0])
            if extinf is None:
                raise bad
            anchor = "extinf"
        elif tag == "#EXT-X-PRELOAD-HINT":
            a = attributes(value)
            t = attr(a, "TYPE")
            if t is None or t[0]:
                raise bad
            if t[1] == "PART":
                if p["hint"] is not None:
                    raise bad
                h = attr_quoted(a, "URI")
                if h is None:
                    raise bad
                p["hint"] = h
        elif tag == "#EXT-X-ENDLIST":
            p["end"] = True
        elif tag in ("#EXT-X-SKIP", "#EXT-X-STREAM-INF", "#EXT-X-I-FRAME-STREAM-INF", "#EXT-X-MEDIA"):
            raise bad
    if extinf is not None:
        raise bad
    if pending != new_extra():
        p["open"] = pending
    return p


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


def render_hls(segs, r, extras):
    """The rendering of §8.6: header, then each segment (discontinuity, date-time, parts
    with their layer lines, EXTINF, its layer lines, URI), the open segment, the hint,
    the end."""
    target = r["target"] if r["target"] is not None else max([-(-s["ms"] // 1000) for s in segs], default=0)
    out = ["#EXTM3U", "#EXT-X-VERSION:%d" % r["version"], "#EXT-X-TARGETDURATION:%d" % target]
    if r["part_target"] is not None:
        out.append("#EXT-X-PART-INF:PART-TARGET=" + secs(r["part_target"]))
    ctl = (["CAN-BLOCK-RELOAD=YES"] if r["block"] else []) + (["PART-HOLD-BACK=" + secs(r["hold_back"])] if r["hold_back"] is not None else [])
    if ctl:
        out.append("#EXT-X-SERVER-CONTROL:" + ",".join(ctl))
    out.append("#EXT-X-MEDIA-SEQUENCE:%d" % r["seq"])
    if r["dseq"]:
        out.append("#EXT-X-DISCONTINUITY-SEQUENCE:%d" % r["dseq"])
    if r.get("ptype"):
        out.append("#EXT-X-PLAYLIST-TYPE:" + r["ptype"])
    if r.get("indep"):
        out.append("#EXT-X-INDEPENDENT-SEGMENTS")
    if r.get("map") is not None:
        out.append('#EXT-X-MAP:URI="%s"' % r["map"])

    def parts(ps):
        for q in ps:
            out.append("#EXT-X-PART:DURATION=%s%s,URI=\"%s\"" % (secs(q["ms"]), ",INDEPENDENT=YES" if q["independent"] else "", q["uri"]))
            out.extend("#WAI-LAYER:N=%d,URI=\"%s\"" % (l["n"], l["uri"]) for l in q["layers"])

    for i, s in enumerate(segs):
        e = extras[i] if i < len(extras) else new_extra()
        if s.get("ad") or e["discontinuity"]:
            out.append("#EXT-X-DISCONTINUITY")
        if e["pdt"] is not None:
            out.append("#EXT-X-PROGRAM-DATE-TIME:" + e["pdt"])
        if MUTATE == "parts-after-extinf":
            out.append("#EXTINF:%s," % secs(s["ms"]))
            parts(e["parts"])
        else:
            parts(e["parts"])
            if e.get("gap"):
                out.append("#EXT-X-GAP")
            out.append("#EXTINF:%s," % secs(s["ms"]))
        out.extend("#WAI-LAYER:N=%d,URI=\"%s\"" % (l["n"], l["uri"]) for l in e["layers"])
        out.append(s["uri"])
    if len(extras) > len(segs):
        o = extras[len(segs)]
        if o["discontinuity"]:
            out.append("#EXT-X-DISCONTINUITY")
        if o["pdt"] is not None:
            out.append("#EXT-X-PROGRAM-DATE-TIME:" + o["pdt"])
        parts(o["parts"])
    if extras and extras[-1]["hint"] is not None:
        out.append("#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"%s\"" % extras[-1]["hint"])
    if r["end"]:
        out.append("#EXT-X-ENDLIST")
    return "\n".join(out) + "\n"


def uri_ok(u, quoted):
    return bool(u) and not u.startswith("#") and "\r" not in u and "\n" not in u and not (quoted and '"' in u)


def check_render(segs, r, extras):
    bad = Refusal("playlist_malformed")
    n = len(segs)
    if len(extras) not in (0, n, n + 1) or r["version"] == 0:
        raise bad
    if r["part_target"] is not None and (r["hold_back"] is None or r["part_target"] == 0 or r["hold_back"] < 2 * r["part_target"]):
        raise bad
    if any(not uri_ok(s["uri"], False) for s in segs):
        raise bad
    if r.get("map") is not None and not uri_ok(r["map"], True):
        raise bad

    def layers_ok(ls):
        last = 0
        for l in ls:
            if l["n"] == 0 or l["n"] <= last or not uri_ok(l["uri"], True):
                raise bad
            last = l["n"]

    for i, e in enumerate(extras):
        if e["pdt"] is not None and (not e["pdt"] or not re.fullmatch(r"[A-Za-z0-9:.+-]+", e["pdt"], re.ASCII)):
            raise bad
        if e["parts"] and r["part_target"] is None:
            raise bad
        for q in e["parts"]:
            if not uri_ok(q["uri"], True):
                raise bad
            layers_ok(q["layers"])
        layers_ok(e["layers"])
        if i < n and e.get("gap") and (e["parts"] or e["layers"]) and MUTATE != "gap-carries":
            raise bad
        if e["hint"] is not None and (i + 1 != len(extras) or r["end"] or not uri_ok(e["hint"], True)):
            raise bad
        if i == n:
            if r["end"] or e["layers"] or e.get("gap") or (not e["parts"] and (e["discontinuity"] or e["pdt"] is not None)) or (not e["parts"] and e["hint"] is None):
                raise bad


def render_inputs(p):
    segs = [{"ms": s["ms"], "uri": s["uri"]} for s in p["segments"]]
    r = {"version": p["version"] if p["version"] is not None else 1, "seq": p["seq"], "dseq": p["dseq"], "target": p["target"],
         "part_target": p["part_target"], "hold_back": p["hold_back"], "block": p["block"], "end": p["end"],
         "ptype": p["ptype"], "indep": p["indep"], "map": p["map"]}
    extras = [json.loads(json.dumps(s["extra"])) for s in p["segments"]]
    if p["open"] is not None:
        extras.append(json.loads(json.dumps(p["open"])))
    if p["hint"] is not None:
        if not extras:
            extras.append(new_extra())
        extras[-1]["hint"] = p["hint"]
    if all(e == new_extra() for e in extras):
        extras = []
    return segs, r, extras


def is_rendered_form(p, text):
    segs, r, extras = render_inputs(p)
    try:
        check_render(segs, r, extras)
    except Refusal:
        return False
    return render_hls(segs, r, extras) == text


def misaligned():
    return Refusal("adjacency_misaligned")


def timelines_adjacent(w, c):
    if w["seq"] != c["seq"] or (w["dseq"] != c["dseq"] and MUTATE != "dseq-ignored") or len(w["segments"]) != len(c["segments"]):
        raise misaligned()
    both = w["part_target"] is not None and c["part_target"] is not None
    if both and w["part_target"] != c["part_target"]:
        raise misaligned()
    for a, b in zip(w["segments"], c["segments"]):
        if a["ms"] != b["ms"] or a["extra"]["discontinuity"] != b["extra"]["discontinuity"]:
            raise misaligned()
        if a["extra"]["pdt"] != b["extra"]["pdt"] and MUTATE != "pdt-ignored":
            raise misaligned()
        pa, pb = [q["ms"] for q in a["extra"]["parts"]], [q["ms"] for q in b["extra"]["parts"]]
        if both and pa and pb and pa != pb:
            raise misaligned()
    if w["open"] is not None and c["open"] is not None:
        a, b = w["open"], c["open"]
        if a["discontinuity"] != b["discontinuity"] or a["pdt"] != b["pdt"]:
            raise misaligned()
        if both and [q["ms"] for q in a["parts"]] != [q["ms"] for q in b["parts"]]:
            raise misaligned()


def check_adjacent(wai, classical):
    w = parse_media_playlist(wai)
    c = parse_media_playlist(classical)
    timelines_adjacent(w, c)


def wai_media_playlist(c, segs, open_, prefix, end):
    """The WAI media playlist on classical timeline `c` (§11.1-§11.4)."""
    if len(segs) != len(c["segments"]):
        raise misaligned()
    has_parts = any(s["parts"] for s in segs) or (open_ is not None and open_["parts"])
    if has_parts and c["part_target"] is None:
        raise misaligned()
    lay = lambda ls: [{"n": l["n"], "uri": "%s%s.wai#n=%d" % (prefix, l["cid"].hex(), l["len"])} for l in ls]
    prt = lambda ps: [{"ms": q["ms"], "independent": q["independent"], "uri": "%s#cid=%s&n=%d" % (q["alias"], q["cid"].hex(), q["n"]), "layers": lay(q["layers"])} for q in ps]
    out_segs, extras = [], []
    for i, (w, k) in enumerate(zip(segs, c["segments"])):
        if w["ms"] != k["ms"]:
            raise misaligned()
        if w["parts"] and k["extra"]["parts"] and [q["ms"] for q in w["parts"]] != [q["ms"] for q in k["extra"]["parts"]]:
            raise misaligned()
        h = "&h=%d" % w["h"] if w["h"] is not None else ""
        out_segs.append({"ms": w["ms"], "uri": "%s%s.wai#n=%d%s" % (prefix, w["cid"].hex(), w["n"], h)})
        e = new_extra()
        e.update(discontinuity=k["extra"]["discontinuity"], gap=w.get("gap", False), pdt=k["extra"]["pdt"], parts=prt(w["parts"]), layers=lay(w["layers"]))
        extras.append(e)
    hint = None
    if open_ is not None:
        if not open_["parts"]:
            hint = open_["hint"]
        else:
            ko = c["open"]
            if ko is None:
                raise misaligned()
            if ko["parts"] and [q["ms"] for q in open_["parts"]] != [q["ms"] for q in ko["parts"]]:
                raise misaligned()
            e = new_extra()
            e.update(discontinuity=ko["discontinuity"], pdt=ko["pdt"], parts=prt(open_["parts"]), hint=open_["hint"])
            extras.append(e)
    if all(e == new_extra() for e in extras):
        extras = []
    if hint is not None:
        if not extras:
            extras = [new_extra() for _ in range(max(len(segs), 1))]
        extras[-1]["hint"] = hint
    pt = c["part_target"]
    r = {"version": c["version"] if c["version"] is not None else 6, "seq": c["seq"], "dseq": c["dseq"], "target": c["target"],
         "part_target": pt, "hold_back": c["hold_back"] if c["hold_back"] is not None else (pt * 3 if pt is not None else None),
         "block": c["block"], "end": end, "ptype": c["ptype"], "indep": c["indep"], "map": None}
    check_render(out_segs, r, extras)
    text = render_hls(out_segs, r, extras)
    timelines_adjacent(parse_media_playlist(text), c)
    return text


def bandwidths(objs):
    if not objs or any(d == 0 for _, d in objs):
        return None
    rate = lambda n, d: min(-(-(8000 * n) // d), U64)
    peak = max(rate(n, d) for n, d in objs)
    sn, sd = sum(n for n, _ in objs), sum(d for _, d in objs)
    return peak, (rate(sn, sd) if MUTATE != "average-floor" else 8000 * sn // sd)


SESSION_DATA = '#EXT-X-SESSION-DATA:DATA-ID="science.transaction.wai.multivariant",VALUE="'


def session_data_position(original):
    """§11.1: the offset after the first line and the EXT-X-VERSION and
    EXT-X-INDEPENDENT-SEGMENTS lines that directly follow it; None when there is none."""
    at = 0
    lines = original.split("\n")
    lines = [l + "\n" for l in lines[:-1]] + ([lines[-1]] if lines[-1] else [])
    for i, l in enumerate(lines):
        bare = l[:-1] if l.endswith("\n") else l
        bare = bare[:-1] if l.endswith("\n") and bare.endswith("\r") else bare
        if i > 0 and not (bare.startswith("#EXT-X-VERSION:") or bare == "#EXT-X-INDEPENDENT-SEGMENTS") and MUTATE != "session-data-anywhere":
            break
        if not l.endswith("\n"):
            return None
        at += len(l.encode("utf-8"))
        if MUTATE == "session-data-anywhere":
            break
    return at if at > 0 else None


def classical_unchanged(original, published):
    if original == published:
        return
    o, p = original.encode("utf-8"), published.encode("utf-8")
    at = session_data_position(original)
    if at is None or not (p.startswith(o[:at]) and p.endswith(o[at:]) and len(p) > len(o)):
        raise Refusal("classical_changed")
    inserted = p[at:len(p) - (len(o) - at)].decode("utf-8")
    if not inserted.endswith("\n"):
        raise Refusal("classical_changed")
    bare = inserted[:-1]
    bare = bare[:-1] if bare.endswith("\r") else bare
    if "\n" in bare or not (bare.startswith(SESSION_DATA) and bare.endswith('"')):
        raise Refusal("classical_changed")
    uri = bare[len(SESSION_DATA):-1]
    if not uri or any(ch in uri for ch in '"\r\n'):
        raise Refusal("classical_changed")


def listing_lines(p):
    """Every part, layer and segment line of a media playlist, in the order a client
    reads them for a cid (§5.2, §11.4)."""
    lines = []
    for s in p["segments"]:
        for q in s["extra"]["parts"]:
            lines.append(q["uri"])
            lines += [l["uri"] for l in q["layers"]]
        lines += [l["uri"] for l in s["extra"]["layers"]]
        lines.append(s["uri"])
    if p["open"] is not None:
        for q in p["open"]["parts"]:
            lines.append(q["uri"])
            lines += [l["uri"] for l in q["layers"]]
    return lines


def uri_cid(line):
    """The cid a listed URI names: its cid fragment, or the <cid>.wai its path ends in."""
    base, _, frag = line.partition("#")
    for kv in frag.split("&") if frag else []:
        k, eq, v = kv.partition("=")
        if eq and k == "cid" and re.fullmatch(r"[0-9a-f]{64}", v):
            return bytes.fromhex(v)
    name = base.rsplit("/", 1)[-1]
    if name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
        return bytes.fromhex(name[:-4])
    return None


def listed_cid(p, uri):
    line = next((l for l in listing_lines(p) if l.split("#", 1)[0] == uri), None)
    return None if line is None else uri_cid(line)


RETENTION = 256
URI_PARTS = re.compile(r"^(([^:/?#]+):)?(//([^/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?$", re.S)


def remove_dot_segments(path):
    """RFC 3986 §5.2.4."""
    inp, out = path, ""
    while inp:
        if inp.startswith("../"):
            inp = inp[3:]
        elif inp.startswith("./"):
            inp = inp[2:]
        elif inp.startswith("/./"):
            inp = inp[2:]
        elif inp == "/.":
            inp = "/"
        elif inp.startswith("/../") or inp == "/..":
            inp = "/" + inp[4:] if inp.startswith("/../") else "/"
            out = out[:out.rfind("/")] if "/" in out else ""
        elif inp in (".", ".."):
            inp = ""
        else:
            j = inp.find("/", 1 if inp.startswith("/") else 0)
            j = len(inp) if j < 0 else j
            out, inp = out + inp[:j], inp[j:]
    return out


def pct_normal(s):
    """RFC 3986 §6.2.2.2: upper-case hexadecimal digits, unreserved characters decoded."""
    def one(m):
        c = chr(int(m.group(1), 16))
        return c if re.fullmatch(r"[A-Za-z0-9._~-]", c) else "%" + m.group(1).upper()
    return re.sub(r"%([0-9A-Fa-f]{2})", one, s)


def rfc3986_resolve(base, ref):
    """ref resolved against base (RFC 3986 §5.2.2, §5.2.3) and normalized (§6.2.2), its
    fragment dropped."""
    def parts(u):
        m = URI_PARTS.match(u)
        return m.group(2), m.group(4) if m.group(3) is not None else None, m.group(5), m.group(7) if m.group(6) is not None else None
    bs, ba, bp, bq = parts(base)
    rs, ra, rp, rq = parts(ref)
    if MUTATE == "uri-unresolved":
        return ref.split("#", 1)[0]
    if rs is not None:
        ts, ta, tp, tq = rs, ra, remove_dot_segments(rp), rq
    elif ra is not None:
        ts, ta, tp, tq = bs, ra, remove_dot_segments(rp), rq
    elif rp == "":
        ts, ta, tp, tq = bs, ba, bp, rq if rq is not None else bq
    elif rp.startswith("/"):
        ts, ta, tp, tq = bs, ba, remove_dot_segments(rp), rq
    else:
        if ba is not None and bp == "":
            merged = "/" + rp
        else:
            merged = bp[:bp.rfind("/") + 1] + rp
        ts, ta, tp, tq = bs, ba, remove_dot_segments(merged), rq
    out = ""
    if ts is not None:
        out += ts.lower() + ":"
    if ta is not None:
        a = pct_normal(ta)
        at = a.rfind("@")
        out += "//" + (a[:at + 1] + a[at + 1:].lower() if at >= 0 else a.lower())
    out += remove_dot_segments(pct_normal(tp))
    if tq is not None:
        out += "?" + pct_normal(tq)
    return out


def new_bindings(url):
    return {"base": url, "uris": {}, "seqs": {}, "rev": 0}


def bind(b, p):
    """§5.2, §11.5: the bindings b with playlist p's added as one more revision: each URL
    a line names (resolved against the playlist's URL) and each media sequence number to
    its cid. Refusal when p lists another cid for one bound, or a gap on a bound sequence
    number; then what none of the last RETENTION revisions listed is forgotten."""
    rev = b["rev"] + 1
    uris, seqs = dict(b["uris"]), dict(b["seqs"])
    for line in listing_lines(p):
        cid = uri_cid(line)
        if cid is None:
            continue
        key = rfc3986_resolve(b["base"], line)
        if key in uris and uris[key][0] != cid and MUTATE != "rebind-uri":
            raise Refusal("playlist_listing_rebound")
        uris[key] = (cid, rev)
    for i, sg in enumerate(p["segments"]):
        n = p["seq"] + i
        if n > U64:
            raise Refusal("playlist_malformed")
        if sg["extra"]["gap"]:
            if n in seqs and MUTATE != "gap-withdraws":
                raise Refusal("playlist_listing_rebound")
            continue
        cid = uri_cid(sg["uri"])
        if cid is None:
            continue
        if n in seqs and seqs[n][0] != cid and MUTATE != "rebind-sequence":
            raise Refusal("playlist_listing_rebound")
        seqs[n] = (cid, rev)
    keep = (lambda last: True) if MUTATE == "bindings-unbounded" else (lambda last: last + RETENTION > rev)
    return {"base": b["base"], "uris": {k: v for k, v in uris.items() if keep(v[1])}, "seqs": {k: v for k, v in seqs.items() if keep(v[1])}, "rev": rev}


def cid_of(b, uri):
    v = b["uris"].get(rfc3986_resolve(b["base"], uri))
    return None if v is None else v[0]


def xml_char(c):
    o = ord(c)
    return o in (9, 10, 13) or 0x20 <= o <= 0xD7FF or 0xE000 <= o <= 0xFFFD or 0x10000 <= o <= 0x10FFFF


def esc_xml(s):
    """An attribute value: & < > " as entities, tab, LF and CR as character references;
    Refusal for a character that is not an XML Char."""
    out = []
    for c in s:
        if c in '&<>"\t\n\r':
            out.append({"&": "&amp;", "<": "&lt;", ">": "&gt;", '"': "&quot;", "\t": "&#9;", "\n": "&#10;", "\r": "&#13;"}[c])
        elif xml_char(c) or MUTATE == "xml-char-unchecked":
            out.append(c)
        else:
            raise Refusal("playlist_malformed")
    return "".join(out)


# The characters of Unicode's White_Space property: an MPD @id holds none.
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0\u1680\u2028\u2029\u202f\u205f\u3000") | {chr(c) for c in range(0x2000, 0x200B)}


def id_ok(s):
    return s != "" and not any(c in WHITE_SPACE for c in s)


def xs_seconds(ms):
    return "PT%dS" % (ms // 1000) if ms % 1000 == 0 else "PT%sS" % secs(ms)


def wai_mpd(t, set_id, union, floor, rep_id, bandwidth, media):
    """wai.mpd (§11.1) on the timeline t = (period id or None, start ms, timescale,
    [(t, d)]); media[i] None leaves segment i out (a gap)."""
    period_id, start_ms, ts, segs = t
    if not segs or ts == 0 or len(media) != len(segs) or union[0] != floor[0]:
        raise Refusal("playlist_malformed")
    if not id_ok(rep_id) or (period_id is not None and not id_ok(period_id)):
        raise Refusal("playlist_malformed")
    offered = [(sg, m) for sg, m in zip(segs, media) if m is not None]
    if not offered:
        raise Refusal("playlist_malformed")
    to_ms = lambda d: min(d * 1000 // ts, U64)
    total = min(sum(to_ms(d) for _, d in segs), U64)
    longest = max(to_ms(d) for _, d in segs)
    period = ('  <Period id="%s" start="%s">' % (esc_xml(period_id), xs_seconds(start_ms))) if period_id is not None else ('  <Period start="%s">' % xs_seconds(start_ms))
    out = ['<?xml version="1.0" encoding="UTF-8"?>',
           '<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" profiles="urn:mpeg:dash:profile:full:2011" type="static" mediaPresentationDuration="PT%sS" minBufferTime="PT%dS">' % (secs(total), -(-longest // 1000)),
           period,
           '    <AdaptationSet id="%d" mimeType="application/wai" codecs="%s" segmentAlignment="true">' % (set_id, esc_xml(render(union))),
           '      <EssentialProperty schemeIdUri="tag:transaction.science,2026:wai/http-delivery"/>',
           '      <SupplementalProperty schemeIdUri="tag:transaction.science,2026:wai/floor" value="%s"/>' % esc_xml(render(floor)),
           '      <Representation id="%s" bandwidth="%d">' % (esc_xml(rep_id), bandwidth),
           '        <SegmentList timescale="%d">' % ts, '          <SegmentTimeline>']
    out += ['            <S t="%d" d="%d"/>' % sg for sg, _ in offered]
    out += ['          </SegmentTimeline>'] + ['          <SegmentURL media="%s"/>' % esc_xml(m) for _, m in offered]
    out += ['        </SegmentList>', '      </Representation>', '    </AdaptationSet>', '  </Period>', '</MPD>']
    return "\n".join(out) + "\n"


# ---------------------------------------------- reading an MPD (Appendix I)

MAX_TIMELINES = 256
MAX_DEPTH = 64
NAME = re.compile(r"[A-Za-z_:][A-Za-z0-9_:.\-]*", re.ASCII)


class Xml:
    """A reader of the XML subset of Appendix I over a normalised text."""

    def __init__(self, s):
        self.s, self.at = s, 0

    def bad(self):
        return Refusal("playlist_malformed")

    def rest(self):
        return self.s[self.at:]

    def eat(self, lit):
        if self.s.startswith(lit, self.at):
            self.at += len(lit)
            return True
        return False

    def space(self):
        n = 0
        while self.at + n < len(self.s) and self.s[self.at + n] in " \t\n":
            n += 1
        self.at += n
        return n > 0

    def name(self):
        m = NAME.match(self.s, self.at)
        if m is None:
            raise self.bad()
        self.at = m.end()
        return m.group(0)

    def reference(self):
        end = self.s.find(";", self.at)
        if end < 0:
            raise self.bad()
        body = self.s[self.at:end]
        named = {"amp": "&", "lt": "<", "gt": ">", "quot": '"', "apos": "'"}
        if body in named:
            c = named[body]
        else:
            if body.startswith("#x") and re.fullmatch(r"[0-9A-Fa-f]{1,8}", body[2:], re.ASCII):
                n = int(body[2:], 16)
            elif body.startswith("#") and re.fullmatch(r"[0-9]{1,10}", body[1:], re.ASCII) and int(body[1:]) <= 0xFFFFFFFF:
                n = int(body[1:])
            else:
                raise self.bad()
            if 0xD800 <= n <= 0xDFFF or n > 0x10FFFF or not xml_char(chr(n)):
                raise self.bad()
            c = chr(n)
        self.at = end + 1
        return c

    def attr_value(self):
        if self.at >= len(self.s) or self.s[self.at] not in "\"'":
            raise self.bad()
        q = self.s[self.at]
        self.at += 1
        out = []
        while True:
            if self.at >= len(self.s):
                raise self.bad()
            c = self.s[self.at]
            self.at += 1
            if c == q:
                return "".join(out)
            if c == "<":
                raise self.bad()
            if c == "&":
                out.append(self.reference())
            elif c in "\t\n":
                out.append(" ")
            else:
                out.append(c)

    def comment(self):
        self.at += 4
        end = self.s.find("--", self.at)
        if end < 0 or not self.s.startswith("-->", end):
            raise self.bad()
        self.at = end + 3

    def misc(self):
        while True:
            if self.space():
                continue
            if self.s.startswith("<!--", self.at):
                self.comment()
                continue
            return

    def start_tag(self):
        name = self.name()
        attrs, names = [], set()
        while True:
            spaced = self.space()
            if self.eat("/>"):
                return {"name": name, "attrs": attrs, "children": []}, True
            if self.eat(">"):
                return {"name": name, "attrs": attrs, "children": []}, False
            if not spaced:
                raise self.bad()
            k = self.name()
            self.space()
            if not self.eat("="):
                raise self.bad()
            self.space()
            v = self.attr_value()
            seen = any(a == k for a, _ in attrs) if MUTATE == "xml-attributes-scanned" else k in names
            if seen and MUTATE != "xml-attribute-twice":
                raise self.bad()
            names.add(k)
            attrs.append((k, v))


def read_xml(text):
    """The root element of a document in the XML subset of Appendix I:
    {"name", "attrs", "children"}; Refusal("playlist_malformed") otherwise."""
    if text.startswith("\ufeff"):
        text = text[1:]
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    if not all(xml_char(c) for c in text):
        raise Refusal("playlist_malformed")
    r = Xml(text)
    if text.startswith("<?xml") and text[5:6] in (" ", "\t", "\n", "?") and text[5:6] != "":
        end = text.find("?>")
        if end < 0:
            raise r.bad()
        r.at = end + 2
    r.misc()
    if not r.eat("<"):
        raise r.bad()
    root, closed = r.start_tag()
    stack, done = ([], root) if closed else ([root], None)
    while done is None:
        if r.at >= len(text):
            raise r.bad()
        c = text[r.at]
        if c == "<":
            if text.startswith("<!--", r.at):
                r.comment()
            elif r.eat("</"):
                name = r.name()
                r.space()
                if not r.eat(">"):
                    raise r.bad()
                e = stack.pop()
                if e["name"] != name:
                    raise r.bad()
                if stack:
                    stack[-1]["children"].append(e)
                else:
                    done = e
            elif text.startswith("<!", r.at) or text.startswith("<?", r.at):
                raise r.bad()
            else:
                r.at += 1
                e, closed = r.start_tag()
                if closed:
                    stack[-1]["children"].append(e)
                else:
                    if len(stack) >= MAX_DEPTH:
                        raise r.bad()
                    stack.append(e)
        elif c == "&":
            r.at += 1
            r.reference()
        else:
            if text.startswith("]]>", r.at):
                raise r.bad()
            r.at += 1
    r.misc()
    if r.at != len(text):
        raise r.bad()
    return done


def local(e):
    return e["name"].rsplit(":", 1)[-1]


def xattr(e, k):
    return next((v for a, v in e["attrs"] if a == k), None)


def duration_ms(s):
    """An xs:duration of days, hours, minutes and seconds in milliseconds; None otherwise."""
    if not s.startswith("P"):
        return None
    rest = s[1:]
    date, sep, time = rest.partition("T")
    time = time if sep else None
    integer = lambda v: int(v) if re.fullmatch(r"[0-9]+", v, re.ASCII) else None
    ms, anything = 0, False
    if date:
        if not date.endswith("D") or integer(date[:-1]) is None:
            return None
        ms, anything = integer(date[:-1]) * 86400000, True
    if time is not None:
        if time == "":
            return None
        for unit, scale in (("H", 3600000), ("M", 60000)):
            i = time.find(unit)
            if i >= 0:
                v = integer(time[:i])
                if v is None:
                    return None
                ms += v * scale
                time, anything = time[i + 1:], True
        if time:
            if not time.endswith("S"):
                return None
            v = decimal_ms(time[:-1])
            if v is None:
                return None
            ms, anything = ms + v, True
    return ms if anything and ms <= U64 else None


def unsigned(v):
    if v is None or not re.fullmatch(r"[0-9]+", v, re.ASCII) or int(v) > U64:
        raise Refusal("playlist_malformed")
    return int(v)


def timeline_of(e, inherited):
    """A timeline as written: {"timescale", "runs": [(t, d, count)]}, one run per S, never
    expanded; None without a SegmentTimeline."""
    st = next((c for c in e["children"] if local(c) == "SegmentTimeline"), None)
    if st is None:
        return None
    ts = next((xattr(x, "timescale") for x in [e] + inherited if xattr(x, "timescale") is not None), None)
    timescale = unsigned(ts) if ts is not None else 1
    if timescale == 0:
        raise Refusal("playlist_malformed")
    runs, nxt = [], 0
    for sel in (c for c in st["children"] if local(c) == "S"):
        d = unsigned(xattr(sel, "d"))
        if d == 0:
            raise Refusal("playlist_malformed")
        t = unsigned(xattr(sel, "t")) if xattr(sel, "t") is not None else nxt
        r = unsigned(xattr(sel, "r")) if xattr(sel, "r") is not None else 0
        end = t + d * (r + 1)
        if end > U64:
            raise Refusal("playlist_malformed")
        if MUTATE == "dash-r-expanded" and r < 1 << 20:
            runs += [(t + i * d, d, 1) for i in range(r + 1)]
        else:
            runs.append((t, d, r + 1))
        nxt = end
    return {"timescale": timescale, "runs": runs}


SEGMENT_INFO = ("SegmentTemplate", "SegmentList")


def read_mpd(text):
    """An MPD's periods: [{"id", "start_ms", "sets": [{"id", "timelines"}]}]."""
    root = read_xml(text)
    if local(root) != "MPD":
        raise Refusal("playlist_malformed")
    periods, held = [], 0
    for i, p in enumerate(c for c in root["children"] if local(c) == "Period"):
        st = xattr(p, "start")
        if st is not None:
            start = duration_ms(st)
            if start is None:
                raise Refusal("playlist_malformed")
        elif i == 0:
            start = 0
        else:
            raise Refusal("playlist_malformed")
        p_info = [c for c in p["children"] if local(c) in SEGMENT_INFO]
        sets = []
        for a in (c for c in p["children"] if local(c) == "AdaptationSet"):
            tls = []
            a_info = [c for c in a["children"] if local(c) in SEGMENT_INFO]
            for e in a_info:
                tl = timeline_of(e, [x for x in p_info if local(x) == local(e)])
                if tl is not None:
                    tls.append(tl)
            for rep in (c for c in a["children"] if local(c) == "Representation"):
                for e in (c for c in rep["children"] if local(c) in SEGMENT_INFO):
                    tl = timeline_of(e, [x for x in a_info + p_info if local(x) == local(e)])
                    if tl is not None:
                        tls.append(tl)
            held += len(tls)
            if held > MAX_TIMELINES and MUTATE != "dash-timelines-unbounded":
                raise Refusal("playlist_malformed")
            sets.append({"id": xattr(a, "id"), "timelines": tls})
        periods.append({"id": xattr(p, "id"), "start_ms": start, "sets": sets})
    if not periods:
        raise Refusal("playlist_malformed")
    return periods


def within(wai, classical):
    """Whether the WAI runs list only segments of the classical runs, each with its t and
    d, in order: for each WAI segment, the first classical segment at or after the one
    matched before with its t has its d too. Run by run, never expanded."""
    ci, co = 0, 0
    for wt, wd, wn in wai:
        t, left = wt, wn
        while left > 0:
            while True:
                if ci == len(classical):
                    return False
                ct, cd, cn = classical[ci]
                if t >= ct + co * cd and (t - ct) % cd == 0 and (t - ct) // cd < cn:
                    co = (t - ct) // cd
                    break
                ci, co = ci + 1, 0
            ct, cd, cn = classical[ci]
            if cd != wd:
                return False
            take = min(left, cn - co)
            left, co, t = left - take, co + take, t + take * wd
            if co == cn:
                ci, co = ci + 1, 0
    return True


def check_dash_adjacent(wai, classical):
    """§11.2 in DASH: Refusal("playlist_malformed") when either does not read,
    Refusal("adjacency_misaligned") when not adjacent."""
    w, k = read_mpd(wai), read_mpd(classical)
    mis = Refusal("adjacency_misaligned")
    for wp in w:
        kp = next((p for p in k if p["id"] == wp["id"]), None)
        if kp is None or (kp["start_ms"] != wp["start_ms"] and MUTATE != "dash-start-ignored"):
            raise mis
        if not wp["sets"]:
            raise mis
        for st in wp["sets"]:
            if not st["timelines"]:
                raise mis
            for tl in st["timelines"]:
                if not tl["runs"] or not any((c["timescale"] == tl["timescale"] or MUTATE == "dash-timescale-ignored") and within(tl["runs"], c["runs"])
                                             for cs in kp["sets"] for c in cs["timelines"]):
                    raise mis


def mpd_timeline(text, period_id, set_id, max_segments):
    """The classical MPD's timeline (§11.1): (period id, start ms, timescale, segments),
    expanded only when it lists at most max_segments segments."""
    periods = read_mpd(text)
    p = next((x for x in periods if x["id"] == period_id), None) if period_id is not None else periods[0]
    if p is None:
        raise Refusal("playlist_malformed")
    st = next((x for x in p["sets"] if x["id"] == set_id), None) if set_id is not None else next((x for x in p["sets"] if x["timelines"]), None)
    if st is None or not st["timelines"]:
        raise Refusal("playlist_malformed")
    tl = st["timelines"][0]
    if sum(n for _, _, n in tl["runs"]) > max_segments:
        raise Refusal("playlist_malformed")
    return (p["id"], p["start_ms"], tl["timescale"], [(t + i * d, d) for t, d, n in tl["runs"] for i in range(n)])


# ------------------------------------------------ playlist receipts (§8.6)

def jcs(v):
    """RFC 8785 over the subset the render digest uses."""
    if v is None:
        return "null"
    if v is True or v is False:
        return "true" if v else "false"
    if isinstance(v, int):
        if not 0 <= v <= 2 ** 53 - 1:
            raise ValueError("outside the canonical subset")
        return str(v)
    if isinstance(v, str):
        return json.dumps(v, ensure_ascii=False)
    if isinstance(v, list):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
    return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"


# Ed25519, strictly as WAI verifies every signature (SPEC §7 *Signatures*): the key and R
# decoded with y < p (and no sign bit on x = 0), neither of small order, S < L, and
# [S]B = R + [k]A checked by encoding [S]B − [k]A and comparing it with R byte for byte.
# `_ed25519.py` writes it from RFC 8032's definitions, with the standard library only;
# --mutate small-order-key and small-order-r leave out its two order checks.
import _ed25519 as ED  # noqa: E402


def ed25519_verify(pk, msg, sig):
    return ED.verifies(pk, msg, sig, skip=(MUTATE,))


RECEIPT_KNOWN = {"kind", "manifest_hash", "session_class_hash", "class", "segments", "merkle_root", "work_compose_ops", "viewers_served",
                 "joules_micro", "grant", "parent_receipt_hash", "signer_pubkey", "signer_id", "sig", "render", "extras",
                 "energy_provenance", "energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref"}
LEGACY_CLASSES = {"HwShunt": 1, "ModelBased": 2, "Estimator": 3}


MAX_FILE_NUMBER = 2 ** 53 - 1


class ReceiptMalformed(Exception):
    pass


class NotUnsigned:
    """A JSON number with a sign, a fraction or an exponent: never an unsigned integer of
    the receipt file, whatever its value (so "-0" and "1.0" are not 0 and 1)."""


def read_receipt(text):
    """The receipt file (§8.6), read strictly; ReceiptMalformed otherwise. A number is an
    unsigned integer written as digits alone."""
    def pairs(kv):
        # A member name twice in one object (I-JSON, RFC 7493 §2.3), names compared as read.
        if len({k for k, _ in kv}) != len(kv) and MUTATE != "receipt-duplicates":
            raise ReceiptMalformed()
        return dict(kv)

    try:
        # A number is an unsigned integer of at most 2^53 - 1, written as digits alone.
        o = json.loads(text, object_pairs_hook=pairs, parse_float=lambda t: NotUnsigned(),
                       parse_int=lambda t: NotUnsigned() if t.startswith("-") or (int(t) > MAX_FILE_NUMBER and MUTATE != "file-numbers-unbounded") else int(t))
    except (ValueError, RecursionError):
        raise ReceiptMalformed()
    bad = ReceiptMalformed

    def texts(v):
        if isinstance(v, str):
            yield v
        elif isinstance(v, dict):
            for k, x in v.items():
                yield k
                yield from texts(x)
        elif isinstance(v, list):
            for x in v:
                yield from texts(x)

    # A \u escape that leaves a surrogate unpaired denotes no Unicode text (RFC 8259 §8.2).
    if any(0xD800 <= ord(c) <= 0xDFFF for t in texts(o) for c in t) and MUTATE != "receipt-surrogates":
        raise bad()

    def obj(v, n):
        if not isinstance(v, dict) or len(v) != n:
            raise bad()
        return v

    def u64(v):
        if isinstance(v, bool) or not isinstance(v, int) or not 0 <= v <= U64:
            raise bad()
        return v

    def s(v):
        if not isinstance(v, str):
            raise bad()
        return v

    def hx(v, n):
        if not isinstance(v, str) or not re.fullmatch("[0-9a-f]{%d}" % (2 * n), v):
            raise bad()
        return bytes.fromhex(v)

    def opt(v, f):
        return None if v is None else f(v)

    if not isinstance(o, dict) or not set(o) <= RECEIPT_KNOWN or o.get("kind") != "wai.video.manifest":
        raise bad()
    for k in ("manifest_hash", "session_class_hash", "class", "segments", "merkle_root", "work_compose_ops", "viewers_served",
              "joules_micro", "grant", "parent_receipt_hash", "signer_pubkey", "signer_id", "sig"):
        if k not in o:
            raise bad()
    cl = obj(o["class"], 6)
    if not isinstance(cl.get("blackouts"), list):
        raise bad()
    klass = {"content_id": s(cl.get("content_id")), "entitlement": s(cl.get("entitlement")), "geo": s(cl.get("geo")),
             "blackouts": [s(b) for b in cl["blackouts"]], "ad_pod_id": s(cl.get("ad_pod_id")), "window": u64(cl.get("window"))}
    segs = []
    if not isinstance(o["segments"], list):
        raise bad()
    for x in o["segments"]:
        x = obj(x, 5)
        if x.get("kind") not in ("content", "ad") or not (x.get("ad_id") is None or isinstance(x.get("ad_id"), str)) or "ad_id" not in x:
            raise bad()
        segs.append({"index": u64(x.get("index")), "ad": x["kind"] == "ad", "uri": s(x.get("uri")), "ms": u64(x.get("duration_ms")), "ad_id": x["ad_id"]})
    g = obj(o["grant"], 4)
    if "funds_ceiling" not in g or "joule_ceiling_micro" not in g:
        raise bad()
    # No joule ceiling is written null and signs as 2^64 - 1.
    grant = {"hash": hx(g.get("grant_hash"), 32), "capability": s(g.get("capability")),
             "ceiling": U64 if g["joule_ceiling_micro"] is None else u64(g["joule_ceiling_micro"]), "funds": opt(g["funds_ceiling"], u64)}
    label = o.get("energy_provenance")
    if label is None:
        if any(o.get(k) is not None for k in ("energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref")):
            raise bad()
        prov = None
    elif label in LEGACY_CLASSES and all(o.get(k) is None for k in ("energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref")):
        prov = label
    else:
        raise SystemExit("http_verify: an energy class this verifier does not read")
    render_, extras = None, []
    if ("render" in o) != ("extras" in o):
        raise bad()
    if "render" in o:
        r = obj(o["render"], 11)
        if not isinstance(r.get("can_block_reload"), bool) or not isinstance(r.get("end_list"), bool) or not isinstance(r.get("independent_segments"), bool):
            raise bad()
        if "playlist_type" not in r or r["playlist_type"] not in (None, "VOD", "EVENT") or "map_uri" not in r:
            raise bad()
        render_ = {"version": u64(r.get("version")), "seq": u64(r.get("media_sequence")), "dseq": u64(r.get("discontinuity_sequence")),
                   "target": opt(r.get("target_duration_s"), u64), "part_target": opt(r.get("part_target_ms"), u64),
                   "hold_back": opt(r.get("part_hold_back_ms"), u64), "block": r["can_block_reload"], "end": r["end_list"],
                   "ptype": r["playlist_type"], "indep": r["independent_segments"], "map": opt(r["map_uri"], s)}
        for k in ("target_duration_s", "part_target_ms", "part_hold_back_ms"):
            if k not in r:
                raise bad()
        if render_["version"] > 255 or not isinstance(o["extras"], list):
            raise bad()

        def layers(v):
            if not isinstance(v, list):
                raise bad()
            out = []
            for l in v:
                l = obj(l, 2)
                n = u64(l.get("n"))
                if n > 255:
                    raise bad()
                out.append({"n": n, "uri": s(l.get("uri"))})
            return out

        for e in o["extras"]:
            e = obj(e, 6)
            if not isinstance(e.get("gap"), bool) or not isinstance(e.get("discontinuity"), bool) or not isinstance(e.get("parts"), list) or "program_date_time" not in e or "preload_hint" not in e:
                raise bad()
            parts = []
            for q in e["parts"]:
                q = obj(q, 4)
                if not isinstance(q.get("independent"), bool):
                    raise bad()
                parts.append({"ms": u64(q.get("duration_ms")), "independent": q["independent"], "uri": s(q.get("uri")), "layers": layers(q.get("layers"))})
            extras.append({"discontinuity": e["discontinuity"], "gap": e["gap"], "pdt": opt(e["program_date_time"], s), "parts": parts,
                           "layers": layers(e.get("layers")), "hint": opt(e["preload_hint"], s)})
    return {"manifest_hash": hx(o["manifest_hash"], 32), "class_hash": hx(o["session_class_hash"], 32), "class": klass, "segments": segs,
            "merkle_root": hx(o["merkle_root"], 32), "work": u64(o["work_compose_ops"]), "viewers": u64(o["viewers_served"]),
            "joules": u64(o["joules_micro"]), "prov": prov, "grant": grant, "parent": opt(o["parent_receipt_hash"], lambda v: hx(v, 32)),
            "pubkey": hx(o["signer_pubkey"], 32), "signer_id": s(o["signer_id"]), "sig": hx(o["sig"], 64), "render": render_, "extras": extras,
            "render_json": (o.get("render"), o.get("extras"))}


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


def be(n, width):
    return n.to_bytes(width, "big")


def text_bytes(s):
    """A receipt's text as the bytes hashed and compared. The reader has already refused
    an unpaired surrogate (RFC 8259 §8.2) with its reason, so valid text is plain UTF-8;
    encoding never raises, so a reader that admitted one (the receipt-surrogates
    mutation) reaches a verdict over the bytes a lenient reader would hash, never a
    crash."""
    return s.encode("utf-8", "surrogatepass")


def lp(s):
    b = text_bytes(s)
    return be(len(b), 4) + b


def segment_leaf(sg):
    ad = (b"\x01" + lp(sg["ad_id"])) if sg["ad_id"] is not None else b"\x00"
    return b3(b"wai:video-manifest-seg\x01", be(sg["index"], 8), bytes([2 if sg["ad"] else 1]), be(sg["ms"], 8), lp(sg["uri"]), ad)


def merkle_root(hashes):
    if not hashes:
        return bytes(32)
    level = [b3(b"moq-jwp:merkle-leaf\x01", h) for h in hashes]
    while len(level) > 1:
        if len(level) % 2:
            level.append(level[-1])
        level = [b3(b"moq-jwp:merkle-node\x01", level[i], level[i + 1]) for i in range(0, len(level), 2)]
    return level[0]


def class_hash(c):
    bs = sorted(set(c["blackouts"]), key=text_bytes)
    out = [b"wai:video-manifest-class\x01"] + [lp(c[k]) for k in ("content_id", "entitlement", "geo", "ad_pod_id")]
    out.append(be(len(bs), 4))
    out += [lp(b) for b in bs]
    out.append(be(c["window"], 8))
    return b3(*out)


def render_digest(r):
    """BLAKE3 of the render domain and the JCS of {"render", "extras"}, as the receipt
    file carries them; None without a rendering or past the canonical subset."""
    if r["render"] is None:
        return None
    try:
        c = jcs({"render": r["render_json"][0], "extras": r["render_json"][1]})
    except ValueError:
        return None
    return b3(b"wai:video-manifest-render\x01", text_bytes(c))


def signing_payload(r):
    o = [b"wai:video-manifest-receipt\x01", r["manifest_hash"], r["class_hash"], r["merkle_root"], be(r["work"], 8), be(r["viewers"], 8), be(r["joules"], 8),
         bytes([LEGACY_CLASSES[r["prov"]] if r["prov"] else 0]), r["grant"]["hash"], lp(r["grant"]["capability"]), be(r["grant"]["ceiling"], 8),
         (b"\x01" + be(r["grant"]["funds"], 8)) if r["grant"]["funds"] is not None else b"\x00",
         (b"\x01" + r["parent"]) if r["parent"] is not None else b"\x00", be(len(r["segments"]), 8)]
    o += [segment_leaf(sg) for sg in r["segments"]]
    o += [r["pubkey"], lp(r["signer_id"])]
    if r["render"] is not None and MUTATE != "render-digest-off":
        d = render_digest(r)
        o.append(b"\x01" + d if d is not None else b"\x02")
    return b"".join(o)


def receipt_hash(r):
    return b3(b"wai:video-manifest-receipt-id\x01", signing_payload(r), r["sig"])


def rendered(r):
    rr = r["render"] or {"version": 6, "seq": 0, "dseq": 0, "target": None, "part_target": None, "hold_back": None, "block": False, "end": True,
                         "ptype": None, "indep": False, "map": None}
    return render_hls(r["segments"], rr, r["extras"])


def receipt_verifies(r):
    if r["render"] is None and r["extras"]:
        return False
    if r["render"] is not None:
        try:
            check_render(r["segments"], r["render"], r["extras"])
        except Refusal:
            return False
        if render_digest(r) is None:
            return False
    if blake3(text_bytes(rendered(r))) != r["manifest_hash"]:
        return False
    if merkle_root([segment_leaf(sg) for sg in r["segments"]]) != r["merkle_root"] or class_hash(r["class"]) != r["class_hash"]:
        return False
    if r["work"] != len(r["segments"]) or r["viewers"] == 0:
        return False
    if any(sg["index"] != i or sg["ad"] != (sg["ad_id"] is not None) for i, sg in enumerate(r["segments"])):
        return False
    if (r["prov"] is None) != (r["joules"] == 0):
        return False
    return ed25519_verify(r["pubkey"], signing_payload(r), r["sig"])


def check_playlist(playlist, receipt_bytes, name):
    """§8.6, §11.6, in order: the file reads; its name; the bytes; the receipt."""
    try:
        r = read_receipt(receipt_bytes.decode("utf-8"))
    except (ReceiptMalformed, UnicodeDecodeError):
        raise Refusal("playlist_receipt_malformed")
    if name is not None and name != r["manifest_hash"].hex() + ".json" and MUTATE != "name-unchecked":
        raise Refusal("receipt_name_mismatch")
    if blake3(playlist) != r["manifest_hash"] or text_bytes(rendered(r)) != playlist:
        raise Refusal("playlist_receipt_mismatch")
    if not receipt_verifies(r):
        raise Refusal("playlist_receipt_invalid")
    return r


WALK_BOUND = 64
LIVE_URL = "https://origin.example/live/wai-720.m3u8"


def advance(chain, cur, fetch):
    """§8.6: take the receipt cur (checked against its playlist) into the chain a client
    holds, {"head", "bindings", "bound"}; fetch(h) gives the bytes at mr/<h>.json or None.
    Returns ("first" | "unchanged" | "chained" | "reanchored", walked) and updates chain,
    or raises Refusal and leaves it as it was."""
    if chain["head"] is None:
        b = bind_receipt(chain["bindings"], cur)
        chain["head"], chain["bindings"] = cur, b
        return "first", 0
    target = receipt_hash(chain["head"])
    if receipt_hash(cur) == target and MUTATE != "reload-walked":
        return "unchanged", 0
    if cur["pubkey"] != chain["head"]["pubkey"] and MUTATE != "key-unchecked":
        raise Refusal("playlist_receipt_unchained")
    walked, link = [], cur["parent"]
    while MUTATE != "chain-unchecked":
        if link is None:
            raise Refusal("playlist_receipt_unchained")
        if link == target:
            break
        if len(walked) == chain["bound"] and MUTATE != "walk-unbounded":
            # Past the bound: a revision ahead of the head re-anchors the chain, its
            # bindings kept; one that is not ahead is refused.
            if last_segment(cur) <= last_segment(chain["head"]) and MUTATE != "reanchor-backwards":
                raise Refusal("playlist_receipt_walk_exceeded")
            if MUTATE == "walk-exceeded-wedges":
                raise Refusal("playlist_receipt_walk_exceeded")
            chain["bindings"] = bind_receipt(chain["bindings"], cur)
            chain["head"] = cur
            return "reanchored", len(walked)
        raw = fetch(link)
        if raw is None:
            raise Refusal("playlist_receipt_parent_missing")
        try:
            r = read_receipt(raw.decode("utf-8"))
        except (ReceiptMalformed, UnicodeDecodeError):
            raise Refusal("playlist_receipt_malformed")
        if receipt_hash(r) != link and MUTATE != "walk-name-unchecked":
            raise Refusal("receipt_name_mismatch")
        if not receipt_verifies(r) and MUTATE != "walk-verify-unchecked":
            raise Refusal("playlist_receipt_invalid")
        if r["pubkey"] != cur["pubkey"] and MUTATE != "key-unchecked":
            raise Refusal("playlist_receipt_unchained")
        link = r["parent"]
        walked.append(r)
    b = chain["bindings"]
    for r in list(reversed(walked)) + [cur]:
        b = bind_receipt(b, r)
    chain["head"], chain["bindings"] = cur, b
    return "chained", len(walked)


def last_segment(r):
    """The media sequence number after the last complete segment r lists."""
    return (r["render"]["seq"] if r["render"] is not None else 0) + len(r["segments"])


def bind_receipt(b, r):
    try:
        p = parse_media_playlist(rendered(r))
    except Refusal:
        raise Refusal("playlist_receipt_malformed")
    try:
        return bind(b, p)
    except Refusal as e:
        raise Refusal("playlist_listing_rebound" if e.code == "playlist_listing_rebound" else "playlist_receipt_malformed")


# ---------------------------------------------------- the playlist-side cases

def edit_text(t, edits):
    for e in edits or []:
        if e[0] not in t:
            raise SystemExit("http_verify: a case edits text its file does not hold")
        t = t.replace(e[0], e[1], 1)
    return t


def read_text(d, path):
    return open(os.path.join(d, path), "rb").read().decode("utf-8")


def layer_of(env):
    h = whole_head(env)
    if not is_layer(h):
        raise Refusal("not_a_layer")
    try:
        r = SV.parse_refines(h["manifest"].get("refines"))
    except SV.Malformed:
        raise Refusal("not_a_layer")
    return {"n": r["layer"], "cid": content_hash(env), "len": len(env)}


def parts_of(objs, ps, cms):
    out = []
    for j, p in enumerate(ps or []):
        env = objs[p["object"]]
        ms = cms[j] if j < len(cms) else p.get("duration_ms")
        if not isinstance(ms, int):
            raise SystemExit("http_verify: a part with no duration")
        whole_head(env)
        out.append({"alias": p["alias"], "cid": content_hash(env), "n": len(env), "ms": ms, "independent": p.get("independent") is True,
                    "layers": [layer_of(objs[x]) for x in p.get("layers") or []]})
    return out


def render_media_case(d, objs, inp):
    c = parse_media_playlist(read_text(d, inp["classical"]))
    segs = []
    for i, s in enumerate(inp["segments"]):
        k = c["segments"][i] if i < len(c["segments"]) else None
        env = objs[s["object"]]
        h = whole_head(env)
        w = {"cid": content_hash(env), "n": len(env), "h": covered_len(h["head_len"], len(env)) if h["version"] == "WAI2" else None,
             "ms": s["duration_ms"] if isinstance(s.get("duration_ms"), int) else (k["ms"] if k else 0), "gap": s.get("gap") is True}
        w["layers"] = [layer_of(objs[x]) for x in s.get("layers") or []]
        w["parts"] = parts_of(objs, s.get("parts"), [q["ms"] for q in k["extra"]["parts"]] if k else [])
        segs.append(w)
    open_ = None
    if inp.get("open") is not None:
        o = inp["open"]
        cms = [q["ms"] for q in c["open"]["parts"]] if c["open"] else []
        open_ = {"parts": parts_of(objs, o.get("parts"), cms), "hint": o.get("preload_hint")}
    end = inp["end_list"] if isinstance(inp.get("end_list"), bool) else c["end"]
    return wai_media_playlist(c, segs, open_, inp.get("uri_prefix") or "", end)


def resolve_uri(objs, uri):
    base, _, frag = uri.partition("#")
    cid = None
    for kv in frag.split("&") if frag else []:
        k, eq, v = kv.partition("=")
        if eq and k == "cid" and re.fullmatch(r"[0-9a-f]{64}", v):
            cid = v
    if cid is None:
        name = base.rsplit("/", 1)[-1]
        if name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
            cid = name[:-4]
    b = next((b for b in objs.values() if cid is not None and content_hash(b).hex() == cid), None)
    if b is None:
        raise SystemExit("http_verify: %s names no corpus object" % uri)
    return b


def listed_objects(objs, p):
    segs, parts, every = [], [], []

    def extra(e):
        for q in e["parts"]:
            b = resolve_uri(objs, q["uri"])
            parts.append((len(b), q["ms"]))
            every.append(b)
            every.extend(resolve_uri(objs, l["uri"]) for l in q["layers"])
        every.extend(resolve_uri(objs, l["uri"]) for l in e["layers"])

    for s in p["segments"]:
        if s["extra"]["gap"]:
            continue
        b = resolve_uri(objs, s["uri"])
        every.append(b)
        segs.append((len(b) + sum(len(resolve_uri(objs, l["uri"])) for l in s["extra"]["layers"]), s["ms"]))
        extra(s["extra"])
    if p["open"] is not None:
        extra(p["open"])
    return segs, parts, every


def union_floor_of(every):
    so = [seq_object(e) for e in every]
    return union_of(so), floor_of(so)


def multivariant_case(d, objs, inp):
    out = ["#EXTM3U"]
    for v in inp["variants"]:
        p = parse_media_playlist(read_text(d, v["playlist"]))
        segs, parts, every = listed_objects(objs, p)
        union, floor = union_floor_of(every)
        bw = bandwidths(segs if segs else parts)
        if bw is None:
            raise SystemExit("http_verify: a variant with no duration")
        uri = v["uri"]
        if not uri_ok(uri, False) or union[0] != floor[0]:
            raise Refusal("playlist_malformed")
        line = '#EXT-X-STREAM-INF:BANDWIDTH=%d,AVERAGE-BANDWIDTH=%d,CODECS="%s",X-WAI-FLOOR="%s"' % (bw[0], bw[1], render(union), render(floor))
        if v.get("resolution"):
            line += ",RESOLUTION=%dx%d" % tuple(v["resolution"])
        if v.get("frame_rate_milli") is not None:
            if v["frame_rate_milli"] == 0:
                raise Refusal("playlist_malformed")
            line += ",FRAME-RATE=" + secs(v["frame_rate_milli"])
        out += [line, uri]
    return "\n".join(out) + "\n"


def mpd_case(d, objs, inp):
    p = parse_media_playlist(read_text(d, inp["playlist"]))
    segs, _, every = listed_objects(objs, p)
    union, floor = union_floor_of(every)
    peak = bandwidths(segs)[0]
    if isinstance(inp.get("classical_mpd"), str):
        t = mpd_timeline(read_text(d, inp["classical_mpd"]), inp.get("period_id"), inp.get("classical_set"), len(p["segments"]))
    else:
        c = parse_media_playlist(read_text(d, inp["classical"]))
        at, segs_t = 0, []
        for s in c["segments"]:
            segs_t.append((at, s["ms"]))
            at += s["ms"]
            if at > U64:
                raise Refusal("playlist_malformed")
        t = (inp.get("period_id") or "p0", 0, 1000, segs_t)
    media = [None if s["extra"]["gap"] else s["uri"] for s in p["segments"]]
    return wai_mpd(t, inp.get("set_id") or 1, union, floor, inp["rep_id"], peak, media)


def rendered_case(d, inp, t):
    b = t.encode("utf-8")
    o = {"blake3": blake3(b).hex(), "bytes": len(b)}
    if "file" in inp:
        o["same_as_file"] = os.path.exists(os.path.join(d, inp["file"])) and open(os.path.join(d, inp["file"]), "rb").read() == b
    return o


def parsed_case(p, t):
    parts = lambda ps: [[q["ms"], q["independent"], [l["n"] for l in q["layers"]]] for q in ps]
    return {"version": p["version"], "target_duration_s": p["target"], "media_sequence": p["seq"], "discontinuity_sequence": p["dseq"],
            "part_target_ms": p["part_target"], "part_hold_back_ms": p["hold_back"], "can_block_reload": p["block"], "end_list": p["end"],
            "playlist_type": p["ptype"], "independent_segments": p["indep"], "map_uri": p["map"],
            "segments": [[s["ms"], s["extra"]["discontinuity"], s["extra"]["pdt"], parts(s["extra"]["parts"]), [l["n"] for l in s["extra"]["layers"]], s["uri"], s["extra"]["gap"]] for s in p["segments"]],
            "open": None if p["open"] is None else [p["open"]["discontinuity"], p["open"]["pdt"], parts(p["open"]["parts"])],
            "preload_hint": p["hint"], "rendered_form": is_rendered_form(p, t)}


def live_case(d, objs, inp):
    """§8.6, §11.4, §11.6 over a live playlist's revisions in the order a client fetched
    them, and parts fetched by alias."""
    revisions = inp["revisions"]
    parents = inp.get("parents")
    if parents is None:
        parents = (revisions[0]["wai"].rsplit("/", 1)[0] if "/" in revisions[0]["wai"] else "") + "/mr" if revisions else "/mr"

    def fetch(h):
        f = os.path.join(d, parents, h.hex() + ".json")
        return open(f, "rb").read() if os.path.isfile(f) else None

    url = inp["url"] if isinstance(inp.get("url"), str) else LIVE_URL
    chain = {"head": None, "bindings": new_bindings(url), "bound": inp["walk_bound"] if isinstance(inp.get("walk_bound"), int) else WALK_BOUND}
    listed_b = new_bindings(url)
    revs, listed = [], []
    for r in revisions:
        w = edit_text(read_text(d, r["wai"]), r.get("edit_wai"))
        k = read_text(d, r["classical"])
        try:
            check_adjacent(w, k)
            adjacent = True
        except Refusal as e:
            adjacent = e.code
        p = parse_media_playlist(w)
        try:
            listed_b = bind(listed_b, p)
        except Refusal as e:
            revs.append({"adjacent": adjacent, "listing": e.code, "receipt": None, "chained": None, "walked": None})
            listed.append(None)
            continue
        listed.append(p)
        rpath = r["receipt"]
        name = None if rpath.startswith("cases/") else rpath.rsplit("/", 1)[-1]
        try:
            rr = check_playlist(w.encode("utf-8"), open(os.path.join(d, rpath), "rb").read(), name)
        except Refusal as e:
            revs.append({"adjacent": adjacent, "listing": True, "receipt": e.code, "chained": None, "walked": None})
            continue
        try:
            step, walked = advance(chain, rr, fetch)
            ch = step if step in ("unchanged", "reanchored") else True
        except Refusal as e:
            ch, walked = e.code, None
        revs.append({"adjacent": adjacent, "listing": True, "receipt": "signed", "chained": ch, "walked": walked})
    fetches = []
    for f in inp.get("fetches") or []:
        served = apply_patch(objs[f["object"]], f.get("patch_object"))
        first = next(((k, c) for k, p in enumerate(listed) if p is not None for c in [listed_cid(p, f["alias"])] if c is not None), None)
        if first is None:
            fetches.append({"alias": f["alias"], "listed_in": None, "outcome": "not_listed", "held": None})
        elif content_hash(served) == first[1]:
            held = "signed" if cid_of(chain["bindings"], f["alias"]) == first[1] else "listed"
            fetches.append({"alias": f["alias"], "listed_in": first[0], "outcome": "accepted", "held": held})
        else:
            fetches.append({"alias": f["alias"], "listed_in": first[0], "outcome": "object_digest_mismatch", "held": None})
    return {"revisions": revs, "fetches": fetches}


def playlist_evaluate(d, objs, op, inp, obj):
    """The playlist and adaptation ops; None for any other op."""
    if op == "media_playlist":
        return rendered_case(d, inp, render_media_case(d, objs, inp))
    if op == "multivariant":
        return rendered_case(d, inp, multivariant_case(d, objs, inp))
    if op == "mpd":
        return rendered_case(d, inp, mpd_case(d, objs, inp))
    if op == "playlist_parse":
        t = inp["text"] if "text" in inp else read_text(d, inp["file"])
        return parsed_case(parse_media_playlist(t), t)
    inline = lambda k: inp[k + "_text"] if isinstance(inp.get(k + "_text"), str) else read_text(d, inp[k])
    if op == "uri_resolve":
        return {"url": rfc3986_resolve(inp["base"], inp["reference"])}
    if op == "dash_adjacency":
        check_dash_adjacent(edit_text(inline("wai"), inp.get("edit_wai")), edit_text(inline("classical"), inp.get("edit_classical")))
        return {"adjacent": True}
    if op == "adjacency":
        check_adjacent(edit_text(inline("wai"), inp.get("edit_wai")), edit_text(inline("classical"), inp.get("edit_classical")))
        return {"adjacent": True}
    if op == "unchanged":
        classical_unchanged(inline("original"), edit_text(inline("published"), inp.get("edit_published")))
        return {"unchanged": True}
    if op == "render":
        segs = [{"ms": x[0], "uri": x[1], "ad": x[2] is True} for x in inp["segments"]]
        r = inp["render"]
        rr = {"version": r["version"] % 256, "seq": r["media_sequence"], "dseq": r["discontinuity_sequence"], "target": r["target_duration_s"],
              "part_target": r["part_target_ms"], "hold_back": r["part_hold_back_ms"], "block": r["can_block_reload"] is True, "end": r["end_list"] is True,
              "ptype": r.get("playlist_type") if r.get("playlist_type") in ("VOD", "EVENT") else None, "indep": r.get("independent_segments") is True,
              "map": r.get("map_uri") if isinstance(r.get("map_uri"), str) else None}
        txt = lambda v: v if isinstance(v, str) else None
        lays = lambda ls: [{"n": l["n"] % 256, "uri": l["uri"]} for l in ls or []]
        extras = [{"discontinuity": e.get("discontinuity") is True, "gap": e.get("gap") is True, "pdt": txt(e.get("program_date_time")),
                   "parts": [{"ms": q["duration_ms"], "independent": q.get("independent") is True, "uri": q["uri"], "layers": lays(q.get("layers"))} for q in e.get("parts") or []],
                   "layers": lays(e.get("layers")), "hint": txt(e.get("preload_hint"))} for e in inp.get("extras") or []]
        check_render(segs, rr, extras)
        t = render_hls(segs, rr, extras).encode("utf-8")
        return {"blake3": blake3(t).hex(), "bytes": len(t)}
    if op == "ilog2":
        return {"log2_q16": ilog2_q16(inp["a"], inp["b"])}
    if op == "playlist_receipt":
        pl = edit_text(read_text(d, inp["playlist"]), inp.get("edit_playlist")).encode("utf-8")
        rc = edit_text(read_text(d, inp["receipt"]), inp.get("edit_receipt")).encode("utf-8")
        name = inp["name"] if isinstance(inp.get("name"), str) else (inp["receipt"].rsplit("/", 1)[-1] if inp.get("name") is None else None)
        r = check_playlist(pl, rc, name)
        return {"signed": receipt_hash(r).hex(), "parent": r["parent"].hex() if r["parent"] is not None else None}
    if op == "live":
        return live_case(d, objs, inp)
    if op == "abr":
        k, b = decide(inp["policy"], abr_state(inp["state"]))
        return {"choice": k, "budget": b}
    if op == "abr_order":
        return {"order": order_entries(whole_head(obj(inp)), inp.get("advertises") or [], inp["policy"], abr_state(inp["state"]))}
    if op == "depth":
        layers = [len(objs[i]) for i in inp["objects"]] if "objects" in inp else inp["layers_n"]
        k, b = depth(layers, inp["policy"], abr_state(inp["state"]))
        return {"depth": k, "budget": b}
    return None


def refusal(e):
    out = {"error": e.code}
    if e.skipped is not None:
        out["skipped"] = e.skipped
    return out


RECEIPT_OPS = ("receipt_file", "receipts", "key_log", "grants_file", "param_set")


def receipt_evaluate(d, objs, op, inp):
    """§8.1-§8.5's cases, which http_receipts.py checks. They need the `cryptography`
    package, through the receipt and key verifiers it reads; nothing else here does."""
    import http_receipts as RC
    RC.MUTATE = MUTATE

    def fetch_object(env, f):
        """What a client holds of `env` after fetching it as the `fetch` op does: the cid
        and how it was established (§8.3), or nothing; and the fetch's verdict."""
        t = f.get("tamper") or {}
        served = apply_patch(env, t.get("patch_object"))
        try:
            r = fetch(env, served, frag_of(f.get("fragment")), sink_of(d, f, objs), f.get("threshold_permille", 900), t,
                      f.get("listed_as_layer", False))
        except Refusal as e:
            return None, e.code
        if r["role"] == "layer" and 0 in r["abandoned"]:
            return None, r["verified_by"]
        return (content_hash(env).hex(), "head-proof" if r["verified_by"] == "head-proof" else "envelope"), r["verified_by"]

    try:
        return RC.evaluate(d, objs, op, inp, fetch_object, content_hash)
    except RC.Fault as e:
        return {"fault": str(e)}


def pad_values(b, n):
    """SPEC §2, *Reader limits*: the envelope `b` with a member no revision defines added
    at the end of its manifest, "pad", an array of `n` zeros, and its man_len set to the
    manifest's new length; the manifest holds n + 1 values more."""
    ml = u32(b, 4)
    m = b[8:8 + ml - 1] + b',"pad":[0' + b",0" * (n - 1) + b"]}"
    return b[:4] + len(m).to_bytes(4, "little") + m + b[8 + ml:]


def evaluate(d, objs, op, inp):
    def obj(i, patch=True):
        if "envelope_hex" in i:
            b = bytes.fromhex(i["envelope_hex"])
        elif "object" in i:
            b = objs[i["object"]]
        else:
            b = open(os.path.join(d, i["file"]), "rb").read()
        if "pad_values" in i:
            b = pad_values(b, i["pad_values"])
        return apply_patch(b, i.get("patch_object")) if patch else b

    try:
        if op == "parse_head":
            b = obj(inp)
            n = inp.get("envelope_len", len(b))
            p = min(inp.get("prefix", len(b)), len(b))
            h = parse_head(b[:p], n)
            if isinstance(h, tuple):
                return {"need": h[1]}
            return {"version": h["version"], "head_len": h["head_len"], "envelope_len": h["envelope_len"],
                    "covered_len": covered_len(h["head_len"], n), "components": [list(c) for c in h["components"]]}
        if op == "head_proof":
            b = obj(inp)
            original = obj(inp, patch=False)
            cid = content_hash(original)
            pf = open(os.path.join(d, inp["proof"]), "rb").read() if "proof" in inp else make_proof_file(original)
            p = read_proof(pf)
            verify_proof(p, cid, b)
            return {"covered_len": p["covered"], "n_cv": len(p["cvs"]), "envelope_len": p["n"]}
        if op == "plan":
            b = obj(inp)
            h = parse_head(b, len(b))
            pl = plan(h, [tuple(w) for w in inp["wanted"]], [tuple(r) for r in inp.get("held", [])], inp.get("threshold_permille", 900))
            if pl[0] == "rest":
                return {"whole_rest": list(pl[1]), "why": pl[2]}
            return {"requests": [list(r) for r in pl[1]], "dropped": pl[2]}
        if op == "response":
            r = check_response(inp.get("request"), inp["status"], inp.get("content_range"), inp.get("content_encoding"), inp["body_len"], inp["envelope_len"])
            return {"ok": list(r)}
        if op == "select":
            b = obj(inp)
            h = whole_head(b)
            pay = [b[s:s + ln] for s, ln in h["components"]]
            adv, _, held, order, _, _ = sink_of(d, inp, objs)
            order = order(h) if callable(order) else (order if order is not None else list(range(len(pay))))
            try:
                i, sk = select(h["manifest"], order, adv, held, lambda i, e: payload_ok(e, pay[i]))
                return {"pick": i, "skipped": sk}
            except Refusal as e:
                return {"inert": e.skipped}
        if op in ("fetch", "same_pick"):
            env = obj(inp, patch=False)
            t = inp.get("tamper") or {}
            served = apply_patch(env, t.get("patch_object"))
            r = fetch(env, served, frag_of(inp.get("fragment")), sink_of(d, inp, objs), inp.get("threshold_permille", 900), t,
                      inp.get("listed_as_layer", False))
            if op == "fetch":
                return r
            whole_pick = evaluate(d, objs, "select", inp)
            if "pick" not in whole_pick or whole_pick["pick"] != r["pick"]:
                return {"paths_differ": [whole_pick, r]}
            return {"pick": r["pick"]}
        if op == "fetch_timed":
            env = obj(inp, patch=False)
            sk = sink_of(d, inp, objs)
            return fetch_timed(env, frag_of(inp.get("fragment")), sk, inp.get("threshold_permille", 900), inp.get("deadline_ms"),
                               inp["script"], inp.get("flip"))
        if op == "codecs":
            return {"elements": parse_codecs(inp["value"])}
        if op == "union_floor":
            envs = [bytes.fromhex(x) for x in inp["envelopes_hex"]] if "envelopes_hex" in inp else [objs[i] for i in inp["objects"]]
            os_ = [seq_object(e) for e in envs]
            return {"union": render(union_of(os_)), "floor": render(floor_of(os_))}
        if op == "supports":
            return {"supports": supports(parse_element(inp["floor"]), inp["containers"], inp["caps"])}
        if op == "layer_tag":
            return parse_layer_tag(inp["line"])
        if op in RECEIPT_OPS:
            return receipt_evaluate(d, objs, op, inp)
        r = playlist_evaluate(d, objs, op, inp, obj)
        if r is not None:
            return r
    except Refusal as e:
        if op == "same_pick" and e.code == "inert":
            try:
                evaluate(d, objs, "select", inp)["inert"]
                return {"inert": True}
            except KeyError:
                return {"paths_differ": "range inert, whole picks"}
        return refusal(e)
    raise SystemExit("unknown op " + op)


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

def self_test():
    v = json.load(open(os.path.join(HERE, "blake3_vectors.json")))
    for c in v["cases"]:
        data = bytes(i % 251 for i in range(c["input_len"]))
        got = blake3(data)
        if MUTATE == "blake3-iv":
            got = bytes([got[0] ^ 1]) + got[1:]
        if got.hex() != c["hash"]:
            print("FAIL self-test: BLAKE3 of %d bytes" % c["input_len"])
            return False
        if c["input_len"] > 1024:
            n = c["input_len"]
            l = 1024 * 2 ** ((((n - 1) // 1024)).bit_length() - 1)
            if B3.parent_cv(B3.subtree_cv(data[:l], 0), B3.subtree_cv(data[l:], l), True) != got:
                print("FAIL self-test: the tree functions over %d bytes" % n)
                return False
    return mpd_bounds_hold()


def mpd_bounds_hold():
    """Appendix I's reader holds what a document writes, not what its @r says, and reads
    an element's attributes in linear time: 8 timelines of 2^16 segments each in a
    kilobyte, read in a footprint bounded by its size (tracemalloc's peak); and one
    element of 40000 attributes, read in bounded CPU time of this thread (never wall
    time)."""
    import tracemalloc
    bomb = "<MPD><Period id=\"p\"><AdaptationSet>%s</AdaptationSet></Period></MPD>" % "".join(
        '<Representation id="r%d"><SegmentList timescale="1"><SegmentTimeline><S t="0" d="1" r="65535"/></SegmentTimeline></SegmentList></Representation>' % i
        for i in range(8))
    tracemalloc.start()
    try:
        periods = read_mpd(bomb)
        peak = tracemalloc.get_traced_memory()[1]
    finally:
        tracemalloc.stop()
    if sum(n for tl in periods[0]["sets"][0]["timelines"] for _, _, n in tl["runs"]) != 8 << 16 or peak > 64 * len(bomb) + (1 << 20):
        print("FAIL self-test: an MPD of %d bytes held %d bytes" % (len(bomb), peak))
        return False
    wide = "<MPD %s/>" % " ".join('a%d=""' % i for i in range(40000))
    t0 = time.thread_time()
    try:
        read_mpd(wide)
    except Refusal:
        pass
    spent = time.thread_time() - t0
    if spent > 2.0:
        print("FAIL self-test: one element of 40000 attributes read in %.2f s of CPU" % spent)
        return False
    return bindings_retention_holds()


def bindings_retention_holds():
    """§11.5: a binding is kept while one of the last RETENTION revisions lists it, then
    forgotten, so a rebinding is refused at the span's last revision and not after it."""
    cid = lambda b: bytes([b]) * 32
    first = parse_media_playlist('#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXT-X-PART-INF:PART-TARGET=0.5\n#EXT-X-PART:DURATION=0.5,URI="p.wai#cid=%s"\n' % cid(1).hex())
    again = parse_media_playlist('#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXT-X-PART-INF:PART-TARGET=0.5\n#EXT-X-PART:DURATION=0.5,URI="./p.wai#cid=%s"\n' % cid(2).hex())
    other = parse_media_playlist("#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXTINF:1,\n%s.wai\n" % cid(3).hex())
    b = bind(new_bindings("https://origin.example/live/p.m3u8"), first)
    for _ in range(RETENTION - 1):
        b = bind(b, other)
    try:
        bind(b, again)
        print("FAIL self-test: a rebinding within the retention was taken")
        return False
    except Refusal:
        pass
    b = bind(b, other)
    try:
        bind(b, again)
    except Refusal:
        print("FAIL self-test: a binding outlived the retention")
        return False
    return True


def run_cases(d, path):
    """Evaluate every case of a case file (the corpus's format) against the corpus at d,
    and require each committed outcome."""
    load_registry()
    objs = {i: open(os.path.join(d, o["path"]), "rb").read() for i, o in json.load(open(os.path.join(d, "objects.json"))).items()}
    cases = json.load(open(path))
    bad = 0
    for c in cases:
        got = evaluate(d, objs, c["op"], c["input"])
        if got != c["expect"]:
            bad += 1
            if bad <= 20:
                print("DIFF %s: rust %s, python %s" % (c["name"], json.dumps(c["expect"]), json.dumps(got)))
    print("%d/%d cases agree" % (len(cases) - bad, len(cases)))
    return bad == 0


def run(d):
    load_registry()
    ok = self_test()
    listing = json.load(open(os.path.join(d, "objects.json")))
    objs = {}
    for i, o in listing.items():
        b = open(os.path.join(d, o["path"]), "rb").read()
        if content_hash(b).hex() != o["cid"]:
            print("FAIL object %s: its bytes do not hash to its cid" % i)
            ok = False
        if open(os.path.join(d, o["path"] + ".hp"), "rb").read() != make_proof_file(b):
            print("FAIL object %s: its head proof is not the one the text gives" % i)
            ok = False
        objs[i] = b
    cases = json.load(open(os.path.join(d, "cases.json")))
    passed = 0
    for c in cases:
        got = evaluate(d, objs, c["op"], c["input"])
        if got == c["expect"]:
            passed += 1
        else:
            print("FAIL %s: expected %s, got %s" % (c["name"], json.dumps(c["expect"]), json.dumps(got)))
            ok = False
    print("%d/%d cases give their outcome%s" % (passed, len(cases), " (mutation %s)" % MUTATE if MUTATE else ""))
    return ok


def diff(path, flip, seed):
    load_registry()
    lines = [json.loads(l) for l in open(path) if l.strip()]
    rng = random.Random(seed)
    candidates = [i for i, l in enumerate(lines) if len(bytes.fromhex(l["proof"])) > 53]
    flipped = set(rng.sample(candidates, flip)) if flip else set()
    refused = set()
    for i, l in enumerate(lines):
        proof = bytearray(bytes.fromhex(l["proof"]))
        if i in flipped:
            k = proof[52]
            at = 53 + 32 * rng.randrange(k) + rng.randrange(32)
            proof[at] ^= 1 << rng.randrange(8)
        try:
            p = read_proof(bytes(proof))
            if p["n"] != l["envelope_len"]:
                raise Refusal("head_proof_malformed")
            verify_proof(p, bytes.fromhex(l["cid"]), bytes.fromhex(l["prefix"]))
        except Refusal:
            refused.add(i)
    print("%d proofs; %d flipped; %d refused" % (len(lines), len(flipped), len(refused)))
    if refused != flipped:
        print("FAIL: refused %s, flipped %s" % (sorted(refused - flipped)[:10], sorted(flipped - refused)[:10]))
        return False
    return True


def main(argv):
    global MUTATE
    if "--mutate" in argv:
        i = argv.index("--mutate")
        MUTATE = argv[i + 1]
        argv = argv[:i] + argv[i + 2:]
    if argv and argv[0] == "cases":
        return 0 if run_cases(argv[2] if len(argv) > 2 else HERE, argv[1]) else 1
    if argv and argv[0] == "diff":
        flip = int(argv[argv.index("--flip") + 1]) if "--flip" in argv else 0
        seed = int(argv[argv.index("--seed") + 1]) if "--seed" in argv else 1
        return 0 if diff(argv[1], flip, seed) else 1
    d = argv[0] if argv else HERE
    return 0 if run(d) else 1


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