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

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

- the co-watch policy (§3): its layout, rules, signature and chain, and the object-sequence
  source digest (§3.5);
- the statements (§4): reading, who signs, the control attribute (§4.4);
- controls (§5): who may control, the lead and the horizon, the range, and the order of
  application with the overridden ones;
- the controlled position (§6);
- presentation statements (§7);
- windows, due slots, samples, settling, percentiles, verdicts, the skew statement and
  completeness (§8);
- the catch-up rule (§9).

For every room it requires the outcome, reason and place, and everything else the corpus
states (§11.3); for every catch-up case, its correction; of the corpus, that every reason of
§12 appears in it and that no other file is there. Then it mutates copies of the corpus, one
defect each, and requires every copy to fail; and it drops each of its own checks in turn and
requires the corpus to catch the weakened verifier.

    python3 co_watch_verify.py <live-conformance/cowatch>
    python3 co_watch_verify.py --room <room.json>

With `--room`, it verifies one room given in a vector's form (descriptor, grants, chains,
policies, sources) and prints what §11.3 states of it, as JSON: a room a deployment kept,
checked from the text alone.

Exit status 0 only if every case verifies, every mutated copy fails and every weakened
verifier is caught. 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"))
import live_events_verify as LEV  # noqa: E402
import live_rooms_verify as LRV  # noqa: E402
import _ed25519 as ED  # noqa: E402

Refuse = LEV.Refuse

# Ed25519 is a pure function of its inputs: each signature is verified once, and every later
# pass over the corpus (its mutated copies, the weakened verifiers) reads the answer back.
_verified = {}
_verify = ED.verify


def _memo_verify(pk, msg, sig, skip=()):
    k = (bytes(pk), bytes(msg), bytes(sig), tuple(skip))
    if k not in _verified:
        _verified[k] = _verify(pk, msg, sig, skip)
    return _verified[k]


ED.verify = _memo_verify
EXTENSION = os.path.join(HERE, "..", "extensions", "co-watch.md")

POLICY_MAGIC, CONTROL_MAGIC, PRESENTATION_MAGIC, SKEW_MAGIC = b"WXP1", b"WXC1", b"WXS1", b"WXK1"
POLICY_DOMAIN = b"wai:co-watch-policy\x01"                 # §3.2
PRESENTATION_DOMAIN = b"wai:co-watch-presentation\x01"     # §7.1
SOURCE_DOMAIN = b"wai:co-watch-source\x01"                 # §3.5
ZERO = bytes(32)
NORMAL = 1000000
META, JOIN, LEAVE, EXCHANGE = 0x01, 0x10, 0x11, 0x14
DIRECT = 2                                                 # live-rooms §3.3, bit 1
PLAY, PAUSE, SEEK, RATE = 1, 2, 3, 4
ACTIONS = {1: "play", 2: "pause", 3: "seek", 4: "rate"}
STATES = {1: "playing", 2: "paused", 3: "stalled", 4: "ended"}
VERDICTS = {1: "in_sync", 2: "out_of_sync", 3: "unknown"}
IN_SYNC, OUT_OF_SYNC, UNKNOWN = 1, 2, 3
U32 = (1 << 32) - 1

# Each check this verifier makes, by name; the self-test drops one at a time (§13).
DROP = set()


def on(name):
    return name not in DROP


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


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


