#!/usr/bin/env python3
"""wavelab-verify: recompute the wavelab ledger from its published segments and the exchange's
public archive, and say where the operator's record disagrees.

    python verify.py path/to/wavelab-ledger                     # everything, fetching archives
    python verify.py path/to/wavelab-ledger --offline           # chain, heads, draws, costs only
    python verify.py path/to/wavelab-ledger --archive-mirror M  # archives from a local mirror

Standard library only, and no code shared with the operator. Every rule the operator committed to
is read from the ledger's own genesis rows, so what it checks is that the operator followed the
rules it committed to before the first signal. A few thresholds are this file's own, fixed with
its hash in that same genesis: an arm with no outcome 31 days past its horizon FAILS as withheld
(3 days WARNS); a Bitcoin anchor more than a day before its head FAILS (more than 7 days after it
WARNS); a manifest written more than 60 s into its bar WARNS; and so does a stretch of more than
two bars with no bar promised while heads are still being sealed. The exit walk it implements is
checked against the committed exit template, not assumed.

WHAT IT RECOMPUTES
  chain      every row hash from its body; seq gapless from 1; each row names its predecessor;
             time never runs backwards along the chain
  bodies     the encoding is canonical: one value has exactly one spelling
  heads      each covers exactly the rows since the last; its segment file holds exactly those rows;
             the pre-registered cadence is kept from the genesis on, or an incident names the
             gap; git history only adds segments (read from a full clone; anything else WARNS)
  proofs     every OpenTimestamps proof is for its head's own hash, every Bitcoin time recorded is
             the time of the block that proof reaches, every git proof binds its segment's bytes
  genesis    the pre-registration, the committed rules and their source, every strategy's code
             and parameters, and the sealed modules that code lives in, each hash to its claim
  manifests  each bar was promised while open, with the roster the bindings imply; every promise
             was delivered in full, or the shortfall is disclosed by an incident naming it
  timing     decisions written after their bar closed, within the committed skew, `late` honest,
             from a clean build, citing their strategy's sealed code
  controls   each actionable plan has its control immediately after it; the side is the committed
             coin flip over the bar's manifest, the strategy, its direction and the close; the
             geometry mirrors the treatment's; costs recomputed
  plans      from the archive: the close, the exact Wilder ATR over the committed window, and from
             them the stop, target, R multiple, cost and verdict — to the digit
  outcomes   from the archive: every resolution's barrier, R, excursions and bar count, and the
             archive files it cites; and that no outcome is withheld past its horizon
  evaluation the pre-registered decision rules themselves: the family endpoint (evaluate_family:
             pooled episodes, cluster bootstrap over 8-day blocks with the committed counter RNG)
             and the per-strategy one (evaluate: episodes, Student t bound, one-sided and deadline
             rules). Both are reference implementations of the sealed text

WHAT IT DOES NOT
  It does not re-run the 88 strategies. Their code — the sealed modules and the runtime they ran
  under — is in the genesis, and anyone can run it; this file checks everything that follows from
  a signal, not the signal itself. A strategy that was published but was not what ran would not be
  caught here. The provenance column records which build of the operator's engine wrote each row.

Exit status: 0 when nothing FAILS, 1 otherwise. WARN is information a reader should see.
"""

from __future__ import annotations

import argparse
import csv
import functools
import hashlib
import io
import json
import math
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import UTC, datetime
from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal
from fractions import Fraction
from pathlib import Path

VERIFIER_VERSION = "1.0.0"

DOMAIN = "wavelab-journal-chain"
CHAIN_VERSION = "3"
HEADER = ("seq", "prev", "kind", "written_ms", "writer_id", "engine_sha", "provenance",
          "clock_source", "clock_skew_ms", "schema_sha256")
ZERO_HASH = "0" * 64
UNRESERVED = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~")
ARCHIVE = "https://data.binance.vision/data/spot/daily/klines"
DAY_MS = 86_400_000
MICRO = 1_000_000
DAY = 86_400_000
#: The operator writes an arm unresolvable if the archive has not published its days 30 days after
#: its horizon; a day's grace on top, and an arm still without a row has been withheld.
WITHHELD_AFTER_MS = 31 * DAY
OVERDUE_AFTER_MS = 3 * DAY
Q8 = Decimal("0.00000001")


def f8(x: Decimal) -> str:
    """Eight decimals in plain notation. str(Decimal) switches to exponents below 1e-6 — an
    exactly-zero excursion printed as '0E-8' — and would FAIL honest rows written as
    '0.00000000'."""
    return format(x.quantize(Q8), "f")


def sha256(data: bytes | str) -> str:
    return hashlib.sha256(data.encode("utf-8") if isinstance(data, str) else data).hexdigest()


# ---------------------------------------------------------------------------- findings

@dataclass
class Report:
    fails: list[tuple[str, int | None, str]] = field(default_factory=list)
    warns: list[tuple[str, int | None, str]] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    checked: dict[str, int] = field(default_factory=lambda: defaultdict(int))

    def fail(self, check: str, seq: int | None, msg: str) -> None:
        self.fails.append((check, seq, msg))

    def warn(self, check: str, seq: int | None, msg: str) -> None:
        self.warns.append((check, seq, msg))

    def ok(self, check: str, n: int = 1) -> None:
        self.checked[check] += n


# ---------------------------------------------------------------------------- the body format

def pct(v: str) -> str:
    return "".join(ch if ch in UNRESERVED else "".join(f"%{b:02X}" for b in ch.encode("utf-8"))
                   for ch in v)


def unpct(s: str) -> str:
    """Decode, and refuse any spelling that is not the canonical one: a value with two spellings
    would have two hashes, and a row could be re-spelled into a different chain."""
    v = urllib.parse.unquote(s, errors="strict")
    if pct(v) != s:
        raise ValueError(f"non-canonical percent-encoding: {s[:60]!r}")
    return v


@dataclass
class Row:
    seq: int
    row_hash: str
    body: str
    h: dict[str, str]
    c: dict[str, str | None]
    order: list[str]

    @property
    def kind(self) -> str:
        return self.h["kind"]

    def i(self, col: str) -> int | None:
        v = self.c[col]
        return None if v is None else int(v)

    def d(self, col: str) -> Decimal | None:
        v = self.c[col]
        return None if v is None else Decimal(v)


def parse_body(body: str) -> tuple[dict[str, str], dict[str, str | None], list[str]]:
    if not body.endswith("\n"):
        raise ValueError("body does not end with a newline")
    lines = body[:-1].split("\n")
    if lines[0] != DOMAIN or lines[1] != f"v={CHAIN_VERSION}":
        raise ValueError(f"unknown domain or version: {lines[:2]}")
    h: dict[str, str] = {}
    for key, line in zip(HEADER, lines[2:2 + len(HEADER)], strict=True):
        k, sep, v = line.partition("=")
        if k != key or not sep:
            raise ValueError(f"header line {line[:40]!r} where {key!r} belongs")
        h[k] = unpct(v)
    c: dict[str, str | None] = {}
    order: list[str] = []
    for line in lines[2 + len(HEADER):]:
        if line.endswith(":null") and "=" not in line:
            k, v = line[:-5], None
        else:
            k, sep, raw = line.partition("=")
            if not sep:
                raise ValueError(f"malformed line {line[:40]!r}")
            v = unpct(raw)
        if k in c:
            raise ValueError(f"column {k!r} appears twice")
        c[k] = v
        order.append(k)
    return h, c, order


# ---------------------------------------------------------------------------- loading

@dataclass
class Segment:
    path: str
    first: int
    last: int
    data: bytes


def load(ledger_dir: Path, rep: Report) -> tuple[list[Row], list[Segment]]:
    files = sorted((ledger_dir / "segments").rglob("*.ndjson"), key=lambda p: p.name)
    segs: list[Segment] = []
    rows: list[Row] = []
    crlf = []
    for f in files:
        a, _, b = f.stem.partition("-")
        data = f.read_bytes()
        if b"\r\n" in data:
            # Git for Windows converts line endings on checkout by default (core.autocrlf). The
            # operator publishes LF only, so this is the reader's checkout, not the ledger: read
            # the bytes as committed. Content cannot differ between the two spellings.
            crlf.append(f.name)
            data = data.replace(b"\r\n", b"\n")
        segs.append(Segment(f.relative_to(ledger_dir).as_posix(), int(a), int(b), data))
    if crlf:
        rep.warn("segments", None, f"{len(crlf)} segment files were checked out with CRLF line "
                                   "endings (git's core.autocrlf) and were read back as LF; "
                                   "clone with `git -c core.autocrlf=false clone ...` to avoid it")
    for s in segs:
        seqs = []
        for n, line in enumerate(s.data.decode("utf-8").splitlines(), 1):
            try:
                obj = json.loads(line)
                h, c, order = parse_body(obj["body"])
                rows.append(Row(int(obj["seq"]), obj["row_hash"], obj["body"], h, c, order))
                seqs.append(int(obj["seq"]))
            except (ValueError, KeyError, TypeError) as e:
                rep.fail("bodies", None, f"{s.path}:{n}: {e}")
        if seqs != list(range(s.first, s.last + 1)):
            rep.fail("segments", s.first, f"{s.path} does not hold exactly rows {s.first}-{s.last}")
    return rows, segs


