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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. It is a second native sink: for every case it opens a session from the case's
configuration, derives the session key from the case's seed (RFC 8032 §5.1.5), and gives each call
the answer it reaches by the extension's rules, from the standard library and these second
implementations of the texts the extension cites:

- a playlist receipt and the chain of a live playlist's receipts (http-delivery §8.6, §11.5), and
  RFC 3986 resolution, with ../http-conformance/http_verify.py;
- the bundle (stream-audit §9, §10) with stream_audit_verify.py, whose readers of acknowledgements,
  claims and the session record are settlement_verify.py's, ad_presentation_verify.py's and its
  own; every signature with ../claim-conformance/_ed25519.py, WAI's strict rule.

Everything else is here: the configuration (§3.1), reading a playlist (§4.1), a playlist's verdict
(§4.2), an object's (§4.3), the observations, spans and claim spans (§5.1, stream-audit §5.2,
ad-presentation §10), the log (§5.2), the termination rule (§6.1), the energy entries (§7) and the
refused calls (§3.3, §8). For the end of a session it requires the bundle the corpus states to be
the one its own session holds: every acknowledgement's fields and item lists, every claim's fields,
the session record's fields, the log and the notes byte for byte, and the collector's outcome
(stream-audit §10.3) the case states. It does not sign: Ed25519 is deterministic, and every byte a
signature covers is checked.

Then it mutates copies of the corpus, one defect each, and requires every copy to fail; and it
drops each of its own checks in turn and requires the corpus to catch the weakened verifier.

    python3 -I native_sink_verify.py <live-conformance/native-sink>

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

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

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
sys.path.insert(0, os.path.join(HERE, "..", "http-conformance"))
import _ed25519 as ED  # noqa: E402
import _blake3 as B3  # noqa: E402
import http_verify as HV  # noqa: E402
import settlement_verify as SV  # noqa: E402
import ad_presentation_verify as APV  # noqa: E402
import stream_audit_verify as SAV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "native-sink.md")

CHANNEL_DOMAIN = b"wai:audit-channel\x01"            # stream-audit §3.1
KEY_SESSION_DOMAIN = b"wai:ad-presentation-session\x01"  # stream-audit §3.2
ORIGIN_DOMAIN = b"wai:settle-origin\x01"             # settlement §14.2
MARKER_DOMAIN = b"wai:audit-marker\x01"              # stream-audit §6.1
CREATIVE_DOMAIN = b"wai:ad-creative\x01"             # ad-presentation §4
ITEMS_DOMAIN = b"wai:settle-items\x01"               # settlement §3.2
OBJECT_DOMAIN = b"moq-jwp:object\x01"                # jwp-receipts §2.1
LIVE, ON_DEMAND = 1, 2
ENDED = {"ended": 1, "user": 2, "hidden": 3, "idle": 4, "error": 5}
EVENTS = ["start", "present", "stall-start", "stall-end", "end", "abandon"]
ZERO = bytes(32)
NO_STARTUP = 0xFFFFFFFF
NOT_EXPOSED = 0xFFFFFFFF
U64 = 2 ** 64 - 1

DROP = set()  # the checks dropped, for the weakened-verifier pass


class Refused(Exception):
    def __init__(self, code):
        super().__init__(code)
        self.code = code


class Disagree(Exception):
    pass


def sha(*parts):
    return hashlib.sha256(b"".join(parts)).digest()


def unhex(s):
    return bytes.fromhex(s)