def reasons_of_extension():
    """§12's table: the codes this extension adds, in order."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    sec = text[text.index("\n## 12. Reasons"):]
    sec = sec[:sec.index("\n## ", 1)]
    return re.findall(r"^\| `([a-z_0-9]+)` \|", sec, re.M)


REASONS = reasons_of_extension()


class Rd:
    def __init__(self, b, p=0):
        self.b, self.p = b, p

    def take(self, n, why):
        if self.p + n > len(self.b):
            raise Refuse(why)
        s = self.b[self.p:self.p + n]
        self.p += n
        return s

    def u8(self, why):
        return self.take(1, why)[0]

    def u16(self, why):
        return struct.unpack("<H", self.take(2, why))[0]

    def u32(self, why):
        return struct.unpack("<I", self.take(4, why))[0]

    def u64(self, why):
        return struct.unpack("<Q", self.take(8, why))[0]


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

def read_policy(b):
    """§3.3: steps 1 to 3, then the rules of §3.1's table."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != POLICY_MAGIC:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] != 1:
        raise Refuse("unknown_version")
    r, t = Rd(b, 5), "truncated"
    p = {"room": r.take(32, t), "prev": r.take(32, t), "effective": r.u64(t), "source_type": r.u8(t),
         "source": r.take(32, t), "duration_us": r.u64(t)}
    for k in ("lead_ms", "horizon_ms", "min_rate", "max_rate", "period_ms", "window_ms", "settle_ms", "bound_us"):
        p[k] = r.u32(t)
    p["quantile"] = r.u8(t)
    for k in ("tolerance_us", "nudge", "seek_us"):
        p[k] = r.u32(t)
    p["body"] = b[:r.p]
    p["sig"] = r.take(64, t)
    if r.p != len(b):
        raise Refuse("malformed")
    if on("policy_rules"):
        ok = (p["source_type"] in (1, 2, 3) and p["duration_us"] > 0 and p["lead_ms"] > 0 and p["period_ms"] > 0
              and p["bound_us"] > 0 and p["nudge"] > 0 and p["horizon_ms"] >= p["lead_ms"]
              and 0 < p["min_rate"] <= NORMAL <= p["max_rate"]
              and p["window_ms"] % p["period_ms"] == 0 and 1 <= p["window_ms"] // p["period_ms"] <= 1000
              and 1 <= p["quantile"] <= 100 and p["tolerance_us"] <= p["bound_us"] and p["seek_us"] > p["bound_us"])
        if not ok:
            raise Refuse("cw_policy_malformed")
    p["digest"] = sha(p["body"])
    return p


def source_digest(objects):
    """§3.5: an object sequence's digest."""
    out = SOURCE_DOMAIN + struct.pack("<I", len(objects))
    for o in objects:
        out += bytes.fromhex(o["content_hash"]) + struct.pack("<I", o["duration_ms"])
    return sha(out)


def check_policies(d, blobs, sources):
    """§3.3, §3.4, then §3.5 for the sources given: the policies, or Stop at ("policy", i)."""
    held = []
    for i, b in enumerate(blobs):
        try:
            p = read_policy(b)
            if p["room"] != d["room"]:
                raise Refuse("wrong_room")
            bad = ED.verify(d["authority"], POLICY_DOMAIN + p["body"], p["sig"])
            if bad:
                raise Refuse(bad)
            if on("policy_link"):
                if not held:
                    linked = p["prev"] == ZERO and p["effective"] == 0
                else:
                    q = held[-1]
                    same = (p["source_type"], p["source"]) == (q["source_type"], q["source"])
                    linked = p["prev"] == q["digest"] and p["effective"] > q["effective"] and (not same or p["duration_us"] == q["duration_us"])
                if not linked:
                    raise Refuse("cw_policy_unlinked")
            objs = sources[i] if i < len(sources) else None
            if objs is not None and on("sources"):
                total = sum(o["duration_ms"] for o in objs) * 1000
                if p["source_type"] != 2 or source_digest(objs) != p["source"] or total != p["duration_us"]:
                    raise Refuse("cw_source_mismatch")
        except Refuse as e:
            raise Stop(e.reason, ["policy", i])
        held.append(p)
    return held


def policy_at(ps, t):
    """§3.4: the index of the last policy whose effective is at or below t, or None."""
    k = None
    for i, p in enumerate(ps):
        if p["effective"] <= t:
            k = i
    return k


def starts_of(ps):
    """§3.4, §6.4: (instant, policy index) of each source's start."""
    out = []
    for i, p in enumerate(ps):
        if i == 0 or (p["source_type"], p["source"]) != (ps[i - 1]["source_type"], ps[i - 1]["source"]):
            out.append((p["effective"], i))
    return out


# ---------------------------------------------------------------- §4 statements

def control_bits(g):
    """§4.4: the actions `watch_control` allows, None without it; Refuse when malformed."""
    for k, v in LRV.read_attrs(g["attrs"]) or []:
        if k == "watch_control":
            if not re.fullmatch(r"[1-9][0-9]?", v) or int(v) > 15:
                raise Refuse("cw_attr_malformed")
            return int(v)
    return None