# ---------------------------------------------------------------------------- chain and heads

def check_chain(rows: list[Row], rep: Report) -> None:
    prev = ZERO_HASH
    for n, r in enumerate(rows, 1):
        if r.seq != n:
            rep.fail("chain", r.seq, f"expected seq {n}: the chain has a gap or a reordering")
            return
        if str(r.seq) != r.h["seq"]:
            rep.fail("chain", r.seq, f"the body says seq {r.h['seq']}")
        if sha256(r.body) != r.row_hash:
            rep.fail("chain", r.seq, "row_hash is not the SHA-256 of the body")
        if r.h["prev"] != prev:
            rep.fail("chain", r.seq, "prev does not name the preceding row's hash")
        # Time only moves forward along the chain. Without this, a bar left unpromised could be
        # "promised" hours later by appending rows dated inside it: every other check reads the
        # times each row claims for itself.
        if n > 1 and int(r.h["written_ms"]) < int(rows[n - 2].h["written_ms"]):
            rep.fail("chain", r.seq, "written_ms is earlier than the row before it")
        prev = r.row_hash
        rep.ok("chain")


def check_heads(rows: list[Row], segs: list[Segment], ledger_dir: Path, rep: Report,
                rules: Rules | None = None) -> None:
    by_seq = {r.seq: r for r in rows}
    heads = [r for r in rows if r.kind == "head"]
    covered = 0
    # The genesis is the zeroth head: the cadence is promised from the moment the rules are sealed,
    # so a first head that came late needs its incident like any other.
    last_pub: int | None = rows[0].i("created_ms") if rows and rows[0].kind == "genesis" else None
    last_head_seq = 0
    seg_by_range = {(s.first, s.last): s for s in segs}
    for hd in heads:
        frm, to = hd.i("covers_from_seq"), hd.i("head_seq")
        if frm != covered + 1:
            rep.fail("heads", hd.seq, f"covers from {frm}, but the previous head ended at {covered}")
        if not to < hd.seq:
            rep.fail("heads", hd.seq, "a head covers itself or rows after it")
        if to in by_seq and hd.c["head_row_hash"] != by_seq[to].row_hash:
            rep.fail("heads", hd.seq, f"head_row_hash is not the hash of row {to}")
        pub = hd.i("published_ms")
        interval = hd.i("max_head_interval_ms")
        if rules is not None and interval != int(rules.params["max_head_interval_ms"]):
            rep.fail("heads", hd.seq, "max_head_interval_ms is not the pre-registered cadence")
            interval = int(rules.params["max_head_interval_ms"])
        if pub != int(hd.h["written_ms"]):
            rep.fail("heads", hd.seq, "published_ms is not the time the head was written")
        if last_pub is not None:
            if pub < last_pub:
                rep.fail("heads", hd.seq, "published_ms went backwards")
            elif pub - last_pub > interval:
                lapse = [r for r in rows if r.kind == "incident" and last_head_seq < r.seq < hd.seq
                         and r.c["category"] == "timestamping_lapse"]
                if not lapse:
                    since = "the last head" if last_head_seq else "the genesis"
                    rep.fail("heads", hd.seq, f"{(pub - last_pub) // 60_000} min since {since} "
                                              "and no timestamping_lapse incident says why")
                else:
                    rep.warn("heads", hd.seq, f"late head, disclosed: {lapse[0].c['detail']}")
        seg = seg_by_range.get((frm, to))
        if seg is None:
            if to <= rows[-1].seq:
                rep.fail("heads", hd.seq, f"no segment file holds rows {frm}-{to}")
        else:
            want = "segments/" + seg.path.split("segments/", 1)[1]
            if not hd.c["publish_uri"].endswith(want):
                rep.fail("heads", hd.seq, f"publish_uri does not point at {want}")
        covered, last_pub, last_head_seq = to, pub, hd.seq
        rep.ok("heads")
    ranges = sorted((s.first, s.last) for s in segs)
    expect = 1
    for a, b in ranges:
        if a != expect:
            rep.fail("segments", a, f"segment starts at {a}, expected {expect}")
        expect = b + 1
    head_ranges = {(h.i("covers_from_seq"), h.i("head_seq")) for h in heads}
    for s in segs:
        if (s.first, s.last) in head_ranges:
            continue
        # The newest segment's head row is written AFTER the rows it covers, so it lands in the
        # next segment: until that is published, the last file has no head among the rows.
        newest = s is max(segs, key=lambda x: x.first)
        if newest and s.first == covered + 1 and rows and s.last == rows[-1].seq:
            rep.notes.append(f"{s.path}: its head is published with the next segment")
        else:
            rep.fail("segments", s.first, f"{s.path} is not the range of any head")
    if heads and rows and heads[-1].seq < rows[-1].seq:
        rep.notes.append(f"rows after the last head: {rows[-1].seq - heads[-1].seq} (sealed by "
                         "the next one)")
    _git_append_only(ledger_dir, rep)


def _git_append_only(ledger_dir: Path, rep: Report) -> None:
    """Only a full clone of THIS repository can answer. A ZIP download has no history, a shallow
    clone has only its tip, and a folder inside some other repository would have git answer about
    that one: each would otherwise print a clean `ok` about history nobody read."""
    def git(*args: str) -> str:
        # UTF-8 whatever the reader's code page: git writes paths and file names in it.
        return subprocess.run(["git", "-C", str(ledger_dir), *args], capture_output=True,
                              encoding="utf-8", errors="replace", timeout=120,
                              check=True).stdout.strip()
    try:
        # `--show-cdup` is empty exactly at the top of a working tree, and prints no path, so
        # neither letter case, Unicode form nor a symlink can make a real clone look like another.
        cdup, shallow = git("rev-parse", "--show-cdup"), git("rev-parse", "--is-shallow-repository")
    except (OSError, ValueError, subprocess.SubprocessError):
        cdup, shallow = None, None
    if cdup != "":
        rep.warn("git", None, "not the top of a git clone of the ledger (a ZIP download, a "
                              "folder inside another repository, or no git): the append-only "
                              "history was NOT checked; clone it with git")
        return
    if shallow == "true":
        rep.warn("git", None, "a shallow clone: the append-only history was NOT checked; clone "
                              "without --depth")
        return
    try:
        out = git("log", "--diff-filter=MDR", "--name-only", "--format=%H", "--", "segments")
    except (OSError, ValueError, subprocess.SubprocessError) as e:
        rep.warn("git", None, f"git log failed; the append-only history was NOT checked: {e}")
        return
    if out.strip():
        rep.fail("git", None, "a commit modified, deleted or renamed a published segment:\n" + out)
    else:
        rep.ok("git")


# ---------------------------------------------------------------------------- witnesses

OTS_MAGIC = b"\x00OpenTimestamps\x00\x00Proof\x00\xbf\x89\xe2\xe8\x84\xe8\x92\x94"
OTS_BITCOIN = bytes.fromhex("0588960d73d71901")
OTS_PENDING = bytes.fromhex("83dfe30d2ef90c8e")
EXPLORERS = ("https://blockstream.info/api", "https://mempool.space/api")


class _Reader:
    def __init__(self, b: bytes) -> None:
        self.b, self.i = b, 0

    def take(self, n: int) -> bytes:
        if self.i + n > len(self.b):
            raise ValueError("truncated proof")
        out = self.b[self.i:self.i + n]
        self.i += n
        return out

    def byte(self) -> int:
        return self.take(1)[0]

    def varuint(self) -> int:
        v = shift = 0
        while True:
            b = self.byte()
            v |= (b & 0x7F) << shift
            if not b & 0x80:
                return v
            shift += 7

    def varbytes(self) -> bytes:
        return self.take(self.varuint())


def _ots_op(tag: int, msg: bytes, r: _Reader) -> bytes:
    if tag == 0xF0:
        return msg + r.varbytes()
    if tag == 0xF1:
        return r.varbytes() + msg
    if tag == 0x08:
        return hashlib.sha256(msg).digest()
    if tag == 0x02:
        return hashlib.sha1(msg).digest()
    if tag == 0x03:
        return hashlib.new("ripemd160", msg).digest()
    if tag == 0xF2:
        return msg[::-1]
    if tag == 0xF3:
        return msg.hex().encode()
    raise ValueError(f"unknown OpenTimestamps op 0x{tag:02x}")