def is_uint(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= U64


def public_key(seed):
    """RFC 8032 §5.1.5: the public key of a 32-byte seed."""
    h = hashlib.sha512(seed).digest()
    a = int.from_bytes(h[:32], "little")
    a &= (1 << 254) - 8
    a |= 1 << 254
    q, p = (0, 1, 1, 0), ED.B
    while a:
        if a & 1:
            q = ED._add(q, p)
        p = ED._add(p, p)
        a >>= 1
    return ED.encode(q)


# ------------------------------------------------------------------ §3.1 the configuration

def channel_id(name):
    if not isinstance(name, str) or not 1 <= len(name) <= 255 or not all(0x21 <= ord(c) <= 0x7E for c in name):
        return None
    return sha(CHANNEL_DOMAIN, name.encode())


def read_config(v):
    members = ["channel", "owner", "presentation", "window_us", "ad_classes", "content_addressed", "require", "unseen_grace_ms", "idle_ms"]
    bad = Refused("native_config_invalid")
    if not isinstance(v, dict) or ("config-members" not in DROP and any(k not in members for k in v)):
        raise bad
    if channel_id(v.get("channel")) is None:
        raise bad
    owner = v.get("owner")
    if not isinstance(owner, str) or not re.fullmatch(r"[0-9a-f]{64}", owner):
        raise bad
    pres = v.get("presentation")
    if pres not in (None, "live", "on_demand"):
        raise bad

    def num(k, d):
        x = v.get(k)
        if x is None:
            return d
        if not is_uint(x):
            raise bad
        return x
    cfg = {"channel": v["channel"], "owner": unhex(owner), "presentation": {None: None, "live": LIVE, "on_demand": ON_DEMAND}[pres],
           "window_us": num("window_us", 10_000_000), "unseen_grace_ms": num("unseen_grace_ms", 10_000), "idle_ms": num("idle_ms", 0)}
    if cfg["window_us"] < 1_000_000 and "window-floor" not in DROP:
        raise bad
    ac = v.get("ad_classes")
    if ac is None:
        ac = []
    if not isinstance(ac, list) or not all(isinstance(x, str) for x in ac):
        raise bad
    cfg["ad_classes"] = ac
    ca = v.get("content_addressed")
    if ca is None:
        ca = False
    if not isinstance(ca, bool):
        raise bad
    cfg["content_addressed"] = ca
    req = v.get("require")
    if req is None:
        req = "nothing"
    if req not in ("nothing", "signed"):
        raise bad
    cfg["require"] = req
    return cfg


# ------------------------------------------------------------------ §4.1 reading a playlist

def attributes(s):
    out = {}
    i, n = 0, len(s)
    while i < n:
        j = i
        while j < n and (s[j].isupper() and s[j].isascii() or s[j].isdigit() and s[j].isascii() or s[j] == "-"):
            j += 1
        if j == i or j >= n or s[j] != "=":
            i = j if j > i else i + 1
            continue
        name = s[i:j]
        j += 1
        if j < n and s[j] == '"':
            e = s.find('"', j + 1)
            if e >= 0:
                out[name] = s[j + 1:e]
                i = e + 1
                continue
        e = s.find(",", j)
        e = n if e < 0 else e
        out[name] = s[j:e]
        i = e
    return out


def days_from_civil(y, m, d):
    y -= m <= 2
    era = y // 400
    yoe = y - era * 400
    doy = (153 * ((m + 9) % 12) + 2) // 5 + d - 1
    doe = yoe * 365 + yoe // 4 - yoe // 100 + doy
    return era * 146097 + doe - 719468


def date_ms(s):
    """RFC 3339 §5.6 date-time, the fraction truncated to the millisecond."""
    m = re.fullmatch(r"(\d{4})-(\d{2})-(\d{2})[Tt](\d{2}):(\d{2}):(\d{2})(\.(\d+))?([Zz]|([+-])(\d{2}):(\d{2}))", s)
    if not m:
        return None
    y, mo, d, h, mi, se = (int(m.group(k)) for k in range(1, 7))
    leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
    dim = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    if not 1 <= mo <= 12 or not 1 <= d <= dim[mo - 1] or h > 23 or mi > 59 or se > 59:
        return None
    ms = int((m.group(8) or "0")[:3].ljust(3, "0"))
    off = 0
    if m.group(10):
        oh, om = int(m.group(11)), int(m.group(12))
        if oh > 23 or om > 59:
            return None
        off = (oh * 60 + om) * 60_000 * (1 if m.group(10) == "+" else -1)
    return ((days_from_civil(y, mo, d) * 24 + h) * 60 + mi) * 60_000 + se * 1000 + ms - off


def seconds_ms(s):
    """A decimal number of seconds read exactly, rounded half up to the millisecond."""
    m = re.match(r"\s*(\d*)(?:\.(\d*))?", s)
    whole, frac = m.group(1), m.group(2)
    if not whole and frac is None:
        return None
    if not whole:
        whole = "0"
    if frac is not None and frac == "" and m.group(1) == "":
        return None
    f = (frac or "").ljust(4, "0")
    ms = int(whole) * 1000 + int(f[:3])
    return ms + (1 if int(f[3]) >= 5 else 0)


def resolve(base, ref):
    return HV.rfc3986_resolve(base, ref)


def read_playlist(text, base, ad_classes):
    p = {"segments": [], "map": None, "keys": [], "ads": [], "on_demand": False, "multivariant": False}
    duration, program = None, None
    for raw in text.split("\n"):
        line = raw.rstrip("\r \t")
        if line.startswith("#EXTINF:"):
            d = seconds_ms(line[8:])
            duration = max(1, d or 0)
        elif line.startswith("#EXT-X-PROGRAM-DATE-TIME:"):
            program = date_ms(line[25:].strip())
        elif line.startswith("#EXT-X-ENDLIST"):
            p["on_demand"] = True
        elif line.startswith("#EXT-X-STREAM-INF:") or line.startswith("#EXT-X-I-FRAME-STREAM-INF:"):
            p["multivariant"] = True
        elif line.startswith("#EXT-X-MAP:"):
            u = attributes(line[11:]).get("URI")
            if u is not None:
                p["map"] = resolve(base, u)
        elif line.startswith("#EXT-X-KEY:") or line.startswith("#EXT-X-SESSION-KEY:"):
            u = attributes(line[line.index(":") + 1:]).get("URI")
            if u is not None:
                p["keys"].append(resolve(base, u))
        elif line.startswith("#EXT-X-DATERANGE:"):
            a = attributes(line[17:])
            is_ad = "SCTE35-OUT" in a or "SCTE35-CMD" in a or a.get("CLASS") in ad_classes
            dv = a.get("DURATION", a.get("PLANNED-DURATION"))
            d = seconds_ms(dv) if dv is not None else None
            st = date_ms(a["START-DATE"]) if "START-DATE" in a else None
            if is_ad and d is not None and st is not None:
                p["ads"].append({"marker": line, "start": st, "duration": d})
        elif line and not line.startswith("#"):
            if not p["multivariant"]:
                p["segments"].append({"url": resolve(base, line), "duration": duration or 1, "program": program})
                if program is not None and duration is not None:
                    program += duration
            duration = None
    if p["multivariant"]:
        p["segments"], p["map"] = [], None
    return p


def origin_of(url):
    m = re.match(r"([^:/?#]+)://([^/?#]*)", url)
    if not m:
        return None
    scheme, auth = m.group(1).lower(), m.group(2)
    auth = auth.rsplit("@", 1)[-1]
    if not auth:
        return None
    if auth.startswith("["):
        host, _, rest = auth[1:].partition("]")
        host, port = "[" + host.lower() + "]", rest[1:] if rest.startswith(":") else None
    elif ":" in auth:
        host, port = auth.rsplit(":", 1)
        host = host.lower()
    else:
        host, port = auth.lower(), None
    default = {"http": "80", "https": "443"}.get(scheme, "")
    return f"{scheme}://{host}:{port}" if port and port != default else f"{scheme}://{host}"


def stated_cid(line):
    """http-delivery §3.1, §5.2: the cid a URI states, its fragment's `cid`, else a last path
    segment `<cid>.wai`."""
    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 unhex(v)
    name = last_segment(base)
    if name and name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
        return unhex(name[:-4])
    return None


def last_segment(u):
    u = re.split(r"[?#]", u, maxsplit=1)[0]
    s = u.rsplit("/", 1)[-1]
    return s or None


def stated_sha256(url):
    name = last_segment(url)
    if name is None:
        return None
    m = re.fullmatch(r"([0-9a-f]{64})\.([A-Za-z0-9]+)", name)
    if not m or m.group(2) == "wai":
        return None
    return unhex(m.group(1))


def content_hash(b):
    return B3.blake3(OBJECT_DOMAIN + b)


# ------------------------------------------------------------------ §5 spans, ad spans, the log

class ContentSpans:
    def __init__(self, start):
        self.start, self.done, self.cur = start, [], None

    def ms(self, at):
        return max(0, (at - self.start) // 1000)

    def close(self, at):
        if self.cur is not None:
            end = self.ms(at)
            if end > self.cur["s"]:
                self.done.append({"start": self.cur["s"], "length": end - self.cur["s"], "visible": self.cur["v"], "flags": (1 if self.cur["a"] else 0) | (2 if self.cur["ad"] else 0)})
        self.cur = None

    def observe(self, at, playing, vis, audible, ad):
        if not playing:
            return self.close(at)
        cls = vis // 100
        if self.cur is not None and (cls != self.cur["c"] or audible != self.cur["a"] or ad != self.cur["ad"]):
            self.close(at)
        if self.cur is not None:
            self.cur["v"] = min(self.cur["v"], vis)
            return
        self.cur = {"s": self.ms(at), "v": vis, "c": cls, "a": audible, "ad": ad}


class AdSpans:
    def __init__(self):
        self.done, self.cur, self.max_end, self.resume_next = [], None, 0, False

    def close(self):
        c = self.cur
        if c is not None and c["e"] > c["s"]:
            self.done.append({"start": c["s"], "length": c["e"] - c["s"], "visible": c["v"], "flags": (1 if c["a"] else 0) | (2 if c["r"] else 0)})
            self.max_end = max(self.max_end, c["e"])
        self.cur = None

    def observe(self, pos, playing, vis, audible, seeked):
        pos = max(0, pos)
        if not playing or seeked:
            self.close()
            if self.done:
                self.resume_next = True
            if not playing:
                return
        cls = vis // 100
        c = self.cur
        if c is not None:
            if pos < c["e"]:
                return
            if cls != c["c"] or audible != c["a"]:
                c["e"] = pos
                self.close()
            else:
                c["e"], c["v"] = pos, min(c["v"], vis)
                return
        if pos < self.max_end:
            return
        continues = not self.resume_next and bool(self.done) and pos == self.max_end
        self.cur = {"s": pos, "e": pos, "v": vis, "c": cls, "a": audible, "r": bool(self.done) and not continues}
        self.resume_next = False


OBJECT_NULLS = ["ot", "url", "content_hash", "capability", "rendition", "layer", "layers_total", "layer_cap", "counted", "transferred", "d_ms", "dl_ms",
                "ttfb_ms", "ttlb_ms", "rc", "path", "receipt"]


def observation(sid, sn, at, event, st, content_hash=None, rc=None):
    obj = None
    if event == "response":
        obj = {k: None for k in OBJECT_NULLS}
        obj["content_hash"], obj["rc"] = content_hash, rc
    present = {"unit": 0, "layers": 1, "first": True, "late_ms": None, "decode_us": None} if event == "present" else None
    o = {"v": 1, "sid": sid, "sn": sn, "ts_ms": at // 1000, "event": event, "cid": None, "st": st, "sf": "h", "object": obj, "present": present}
    for k in ["switch", "layer_change", "capable", "cohort", "metrics", "claim"]:
        o[k] = None
    return o


def cmcd_report(o):
    """delivery-telemetry §3, §4: the Event-mode report of an observation."""
    if o["event"] == "present" and not o["present"]["first"]:
        return None
    sf = lambda s: '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'  # noqa: E731
    pre = "science.transaction.wai-"
    m = {"v": "2", "sid": sf(o["sid"]), "sn": str(o["sn"]), "st": o["st"], "sf": o["sf"], "ts": str(o["ts_ms"])}
    if o["object"] is not None:
        if o["object"]["content_hash"] is not None:
            m[pre + "ch"] = sf(o["object"]["content_hash"])
        if o["object"]["rc"] is not None:
            m["rc"] = str(o["object"]["rc"])
    e, sta = {"response": ("rr", None), "start": ("ps", "s"), "present": ("ps", "p"), "stall-end": ("ps", "p"), "stall-start": ("ps", "r"),
              "abandon": ("ps", "q"), "end": ("ps", "e")}[o["event"]]
    m["e"] = e
    if sta:
        m["sta"] = sta
    if o["present"] is not None:
        m[pre + "u"], m[pre + "pl"] = sf(str(o["present"]["unit"])), sf(str(o["present"]["layers"]))
    return ",".join(f"{k}={m[k]}" for k in sorted(m))


def class_tuple(e):
    """An entry's class as energy-measurement §2.7 encodes it: its tag, then its evidence."""
    lab, u = e["energy_provenance"], e.get("energy_uncertainty") or {}
    if lab == "ModelBased":
        return (2,)
    if lab == "Estimator":
        return (3,)
    if lab == "OnChipCounter":
        return (4, u.get("relative_ppm"), u.get("window_us"), {"Unknown": 0, "Off": 1, "On": 2}.get(e.get("energy_counter_filtering")))
    return (5, u.get("relative_ppm"), u.get("window_us"), e.get("energy_calibration_ref"))


# ------------------------------------------------------------------ the sink

class Sink:
    def __init__(self, cfg, seed, start):
        self.cfg, self.start = cfg, start
        self.player = public_key(seed)
        self.channel = channel_id(cfg["channel"])
        self.session = sha(KEY_SESSION_DOMAIN, self.channel, self.player)[:16]
        self.sid = self.session.hex()
        self.presentation = cfg["presentation"]
        self.received, self.by_url, self.notes, self.log, self.cmcd_taken = [], {}, [], [], 0
        self.spans = ContentSpans(start)
        self.segments, self.ads, self.keys = {}, [], set()
        self.roles = {3: (0, None), 4: (0, None)}
        self.attachments, self.last = [], None
        self.holds, self.listed, self.given = {}, {}, {}
        self.unseen_since, self.last_interaction, self.ended = None, None, None
        if "log-start" not in DROP:
            self.log_event("start", start)

    def st(self):
        return "l" if (self.presentation or LIVE) == LIVE else "v"

    def log_event(self, name, at):
        self.log_push(name, at)

    def log_push(self, event, at, content_hash=None, rc=None):
        """§5.2: nothing after `end` or `abandon`; times never decrease."""
        if self.has(["end", "abandon"]) and "log-closed" not in DROP:
            return
        last = self.log[-1]["ts_ms"] if self.log else 0
        if "log-order" not in DROP:
            at = max(at, last * 1000)
        self.log.append(observation(self.sid, len(self.log), at, event, self.st(), content_hash, rc))

    def has(self, names):
        return any(o["event"] in names for o in self.log)

    def live(self):
        if self.ended is not None and "after-end" not in DROP:
            raise Refused("native_session_ended")

    # §5.2
    def event(self, name, at):
        self.live()
        if name not in EVENTS:
            raise Refused("native_input_invalid")
        if name == "present" and self.has(["present"]):
            return
        self.log_event(name, at)

    def hold(self, k):
        return self.holds.setdefault(k, {"chain": {"head": None, "bindings": HV.new_bindings(k), "bound": HV.WALK_BOUND}, "parents": {}})

    # §4.2
    def playlist(self, url, b, receipt):
        self.live()
        v = self.take_playlist(url, b, receipt)
        if v["need_parent"] is None:
            self.given[(resolve(url, url), B3.blake3(b))] = (v["listing"], v["refused"])
        v["receipt_name"] = "mr/" + B3.blake3(b).hex() + ".json"
        v["present"] = v["refused"] is None and v["need_parent"] is None
        return {k: v[k] for k in ["kind", "listing", "receipt", "receipt_name", "step", "need_parent", "refused", "present", "segments", "ads"]}

    def take_playlist(self, url, b, receipt):
        v = {"kind": "media", "listing": "listed", "receipt": None, "step": None, "need_parent": None, "refused": None, "segments": 0, "ads": 0}
        try:
            text = b.decode("utf-8")
        except UnicodeDecodeError:
            text = None
        if text is None or len(b) > 16 * 1024 * 1024 or not text.lstrip("").startswith("#EXTM3U"):
            v["refused"] = "playlist_malformed"
            return v
        k = resolve(url, url)
        p = read_playlist(text, k, self.cfg["ad_classes"])
        if p["multivariant"]:
            v["kind"] = "multivariant"
            return v
        signed = False
        if receipt is not None and "receipts" not in DROP:
            try:
                r = HV.check_playlist(b, receipt, None)
            except HV.Refusal as e:
                v["refused"] = e.code
                return v
            h = self.hold(k)
            needed = []

            def fetch(x):
                if x in h["parents"]:
                    return h["parents"][x]
                needed.append(x)
                return None
            if "chain" not in DROP:
                try:
                    step, _ = HV.advance(h["chain"], r, fetch)
                except HV.Refusal as e:
                    if e.code == "playlist_receipt_parent_missing" and needed:
                        v["need_parent"] = "mr/" + needed[0].hex() + ".json"
                        return v
                    v["refused"] = e.code
                    return v
                v["step"] = step
            v["receipt"] = HV.receipt_hash(r).hex()
            signed = True
        elif self.cfg["require"] == "signed" and "require-signed" not in DROP:
            v["refused"] = "playlist_unsigned"
            return v
        v["listing"] = "signed" if signed else "listed"
        if self.presentation is None:
            self.presentation = ON_DEMAND if p["on_demand"] else LIVE
        for x in p["keys"]:
            self.keys.add(resolve(x, x))
        cids = {}
        for raw in text.split("\n"):
            line = raw.rstrip("\r \t")
            if line and not line.startswith("#"):
                c = stated_cid(line)
                if c is not None:
                    cids[resolve(k, line)] = c

        def list_(u, media):
            e = self.listed.setdefault(u, {"signed": signed, "cid": cids.get(u), "media": media})
            e["signed"] = e["signed"] or signed
            e["media"] = e["media"] or media
            if e["cid"] is None:
                e["cid"] = cids.get(u)
        if p["map"] is not None:
            list_(p["map"], True)
        for s in p["segments"]:
            list_(s["url"], True)
            self.segments[s["url"]] = s
        for a in p["ads"]:
            if not any(x["ad"]["marker"] == a["marker"] for x in self.ads):
                self.ads.append({"ad": a, "rec": AdSpans()})
        v["segments"], v["ads"] = len(p["segments"]), len(p["ads"])
        return v

    def parent(self, url, name, b):
        self.live()
        m = re.fullmatch(r"mr/([0-9a-f]{64})\.json", name)
        self.hold(resolve(url, url))["parents"][unhex(m.group(1))] = b

    # §4.3
    def object(self, meta, b):
        self.live()
        members = ["url", "t_us", "media", "status", "cmsd_static", "cmsd_dynamic"]
        if not isinstance(meta, dict) or any(k not in members for k in meta) or not isinstance(meta.get("url"), str) or not is_uint(meta.get("t_us")):
            raise Refused("native_input_invalid")
        for k in ["cmsd_static", "cmsd_dynamic"]:
            if meta.get(k) is not None and not isinstance(meta[k], str):
                raise Refused("native_input_invalid")
        if meta.get("media") is not None and not isinstance(meta["media"], bool):
            raise Refused("native_input_invalid")
        status = 200 if meta.get("status") is None else meta["status"]
        if not is_uint(status):
            raise Refused("native_input_invalid")
        url, at = meta["url"], meta["t_us"]
        k = resolve(url, url)
        s256 = sha(b)
        listed = self.listed.get(k)
        given = None if "playlist-body" in DROP else self.given.get((k, B3.blake3(b)))
        media = meta["media"] if meta.get("media") is not None else bool(listed and listed["media"])
        listing = given[0] if given else ("signed" if listed and listed["signed"] else "listed" if listed else "unlisted")
        v = {"sha256": s256.hex(), "cid": None, "listing": listing, "bound": "none", "media": media, "key": False, "acknowledged": True,
             "refused": given[1] if given else None}
        if k in self.keys and "keys" not in DROP:
            v["key"], v["acknowledged"] = True, False
            v["present"] = True
            return v
        stated = (listed or {}).get("cid") or stated_cid(url)
        if stated is not None and "cid-name" not in DROP:
            v["bound"] = "cid"
            h = content_hash(b)
            v["cid"] = h.hex()
            if h != stated and v["refused"] is None:
                v["refused"] = "object_digest_mismatch"
        elif stated_sha256(k) is not None and (self.cfg["content_addressed"] or listing == "signed") and "sha256-name" not in DROP:
            v["bound"] = "sha256"
            if stated_sha256(k) != s256 and v["refused"] is None:
                v["refused"] = "object_digest_mismatch"
        if v["refused"] is None and self.cfg["require"] == "signed" and given is None and listing != "signed" and "require-signed" not in DROP:
            v["refused"] = "object_unsigned"
        origin = origin_of(k)
        if origin is None:
            raise Refused("native_input_invalid")
        self.received.append({"at": at, "origin": origin, "media": media, "len": len(b), "hash": s256})
        self.log_push("response", at, s256.hex(), status)
        if meta.get("cmsd_static") is not None or meta.get("cmsd_dynamic") is not None:
            self.notes.append({"content_hash": s256.hex(), "cmsd_static": meta.get("cmsd_static"), "cmsd_dynamic": meta.get("cmsd_dynamic")})
        self.by_url[k] = s256
        v["present"] = v["refused"] is None
        return v

    # §5.1, §6.1
    def ad_at(self, program):
        if program is None:
            return None
        for i, x in enumerate(self.ads):
            if x["ad"]["start"] <= program < x["ad"]["start"] + x["ad"]["duration"]:
                return i
        return None

    def ad_segments(self, ad):
        segs = [s for s in self.segments.values() if s["program"] is not None and s["program"] < ad["start"] + ad["duration"] and s["program"] + s["duration"] > ad["start"]]
        return sorted(segs, key=lambda s: s["program"])

    def observe(self, o):
        self.live()
        members = ["t_us", "playing", "visible_bp", "audible", "program_ms", "seeked", "interaction"]
        if not isinstance(o, dict) or any(k not in members for k in o) or not is_uint(o.get("t_us")):
            raise Refused("native_input_invalid")
        vis = o.get("visible_bp")
        if isinstance(vis, bool) or not isinstance(vis, (int, float)):
            raise Refused("native_input_invalid")
        flags = {}
        for k in ["playing", "audible", "seeked", "interaction"]:
            x = o.get(k)
            if x is not None and not isinstance(x, bool):
                raise Refused("native_input_invalid")
            flags[k] = bool(x)
        prog = o.get("program_ms")
        if prog is not None and (isinstance(prog, bool) or not isinstance(prog, int)):
            raise Refused("native_input_invalid")
        at, playing, audible, seeked = o["t_us"], flags["playing"], flags["audible"], flags["seeked"]
        vis = int(min(10000, max(0, (vis + 0.5) // 1)))
        if playing and not self.has(["present"]) and "log-present" not in DROP:
            self.log_event("present", at)
        ad = self.ad_at(prog)
        if "spans" not in DROP:
            self.spans.observe(at, playing, vis, audible, ad is not None)
        last = self.last
        for i, x in enumerate(self.ads):
            segs = self.ad_segments(x["ad"])
            base = segs[0]["program"] if segs else x["ad"]["start"]
            rec = x["rec"]
            if i == ad:
                if rec.cur is None and not rec.done and playing and not seeked and last and last[0] and last[1] is not None and last[1] < base:
                    rec.observe(0, True, vis, audible, False)
                rec.observe(prog - base, playing, vis, audible, seeked)
            elif rec.cur is not None:
                if playing and not seeked and prog is not None:
                    d = sum(s["duration"] for s in segs) if segs else x["ad"]["duration"]
                    rec.observe(min(prog - base, d), True, rec.cur["v"], rec.cur["a"], False)
                rec.observe(0, False, 0, False, False)
        self.last = (playing, prog)
        if flags["interaction"]:
            self.last_interaction = at
        if playing and vis == 0:
            if self.unseen_since is None:
                self.unseen_since = at
        elif vis > 0:
            self.unseen_since = None
        g = self.cfg["unseen_grace_ms"]
        if g > 0 and self.unseen_since is not None and at - self.unseen_since >= g * 1000 and "rule-hidden" not in DROP:
            return {"end": "hidden"}
        idle = self.cfg["idle_ms"]
        since = self.last_interaction if self.last_interaction is not None else self.start
        if idle > 0 and at - since >= idle * 1000 and "rule-idle" not in DROP:
            return {"end": "idle"}
        return {"end": None}

    # §7
    def energy(self, e):
        self.live()
        role = e.get("role")
        if not is_uint(role) or not 1 <= role <= 4:
            raise Refused("native_input_invalid")
        if e.get("absent") is True:
            self.roles.pop(role, None)
            return
        j = e.get("joules_micro")
        label = e.get("energy_provenance")
        if not is_uint(j) or label not in (None, "OnChipCounter", "CalibratedInstrument", "ModelBased", "Estimator"):
            raise Refused("native_input_invalid")
        if "energy-cohere" not in DROP and (label is None) != (j == 0):
            raise Refused("native_input_invalid")
        self.roles[role] = (j, None if label is None else e)

    def attach(self, name, b):
        self.live()
        if not name:
            raise Refused("native_input_invalid")
        self.attachments.append((name, b))

    def take_cmcd(self):
        self.live()
        fresh = self.log[self.cmcd_taken:]
        self.cmcd_taken = len(self.log)
        return "\n".join(r for r in (cmcd_report(o) for o in fresh) if r is not None)

    # stream-audit §4.2: what each acknowledgement holds
    def windows(self, end):
        w = self.cfg["window_us"]
        media_class = 1 if (self.presentation or LIVE) == LIVE else 2
        wins, seen, rep = {}, set(), [0, 0]
        for r in self.received:
            kk = (r["at"] - self.start) // w
            if (r["origin"], kk, r["hash"]) in seen:
                rep[0] += 1
                rep[1] += r["len"]
                continue
            seen.add((r["origin"], kk, r["hash"]))
            c = media_class if r["media"] else 4
            wins.setdefault((r["origin"], kk), {}).setdefault(c, []).append(r)
        out = []
        for (origin, kk) in sorted(wins):
            ws = self.start + kk * w
            out.append({"origin": origin, "start": ws, "end": min(ws + w, end), "classes": sorted(wins[(origin, kk)].items())})
        return out, rep

    def end(self, name, at, bundle_text):
        """The bundle the corpus states is checked against what this session holds."""
        self.live()
        if name not in ENDED or at <= self.start:
            raise Refused("native_input_invalid")
        ended = ENDED[name]
        if not self.has(["end", "abandon"]) and "log-end" not in DROP:
            self.log_event("end" if ended == 1 else "abandon", at)
        self.ended = ended
        spans = [] if "spans" in DROP else (self.spans.close(at) or self.spans.done)
        try:
            b = json.loads(bundle_text)
        except ValueError:
            raise Disagree("the bundle is not JSON")
        if set(b) != {"v", "session", "acks", "claims", "log", "notes", "attachments"}:
            raise Disagree("the bundle's members")
        log = json.dumps([self.log], separators=(",", ":"), ensure_ascii=False)
        notes = json.dumps(self.notes, separators=(",", ":"), ensure_ascii=False) if self.notes else None
        if b["log"] != log:
            raise Disagree("the log")
        if b["notes"] != notes:
            raise Disagree("the notes")
        # The acknowledgements: one per origin and window, chained, in order.
        wins, rep = self.windows(at)
        if len(b["acks"]) != len(wins):
            raise Disagree("the number of acknowledgements")
        prev, ack_digests = {}, []
        for a, w in zip(b["acks"], wins):
            raw = unhex(a["record"])
            r = SV.read_ack(raw)
            want_items = [b"".join(x["hash"] + struct.pack("<QQ", 0, x["len"]) for x in items) for _, items in w["classes"]]
            if [unhex(x) for x in a["items"]] != want_items:
                raise Disagree("an acknowledgement's items")
            if (r["mode"], r["role"], r["upstream"], r["downstream"], r["start"], r["end"], r["prev"]) != (1, 2, sha(ORIGIN_DOMAIN, w["origin"].encode()), self.player, w["start"], w["end"], prev.get(w["origin"], ZERO)):
                raise Disagree("an acknowledgement's fields")
            ents = [(e["class"], e["grant"], e["proof"], e["objects"], e["bytes"], e["items"]) for e in r["entries"]]
            want = [(c, ZERO, ZERO, len(items), sum(x["len"] for x in items), sha(ITEMS_DOMAIN, il)) for (c, items), il in zip(w["classes"], want_items)]
            if ents != want:
                raise Disagree("an acknowledgement's entries")
            d = sha(raw[:-64])
            prev[w["origin"]] = d
            ack_digests.append(d)
        # The claims.
        want_claims = []
        for x in self.ads:
            x["rec"].close()
            sp = x["rec"].done
            segs = self.ad_segments(x["ad"])
            if not sp or not segs or any(s["url"] not in self.by_url for s in segs):
                continue
            creative = [(self.by_url[s["url"]], min(s["duration"], 0xFFFFFFFF)) for s in segs]
            cd = sha(CREATIVE_DOMAIN, struct.pack("<I", len(creative)), *(h + struct.pack("<I", d) for h, d in creative))
            touched, t0 = [], 0
            for h, d in creative:
                if any(s["start"] < t0 + d and s["start"] + s["length"] > t0 for s in sp):
                    touched.append(h)
                t0 += d
            want_claims.append({"marker": x["ad"]["marker"], "creative": creative, "cd": cd, "objects": touched, "spans": sp})
        if len(b["claims"]) != len(want_claims):
            raise Disagree("the number of claims")
        claim_digests = []
        for c, w in zip(b["claims"], want_claims):
            r = APV.read_claim(unhex(c["claim"]))
            got = (r["room"], r["cue"], r["session"], r["slot"], r["creative_hash"], r["player"], r["geometry"], r["frames"], r["objects"], r["spans"], r["attestation"])
            if got != (self.channel, sha(MARKER_DOMAIN, w["marker"].encode()), self.session, 0, w["cd"], self.player, 3, NOT_EXPOSED, w["objects"], w["spans"], b""):
                raise Disagree("a claim's fields")
            if c["marker"] != w["marker"] or [(unhex(o["content_hash"]), o["duration_ms"]) for o in c["creative"]] != w["creative"]:
                raise Disagree("a claim's creative")
            claim_digests.append(sha(r["body"]))
        # The attachments.
        if [(a["name"], unhex(a["record"])) for a in b["attachments"]] != self.attachments:
            raise Disagree("the attachments")
        # The record.
        s = SAV.read_session(unhex(b["session"]))
        origins = sorted({r["origin"] for r in self.received})
        got_energy = [(e["role"], e["joules"], e["class"]) for e in s["energy"]]
        want_energy = [(role, j, None if e is None else class_tuple(e)) for role, (j, e) in sorted(self.roles.items())]
        if got_energy != want_energy:
            raise Disagree("the record's energy")
        fields = (s["channel"], s["owner"], s["player"], s["session"], s["presentation"], s["start"], s["end"], s["ended"],
                  [o.decode() for o in s["origins"]], s["acks"], s["claims"], s["spans"], s["repeat_objects"], s["repeat_bytes"],
                  s["log"], s["notes"], s["attachments"])
        want = (self.channel, self.cfg["owner"], self.player, self.session, self.presentation or LIVE, self.start, at, ended,
                origins, sorted(ack_digests), sorted(claim_digests), spans, rep[0], rep[1],
                sha(log.encode()), sha(notes.encode()) if notes else ZERO, sorted(sha(x) for _, x in self.attachments))
        for i, (g, wv) in enumerate(zip(fields, want)):
            if g != wv:
                raise Disagree(f"the record's field {i}")
        return {"bundle": bundle_text}


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

def run_call(sink, objects, d, answer):
    body = lambda k: None if d.get(k) is None else unhex(objects[d[k]])  # noqa: E731
    try:
        c = d["call"]
        if c == "playlist":
            return sink.playlist(d["url"], body("body"), body("receipt"))
        if c == "parent":
            sink.parent(d["url"], d["name"], body("body"))
            return {"ok": True}
        if c == "object":
            return sink.object(d["meta"], body("body"))
        if c == "event":
            sink.event(d["name"], d["t_us"])
            return {"ok": True}
        if c == "observe":
            return sink.observe(d["obs"])
        if c == "energy":
            sink.energy(d["entry"])
            return {"ok": True}
        if c == "attach":
            sink.attach(d["name"], body("body"))
            return {"ok": True}
        if c == "cmcd":
            return {"cmcd": sink.take_cmcd()}
        if c == "end":
            if not isinstance(answer, dict) or "bundle" not in answer:
                sink.live()
                raise Disagree("an end the corpus states no bundle for")
            return sink.end(d["ended"], d["t_us"], answer["bundle"])
        raise Disagree(f"an unknown call {c}")
    except Refused as e:
        return {"error": e.code}


def check(doc, quiet=False):
    fails = []
    objects = doc["objects"]
    for c in doc["configs"]:
        try:
            read_config(c["config"])
            got = None
        except Refused as e:
            got = e.code
        if got != c["refused"]:
            fails.append(f"config {c['note']}: {got}")
    for case in doc["cases"]:
        name = case["name"]
        try:
            cfg = read_config(case["config"])
            sink = Sink(cfg, unhex(case["seed"]), case["start_us"])
            bundle = None
            for i, call in enumerate(case["calls"]):
                got = run_call(sink, objects, call["do"], call["answer"])
                if got != call["answer"]:
                    raise Disagree(f"call {i}: {json.dumps(got)[:300]} where the corpus states {json.dumps(call['answer'])[:300]}")
                if bundle is None and isinstance(got, dict) and "bundle" in got:
                    bundle = got["bundle"]
            if bundle is None:
                raise Disagree("no bundle")
            out = SAV.verify_bundle(bundle.encode(), unhex(doc["owner"]), doc["channels"])
            if out != case["outcome"]:
                raise Disagree(f"the collector's outcome {json.dumps(out)[:300]}")
            if not quiet:
                print(f"PASS {name}")
        except Exception as e:  # a verifier reports every failure as a failure, never a crash
            fails.append(f"{name}: {e!r}"[:500])
    return fails


def reasons_of_extension():
    text = open(EXTENSION, encoding="utf-8").read()
    sec = text.split("## 8. Reasons", 1)[1].split("\n## ", 1)[0]
    return {m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", sec, re.M)}


def reasons_given(doc):
    out = {c["refused"] for c in doc["configs"]}
    for case in doc["cases"]:
        for call in case["calls"]:
            a = call["answer"]
            for k in ("refused", "error"):
                if isinstance(a, dict) and isinstance(a.get(k), str):
                    out.add(a[k])
    return out


def flip_hex(h, at):
    i = at % len(h)
    return h[:i] + ("0" if h[i] != "0" else "1") + h[i + 1:]


def find(doc, name):
    return next(c for c in doc["cases"] if c["name"] == name)


def mutations(doc):
    """Copies of the corpus with one defect each; every one must fail."""
    def m(fn):
        d = copy.deepcopy(doc)
        fn(d)
        return d
    out = []
    out.append(("a fragment's bytes changed", m(lambda d: d["objects"].__setitem__("f2", flip_hex(d["objects"]["f2"], 40)))))
    out.append(("a receipt's bytes changed", m(lambda d: d["objects"].__setitem__("vod-receipt", d["objects"]["vod-receipt"].replace("7075626c6963", "70726976617a", 1)))))
    out.append(("a verdict says present for a refused fragment", m(lambda d: next(c for c in find(d, "tampered-fragment")["calls"] if c["answer"].get("refused"))["answer"].__setitem__("present", True))))
    out.append(("a refusal's reason renamed", m(lambda d: next(c for c in find(d, "unsigned-refused")["calls"] if c["answer"].get("refused"))["answer"].__setitem__("refused", "playlist_receipt_invalid"))))
    out.append(("content addressing turned off", m(lambda d: find(d, "content-addressed")["config"].__setitem__("content_addressed", False))))
    out.append(("a session's seed changed", m(lambda d: find(d, "signed-vod").__setitem__("seed", flip_hex(find(d, "signed-vod")["seed"], 3)))))
    out.append(("two calls swapped", m(lambda d: find(d, "wai-objects")["calls"].insert(1, find(d, "wai-objects")["calls"].pop(0)))))
    out.append(("the hidden rule's grace lengthened", m(lambda d: find(d, "hidden")["config"].__setitem__("unseen_grace_ms", 5000))))
    out.append(("an observation's end dropped", m(lambda d: next(c for c in find(d, "idle")["calls"] if c["answer"].get("end"))["answer"].__setitem__("end", None))))
    def bundle_edit(d, case, fn):
        a = next(c for c in find(d, case)["calls"] if "bundle" in c["answer"])["answer"]
        b = json.loads(a["bundle"])
        fn(b)
        a["bundle"] = json.dumps(b, separators=(",", ":"))
    out.append(("a bundle's record changed", m(lambda d: bundle_edit(d, "signed-vod", lambda b: b.__setitem__("session", flip_hex(b["session"], 300))))))
    out.append(("a bundle's item list changed", m(lambda d: bundle_edit(d, "page-parity", lambda b: b["acks"][0]["items"].__setitem__(0, flip_hex(b["acks"][0]["items"][0], 5))))))
    out.append(("a bundle's log changed", m(lambda d: bundle_edit(d, "hidden", lambda b: b.__setitem__("log", b["log"].replace('"abandon"', '"end"'))))))
    out.append(("an outcome's time presented changed", m(lambda d: find(d, "signed-vod")["outcome"].__setitem__("presented_ms", 1))))
    out.append(("a key acknowledged", m(lambda d: next(c for c in find(d, "third-party-ad")["calls"] if c["answer"].get("key"))["answer"].__setitem__("acknowledged", True))))
    out.append(("an energy entry with no class beside a figure accepted", m(lambda d: next(c for c in find(d, "energy")["calls"] if c["answer"].get("error"))["answer"].update({"ok": True, "error": None}))))
    out.append(("a parent named under another hash", m(lambda d: next(c for c in find(d, "live-chain")["calls"] if c["do"]["call"] == "parent")["do"].__setitem__("name", "mr/" + "0" * 64 + ".json"))))
    out.append(("a configuration's refusal dropped", m(lambda d: d["configs"][0].__setitem__("refused", None))))
    return out


DROPPABLE = ["config-members", "window-floor", "receipts", "chain", "require-signed", "playlist-body", "keys", "cid-name", "sha256-name",
             "rule-hidden", "rule-idle", "energy-cohere", "after-end", "spans", "log-start", "log-present", "log-end", "log-order", "log-closed"]


def main(corpus_dir):
    doc = json.load(open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8"))
    status = 0
    files = sorted(os.listdir(corpus_dir))
    if files != ["README.md", "vectors.json"]:
        print(f"FAIL the corpus holds other files: {files}")
        status = 1
    fails = check(doc)
    for f in fails:
        print(f"FAIL {f}")
    status |= bool(fails)
    want, given = reasons_of_extension(), reasons_given(doc)
    if want != given:
        print(f"FAIL §8 lists {sorted(want - given)} that no vector reaches, and the corpus reaches {sorted(given - want)} that §8 does not list")
        status = 1
    else:
        print(f"PASS every reason of §8 refuses a vector ({len(want)})")
    caught = 0
    for name, d in mutations(doc):
        if check(d, quiet=True):
            caught += 1
        else:
            print(f"FAIL a mutated corpus passed: {name}")
            status = 1
    print(f"{'PASS' if caught == len(mutations(doc)) else 'FAIL'} every mutated copy of the corpus fails ({caught})")
    weak = 0
    for k in DROPPABLE:
        DROP.clear()
        DROP.add(k)
        if check(doc, quiet=True):
            weak += 1
        else:
            print(f"FAIL the corpus does not catch the verifier with its check `{k}` dropped")
            status = 1
    DROP.clear()
    print(f"{'PASS' if weak == len(DROPPABLE) else 'FAIL'} every weakened verifier fails on the corpus ({weak})")
    return status


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