def may_control(g, action):
    """§5.2."""
    if g["powers"] & DIRECT:
        return True
    try:
        bits = control_bits(g)
    except Refuse:
        return False
    return bits is not None and bits >> (action - 1) & 1 == 1


def read_statement(payload):
    """§4.1, §4.2: None for a payload not this extension's; else (kind, fields)."""
    if len(payload) < 4 or payload[:4] not in (CONTROL_MAGIC, PRESENTATION_MAGIC, SKEW_MAGIC):
        return None
    bad = "cw_statement_malformed"
    if len(payload) < 5 or payload[4] != 1:
        raise Refuse(bad)
    r = Rd(payload, 5)
    magic = payload[:4]
    if magic == CONTROL_MAGIC:                                                       # §5.1
        s = ("control", {"action": r.u8(bad), "at": r.u64(bad), "arg": r.u64(bad)})
        a, arg = s[1]["action"], s[1]["arg"]
        if a in (PLAY, PAUSE):
            ok = arg == 0
        elif a == SEEK:
            ok = True
        elif a == RATE:
            ok = 1 <= arg <= U32
        else:
            ok = False
        if not ok:
            raise Refuse(bad)
    elif magic == PRESENTATION_MAGIC:                                                # §7.1
        f = {"room": r.take(32, bad), "player": r.take(32, bad), "source": r.take(32, bad), "at": r.u64(bad),
             "position": r.u64(bad), "state": r.u8(bad), "rate": r.u32(bad)}
        f["body"] = payload[:r.p]
        f["sig"] = r.take(64, bad)
        if f["state"] not in STATES or f["rate"] == 0:
            raise Refuse(bad)
        s = ("presentation", f)
    else:                                                                            # §8.4
        f = {"from": r.u64(bad), "to": r.u64(bad), "bound_us": r.u32(bad), "quantile": r.u8(bad)}
        nv = r.u8(bad)
        if not 1 <= nv <= 8:
            raise Refuse(bad)
        f["viewers"] = []
        for _ in range(nv):
            e = {"principal": r.take(32, bad), "held": r.u64(bad), "due": r.u16(bad), "samples": r.u16(bad),
                 "counted": r.u16(bad), "p50": r.u32(bad), "pq": r.u32(bad), "max": r.u32(bad), "verdict": r.u8(bad)}
            if e["verdict"] not in VERDICTS:
                raise Refuse(bad)
            f["viewers"].append(e)
        keys = [e["principal"] for e in f["viewers"]]
        if any(keys[i] >= keys[i + 1] for i in range(len(keys) - 1)):
            raise Refuse(bad)
        want = [(a, b) for a in range(nv) for b in range(a + 1, nv)]
        if r.u8(bad) != len(want):
            raise Refuse(bad)
        f["pairs"] = []
        for a, b in want:
            e = {"a": r.u8(bad), "b": r.u8(bad), "n": r.u16(bad), "p50": r.u32(bad), "pq": r.u32(bad),
                 "max": r.u32(bad), "verdict": r.u8(bad)}
            if (e["a"], e["b"]) != (a, b) or e["verdict"] not in VERDICTS:
                raise Refuse(bad)
            f["pairs"].append(e)
        s = ("skew", f)
    if r.p != len(payload):
        raise Refuse(bad)
    return s


# ---------------------------------------------------------------- §5.4, §6 the controlled position