def ots_attestations(blob: bytes) -> tuple[bytes, list[tuple[str, bytes, object]]]:
    """(file digest, [(kind, commitment, payload)]) — every attestation in an .ots file, each with
    the message the path from the digest reaches. Pure: no network."""
    r = _Reader(blob)
    if r.take(len(OTS_MAGIC)) != OTS_MAGIC or r.varuint() != 1:
        raise ValueError("not an OpenTimestamps v1 proof")
    if r.byte() != 0x08:
        raise ValueError("the proof does not timestamp a SHA-256 digest")
    digest = r.take(32)
    found: list[tuple[str, bytes, object]] = []

    def branch(tag: int, msg: bytes) -> None:
        if tag == 0x00:
            kind, payload = r.take(8), r.varbytes()
            if kind == OTS_BITCOIN:
                found.append(("bitcoin", msg, _Reader(payload).varuint()))
            elif kind == OTS_PENDING:
                found.append(("pending", msg, _Reader(payload).varbytes().decode()))
            else:
                found.append(("unknown", msg, kind.hex()))
        else:
            walk(_ots_op(tag, msg, r))

    def walk(msg: bytes) -> None:
        while True:
            tag = r.byte()
            if tag == 0xFF:
                branch(r.byte(), msg)
                continue
            branch(tag, msg)
            return

    walk(digest)
    if r.i != len(blob):
        raise ValueError("trailing bytes after the proof")
    return digest, found


@functools.cache
def block_header(height: int, headers: Path | None) -> bytes:
    """The 80-byte header at `height`: from a local directory of `<height>.hex` files if given,
    otherwise from public explorers — checked against its own hash either way."""
    if headers is not None:
        raw = bytes.fromhex((headers / f"{height}.hex").read_text().strip())
        if len(raw) != 80:
            raise ValueError(f"{height}.hex is not an 80-byte header")
        return raw
    last: Exception | None = None
    for base in (*EXPLORERS, *EXPLORERS):      # a second round: explorers rate-limit bursts
        if last is not None and base == EXPLORERS[0]:
            time.sleep(3)
        try:
            with urllib.request.urlopen(f"{base}/block-height/{height}", timeout=20) as f:
                h = f.read().decode().strip()
            with urllib.request.urlopen(f"{base}/block/{h}/header", timeout=20) as f:
                raw = bytes.fromhex(f.read().decode().strip())
            if hashlib.sha256(hashlib.sha256(raw).digest()).digest()[::-1].hex() != h:
                raise ValueError(f"{base} returned a header that is not block {h}")
            return raw
        except Exception as e:  # noqa: BLE001 — try the next explorer
            last = e
    raise ValueError(f"no explorer returned block {height}: {last}")


def check_proofs(rows: list[Row], segs: list[Segment], rep: Report, *, online: bool,
                 headers: Path | None) -> None:
    """Every witness a head claims must be a witness to THAT head.

    OpenTimestamps: the proof must be a proof for the head row's own hash (SHA-256 of its body,
    so `ots verify` works on the body saved as a file); a Bitcoin attestation must reach its
    block's merkle root, and the recorded time must be that block's. git: the proof binds the
    bytes of the segment the head covers."""
    by_seq = {r.seq: r for r in rows}
    seg_by_range = {(s.first, s.last): s for s in segs}
    for pr in rows:
        if pr.kind != "head_proof":
            continue
        head = by_seq.get(pr.i("head_ref_seq"))
        if head is None or head.kind != "head":
            rep.fail("proofs", pr.seq, "a proof for something that is not a head")
            continue
        kind = pr.c["proof_type"]
        if kind == "git_commit":
            seg = seg_by_range.get((head.i("covers_from_seq"), head.i("head_seq")))
            if seg is not None and sha256(seg.data) != pr.c["proof_sha256"]:
                rep.fail("proofs", pr.seq, "the git proof does not bind the published segment")
            rep.ok("proofs")
            continue
        if kind != "opentimestamps":
            rep.ok("proofs")
            continue
        blob = bytes.fromhex(pr.c["proof_blob"] or "")
        if sha256(blob) != pr.c["proof_sha256"]:
            rep.fail("proofs", pr.seq, "proof_blob does not hash to proof_sha256")
            continue
        try:
            digest, atts = ots_attestations(blob)
        except ValueError as e:
            rep.fail("proofs", pr.seq, f"unreadable OpenTimestamps proof: {e}")
            continue
        if digest.hex() != head.row_hash:
            rep.fail("proofs", pr.seq, "the OpenTimestamps proof is for another hash, not this head's")
            continue
        claimed = pr.i("attested_ms")
        if claimed is None:
            rep.ok("proofs")
            continue
        if not online:
            rep.warn("proofs", pr.seq, "Bitcoin time not checked offline")
            continue
        # The LOWEST-HEIGHT attestation, which is the first block that carried this hash. Not the
        # earliest block TIME: Bitcoin timestamps are not monotonic in height, so "earliest time"
        # and "first block" disagree now and then, and the two sides would differ over an honest
        # proof. Reached and unreachable blocks are kept apart: an explorer outage is our problem,
        # not the ledger's, so it warns; a header that contradicts the proof fails.
        reached, missing = [], []
        for k, msg, height in sorted((a for a in atts if a[0] == "bitcoin"), key=lambda a: a[2]):
            try:
                raw = block_header(height, headers)
            except (OSError, ValueError) as e:
                missing.append(height)
                rep.warn("proofs", pr.seq, f"block {height} unavailable, not checked: {e}")
                continue
            if raw[36:68] == msg:
                reached.append((height, int.from_bytes(raw[68:72], "little") * 1000))
            else:
                rep.fail("proofs", pr.seq, f"the attestation to block {height} does not reach it")
        if not reached:
            if not missing:
                rep.fail("proofs", pr.seq, "no Bitcoin attestation in the proof reaches its block")
            continue
        height, when = reached[0]
        if missing and min(missing) < height:
            # The first block that carried this hash is one we could not read, so the claimed
            # time can be neither confirmed nor contradicted by the ones we could.
            rep.warn("proofs", pr.seq, f"attested_ms not checked: the lowest block, "
                                       f"{min(missing)}, is unavailable")
            continue
        if when != claimed:
            rep.fail("proofs", pr.seq, f"attested_ms {claimed} is not block {height}'s time {when}")
        elif claimed < head.i("published_ms") - DAY:
            rep.fail("proofs", pr.seq, "the proof is anchored before the head it timestamps")
        elif claimed - head.i("published_ms") > 7 * DAY:
            rep.warn("proofs", pr.seq, f"anchored {(claimed - head.i('published_ms')) // DAY} days "
                                       "after the head says it was sealed")
        else:
            rep.ok("proofs")
    heads = [r for r in rows if r.kind == "head"]
    anchored = {r.i("head_ref_seq") for r in rows
                if r.kind == "head_proof" and r.i("attested_ms") is not None}
    if heads:
        last = max((h.seq for h in heads if h.seq in anchored), default=None)
        rep.notes.append("no head is anchored in Bitcoin yet" if last is None else
                         f"latest head anchored in Bitcoin: {last} of {heads[-1].seq}")


# ---------------------------------------------------------------------------- genesis

@dataclass
class Rules:
    params: dict
    exit_template: dict
    cost_model: dict
    draw: dict
    verifier_sha256: str
    prereg_seq: int
    effective_from_ms: int = 0


def read_genesis(rows: list[Row], rep: Report) -> Rules | None:
    if not rows or rows[0].kind != "genesis":
        rep.fail("genesis", 1, "row 1 is not a genesis row")
        return None
    g = rows[0]
    kinds = {r.kind: r for r in rows[:200] if r.kind != "strategy_binding"}
    need = ("pre_registration", "exit_template", "cost_model", "fill_rule", "draw_algorithm")
    for k in need:
        if k not in kinds:
            rep.fail("genesis", None, f"no {k} row in the ceremony")
            return None
    pr = kinds["pre_registration"]
    params = json.loads(pr.c["params_json"])
    if sha256(pr.c["params_json"]) != pr.c["params_sha256"]:
        rep.fail("genesis", pr.seq, "params_json does not hash to params_sha256")
    et, cm, da = kinds["exit_template"], kinds["cost_model"], kinds["draw_algorithm"]
    fr = kinds["fill_rule"]
    for k, v in (("tie_break_policy", "pessimistic_sl"), ("vertical_exit", "close_at_market")):
        if et.c.get(k) != v:
            rep.fail("genesis", et.seq, f"the exit template's {k} is {et.c.get(k)!r}; this file "
                                        f"implements only {v!r}")
    for r in (et, cm, fr, da):
        if sha256(r.c["contents_json"]) != r.c["contents_sha256"]:
            rep.fail("genesis", r.seq, "contents_json does not hash to contents_sha256")
    # Every piece of the operator's code the genesis commits to is IN the genesis, readable, and
    # must hash to the commitment: a hash of code nobody can read is not a commitment anyone can
    # check. This file does not run that code — it re-implements the rules — but a reader can.
    committed_code = (
        (da, "source_text", da.c["source_sha256"], "the draw algorithm"),
        (fr, "source_text", fr.c["source_sha256"], "the plan arithmetic"),
        (et, "resolver_source_text", et.c["resolver_sha256"], "the resolver"),
        (et, "atr_source_text", json.loads(et.c["contents_json"])["atr"]["source_sha256"],
         "the ATR"),
    )
    for r, col, want, what in committed_code:
        if sha256(r.c[col]) != want:
            rep.fail("genesis", r.seq, f"the source of {what} does not hash to its commitment")
    # The strategies: each binding's code and params hash to their commitments, each lives in a
    # sealed module that contains it, and each sealed module hashes to its own commitment.
    sources = {r.c["path"]: r for r in rows[:400] if r.kind == "sealed_source"}
    for r in sources.values():
        if sha256(r.c["source_text"]) != r.c["source_sha256"]:
            rep.fail("genesis", r.seq, f"sealed source {r.c['path']} does not hash to its sha")
    for b in (r for r in rows if r.kind == "strategy_binding"):
        if sha256(b.c["code_text"]) != b.c["code_sha256"]:
            rep.fail("genesis", b.seq, f"{b.c['strategy_id']}: code_text does not hash to code_sha256")
        if sha256(b.c["params_json"]) != b.c["params_sha256"]:
            rep.fail("genesis", b.seq, f"{b.c['strategy_id']}: params do not hash to params_sha256")
        mod = sources.get(b.c["code_source_uri"])
        if mod is None or b.c["code_text"] not in mod.c["source_text"]:
            rep.fail("genesis", b.seq, f"{b.c['strategy_id']}: its code is not in a sealed module")
    if not sources:
        rep.fail("genesis", None, "no sealed sources: the strategies' code cannot be re-run")
    rep.notes.append(f"pre-registration: {pr.c['title']}")
    rep.notes.append(f"prior: {pr.c['prior']}")
    rep.notes.append(f"primary endpoint: {pr.c['primary_endpoint']}")
    rep.ok("genesis")
    return Rules(params, json.loads(et.c["contents_json"]), json.loads(cm.c["contents_json"]),
                 json.loads(da.c["contents_json"]), g.c["verifier_sha256"], pr.seq,
                 int(pr.c["effective_from_ms"]))


# ---------------------------------------------------------------------------- manifests, timing

def active_roster(bindings: list[Row], retirements: list[Row], bar_open: int, symbol: str,
                  tf: str) -> list[str]:
    retired = {r.i("binding_seq") for r in retirements if r.i("effective_from_ms") <= bar_open}
    return sorted(b.c["strategy_id"] for b in bindings
                  if b.i("effective_from_ms") <= bar_open and b.seq not in retired
                  and symbol in json.loads(b.c["symbols_json"])
                  and tf in json.loads(b.c["timeframes_json"]))


def check_manifests(rows: list[Row], rules: Rules, rep: Report) -> None:
    bindings = [r for r in rows if r.kind == "strategy_binding"]
    retirements = [r for r in rows if r.kind == "strategy_retirement"]
    manifests = [r for r in rows if r.kind == "bar_manifest"]
    decisions = defaultdict(list)
    for r in rows:
        if r.kind == "decision":
            decisions[r.i("manifest_seq")].append(r)
    max_skew = int(rules.params["max_skew_ms"])
    late_after = int(rules.params["late_after_ms"])
    bind_by_id = {b.c["strategy_id"]: b for b in bindings}
    incidents = [r for r in rows if r.kind == "incident"]
    late_rows = 0
    by_bar: dict[tuple, list[Row]] = defaultdict(list)
    for m in manifests:
        key = (m.c["symbol"], m.c["timeframe"], m.i("bar_open_ms"))
        by_bar[key].append(m)
        if m.i("max_skew_ms") != max_skew:
            rep.fail("manifests", m.seq, "max_skew_ms is not the pre-registered tolerance")
        if m.c["manifest_kind"] == "refusal":
            rep.warn("manifests", m.seq, f"bar {m.i('bar_open_ms')} refused: "
                                         f"{m.c['refusal_reason']}")
            rep.ok("manifests")
            continue
        written = int(m.h["written_ms"])
        if not (m.i("bar_open_ms") - max_skew <= written < m.i("bar_close_ms")):
            rep.fail("manifests", m.seq, "a normal manifest was not written while its bar was open")
        elif written - m.i("bar_open_ms") > 60_000:
            # The operator's own rule: a bar is promised at the close of the one before it. A
            # promise made later, mid-bar, is legal but was made with part of the bar known.
            rep.warn("manifests", m.seq, f"promised {(written - m.i('bar_open_ms')) // 1000} s "
                                         "into its bar, not at its open")
        if abs(int(m.h["clock_skew_ms"])) > max_skew:
            rep.fail("manifests", m.seq, "a normal manifest was written outside the skew tolerance")
        roster = json.loads(m.c["roster_json"])
        if sha256(m.c["roster_json"]) != m.c["roster_sha256"] or m.i("expected_rows") != len(roster):
            rep.fail("manifests", m.seq, "roster, its hash and expected_rows disagree")
        want = active_roster(bindings, retirements, m.i("bar_open_ms"), m.c["symbol"],
                             m.c["timeframe"])
        if roster != want:
            rep.fail("manifests", m.seq, f"roster differs from the active bindings "
                                         f"({len(roster)} vs {len(want)})")
        got = decisions.get(m.seq, [])
        later = any(n.c["symbol"] == m.c["symbol"] and n.c["timeframe"] == m.c["timeframe"]
                    and n.i("bar_open_ms") > m.i("bar_open_ms") for n in manifests)
        # A bar with nothing delivered and nothing promised after it is still open: pending.
        pending = not got and not later
        if not pending and sorted(d.c["strategy_id"] for d in got) != roster:
            # Only an incident that NAMES this manifest excuses it. An open-ended one — a
            # timestamping lapse says affected_from_seq with no end — would otherwise excuse
            # every shortfall that came after it, for ever.
            why = [i for i in incidents
                   if i.i("affected_from_seq") is not None and i.i("affected_to_seq") is not None
                   and i.i("affected_from_seq") <= m.seq <= i.i("affected_to_seq")]
            msg = (f"bar {m.i('bar_open_ms')}: {len(got)} of {len(roster)} promised decisions "
                   f"delivered")
            if why:
                # Disclosed by the operator in the chain itself: a shortfall on the record, with
                # its reason. A shortfall nobody explains is the one that fails.
                rep.warn("coverage", m.seq, f"{msg}; disclosed: {why[0].c['detail'][:200]}")
            else:
                rep.fail("coverage", m.seq, msg)
        for d in got:
            _check_decision_timing(d, m, max_skew, late_after, rep, bind_by_id.get(d.c["strategy_id"]))
            if d.i("late"):
                late_rows += 1
        rep.ok("manifests")
    # Every decision must hang off a NORMAL manifest for its own bar, and be on its roster. The
    # loop above walks manifests, so a decision pointing anywhere else would never be looked at —
    # and the control's coin is drawn from the manifest the decision names.
    by_manifest = {m.seq: m for m in manifests}
    for d in rows:
        if d.kind != "decision":
            continue
        m = by_manifest.get(d.i("manifest_seq"))
        if (m is None or m.c["manifest_kind"] != "normal"
                or m.i("bar_open_ms") != d.i("bar_open_ms")
                or m.c["symbol"] != d.c["symbol"] or m.c["timeframe"] != d.c["timeframe"]
                or d.c["strategy_id"] not in json.loads(m.c["roster_json"])):
            rep.fail("coverage", d.seq, "decision does not hang off a normal manifest for its own "
                                        "bar, or is not on that manifest's roster")
    for key, ms in by_bar.items():
        if len(ms) > 1:
            rep.fail("manifests", ms[1].seq, f"bar {key[2]} has {len(ms)} manifests")
    if late_rows:
        rep.warn("timing", None, f"{late_rows} decisions were written late; the pre-registration "
                                 "excludes them from the endpoint")
    _check_grid(manifests, bindings, rep,
                max((h.i("published_ms") for h in rows if h.kind == "head"), default=None))


def _check_decision_timing(d: Row, m: Row, max_skew: int, late_after: int, rep: Report,
                           binding: Row | None) -> None:
    written = int(d.h["written_ms"])
    if d.i("bar_open_ms") != m.i("bar_open_ms") or d.seq < m.seq:
        rep.fail("timing", d.seq, "decision does not belong to its manifest's bar")
    if written < d.i("bar_close_ms") - max_skew:
        rep.fail("timing", d.seq, "decision written before its bar closed")
    if abs(int(d.h["clock_skew_ms"])) > max_skew:
        rep.fail("timing", d.seq, "decision written outside the skew tolerance")
    if d.i("late") != int(written - d.i("bar_close_ms") > late_after):
        rep.fail("timing", d.seq, "the late flag does not match the times in the row")
    if d.h["provenance"] != "clean" or d.h["engine_sha"] == "?":
        rep.fail("timing", d.seq, "decision written from a tree that was not clean")
    if d.i("max_skew_ms") != max_skew or d.i("late_after_ms") != late_after:
        rep.fail("timing", d.seq, "decision carries limits other than the pre-registered ones")
    b = binding
    if b is None or (d.c["strategy_code_sha256"], d.c["strategy_params_sha256"],
                     d.i("strategy_binding_seq")) != (b.c["code_sha256"], b.c["params_sha256"],
                                                       b.seq):
        rep.fail("timing", d.seq, "decision does not cite its strategy's sealed code and params")
    rep.ok("timing")