def position(st, x, duration):
    """§6.2."""
    if not st["playing"]:
        return st["pos"]
    return min(duration, st["pos"] + (x - st["anchor"]) * st["rate"] // 1000000)


class Timeline:
    """§5.4 and §6 over the accepted controls (each with its event's t and digest)."""

    def __init__(self, policies, controls):
        self.policies = policies
        self.starts = starts_of(policies)
        self.controls = sorted(controls, key=lambda c: (c["at"], c["t"], c["digest"]))
        self.overridden = set()
        aspect = {PLAY: 0, PAUSE: 0, SEEK: 1, RATE: 2}
        for i, c in enumerate(self.controls):
            for d in self.controls[i + 1:]:
                if d["at"] != c["at"]:
                    break
                if aspect[d["action"]] == aspect[c["action"]]:
                    self.overridden.add(c["digest"])
        events = [(s, 0, k, None) for k, (s, _) in enumerate(self.starts)]
        events += [(c["at"], 1, k, c) for k, c in enumerate(self.controls)]
        events.sort(key=lambda e: (e[0], e[1], e[2]))
        self.points, st = [], None
        for x, what, _, c in events:
            if what == 0:
                st = {"playing": False, "rate": NORMAL, "anchor": x, "pos": 0}
            else:
                p = position(st, x, self.duration_at(x))
                st = dict(st, anchor=x, pos=p)
                if c["action"] == PLAY:
                    st["playing"] = True
                elif c["action"] == PAUSE:
                    st["playing"] = False
                elif c["action"] == SEEK:
                    st["pos"] = c["arg"]
                else:
                    st["rate"] = c["arg"]
            self.points.append((x, st))

    def duration_at(self, x):
        k = policy_at(self.policies, x)
        return 0 if k is None else self.policies[k]["duration_us"]

    def state_at(self, x):
        st = None
        for at, s in self.points:
            if at <= x:
                st = s
        return st

    def position_at(self, x):
        st = self.state_at(x)
        return 0 if st is None else position(st, x, self.duration_at(x))

    def settle_points(self):
        return sorted({c["at"] for c in self.controls} | {s for s, _ in self.starts})


# ---------------------------------------------------------------- §9 catch-up

def catch_up(p, playing, rate, controlled, duration, presented):
    """§9."""
    e = presented - controlled
    if not playing or controlled >= duration:
        return {"action": "seek", "value": controlled} if abs(e) > p["tolerance_us"] else {"action": "hold"}
    if abs(e) <= p["tolerance_us"]:
        return {"action": "rate", "value": rate}
    if abs(e) > p["seek_us"]:
        return {"action": "seek", "value": controlled}
    r = rate - p["nudge"] if e > 0 else rate + p["nudge"]
    return {"action": "rate", "value": max(p["min_rate"], min(p["max_rate"], r))}


# ---------------------------------------------------------------- §8 skew

def window_policy(ps, frm, to):
    """§8.1: the policy of the window frm–to, or None when it is not one."""
    if frm >= to:
        return None
    k = policy_at(ps, frm)
    if k is None or policy_at(ps, to - 1) != k:
        return None
    per = ps[k]["period_ms"] * 1000
    n = (to - frm) // per
    return ps[k] if (to - frm) % per == 0 and 1 <= n <= 1000 else None


def percentile(vals, q):
    """§8.5: nearest rank; 0 for none."""
    if not vals:
        return 0
    v = sorted(vals)
    rank = -(-q * len(v) // 100)
    return v[max(rank, 1) - 1]


class Facts:
    """Presence (§8.2), grants, samples (§8.3) and the timeline."""

    def __init__(self, timeline, stretches, grant_changes, samples):
        self.tl, self.stretches, self.grants, self.samples = timeline, stretches, grant_changes, samples

    def subscribes_at(self, w, x):
        last = None
        for g in self.grants:
            if g["principal"] == w and g["t"] <= x:
                last = g
        return last is not None and last["subscribe"]

    def due(self, p, frm, to, w):
        per = p["period_ms"] * 1000
        out = []
        for j in range((to - frm) // per):
            s = frm + j * per
            inside = any(st["principal"] == w and st["join"] <= s and (st["leave"] is None or st["leave"] >= s + per) for st in self.stretches)
            out.append(inside and self.subscribes_at(w, s))
        return out

    def viewers(self, frm, to):
        p = window_policy(self.tl.policies, frm, to)
        if p is None:
            return []
        who = sorted({st["principal"] for st in self.stretches})
        return [w for w in who if any(self.due(p, frm, to, w))]

    def settling(self, p, w, at):
        span = p["settle_ms"] * 1000
        if not on("settle"):
            return False
        near = [x for x in self.tl.settle_points()] + [st["join"] for st in self.stretches if st["principal"] == w]
        return any(x <= at < x + span for x in near)

    def row(self, p, frm, to, w, held):
        due = self.due(p, frm, to, w)
        per = p["period_ms"] * 1000
        sample = [None] * len(due)
        for s in self.samples:
            if s["principal"] != w or s["seq"] >= held or not frm <= s["at"] < to:
                continue
            j = (s["at"] - frm) // per
            if not due[j] or sample[j] is not None:
                continue
            e = s["position"] - self.tl.position_at(s["at"])
            sample[j] = (e, not self.settling(p, w, s["at"]))
        return due, sample

    def entries(self, frm, to, listed):
        p = window_policy(self.tl.policies, frm, to)
        rows = [self.row(p, frm, to, w, held) for w, held in listed]
        bound = p["bound_us"]

        def judge(unknown, pq):
            return UNKNOWN if unknown else (IN_SYNC if pq <= bound else OUT_OF_SYNC)

        viewers = []
        for (w, held), (due, sample) in zip(listed, rows):
            nd = sum(1 for d in due if d)
            ns = sum(1 for s in sample if s is not None)
            errs = [abs(s[0]) for s in sample if s is not None and s[1]]
            p50, pq = percentile(errs, 50), percentile(errs, p["quantile"])
            viewers.append({"principal": w, "held": held, "due": nd, "samples": ns, "counted": len(errs),
                            "p50": min(p50, U32), "pq": min(pq, U32), "max": min(max(errs, default=0), U32),
                            "verdict": judge(ns < nd or not errs, pq)})
        pairs = []
        for a in range(len(listed)):
            for b in range(a + 1, len(listed)):
                d = [abs(x[0] - y[0]) for x, y in zip(rows[a][1], rows[b][1]) if x is not None and y is not None and x[1] and y[1]]
                p50, pq = percentile(d, 50), percentile(d, p["quantile"])
                unknown = viewers[a]["verdict"] == UNKNOWN or viewers[b]["verdict"] == UNKNOWN or not d
                pairs.append({"a": a, "b": b, "n": len(d), "p50": min(p50, U32), "pq": min(pq, U32),
                              "max": min(max(d, default=0), U32), "verdict": judge(unknown, pq)})
        return viewers, pairs


# ---------------------------------------------------------------- §11 verification

def viewer_json(v):
    return {"principal": v["principal"].hex(), "held": v["held"], "due": v["due"], "samples": v["samples"], "counted": v["counted"],
            "p50": v["p50"], "pq": v["pq"], "max": v["max"], "verdict": VERDICTS[v["verdict"]]}


def pair_json(p):
    return {"a": p["a"], "b": p["b"], "n": p["n"], "p50": p["p50"], "pq": p["pq"], "max": p["max"], "verdict": VERDICTS[p["verdict"]]}


def verify(descriptor, grants, chains, policy_blobs, sources):
    """§11.2: a dict of what the corpus states (§11.3)."""
    room = LRV.verify_room(descriptor, grants, None, chains)                              # step 1
    if room["outcome"] == "refuse":
        return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}
    gap = room["outcome"] == "gap"
    d = LRV.read_descriptor(descriptor)
    held_grants, _ = LRV.check_grants(d, grants)
    by_digest = {g["digest"]: g for g in held_grants}
    room_keys = (d["authority"], d["clock"])
    try:
        policies = check_policies(d, policy_blobs, sources)                              # step 2
        for i, g in enumerate(held_grants):                                              # step 3
            if on("attr"):
                try:
                    control_bits(g)
                except Refuse as e:
                    raise Stop(e.reason, ["grant", i])
        records, seen = [], set()
        for c, i, under in room["order"]:
            ev = LEV.read_record(chains[c][i])
            if ev["digest"] in seen:
                continue
            seen.add(ev["digest"])
            ev["chain"], ev["index"], ev["under"] = c, i, bytes.fromhex(under) if under else None
            records.append(ev)
        stretches, grant_changes = [], []                                                # §8.2
        for ev in records:
            if ev["principal"] not in room_keys or ev["kind"] not in (JOIN, LEAVE, EXCHANGE):
                continue
            pl = LEV.read_payload(ev["kind"], ev["payload"], 1 << 20)
            x = LRV.read_detail(ev["kind"], pl["detail"])
            who = pl["subject"]
            if ev["kind"] == JOIN:
                stretches.append({"principal": who, "join": ev["t"], "leave": None})
                grant_changes.append({"principal": who, "t": ev["t"], "subscribe": by_digest[x["grant"]]["subscribe"] == 1})
            elif ev["kind"] == LEAVE:
                for st in reversed(stretches):
                    if st["principal"] == who and st["leave"] is None:
                        st["leave"] = ev["t"]
                        break
            else:
                grant_changes.append({"principal": who, "t": ev["t"], "subscribe": by_digest[x["new"]]["subscribe"] == 1})
        controls, presentations, pending = [], [], []
        for ev in records:                                                               # step 4
            if ev["kind"] != META:
                continue
            at = ["record", ev["chain"], ev["index"]]
            try:
                s = read_statement(ev["payload"])
                if s is None:
                    continue
                kind, f = s
                t = ev["t"]
                k = policy_at(policies, t)
                if k is None:
                    raise Refuse("cw_no_policy")
                p = policies[k]
                is_key = ev["principal"] in room_keys
                if kind == "control":
                    if on("control_power") and not is_key and not may_control(by_digest[ev["under"]], f["action"]):
                        raise Refuse("control_not_permitted")
                    if on("lead") and f["at"] < t + p["lead_ms"] * 1000:
                        raise Refuse("control_not_ahead")
                    if on("horizon") and f["at"] > t + p["horizon_ms"] * 1000:
                        raise Refuse("control_beyond_horizon")
                    q = policies[policy_at(policies, f["at"])]
                    if on("range") and ((f["action"] == SEEK and f["arg"] > q["duration_us"])
                                        or (f["action"] == RATE and not q["min_rate"] <= f["arg"] <= q["max_rate"])):
                        raise Refuse("control_out_of_range")
                    controls.append(dict(f, t=t, digest=ev["digest"], author=ev["principal"]))
                elif kind == "presentation":
                    if on("who_presents") and is_key:
                        raise Refuse("cw_not_permitted")
                    if f["room"] != d["room"]:
                        raise Refuse("wrong_room")
                    if on("wrong_key") and f["player"] != ev["principal"]:
                        raise Refuse("presentation_wrong_key")
                    bad = ED.verify(f["player"], PRESENTATION_DOMAIN + f["body"], f["sig"]) if on("pres_sig") else None
                    if bad:
                        raise Refuse(bad)
                    if on("ahead") and f["at"] > t:
                        raise Refuse("presentation_ahead")
                    if on("source") and f["source"] != policies[policy_at(policies, f["at"])]["source"]:
                        raise Refuse("presentation_wrong_source")
                    presentations.append(dict(f, digest=ev["digest"], seq=ev["seq"]))
                else:
                    if on("who_skews") and not is_key:
                        raise Refuse("cw_not_permitted")
                    wp = window_policy(policies, f["from"], f["to"])
                    if wp is None or f["to"] > t:
                        raise Refuse("cw_statement_malformed")
                    if on("skew_bound") and (f["bound_us"], f["quantile"]) != (wp["bound_us"], wp["quantile"]):
                        raise Refuse("skew_mismatch")
                    pending.append((at, ev["digest"], f))
            except Refuse as e:
                raise Stop(e.reason, at)
        tl = Timeline(policies, controls)
        facts = Facts(tl, stretches, grant_changes,
                      [{"principal": p["player"], "seq": p["seq"], "at": p["at"], "position": p["position"]} for p in presentations])
        skews = []
        counts = {}
        for ev in records:
            counts[ev["principal"]] = counts.get(ev["principal"], 0) + 1
        for at, digest, f in pending:                                                    # §8.4 steps 2 and 3
            if gap:
                skews.append({"digest": digest.hex(), "from": f["from"], "to": f["to"], "viewers": "unchecked", "pairs": "unchecked"})
                continue
            vs = set(facts.viewers(f["from"], f["to"]))
            if on("skew_viewers") and any(v["principal"] not in vs or v["held"] > counts.get(v["principal"], 0) for v in f["viewers"]):
                raise Stop("skew_mismatch", at)
            viewers, pairs = facts.entries(f["from"], f["to"], [(v["principal"], v["held"]) for v in f["viewers"]])
            if on("skew_recompute") and (viewers != f["viewers"] or pairs != f["pairs"]):
                raise Stop("skew_mismatch", at)
            skews.append({"digest": digest.hex(), "from": f["from"], "to": f["to"],
                          "viewers": [viewer_json(v) for v in viewers], "pairs": [pair_json(p) for p in pairs]})
        if not gap and on("complete"):                                                   # step 5, §8.6
            windows = {}
            for at, _, f in pending:
                w = windows.setdefault((f["from"], f["to"]), [at, set()])
                w[1].update(v["principal"] for v in f["viewers"])
            for (frm, to), (at, listed) in windows.items():
                if any(w not in listed for w in facts.viewers(frm, to)):
                    raise Stop("skew_incomplete", at)
    except Stop as s:
        return {"outcome": "refuse", "reason": s.reason, "at": s.at}
    out = {"outcome": "gap" if gap else "accept", "room": d["room"].hex(), "gaps": room.get("gaps", []),
           "policy_digests": [p["digest"].hex() for p in policies],
           "starts": [{"at": s, "policy": i} for s, i in tl.starts],
           "controls": [{"digest": c["digest"].hex(), "author": c["author"].hex(), "t": c["t"], "action": ACTIONS[c["action"]],
                         "at": c["at"], "arg": c["arg"], "applied": c["digest"] not in tl.overridden} for c in controls],
           "timeline": [{"at": x, "playing": s["playing"], "rate": s["rate"], "position": position(s, x, tl.duration_at(x))} for x, s in tl.points],
           "presentations": [], "skews": skews}
    for p in presentations:
        controlled = tl.position_at(p["at"])
        pol = policies[policy_at(policies, p["at"])]
        out["presentations"].append({"digest": p["digest"].hex(), "player": p["player"].hex(), "at": p["at"], "position": p["position"],
                                     "state": STATES[p["state"]], "controlled": controlled, "error": p["position"] - controlled,
                                     "settling": facts.settling(pol, p["player"], p["at"])})
    return out


def run_vector(v):
    sources = []
    for s in v["sources"]:
        sources.append(None if s is None else s)
    return verify(bytes.fromhex(v["descriptor"]), [bytes.fromhex(g) for g in v["grants"]],
                  [[bytes.fromhex(r) for r in c] for c in v["chains"]], [bytes.fromhex(p) for p in v["policies"]], sources)


STATED = ("outcome", "reason", "at", "room", "gaps", "policy_digests", "starts", "controls", "timeline", "presentations", "skews")


def check(doc, quiet=False, first=False):
    """Every vector and catch-up case of `doc` against this verifier: the failures (with
    `first`, the first alone)."""
    fails, seen = [], set()
    for v in doc["vectors"]:
        if first and fails:
            return fails, seen
        try:
            got = run_vector(v)
        except Exception as e:  # a copy that does not read is a failure to verify it
            fails.append("%s: does not verify (%r)" % (v["name"], e))
            continue
        bad = [k for k in STATED if (k in got or k in v) and got.get(k) != v.get(k)]
        if got.get("reason"):
            seen.add(got["reason"])
        if not quiet:
            what = got["outcome"] + ("" if "reason" not in got else " " + got["reason"] + " at %s" % got["at"])
            print(("PASS" if not bad else "FAIL") + " room %s: %s" % (v["name"], what))
        for k in bad:
            fails.append("%s: %s is %s, the corpus states %s" % (v["name"], k, json.dumps(got.get(k))[:200], json.dumps(v.get(k))[:200]))
    cu = doc["catch_up"]
    for i, c in enumerate(cu["cases"]):
        got = catch_up(cu["params"], c["playing"], c["rate"], c["controlled"], c["duration"], c["presented"])
        ok = got == c["correction"]
        if not quiet:
            print(("PASS" if ok else "FAIL") + " catch-up %d: %s -> %s" % (i, c["about"], json.dumps(got)))
        if not ok:
            fails.append("catch-up case %d: %s, the corpus states %s" % (i, got, c["correction"]))
    return fails, seen


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier must catch."""
    vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
    party, drift, nxt, gapv = vn["watch_party"], vn["drift"], vn["next_episode"], vn["gap"]

    def mutate(label, f):
        d = copy.deepcopy(doc)
        f(d)
        return label, d

    def vs(d):
        return d["vectors"]

    def flip(h, i):
        b = bytearray(bytes.fromhex(h))
        b[i] ^= 1
        return b.hex()

    def drop_chain(d, k, principal):
        chains = vs(d)[k]["chains"]
        vs(d)[k]["chains"] = [c for c in chains if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() != principal]

    late = doc["principals"]["late"]
    return [
        mutate("a stated 95th percentile", lambda d: vs(d)[party]["skews"][2]["viewers"][0].update(pq=vs(d)[party]["skews"][2]["viewers"][0]["pq"] + 1)),
        mutate("a stated verdict", lambda d: vs(d)[drift]["skews"][0]["viewers"][2].update(verdict="in_sync")),
        mutate("a stated pair", lambda d: vs(d)[drift]["skews"][2]["pairs"][0].update(n=0)),
        mutate("a control stated applied", lambda d: vs(d)[party]["controls"][6].update(applied=True)),
        mutate("a stated error", lambda d: vs(d)[party]["presentations"][9].update(error=vs(d)[party]["presentations"][9]["error"] + 1)),
        mutate("a statement stated settling", lambda d: vs(d)[party]["presentations"][7].update(settling=not vs(d)[party]["presentations"][7]["settling"])),
        mutate("a stated position of the timeline", lambda d: vs(d)[party]["timeline"][3].update(position=1)),
        mutate("the second source's start unstated", lambda d: vs(d)[nxt]["starts"].pop()),
        mutate("a byte of a presentation statement's own signature", lambda d: vs(d)[party]["chains"][2].__setitem__(4, flip(vs(d)[party]["chains"][2][4], -70))),
        mutate("the late joiner's chain withheld", lambda d: drop_chain(d, party, late)),
        mutate("an object of the second source one millisecond longer", lambda d: vs(d)[nxt]["sources"][1][0].update(duration_ms=vs(d)[nxt]["sources"][1][0]["duration_ms"] + 1)),
        mutate("the gap's skew statements stated checked", lambda d: vs(d)[gapv]["skews"][0].update(viewers=[])),
        mutate("a refusal's reason", lambda d: vs(d)[vn["control_short_lead"]].update(reason="control_beyond_horizon")),
        mutate("a refusal's place", lambda d: vs(d)[vn["skew_leaves_a_viewer_out"]].update(at=["record", 0, 4])),
        mutate("a catch-up correction", lambda d: d["catch_up"]["cases"][3]["correction"].update(value=1040000)),
        mutate("the catch-up nudge", lambda d: d["catch_up"]["params"].update(nudge=40000)),
    ]


CHECKS = ("policy_rules", "policy_link", "sources", "attr", "control_power", "lead", "horizon", "range", "who_presents", "wrong_key",
          "pres_sig", "ahead", "source", "who_skews", "skew_bound", "skew_viewers", "skew_recompute", "complete", "settle")


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    fails += ["ed25519 self-test"] if st else []
    ok = len(REASONS) == 16
    print(("PASS" if ok else "FAIL") + " §12 read from the extension: %d reasons" % len(REASONS))
    fails += [] if ok else ["§12 does not read"]
    names = sorted(n for n in os.listdir(corpus_dir) if n != "README.md")
    ok = names == ["vectors.json"]
    print(("PASS" if ok else "FAIL") + " the corpus holds vectors.json and README.md alone")
    fails += [] if ok else ["other files: %s" % names]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    f2, seen = check(doc)
    fails += f2
    missing = [r for r in REASONS if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of §12 refuses a vector")
    fails += ["no vector refused with %s" % ", ".join(missing)] if missing else []
    for label, d in mutations(doc):
        caught, _ = check(d, quiet=True, first=True)
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for name in CHECKS:
        DROP.add(name)
        try:
            caught, _ = check(doc, quiet=True, first=True)
        finally:
            DROP.discard(name)
        print(("PASS" if caught else "FAIL") + " the corpus catches the check dropped: %s%s" % (name, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("the corpus does not catch the check dropped: %s" % name)
    print("co_watch_verify: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
    for f in fails:
        print("  " + f)
    return 0 if not fails else 1


def room_main(path):
    with open(path, encoding="utf-8") as f:
        v = json.load(f)
    v.setdefault("sources", [])
    print(json.dumps(run_vector(v)))
    return 0


if __name__ == "__main__":
    if len(sys.argv) == 3 and sys.argv[1] == "--room":
        sys.exit(room_main(sys.argv[2]))
    if len(sys.argv) != 2:
        print("usage: co_watch_verify.py <live-conformance/cowatch> | --room <room.json>", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))