def _check_grid(manifests: list[Row], bindings: list[Row], rep: Report,
                newest_seal: int | None = None) -> None:
    if not manifests or not bindings:
        return
    tf_ms = manifests[0].i("tf_ms")
    start = min(b.i("effective_from_ms") for b in bindings)
    start = -(-start // tf_ms) * tf_ms
    have = {m.i("bar_open_ms") for m in manifests}
    end = max(have)
    missing = [t for t in range(start, end + 1, tf_ms) if t not in have]
    for t in missing:
        rep.warn("missing", None, f"bar {t} ({_iso(t)}) has no manifest of either kind")
    # Bars after the last manifest are not "missing from the grid" — the grid ends there — so an
    # emitter that simply stopped would leave no trace at all without this.
    if newest_seal is not None and newest_seal - end > 2 * tf_ms:
        rep.warn("missing", None, f"no bar has been promised since {_iso(end)}, while the record "
                                  f"was still being sealed at {_iso(newest_seal)}")
    rep.ok("grid", (end - start) // tf_ms + 1)


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

def ceil_div(a: int, b: int) -> int:
    return -((-a) // b)


def ticks(price: str | Decimal, tick_micro: int) -> int:
    t = Decimal(price) * MICRO / tick_micro
    if t != t.to_integral_value():
        raise ValueError(f"price {price} is off the tick grid")
    return int(t)


def px(t: int, tick_micro: int) -> str:
    return f8(Decimal(t) * tick_micro / MICRO)


def micro(v: int) -> str:
    return f8(Decimal(v) / MICRO)


def cost_micro(entry: int, stop: int, target: int, risk: int, rate: int) -> int:
    """ceil(rate/1e5 * (entry + exit) / risk * 1e6), on the dearer leg. Each leg at its own notional."""
    return max(ceil_div(rate * (entry + x) * 10, risk) for x in (stop, target))


@dataclass
class Plan:
    state: str
    entry: int
    stop: int | None = None
    target: int | None = None
    risk: int | None = None
    tp_r: int | None = None
    cost: int | None = None
    p_req: int | None = None


def plan(direction: int, entry: int, atr: int, rules: Rules) -> Plan:
    p, et = rules.params, rules.exit_template
    if atr <= 0:
        return Plan("no_atr", entry)
    risk = int(et["sl_atr_micro"]) * atr // MICRO
    if risk < int(p["min_risk_ticks"]):
        return Plan("risk_below_tick_floor", entry)
    stop = entry - direction * risk
    reward = ceil_div(int(et["tp_r_micro"]) * risk, MICRO)
    target = entry + direction * reward
    if stop <= 0 or target <= 0:
        return Plan("risk_below_tick_floor", entry)
    tp_r = reward * MICRO // risk
    rate = int(rules.cost_model["fee_tenth_bps"]) + int(rules.cost_model["non_fee_tenth_bps"])
    cost = cost_micro(entry, stop, target, risk, rate)
    if cost > int(p["max_cost_r_micro"]):
        return Plan("cost_over_cap", entry, stop, target, risk, tp_r, cost)
    loss, ev = int(p["expected_loss_r_micro"]), int(p["ev_min_r_micro"])
    p_req = ceil_div((loss + cost + ev) * MICRO, tp_r + loss)
    return Plan("planned", entry, stop, target, risk, tp_r, cost, p_req)


def wilder_atr(bars: list[Kline], period: int, tick_micro: int) -> int:
    """Exact: rationals over integer ticks, seeded with the mean of the first `period` ranges."""
    h = [ticks(b.high, tick_micro) for b in bars]
    lo = [ticks(b.low, tick_micro) for b in bars]
    c = [ticks(b.close, tick_micro) for b in bars]
    if len(c) <= period:
        return 0
    tr = [max(h[i] - lo[i], abs(h[i] - c[i - 1]), abs(lo[i] - c[i - 1])) for i in range(1, len(c))]
    atr = Fraction(sum(tr[:period]), period)
    for x in tr[period:]:
        atr = (atr * (period - 1) + x) / period
    return math.floor(atr)


# ---------------------------------------------------------------------------- the draw

def draw(manifest_row_hash: str, decision: Row, rules: Rules) -> tuple[str, int]:
    """The committed coin: SHA-256 over the bar's manifest (sealed while the bar was open), the
    strategy, its direction and the bar's close — nothing the operator could vary afterwards."""
    dom = rules.draw["seed_domain"]
    n = int(rules.draw["seed_hex_chars"])
    c = decision.c
    seed = sha256(f"{dom}\n{manifest_row_hash}\n{c['strategy_id']}\n{c['direction']}\n"
                  f"{c['mark_price']}\n")[:n]
    return seed, (1 if int(seed, 16) & 1 == 0 else -1)


def check_controls(rows: list[Row], rules: Rules, rep: Report) -> None:
    by_seq = {r.seq: r for r in rows}
    tick = int(rules.params["tick_size_micro"])
    rate = int(rules.cost_model["fee_tenth_bps"]) + int(rules.cost_model["non_fee_tenth_bps"])
    for d in rows:
        if d.kind != "decision" or d.c["verdict"] != "actionable":
            continue
        c = by_seq.get(d.seq + 1)
        if c is None or c.kind != "control" or c.i("decision_seq") != d.seq:
            rep.fail("controls", d.seq, "an actionable plan without its control immediately after")
            continue
        m = by_seq.get(d.i("manifest_seq"))
        seed, side = draw(m.row_hash if m else "", d, rules)
        if c.c["draw_seed"] != seed or c.i("direction") != side:
            rep.fail("controls", c.seq, "the control's side is not the committed draw")
        # The geometry is checked against the control's OWN side, so that a wrong draw and a
        # wrong geometry are reported as what they are rather than one masking the other.
        own = c.i("direction")
        entry = ticks(d.c["fill_price"], tick)
        risk = abs(entry - ticks(d.c["stop"], tick))
        reward = abs(ticks(d.c["resolution_target"], tick) - entry)
        stop, target = entry - own * risk, entry + own * reward
        if (c.c["entry_price"], c.c["stop"], c.c["target"]) != (px(entry, tick), px(stop, tick),
                                                                px(target, tick)):
            rep.fail("controls", c.seq, "the control's geometry does not mirror its treatment's")
        if c.c["cost_r"] != micro(cost_micro(entry, stop, target, risk, rate)):
            rep.fail("controls", c.seq, "the control's cost_r does not recompute")
        if c.i("entry_ts_ms") != d.i("fill_ts_ms"):
            rep.fail("controls", c.seq, "the control does not enter where its treatment did")
        rep.ok("controls")
    for c in rows:
        if c.kind == "control":
            d = by_seq.get(c.i("decision_seq"))
            if d is None or d.c["verdict"] != "actionable" or c.seq != d.seq + 1:
                rep.fail("controls", c.seq, "a control that does not follow an actionable plan")


# ---------------------------------------------------------------------------- the archive

@dataclass(frozen=True)
class Kline:
    open_ms: int
    high: str
    low: str
    close: str


class Unpublished(Exception):
    pass


class Archive:
    """Binance's daily kline files, each checked against the checksum published beside it."""

    def __init__(self, cache: Path, mirror: Path | None = None) -> None:
        self.cache, self.mirror = cache, mirror
        self.days: dict[str, tuple[list[Kline], str]] = {}

    def _get(self, url: str) -> bytes | None:
        if self.mirror is not None:
            p = self.mirror / url.removeprefix(ARCHIVE + "/")
            return p.read_bytes() if p.exists() else None
        try:
            with urllib.request.urlopen(url, timeout=30) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return None
            raise

    def day(self, symbol: str, tf: str, day_ms: int) -> tuple[list[Kline], str]:
        d = datetime.fromtimestamp(day_ms / 1000, tz=UTC).date().isoformat()
        name = f"{symbol}-{tf}-{d}.zip"
        if name in self.days:
            return self.days[name]
        local = self.cache / symbol / tf / name
        if local.exists() and local.with_suffix(".zip.CHECKSUM").exists():
            blob, chk = local.read_bytes(), local.with_suffix(".zip.CHECKSUM").read_bytes()
        else:
            url = f"{ARCHIVE}/{symbol}/{tf}/{name}"
            blob, chk = self._get(url), self._get(url + ".CHECKSUM")
            if blob is None or chk is None:
                raise Unpublished(name)
        want = chk.decode().split()[0].lower()
        if sha256(blob) != want:
            raise ValueError(f"{name} does not match its published checksum")
        if not local.exists():
            local.parent.mkdir(parents=True, exist_ok=True)
            local.write_bytes(blob)
            local.with_suffix(".zip.CHECKSUM").write_bytes(chk)
        with zipfile.ZipFile(io.BytesIO(blob)) as z:
            text = z.read(next(n for n in z.namelist() if n.endswith(".csv"))).decode()
        ks = []
        for row in csv.reader(io.StringIO(text)):
            if row and row[0].strip().isdigit():
                t = int(row[0])
                ks.append(Kline(t // 1000 if t > 10**14 else t, row[2], row[3], row[4]))
        self.days[name] = (ks, want)
        return self.days[name]

    def bars(self, symbol: str, tf: str, frm: int, to: int) -> tuple[dict[int, Kline], list[dict]]:
        out: dict[int, Kline] = {}
        files = []
        d = frm - frm % DAY_MS
        while d < to:
            ks, digest = self.day(symbol, tf, d)
            out.update({k.open_ms: k for k in ks if frm <= k.open_ms < to})
            name = f"{symbol}-{tf}-{datetime.fromtimestamp(d / 1000, tz=UTC).date()}.zip"
            files.append({"name": name, "sha256": digest})
            d += DAY_MS
        return out, files


# ---------------------------------------------------------------------------- plans from the archive

def check_plans(rows: list[Row], rules: Rules, arc: Archive, rep: Report) -> None:
    atr_rule = rules.exit_template["atr"]
    window, period = int(atr_rule["window_bars"]), int(atr_rule["period"])
    tick = int(rules.params["tick_size_micro"])
    per_bar: dict[tuple, tuple[int, str] | None] = {}
    for d in rows:
        if d.kind != "decision" or d.c["verdict"] not in ("actionable", "watch", "no_trade"):
            continue
        key = (d.c["symbol"], d.c["timeframe"], d.i("bar_open_ms"))
        if key not in per_bar:
            tf_ms = d.i("tf_ms")
            try:
                ks, _ = arc.bars(d.c["symbol"], d.c["timeframe"],
                                 d.i("bar_open_ms") - (window - 1) * tf_ms, d.i("bar_close_ms"))
            except Unpublished as e:
                rep.warn("plans", d.seq, f"archive not published yet ({e}); plan not checked")
                per_bar[key] = None
                continue
            except (OSError, ValueError) as e:
                rep.warn("archive", d.seq, f"archive unusable, plan not checked: {e}")
                per_bar[key] = None
                continue
            seq_ = [ks.get(d.i("bar_open_ms") - (window - 1 - j) * tf_ms) for j in range(window)]
            if any(k is None for k in seq_):
                rep.warn("plans", d.seq, f"the archive lacks bars in the {window}-bar window "
                                         f"ending {d.i('bar_open_ms')}; plan not checked")
                per_bar[key] = None
                continue
            per_bar[key] = (wilder_atr(seq_, period, tick), seq_[-1].close)
        got = per_bar[key]
        if got is None:
            continue
        atr, close = got
        if d.c["mark_price"] != f8(Decimal(close)):
            rep.fail("plans", d.seq, f"mark price {d.c['mark_price']} is not the archive's "
                                     f"close {close}")
            continue
        if d.c["verdict"] == "no_trade":
            rep.ok("plans")
            continue
        p = plan(d.i("direction"), ticks(close, tick), atr, rules)
        want = {"actionable": "planned", "watch": "cost_over_cap"}[d.c["verdict"]]
        if p.state != want:
            rep.fail("plans", d.seq, f"verdict {d.c['verdict']} but the committed rules give "
                                     f"{p.state} (ATR {atr} ticks)")
            continue
        exp = {"stop": px(p.stop, tick), "resolution_target": px(p.target, tick),
               "targets_json": json.dumps([px(p.target, tick)], separators=(",", ":")),
               "rr_at_t2": micro(p.tp_r), "cost_r": micro(p.cost)}
        if want == "planned":
            exp |= {"p_required": micro(p.p_req), "fill_price": px(p.entry, tick),
                    "fill_ts_ms": str(d.i("bar_close_ms"))}
        bad = {k: (d.c[k], v) for k, v in exp.items() if d.c[k] != v}
        if bad:
            rep.fail("plans", d.seq, f"does not recompute from ATR {atr}: {bad}")
        rep.ok("plans")


# ---------------------------------------------------------------------------- outcomes

def walk(direction: int, entry: Decimal, stop: Decimal, target: Decimal, bars: dict[int, Kline],
         first: int, tf_ms: int, max_bars: int) -> dict:
    long = direction == 1
    risk = abs(entry - stop)
    mae = mfe = Decimal(0)
    for i in range(max_bars):
        k = bars.get(first + i * tf_ms)
        if k is None:
            return {"barrier": "unresolvable"}
        hi, lo = Decimal(k.high), Decimal(k.low)
        mfe = max(mfe, ((hi - entry) if long else (entry - lo)) / risk)
        mae = max(mae, ((entry - lo) if long else (hi - entry)) / risk)
        tp = hi >= target if long else lo <= target
        sl = lo <= stop if long else hi >= stop
        if tp and sl:
            return {"barrier": "sl", "r": Decimal(-1), "mae": mae, "mfe": mfe, "bars": i + 1,
                    "amb": 1}
        if tp:
            return {"barrier": "tp", "r": abs(target - entry) / risk, "mae": mae, "mfe": mfe,
                    "bars": i + 1, "amb": 0}
        if sl:
            return {"barrier": "sl", "r": Decimal(-1), "mae": mae, "mfe": mfe, "bars": i + 1,
                    "amb": 0}
    last = Decimal(bars[first + (max_bars - 1) * tf_ms].close)
    return {"barrier": "vertical", "r": direction * (last - entry) / risk, "mae": mae, "mfe": mfe,
            "bars": max_bars, "amb": 0}


def check_resolutions(rows: list[Row], rules: Rules, arc: Archive | None,
                      rep: Report) -> list[tuple[Row, Row]]:
    by_seq = {r.seq: r for r in rows}
    max_bars = int(rules.exit_template["max_bars"])
    seen: set[tuple] = set()
    resolved: dict[int, Row] = {}
    for r in rows:
        if r.kind != "resolution":
            continue
        subj = by_seq.get(r.i("subject_seq"))
        key = (r.i("subject_seq"), r.c["resolver_id"], r.c["resolver_sha256"])
        if key in seen:
            rep.fail("outcomes", r.seq, "the same arm resolved twice by the same resolver build")
        seen.add(key)
        if subj is None or subj.kind not in ("decision", "control"):
            rep.fail("outcomes", r.seq, "resolves something that is not an arm")
            continue
        if r.h["provenance"] != "clean":
            rep.fail("outcomes", r.seq, "resolution written from a tree that was not clean")
        is_t = subj.kind == "decision"
        direction = subj.i("direction")
        entry = subj.d("fill_price" if is_t else "entry_price")
        stop = subj.d("stop")
        target = subj.d("resolution_target" if is_t else "target")
        start = subj.i("fill_ts_ms" if is_t else "entry_ts_ms")
        tf_ms = (subj if is_t else by_seq[subj.i("decision_seq")]).i("tf_ms")
        horizon = start + max_bars * tf_ms
        if r.i("horizon_end_ms") != horizon or r.i("resolved_ms") < horizon:
            rep.fail("outcomes", r.seq, "resolved over the wrong horizon, or before it closed")
        if r.c["resolver_role"] == "operator":
            resolved[subj.seq] = r
        if arc is None:
            continue
        sym = (subj if is_t else by_seq[subj.i("decision_seq")]).c["symbol"]
        tf = (subj if is_t else by_seq[subj.i("decision_seq")]).c["timeframe"]
        try:
            bars, files = arc.bars(sym, tf, start, horizon)
        except Unpublished as e:
            rep.warn("outcomes", r.seq, f"archive not published yet ({e})")
            continue
        except (OSError, ValueError) as e:
            # The exchange's file, not the operator's row: a checksum that does not match, or a
            # download that failed, says nothing about this ledger. It is reported and skipped.
            rep.warn("archive", r.seq, f"archive unusable, outcome not checked: {e}")
            continue
        if r.c["barrier"] != "unresolvable" and json.loads(r.c["source_files_json"]) != files:
            rep.fail("outcomes", r.seq, "cites archive files or hashes other than the published ones")
        w = walk(direction, entry, stop, target, bars, start, tf_ms, max_bars)
        if w["barrier"] != r.c["barrier"]:
            rep.fail("outcomes", r.seq, f"barrier {r.c['barrier']}, the archive says {w['barrier']}")
            continue
        if w["barrier"] == "unresolvable":
            rep.ok("outcomes")
            continue
        cost = subj.d("cost_r")
        exp = {"r_realized": f8((w["r"] - cost).quantize(Q8, rounding=ROUND_FLOOR)),
               "mae_r": f8(w["mae"].quantize(Q8, rounding=ROUND_CEILING)),
               "mfe_r": f8(w["mfe"].quantize(Q8, rounding=ROUND_FLOOR)),
               "bars_to_resolve": str(w["bars"]), "intrabar_ambiguous": str(w["amb"])}
        bad = {k: (r.c[k], v) for k, v in exp.items() if r.c[k] != v}
        if bad:
            rep.fail("outcomes", r.seq, f"does not recompute from the archive: {bad}")
        rep.ok("outcomes")
    # Every arm must be scored, or written unscorable, once its horizon is long past. An operator
    # could otherwise simply never write the outcomes that hurt: withheld pairs vanish from every
    # average without a trace. The operator gives the exchange's archive 30 days.
    newest = max((h.i("published_ms") for h in rows if h.kind == "head"), default=0)
    for d in rows:
        if d.kind != "decision" or d.c["verdict"] != "actionable":
            continue
        for arm in (d, by_seq.get(d.seq + 1)):
            if arm is None or arm.seq in resolved:
                continue
            horizon = d.i("fill_ts_ms") + max_bars * d.i("tf_ms")
            overdue = newest - horizon
            if overdue > WITHHELD_AFTER_MS:
                rep.fail("outcomes", arm.seq, f"no outcome {overdue // 86_400_000} days after "
                                              "its horizon closed: withheld")
            elif overdue > OVERDUE_AFTER_MS:
                rep.warn("outcomes", arm.seq, "outcome overdue (the archive may be late)")
    pairs = []
    for d in rows:
        if d.kind == "decision" and d.c["verdict"] == "actionable":
            c = by_seq.get(d.seq + 1)
            if c is not None and d.seq in resolved and c.seq in resolved:
                pairs.append((resolved[d.seq], resolved[c.seq]))
    return pairs


# ---------------------------------------------------------------------------- the summary

def _arm_r(res: Row, arm: Row, m: Decimal) -> Decimal:
    """An arm's R at cost multiplier m: the row carries it net of its modelled cost once."""
    return Decimal(res.c["r_realized"]) - (m - 1) * arm.d("cost_r")


def episodes_of(rows: list[Row], pairs: list[tuple[Row, Row]], rules: Rules,
                cap: int | None, *, until_ms: int | None = None,
                stopped: set[str] | None = None) -> dict[str, list[tuple[Row, Decimal]]]:
    """Each strategy's independent episodes, in chain order, as the decision rule defines them:
    actionable plans that are not late and whose two arms both have a result, taken greedily so
    that no two entries are closer than `eval_min_gap_bars` bars. `cap` stops at that many;
    `until_ms` ignores plans entered at or after it (the deadline). The strategies whose selection
    stopped at a plan still waiting for its outcome are added to `stopped` when it is given: until
    that outcome is written, their count is not final."""
    p = rules.params
    m = Decimal(p["endpoint_cost_multiplier"])
    gap = int(p["eval_min_gap_bars"])
    by_seq = {r.seq: r for r in rows}
    res_of = {t.i("subject_seq"): (t, c) for t, c in pairs}
    out: dict[str, list[tuple[Row, Decimal]]] = {}
    for r in rows:
        if r.kind == "strategy_binding":
            out.setdefault(r.c["strategy_id"], [])
    last: dict[str, int] = {}
    halted: set[str] = set()
    for d in rows:
        if d.kind != "decision" or d.c["verdict"] != "actionable" or d.i("late"):
            continue
        if until_ms is not None and d.i("fill_ts_ms") >= until_ms:
            continue
        sid = d.c["strategy_id"]
        if sid in halted:
            continue
        chosen = out.setdefault(sid, [])
        if cap is not None and len(chosen) >= cap:
            continue
        entry = d.i("fill_ts_ms")
        if sid in last and entry < last[sid] + gap * d.i("tf_ms"):
            continue
        got = res_of.get(d.seq)
        if got is None:
            # It qualifies on timing but has no outcome yet. STOPPING here (rather than skipping
            # to the next) is what keeps a verdict from changing later: an episode can never be
            # inserted behind one already counted.
            halted.add(sid)
            continue
        if got[0].c["r_realized"] is None or got[1].c["r_realized"] is None:
            continue
        t, c = got
        chosen.append((d, _arm_r(t, d, m) - _arm_r(c, by_seq[d.seq + 1], m)))
        last[sid] = entry
    if stopped is not None:
        stopped.update(halted)
    return out


def _bootstrap_block(seed: str, k: int, n_blocks: int) -> int:
    """The committed randomness: a counter through SHA-256, so any language reproduces it."""
    digest = hashlib.sha256(bytes.fromhex(seed) + k.to_bytes(8, "big")).digest()
    return int.from_bytes(digest[:8], "big") % n_blocks


def blocks_of(episodes: list[tuple[int, float]], width_ms: int) -> list[list[float]]:
    """The episodes grouped into fixed blocks of entry time, in order: block index is
    (entry - earliest entry) // width."""
    first = min(t for t, _ in episodes)
    by: dict[int, list[float]] = defaultdict(list)
    for t, x in episodes:
        by[(t - first) // width_ms].append(x)
    return [by[k] for k in sorted(by)]


def family_bound(blocks: list[list[float]], *, seed: str, resamples: int, alpha: float) -> float:
    """The sealed cluster bootstrap: each resample draws as many blocks as there are, with
    replacement, keeps every episode in each drawn block, and averages them; the bound is the
    sorted resampled means at zero-based index floor(alpha * resamples)."""
    n = len(blocks)
    means = []
    k = 0
    for _ in range(resamples):
        pool: list[float] = []
        for _ in range(n):
            pool += blocks[_bootstrap_block(seed, k, n)]
            k += 1
        means.append(sum(pool) / len(pool))
    means.sort()
    return means[int(alpha * resamples)]


def evaluate_family(rows: list[Row], pairs: list[tuple[Row, Row]], rules: Rules) -> dict:
    """THE PRIMARY ENDPOINT: did the 88, as a family, choose sides better than a coin?

    The pooled mean paired difference over every strategy's episodes entered before the deadline,
    with a one-sided 95% lower bound from a cluster bootstrap over 8-day blocks of entry time — the
    length of a position, so an episode can overlap only episodes in its own block or an adjacent
    one. Evaluated once: after the deadline, when every one of those episodes has its outcome.
    """
    p = rules.params
    alpha = float(p["eval_family_alpha"])
    width = int(p["eval_block_days"]) * DAY
    resamples = int(p["eval_bootstrap_resamples"])
    seed = p["eval_bootstrap_seed"]
    deadline = rules.effective_from_ms + int(p["eval_deadline_after_effective_ms"])
    newest = max((h.i("published_ms") for h in rows if h.kind == "head"), default=0)
    waiting: set[str] = set()
    chosen = episodes_of(rows, pairs, rules, None, until_ms=deadline, stopped=waiting)
    eps = [(d.i("fill_ts_ms"), float(x)) for xs in chosen.values() for d, x in xs]
    n = len(eps)
    if not n:
        return {"episodes": 0, "strategies": 0, "mean_diff_r": None, "lower_bound_r": None,
                "blocks": 0, "status": "accruing"}
    blocks = blocks_of(eps, width)
    mean = sum(x for _, x in eps) / n
    lower = (family_bound(blocks, seed=seed, resamples=resamples, alpha=alpha)
             if newest >= deadline and not waiting else None)
    strategies = sum(1 for xs in chosen.values() if xs)
    status = ("accruing" if lower is None else
              "family_edge" if lower > 0 else "no_evidence")
    return {"episodes": n, "strategies": strategies, "mean_diff_r": mean, "lower_bound_r": lower,
            "blocks": len(blocks), "status": status, "deadline_ms": deadline,
            "waiting": newest >= deadline and bool(waiting)}


def evaluate(rows: list[Row], pairs: list[tuple[Row, Row]], rules: Rules) -> dict:
    """THE REFERENCE IMPLEMENTATION of the pre-registered decision rule, per strategy.

    Episodes: a strategy's actionable plans in chain order that are not late and whose treatment
    and control both have a result; the first such plan, then each next one entering at least
    `eval_min_gap_bars` bars after the previous episode's entry. Statistic: mean d and sample sd s
    of the paired differences at the endpoint's cost multiplier over the first `eval_episodes`.
    Verdict: forward edge only if d - t_crit * s / sqrt(n) > 0; 'mostly one-sided' when more than
    `eval_one_sided_max_share` of the episodes are on one side; 'insufficient sample' when fewer
    were entered before the deadline and all of their outcomes are written; 'accruing' until then.
    Plans entered after the deadline are never counted. The pre-registration's text governs where
    this disagrees."""
    p = rules.params
    need = int(p["eval_episodes"])
    t_crit = float(p["eval_t_critical"])
    one_sided = float(p["eval_one_sided_max_share"])
    deadline = rules.effective_from_ms + int(p["eval_deadline_after_effective_ms"])
    newest = max((h.i("published_ms") for h in rows if h.kind == "head"), default=0)
    res_of = {t.i("subject_seq"): (t, c) for t, c in pairs}
    waiting: set[str] = set()
    chosen = episodes_of(rows, pairs, rules, need, until_ms=deadline, stopped=waiting)
    excluded_late: dict[str, int] = defaultdict(int)
    excluded_unscorable: dict[str, int] = defaultdict(int)
    for d in rows:
        if (d.kind == "decision" and d.c["verdict"] == "actionable"
                and d.i("fill_ts_ms") < deadline):
            if d.i("late"):
                excluded_late[d.c["strategy_id"]] += 1
            else:
                got = res_of.get(d.seq)
                if got is not None and (got[0].c["r_realized"] is None
                                        or got[1].c["r_realized"] is None):
                    excluded_unscorable[d.c["strategy_id"]] += 1
    out = {}
    for sid, episodes in sorted(chosen.items()):
        late, unscorable = excluded_late[sid], excluded_unscorable[sid]
        out[sid] = {**verdict([float(x) for _, x in episodes],
                              [d.i("direction") for d, _ in episodes], need=need, t_crit=t_crit,
                              one_sided=one_sided,
                              deadline_passed=newest >= deadline and sid not in waiting),
                    "late_excluded": late, "unscorable_excluded": unscorable}
    return out


def verdict(diffs: list[float], sides: list[int], *, need: int, t_crit: float,
            one_sided: float, deadline_passed: bool) -> dict:
    """The decision rule's arithmetic on one strategy's episodes, in order."""
    n = len(diffs)
    mean = sum(diffs) / n if n else None
    sd = (sum((x - mean) ** 2 for x in diffs) / (n - 1)) ** 0.5 if n > 1 else None
    p_long = sum(1 for x in sides if x == 1) / n if n else None
    lower = mean - t_crit * sd / n ** 0.5 if n == need and sd is not None else None
    if n == need:
        if max(p_long, 1 - p_long) > one_sided:
            status = "mostly_one_sided"
        elif lower is not None and lower > 0:
            status = "forward_edge"
        else:
            status = "no_evidence"
    elif deadline_passed:
        status = "insufficient_sample"
    else:
        status = "accruing"
    return {"episodes": n, "needed": need, "p_long": p_long, "mean_diff_r": mean,
            "lower_bound_r": lower, "status": status}


def summarise(rows: list[Row], pairs: list[tuple[Row, Row]], rules: Rules) -> dict:
    """The evaluation (the pre-registered endpoint, per strategy), and beside it a DESCRIPTIVE
    pooling of every resolved pair — overlapping positions, every strategy, iid standard error.
    The pooled numbers are not the endpoint and must never be read as a test."""
    by_seq = {r.seq: r for r in rows}
    m = Decimal(rules.params.get("endpoint_cost_multiplier", "1"))
    per: dict[str, list[tuple[Decimal, Decimal]]] = defaultdict(list)
    for t, c in pairs:
        if t.c["r_realized"] is None or c.c["r_realized"] is None:
            continue
        ta, ca = by_seq[t.i("subject_seq")], by_seq[c.i("subject_seq")]
        per[ta.c["strategy_id"]].append((_arm_r(t, ta, m), _arm_r(c, ca, m)))

    def stats(xs: list[tuple[Decimal, Decimal]]) -> dict:
        n = len(xs)
        diffs = [float(a - b) for a, b in xs]
        mean = sum(diffs) / n
        sd = (sum((x - mean) ** 2 for x in diffs) / (n - 1)) ** 0.5 if n > 1 else float("nan")
        return {"pairs": n, "mean_r_plan": float(sum(a for a, _ in xs) / n),
                "mean_r_control": float(sum(b for _, b in xs) / n), "mean_edge_r": mean,
                "se_edge_r": sd / n ** 0.5 if n > 1 else None}

    verdicts: dict[str, int] = defaultdict(int)
    for r in rows:
        if r.kind == "decision":
            verdicts[r.c["verdict"]] += 1
    all_pairs = [p for xs in per.values() for p in xs]
    return {"verdicts": dict(verdicts), "cost_multiplier": str(m),
            "family": evaluate_family(rows, pairs, rules),
            "evaluation": evaluate(rows, pairs, rules),
            "all": stats(all_pairs) if all_pairs else None,
            "by_strategy": {k: stats(v) for k, v in sorted(per.items())}}


def _iso(ms: int) -> str:
    return datetime.fromtimestamp(ms / 1000, tz=UTC).strftime("%Y-%m-%d %H:%M UTC")


# ---------------------------------------------------------------------------- main

def _note_verdicts(rows: list[Row], rep: Report) -> None:
    """How many answers of each kind, and why each abstention says it abstained. Reported, not
    judged: a no-trade or an abstention is the operator's own account, and proving any one of them
    would need the strategies and the feed re-run (see WHAT IT DOES NOT). A run of them on one bar
    is at least visible here."""
    kinds: dict[str, int] = defaultdict(int)
    why: dict[str, int] = defaultdict(int)
    for r in rows:
        if r.kind == "decision":
            kinds[r.c["verdict"]] += 1
            if r.c["verdict"] == "abstained":
                why[r.c.get("abstain_code") or "?"] += 1
    if kinds:
        rep.notes.append("decisions: " + ", ".join(f"{k} {v}" for k, v in sorted(kinds.items()))
                         + ("" if not why else " (abstained: " + ", ".join(
                             f"{k} {v}" for k, v in sorted(why.items())) + ")"))


def verify(ledger_dir: Path, *, archive: Archive | None,
           headers: Path | None = None) -> tuple[Report, dict]:
    rep = Report()
    if not (ledger_dir / "segments").is_dir():
        rep.fail("segments", None, f"{ledger_dir} is not a checkout of the ledger: it has no "
                                   "segments/ directory")
        return rep, {}
    rows, segs = load(ledger_dir, rep)
    if not rows:
        rep.notes.append("no rows published yet")
        return rep, {}
    check_chain(rows, rep)
    rules = read_genesis(rows, rep)
    check_heads(rows, segs, ledger_dir, rep, rules)
    check_proofs(rows, segs, rep, online=archive is not None, headers=headers)
    if rules is None:
        return rep, {}
    me = sha256(Path(__file__).read_bytes())
    if me == rules.verifier_sha256:
        rep.notes.append(f"this is the verifier build the ledger committed to ({me[:12]})")
    else:
        rep.notes.append(f"this verifier ({me[:12]}) is not the build committed at genesis "
                         f"({rules.verifier_sha256[:12]}); any verifier may be used")
    check_manifests(rows, rules, rep)
    _note_verdicts(rows, rep)
    check_controls(rows, rules, rep)
    if archive is not None:
        check_plans(rows, rules, archive, rep)
    pairs = check_resolutions(rows, rules, archive, rep)
    return rep, summarise(rows, pairs, rules)


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("ledger", type=Path, help="a checkout of the wavelab-ledger repository")
    ap.add_argument("--offline", action="store_true", help="skip every check that needs the archive")
    ap.add_argument("--archive-cache", type=Path, default=Path(".archive-cache"))
    ap.add_argument("--archive-mirror", type=Path, help="read archives from a local mirror")
    ap.add_argument("--json", type=Path, help="also write the report and summary as JSON")
    ap.add_argument("--block-headers", type=Path,
                    help="read Bitcoin block headers from <dir>/<height>.hex instead of explorers")
    a = ap.parse_args(argv)
    arc = None if a.offline else Archive(a.archive_cache, a.archive_mirror)
    rep, summary = verify(a.ledger, archive=arc, headers=a.block_headers)
    print(f"wavelab-verify {VERIFIER_VERSION}")
    for n in rep.notes:
        print(f"  note  {n}")
    for check, n in sorted(rep.checked.items()):
        print(f"  ok    {check:<10} {n}")
    for check, seq, msg in rep.warns:
        print(f"  WARN  {check:<10} {'' if seq is None else seq}  {msg}")
    for check, seq, msg in rep.fails:
        print(f"  FAIL  {check:<10} {'' if seq is None else seq}  {msg}")
    fam = summary.get("family") or {}
    if fam.get("episodes"):
        print(f"\n  the PRIMARY endpoint, the family: {fam['episodes']} episodes from "
              f"{fam['strategies']} strategies in {fam['blocks']} blocks · mean difference "
              f"{fam['mean_diff_r']:+.4f} R · "
              + (f"lower bound {fam['lower_bound_r']:+.4f} R → {fam['status']}"
                 if fam["lower_bound_r"] is not None else
                 f"no verdict yet: waiting for the outcomes of plans entered before "
                 f"{_iso(fam['deadline_ms'])}" if fam.get("waiting") else
                 f"no verdict before {_iso(fam['deadline_ms'])}"))
    ev = summary.get("evaluation") or {}
    if ev:
        counts: dict[str, int] = defaultdict(int)
        for e in ev.values():
            counts[e["status"]] += 1
        print("\n  the SECONDARY endpoint, per strategy (underpowered by design): "
              + ", ".join(f"{k} {v}" for k, v in sorted(counts.items())))
        for sid, e in sorted(ev.items()):
            if e["status"] in ("forward_edge", "no_evidence", "mostly_one_sided"):
                print(f"    {sid:<40} {e['status']:<18} d {e['mean_diff_r']:+.3f} R, "
                      f"lower bound {e['lower_bound_r']:+.3f} R, long {e['p_long']:.0%}")
    if summary.get("all"):
        s = summary["all"]
        print(f"\n  descriptive only, not the endpoint: {s['pairs']} resolved pairs, all "
              f"strategies, overlapping, at {summary['cost_multiplier']}x cost · plan "
              f"{s['mean_r_plan']:+.4f} R · control {s['mean_r_control']:+.4f} R · difference "
              f"{s['mean_edge_r']:+.4f} R")
    print(f"\n  {'FAILED' if rep.fails else 'PASSED'}: {len(rep.fails)} failures, "
          f"{len(rep.warns)} warnings")
    if a.json:
        a.json.write_text(json.dumps({"fails": rep.fails, "warns": rep.warns, "notes": rep.notes,
                                      "checked": rep.checked, "summary": summary}, indent=2))
    return 1 if rep.fails else 0


if __name__ == "__main__":
    sys.exit(main())