"""Proof serving — find a commitment and PROVE where it belongs.

Gap-to-done #3/#4 (NEOva2 side only): a read-only service over the
durable sources — the RECORD BOOK (confirmed range records + witness
attestations + tier records), BOTH ARCHIVE holders (the raw ordered book
of commitments; one may burn or be hacked — serving falls over to the
other and any disagreement is LOUD, never silent), and the SEQUENCER LOG
(batch identity ↔ position range — the dispute path: Cotrugli Ledger /
NEO Chamber ask "the disputed transaction of firm A and B landed in batch
so-and-so — prove it"). It never touches the hot ingest path.

`ProofStore.bundle_for(commitment)` returns a self-contained PROOF
BUNDLE: the NEOva position, the carrying BATCH identity, the covering
range record with BOTH witness attestations (base + Witness-about-Witness
— two DIFFERENT qualified masters), the qualified public keys, a Merkle
membership proof of the commitment in the range's object root, and the
chained ladder range → hour → day → quarter → year up to the HIGHEST TIER
ACTUALLY CLOSED — open tiers are named, never claimed. No EraSeal is ever
claimed unless a real seal exists (none is automated today).

`verify_bundle(bundle, trusted_registry=None)` is a PURE function of the
bundle (plus an optional PINNED registry): an auditor / the Ledger / the
Chamber confirms membership, both signatures and every ladder step with
no access to the live engine. The embedded registry is convenience; the
real trust anchor travels to the verifier THROUGH Cotrugli Ledger (and
the Chamber for notarized documents) — and becomes POHAT's QWS chain
(QWS(E) bound from EraSeal(E-1), pinned genesis) once eras are live.

The INDEX (kazalo) is extensible: each rubric is a NAMED builder over the
durable sources, registered in `INDEX_BUILDERS` — new rubrics are added
by registering a builder, never by changing the durable formats. All
indexes are derived, disposable and rebuildable.

Fail-closed everywhere: unknown commitment, an unwitnessed position, a
broken prev-hash chain, book corruption, archive disagreement that no
healthy holder can resolve, or an invalid signature all REFUSE.
"""

from __future__ import annotations

import bisect
import sqlite3
from array import array
import struct
import threading
import zlib
from pathlib import Path

from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PublicKey)

from . import closing as CL
from . import witness as W
from .recordbook import RecordBook, RecordBookCorrupt

FORMAT = "neova2-proof-bundle-v1"
HOUR_FORMAT = "neova2-hour-view-v1"
_TIERS_UP = ["HOUR", "DAY", "QUARTER", "ANNUAL"]

# archive frame layout (mirrors archive_holder; read-only parse here)
_A_MAGIC = b"NVAR"
_A_HDR = struct.Struct(">4sQI")
_A_LEN = struct.Struct(">I")
_A_CRC = struct.Struct(">I")
# sequencer log frame layout (mirrors sequencer; read-only parse here)
_S_MAGIC = b"NVQ1"
_S_REC = struct.Struct(">4s8sQQIQ")
OBJ = 32


class ProofError(Exception):
    """Fail-closed refusal with a stable reason string."""


def _hx(b: bytes) -> str:
    return b.hex()


def _period_start(clock: dict, period_id: int) -> str | None:
    """Derived wall time of an HOUR period from the DECLARED clock; None when the
    declaration is incomplete. Never used for verification."""
    try:
        import datetime as _dt
        start = _dt.datetime.fromisoformat(clock["epoch_start_utc"].replace("Z", "+00:00"))
        t = start + _dt.timedelta(seconds=int(clock["seconds_per_hour"]) * int(period_id))
        return t.strftime("%Y-%m-%dT%H:%M:%SZ")
    except (KeyError, ValueError, TypeError):
        return None


def _tier_record_dict(trec) -> dict:
    return {"tier": trec.tier, "period_id": trec.period_id,
            "first_position": trec.first_position,
            "last_position": trec.last_position,
            "object_count": trec.object_count,
            "n_children": trec.n_children,
            "root": _hx(trec.root), "prev_hash": _hx(trec.prev_hash)}


def _tier_record_from(tr: dict):
    return CL.TierRecord(tr["tier"], tr["period_id"], tr["first_position"],
                         tr["last_position"], tr["object_count"],
                         tr["n_children"], _unhx(tr["root"]), _unhx(tr["prev_hash"]))


def _unhx(s: str) -> bytes:
    return bytes.fromhex(s)


# --------------------------------------------------------------- the kazalo
def _build_by_commitment(store):
    """commitment -> position. Derived by scanning the healthy archive
    once; disposable and rebuildable — the durable format is untouched."""
    idx = {}
    for first, count, blob in store._iter_archive_blobs():
        for j in range(count):
            idx[blob[j * OBJ:(j + 1) * OBJ]] = first + j
    return idx


def _build_by_batch(store):
    """(gateway_id, epoch, batch_seq) -> (position_start, count) — the
    dispute rubric: 'the disputed transaction landed in batch so-and-so'."""
    idx = {}
    for gid, epoch, seq, count, start in store._iter_seq_log():
        idx[(gid, epoch, seq)] = (start, count)
    return idx


def _build_by_position_batch(store):
    """sorted position starts -> batch identity (find the batch that
    carries a position)."""
    rows = sorted((start, count, (gid, epoch, seq))
                  for gid, epoch, seq, count, start
                  in store._iter_seq_log())
    return rows


def _build_by_hour(store):
    """hour period_id -> its tier record (and likewise usable for
    position -> hour via the record's position span)."""
    return {t.period_id: t for t in store._tiers["HOUR"]}


def _build_by_day(store):
    return {t.period_id: t for t in store._tiers["DAY"]}


INDEX_BUILDERS = {
    "by_commitment": _build_by_commitment,
    "by_batch": _build_by_batch,
    "by_position_batch": _build_by_position_batch,
    "by_hour": _build_by_hour,
    "by_day": _build_by_day,
}


def register_index(name, builder):
    """New rubric = new named builder over the durable sources. Nothing
    hardcoded; nothing about the durable formats changes."""
    INDEX_BUILDERS[name] = builder


class ProofIndex:
    """Persistent commitment -> position index (sqlite, WAL).

    Derived, disposable, rebuildable — the durable archive format is untouched.
    INCREMENTAL: it remembers the highest archive position it has indexed and
    resumes from there, so a new object costs one small insert, never a rescan
    of history. Until the first backfill finishes (`ready()` False) lookups
    fall back to the archive tail-scan, exactly the historical behaviour, and
    the on-disk B-tree is created only once at the end of the backfill (bulk
    inserts by ascending position are sequential; building the commitment
    B-tree afterwards is one sort instead of 270M random inserts).

    "Last duplicate wins" is preserved by MAX(position). The index yields a
    CANDIDATE position only; `bundle_for` re-reads the range from an archive
    and re-derives the witnessed root, so a corrupt index cannot forge a
    proof — it can only be caught.
    """

    def __init__(self, path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        # The writer lock serialises WRITES only. Status reads (`ready`,
        # `indexed_max_position`) come from cached attributes and lookups open
        # their own WAL reader connection, so neither /health nor a proof
        # lookup can ever wait behind a multi-minute backfill or B-tree build.
        self._wlock = threading.Lock()
        self._writer = self._connect()
        with self._wlock:
            self._writer.executescript(
                "PRAGMA journal_mode=WAL;"
                "PRAGMA synchronous=NORMAL;"
                "CREATE TABLE IF NOT EXISTS meta ("
                "  key TEXT PRIMARY KEY, value INTEGER NOT NULL);"
                "CREATE TABLE IF NOT EXISTS objects ("
                "  position INTEGER PRIMARY KEY, commitment BLOB NOT NULL);"
                "INSERT OR IGNORE INTO meta VALUES ('indexed_max_position', -1);"
                "INSERT OR IGNORE INTO meta VALUES ('ready', 0);")
            # The WAL must not grow without bound: cap it after a checkpoint
            # (64 MB) and checkpoint explicitly after every backfill batch and
            # after the B-tree build. Found on neo-4 on 2026-09-02: an 11.5 GB
            # WAL beside a 22.8 GB index, because long-lived readers kept the
            # auto-checkpoint from ever truncating it.
            self._writer.execute("PRAGMA journal_size_limit=67108864")
            self._head = self._meta(self._writer, "indexed_max_position")
            self._ready = self._meta(self._writer, "ready") == 1

    def _connect(self):
        conn = sqlite3.connect(str(self.path), timeout=30,
                               check_same_thread=False, isolation_level=None)
        conn.execute("PRAGMA journal_mode=WAL")
        return conn

    def _meta(self, conn, key):
        row = conn.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone()
        return int(row[0]) if row else None

    def indexed_max_position(self) -> int:
        return self._head                         # cached; never waits on a writer

    def ready(self) -> bool:
        return self._ready                        # cached; never waits on a writer

    def count(self) -> int:
        conn = self._connect()
        try:
            return conn.execute("SELECT COUNT(*) FROM objects").fetchone()[0]
        finally:
            conn.close()

    def add_frames(self, archive_path, frames, max_objects=2_000_000) -> int:
        """Index frames (first, count, payload_offset) in ascending order,
        skipping anything at or below the indexed head. Bounded per call so the
        caller (a background thread) yields regularly. Returns objects indexed."""
        head = self.indexed_max_position()
        todo = [f for f in frames if f[0] + f[1] - 1 > head]
        if not todo:
            return 0
        done = 0
        with open(archive_path, "rb") as source, self._wlock:
            self._writer.execute("BEGIN")
            try:
                for first, count, boff in todo:
                    if done >= max_objects:
                        break
                    source.seek(boff)
                    blob = source.read(count * OBJ)
                    if len(blob) != count * OBJ:
                        raise ProofError("archive hole during indexing")
                    start = max(0, head + 1 - first)      # partial overlap
                    self._writer.executemany(
                        "INSERT OR REPLACE INTO objects VALUES (?, ?)",
                        ((first + j, blob[j * OBJ:(j + 1) * OBJ])
                         for j in range(start, count)))
                    head = first + count - 1
                    done += count - start
                self._writer.execute(
                    "UPDATE meta SET value=? WHERE key='indexed_max_position'",
                    (head,))
                self._writer.execute("COMMIT")
                self._head = head
            except BaseException:
                self._writer.execute("ROLLBACK")
                raise
            self._writer.execute("PRAGMA wal_checkpoint(PASSIVE)")
        return done

    def finalize(self):
        """Build the commitment B-tree once the backfill has caught up. Holds
        the WRITER lock for minutes on a large archive; readers and status
        never touch it, so the plane stays responsive throughout."""
        with self._wlock:
            self._writer.execute(
                "CREATE INDEX IF NOT EXISTS by_commitment ON objects(commitment)")
            self._writer.execute(
                "UPDATE meta SET value=1 WHERE key='ready'")
            self._ready = True
            self._writer.execute("PRAGMA wal_checkpoint(TRUNCATE)")

    def wal_bytes(self) -> int:
        """Current size of the WAL file — an operational meter, 0 when absent."""
        wal = self.path.with_name(self.path.name + "-wal")
        return wal.stat().st_size if wal.exists() else 0

    def lookup(self, commitment: bytes):
        """MAX(position) for the commitment, or None. Own connection per call
        so readers never queue behind the backfill writer (WAL)."""
        conn = self._connect()
        try:
            row = conn.execute(
                "SELECT MAX(position) FROM objects WHERE commitment=?",
                (commitment,)).fetchone()
            return None if row is None or row[0] is None else int(row[0])
        finally:
            conn.close()


class ProofStore:
    """Read-only view over one record book + BOTH archives + the
    sequencer log. Validates every prev-hash chain at load (fail-closed)
    and mirrors the brain's folding rules exactly."""

    def __init__(self, book_path, archive_paths, qualified,
                 seq_log_path=None, index_path=None, clock=None):
        self.book_path = Path(book_path)
        if isinstance(archive_paths, (str, Path)):
            archive_paths = [archive_paths]
        self.archive_paths = [Path(p) for p in archive_paths]
        self.seq_log_path = Path(seq_log_path) if seq_log_path else None
        self.qualified = dict(qualified)
        self.archive_disagreement = False        # LOUD, never silent
        self._healthy_archive = None
        self._indexes = {}
        self._source_signature = None
        self._arch_state = {}                    # path -> incremental scan cache
        self._book_state = None                  # incremental book replay cache
        # DECLARED (never attested) mapping from HOUR period ids to wall time:
        # {"seconds_per_hour": int, "epoch_start_utc": str}. The book carries no
        # wall clock; period_id counts the brain's HOUR windows since its epoch.
        self.clock = dict(clock) if clock else None
        self.last_book_load = {"mode": "none", "entries": 0}
        self.index = None if index_path is None else ProofIndex(index_path)
        self.refresh()

    # ------------------------------------------------------------- loading
    def refresh(self):
        signature = self._files_signature()
        if signature == self._source_signature:
            return
        # `live` is commonly an atomically flipped snapshot symlink. Refuse a
        # mixed-generation view: all sources must retain the same identity while
        # the indexes are rebuilt. One retry handles a legitimate flip during a
        # refresh; a continuously moving source is weather, not proof.
        for _ in range(2):
            self._load_book()
            self._scan_archives()
            self._indexes.clear()                # derived: rebuild lazily
            after = self._files_signature()
            if after == signature:
                self._source_signature = after
                return
            signature = after
        raise ProofError("proof sources changed during refresh")

    def _files_signature(self):
        paths = [self.book_path, *self.archive_paths]
        if self.seq_log_path is not None:
            paths.append(self.seq_log_path)
        signatures = []
        for path in paths:
            try:
                stat = path.stat()
                signatures.append((stat.st_dev, stat.st_ino, stat.st_size,
                                   stat.st_mtime_ns))
            except FileNotFoundError:
                # Existing readers treat a not-yet-created archive or optional
                # sequencer log as an empty source; retain that fail-closed path
                # and merely make absence part of the cache identity.
                signatures.append(None)
        return tuple(signatures)

    def kazalo(self, name):
        if name not in self._indexes:
            self._indexes[name] = INDEX_BUILDERS[name](self)
        return self._indexes[name]

    # The book is replayed INCREMENTALLY and kept OUT of RAM: per confirmed
    # record we hold three integers (range start, count, byte offset) plus its
    # HOUR assignment; the record and both attestations are re-read from the
    # book by offset when a bundle needs them. Tier records (bounded by the
    # clock, not by traffic) stay in RAM together with their child hashes,
    # except HOUR children, which are re-derived from the records on demand.
    # Measured 2026-09-04: the old whole-book replay cost ~17 µs and ~1.4 KB
    # of RAM per seal, on EVERY refresh — a never-reset book made the proof
    # plane slower with every close, at zero traffic (neova2 #17).
    def _fresh_book_state(self, ino):
        return {"ino": ino, "end": 0, "last_off": -1, "last_frame": b"",
                "prev": {"RANGE": b"\x00" * 32,
                         **{t: b"\x00" * 32 for t in _TIERS_UP}},
                "open_first": 0, "open_hashes": [],
                "pend": {"HOUR": [], "DAY": [], "QUARTER": []},
                "range_starts": array("q"), "range_counts": array("q"),
                "range_offsets": array("q"),
                "rec_hour": array("q"), "rec_j": array("q"),
                "tiers": {t: [] for t in _TIERS_UP},
                "hour_span": [],                 # hour idx -> (first_rec, n)
                "tier_children": {}, "parent_of": {}}

    def _absorb(self, bs, entry, end):
        """Fold ONE book entry into the chain state; raises ProofError on a
        broken chain or a tier whose children do not re-derive its root."""
        if entry[0] == "CONFIRMED":
            rec = entry[1]
            if rec.prev_record_hash != bs["prev"]["RANGE"]:
                raise ProofError("range-record chain broken")
            h = rec.record_hash()
            bs["prev"]["RANGE"] = h
            bs["range_starts"].append(rec.range_start)
            bs["range_counts"].append(rec.count)
            bs["range_offsets"].append(bs["end"])
            bs["rec_hour"].append(-1)
            bs["rec_j"].append(-1)
            bs["open_hashes"].append(h)
        elif entry[0] == "TIER":
            t = entry[1]
            if t.prev_hash != bs["prev"][t.tier]:
                raise ProofError(f"{t.tier} chain broken")
            bs["prev"][t.tier] = t.record_hash()
            if t.tier == "HOUR":
                hashes = bs["open_hashes"]
            elif t.tier == "DAY":
                hashes = [c.record_hash() for c in bs["pend"]["HOUR"]]
            elif t.tier == "QUARTER":
                hashes = [c.record_hash() for c in bs["pend"]["DAY"]]
            else:
                hashes = [c.record_hash() for c in bs["pend"]["QUARTER"]]
            if t.n_children != len(hashes) or (
                    hashes and CL.merkle_root(hashes) != t.root):
                raise ProofError(
                    f"{t.tier} children do not re-derive its root")
            idx = len(bs["tiers"][t.tier])
            bs["tiers"][t.tier].append(t)
            if t.tier == "HOUR":
                first = bs["open_first"]
                for j in range(len(hashes)):
                    bs["rec_hour"][first + j] = idx
                    bs["rec_j"][first + j] = j
                bs["hour_span"].append((first, len(hashes)))
                bs["open_first"] = len(bs["range_starts"])
                bs["open_hashes"] = []
                bs["pend"]["HOUR"].append(t)
            else:
                bs["tier_children"][(t.tier, idx)] = hashes
                for j, h in enumerate(hashes):
                    bs["parent_of"][h] = (t.tier, idx, j)
                if t.tier == "DAY":
                    bs["pend"]["HOUR"] = []
                    bs["pend"]["DAY"].append(t)
                elif t.tier == "QUARTER":
                    bs["pend"]["DAY"] = []
                    bs["pend"]["QUARTER"].append(t)
                else:
                    bs["pend"]["QUARTER"] = []
        bs["last_off"] = bs["end"]
        bs["end"] = end

    def _same_book(self, bs, ino, size) -> bool:
        """Is the file at book_path a continuation of the replayed state?
        Same inode and not shrunk -> yes. A NEW inode (the proof plane reads
        an atomically flipped snapshot copy, so every sync is a new file) is
        accepted when the last absorbed frame is byte-identical at its
        offset in the new file; anything else -> full replay, fail-closed.
        The tail's own chain validation then guards the rest."""
        if bs is None or size < bs["end"]:
            return False
        if bs["ino"] == ino:
            return True
        if bs["last_off"] < 0:
            return size == 0 and bs["end"] == 0
        frame = RecordBook.frame_bytes_at(self.book_path, bs["last_off"])
        if frame and frame == bs["last_frame"]:
            bs["ino"] = ino
            return True
        return False

    def _load_book(self):
        """Incremental: same inode and not shrunk -> replay only the new
        tail with the chain state carried over; a replaced or truncated
        book -> full replay from zero. Fail-closed on any broken chain."""
        try:
            st = self.book_path.stat()
            ino, size = (st.st_dev, st.st_ino), st.st_size
        except FileNotFoundError:
            ino, size = None, 0
        bs = self._book_state
        mode = "incremental"
        if not self._same_book(bs, ino, size):
            bs = self._book_state = self._fresh_book_state(ino)
            mode = "full"
        n = 0
        if size > bs["end"]:
            try:
                for entry, end in RecordBook.replay_from(self.book_path,
                                                         bs["end"]):
                    self._absorb(bs, entry, end)
                    bs["last_frame"] = RecordBook.frame_bytes_at(
                        self.book_path, bs["last_off"])
                    n += 1
            except RecordBookCorrupt as e:
                raise ProofError(f"record book corrupt: {e}")
        self._range_starts = bs["range_starts"]
        self._tiers = bs["tiers"]
        self._tier_children = bs["tier_children"]
        self._parent_of = bs["parent_of"]
        self.last_book_load = {"mode": mode, "entries": n,
                               "records": len(bs["range_starts"]),
                               "book_end": bs["end"]}

    # ------------------------------------------------ records, read on demand
    def _confirmed_count(self) -> int:
        return len(self._book_state["range_starts"])

    def _confirmed_at(self, i: int):
        """(ClosedRecord, base, waw) for confirmed record i, re-read from the
        book by offset — never held in RAM."""
        entry = RecordBook.read_entry_at(self.book_path,
                                         self._book_state["range_offsets"][i])
        if entry[0] != "CONFIRMED":
            raise ProofError("book entry at offset is not a confirmed record")
        return entry[1], entry[2], entry[3]

    def _record_hashes(self, first: int, n: int) -> list:
        return [self._confirmed_at(first + k)[0].record_hash()
                for k in range(n)]

    @staticmethod
    def _scan_one_archive(path):
        """Full scan (compatibility entry point) — see `_scan_archive_from`."""
        return ProofStore._scan_archive_from(path)[0]

    @staticmethod
    def _scan_archive_from(path, start_off=0, frames=None):
        """Index archive frame offsets without materializing the archive,
        INCREMENTALLY from `start_off` (the end of the last validated frame).

        Archive payloads are the durable bulk record and can be many gigabytes.
        The proof index needs only each frame's ``(first, count, payload_offset)``;
        CRC validation is therefore streamed in bounded chunks. Archives are
        append-only, so a later refresh validates only the new tail — a full
        rescan (22 s per 8.6 GB holder) happens once per process, not once per
        lookup. Returns ``(frames, validated_end)``; bytes beyond
        ``validated_end`` (an incomplete or invalid tail frame) are never served —
        fail-closed — and are re-examined on the next refresh.
        """
        frames = [] if frames is None else frames
        if not path.exists():
            return frames, 0
        validated_end = start_off
        with open(path, "rb") as source:
            source.seek(start_off)
            while True:
                off = source.tell()
                length_raw = source.read(_A_LEN.size)
                if not length_raw:
                    break
                if len(length_raw) != _A_LEN.size:
                    break
                (plen,) = _A_LEN.unpack(length_raw)
                if plen < _A_HDR.size:
                    break
                header = source.read(_A_HDR.size)
                if len(header) != _A_HDR.size:
                    break
                magic, first, count = _A_HDR.unpack(header)
                if magic != _A_MAGIC or _A_HDR.size + count * OBJ != plen:
                    break

                crc_actual = zlib.crc32(length_raw)
                crc_actual = zlib.crc32(header, crc_actual)
                remaining = plen - _A_HDR.size
                while remaining:
                    chunk = source.read(min(1024 * 1024, remaining))
                    if not chunk:
                        remaining = -1
                        break
                    crc_actual = zlib.crc32(chunk, crc_actual)
                    remaining -= len(chunk)
                if remaining != 0:
                    break
                crc_raw = source.read(_A_CRC.size)
                if len(crc_raw) != _A_CRC.size:
                    break
                (crc_expected,) = _A_CRC.unpack(crc_raw)
                if crc_actual & 0xFFFFFFFF != crc_expected:
                    break
                frames.append((first, count,
                               off + _A_LEN.size + _A_HDR.size))
                validated_end = source.tell()
        frames.sort()
        return frames, validated_end

    def _scan_archives(self):
        """Incremental: same inode and not shrunk -> scan only the new tail;
        anything else (replaced file, truncation) -> full rescan, fail-closed."""
        arch = []
        for p in self.archive_paths:
            state = self._arch_state.get(p)
            try:
                st = p.stat()
            except FileNotFoundError:
                frames, end, ino = [], 0, None
            else:
                ino = (st.st_dev, st.st_ino)
                if state and state["ino"] == ino and st.st_size >= state["end"]:
                    frames, end = self._scan_archive_from(
                        p, state["end"], list(state["frames"]))
                else:
                    frames, end = self._scan_archive_from(p)
            self._arch_state[p] = {"ino": ino, "frames": frames, "end": end}
            arch.append((p, frames))
        self._arch = arch

    # ------------------------------------------------------- persistent index
    def archive_head_position(self) -> int:
        """Highest position present in the healthiest archive, -1 if empty."""
        _, frames = self._pick_archive()
        return max((f + c - 1 for f, c, _ in frames), default=-1)

    def index_catch_up(self, max_objects=2_000_000) -> int:
        """Advance the persistent index toward the archive head (bounded).
        Returns objects indexed; 0 means caught up."""
        if self.index is None:
            return 0
        path, frames = self._pick_archive()
        return self.index.add_frames(path, frames, max_objects=max_objects)

    def index_status(self) -> dict:
        if self.index is None:
            return {"enabled": False}
        return {"enabled": True, "ready": self.index.ready(),
                "indexed_max_position": self.index.indexed_max_position(),
                "archive_head_position": self.archive_head_position()}

    def _iter_archive_blobs(self):
        """Yield (first, count, blob) from the FIRST archive that serves
        each frame; disagreement between holders is flagged LOUDLY."""
        p, frames = self._pick_archive()
        with open(p, "rb") as f:
            for first, count, boff in frames:
                f.seek(boff)
                yield first, count, f.read(count * OBJ)

    def _pick_archive(self):
        if not self._arch:
            raise ProofError("no archive available")
        best = max(self._arch, key=lambda a: sum(c for _, c, _ in a[1]))
        return best

    def _iter_seq_log(self):
        if self.seq_log_path is None or not self.seq_log_path.exists():
            return
        data = self.seq_log_path.read_bytes()
        off, n = 0, len(data)
        while off + _A_LEN.size <= n:
            (plen,) = _A_LEN.unpack_from(data, off)
            end = off + _A_LEN.size + plen + _A_CRC.size
            if plen != _S_REC.size or end > n:
                break
            (crc,) = _A_CRC.unpack_from(data, off + _A_LEN.size + plen)
            if zlib.crc32(data[off:off + _A_LEN.size + plen]) \
                    & 0xFFFFFFFF != crc:
                break
            magic, gid, epoch, seq, count, start = _S_REC.unpack(
                data[off + _A_LEN.size:off + _A_LEN.size + plen])
            if magic != _S_MAGIC:
                break
            yield gid, epoch, seq, count, start
            off = end

    # ------------------------------------------------------------- lookups
    def _leaves(self, start, end, expect_root):
        """Ordered commitments [start, end) re-read from an archive whose
        content re-derives `expect_root`. A corrupt holder cannot forge a
        proof — and cannot silently deny one either: the other holder is
        tried and the disagreement is flagged."""
        last_err = None
        for ai, (p, frames) in enumerate(self._arch):
            try:
                out = {}
                starts = [f[0] for f in frames]
                i = max(bisect.bisect_right(starts, start) - 1, 0)
                with open(p, "rb") as f:
                    for first, count, boff in frames[i:]:
                        if first >= end:
                            break
                        lo, hi = max(start, first), min(end, first + count)
                        if lo >= hi:
                            continue
                        f.seek(boff + (lo - first) * OBJ)
                        blob = f.read((hi - lo) * OBJ)
                        for pos in range(lo, hi):
                            out[pos] = blob[(pos - lo) * OBJ:
                                            (pos - lo + 1) * OBJ]
                if set(out) != set(range(start, end)):
                    raise ProofError("archive hole")
                leaves = [out[pos] for pos in range(start, end)]
                if W.merkle_root(leaves) != expect_root:
                    raise ProofError("archive root mismatch")
                if ai > 0:
                    self.archive_disagreement = True
                    print(f"proofs: ARCHIVE DISAGREEMENT — "
                          f"{self.archive_paths[0]} failed, served from "
                          f"{p}", flush=True)
                return leaves, ai
            except ProofError as e:
                last_err = e
        raise ProofError(f"no archive can prove the range: {last_err}")

    def find_position(self, commitment: bytes) -> int:
        # Newest first: the UNINDEXED tail (frames above the persistent index's
        # head) is scanned in bounded, object-aligned chunks — exactly the
        # historical "last duplicate wins" direction. Everything older is one
        # B-tree probe in the persistent index. Without an index, or before its
        # first backfill is ready, the tail is the whole archive: the historical
        # behaviour, never less than it.
        path, frames = self._pick_archive()
        use_index = self.index is not None and self.index.ready()
        head = self.index.indexed_max_position() if use_index else -1
        tail = [f for f in frames if f[0] + f[1] - 1 > head]
        objects_per_chunk = (1024 * 1024) // OBJ
        with open(path, "rb") as source:
            for first, count, blob_offset in reversed(tail):
                remaining = count
                while remaining > 0:
                    chunk_start = max(0, remaining - objects_per_chunk)
                    n_objects = remaining - chunk_start
                    source.seek(blob_offset + chunk_start * OBJ)
                    blob = source.read(n_objects * OBJ)
                    if len(blob) != n_objects * OBJ:
                        raise ProofError("archive hole")
                    offset = blob.rfind(commitment)
                    while offset >= 0:
                        if offset % OBJ == 0:
                            pos = first + chunk_start + offset // OBJ
                            if pos > head:            # below head = indexed
                                return pos
                        offset = blob.rfind(commitment, 0, offset)
                    remaining = chunk_start
        if use_index:
            pos = self.index.lookup(commitment)
            if pos is not None:
                return pos
        raise ProofError("unknown commitment")

    def batch_of(self, position: int):
        rows = self.kazalo("by_position_batch")
        starts = [r[0] for r in rows]
        i = bisect.bisect_right(starts, position) - 1
        if i >= 0:
            start, count, ident = rows[i]
            if start <= position < start + count:
                return ident, start, count
        return None, None, None

    def batch_range(self, gateway_id: bytes, epoch: int, seq: int):
        """The dispute entry point: batch so-and-so -> its position range
        and closing status."""
        got = self.kazalo("by_batch").get((gateway_id, epoch, seq))
        if got is None:
            raise ProofError("unknown batch")
        start, count = got
        i = bisect.bisect_right(self._range_starts, start) - 1
        closed = "UNWITNESSED"
        if i >= 0:
            bs = self._book_state
            rs, rc = bs["range_starts"][i], bs["range_counts"][i]
            if rs <= start < rs + rc:
                closed = "WITNESSED"
        return {"gateway_id": _hx(gateway_id), "gateway_epoch": epoch,
                "batch_seq": seq, "position_start": start, "count": count,
                "status": closed}

    # -------------------------------------------------------------- ladder
    def _ladder_from(self, child_hash: bytes, first_step=None):
        """Walk the closing ladder upward from `child_hash`. `first_step` is an
        optional pre-resolved ("HOUR", idx, j, hashes) for a confirmed record
        (HOUR children are re-derived, not indexed); tier records above are
        resolved through `_parent_of`. Returns (ladder, closed_up_to)."""
        ladder = []
        closed_up_to = "RANGE"
        while first_step is not None or child_hash in self._parent_of:
            if first_step is not None:
                tier, ti, j, hashes = first_step
                first_step = None
            else:
                tier, ti, j = self._parent_of[child_hash]
                hashes = self._tier_children[(tier, ti)]
            trec = self._tiers[tier][ti]
            ladder.append({
                "tier": tier,
                "record": _tier_record_dict(trec),
                "child_index": j,
                "proof": [[r, _hx(h)] for r, h in W.merkle_proof(hashes, j)],
            })
            closed_up_to = tier
            child_hash = trec.record_hash()
        return ladder, closed_up_to

    # ----------------------------------------------------------- hour view
    def hour_view(self, period_id: int | None = None) -> dict:
        """The public HOURLY unit: one closed HOUR tier record, the witnessed
        positions it folds (their record hashes are its Merkle children), and
        the ladder above it. This is what a catalogue references — one entry
        per clock hour, however many seals the hour held (zero included: an
        empty hour is an attested heartbeat, still chained)."""
        hours = self._tiers["HOUR"]
        if not hours:
            raise ProofError("no HOUR tier closed yet")
        if period_id is None:
            idx = len(hours) - 1
        else:
            idx = next((k for k, t in enumerate(hours) if t.period_id == period_id), None)
            if idx is None:
                raise ProofError("unknown HOUR period")
        t = hours[idx]
        first, n = self._book_state["hour_span"][idx]
        bs = self._book_state
        positions = []
        for k, h in enumerate(self._record_hashes(first, n)):
            positions.append({"range_start": bs["range_starts"][first + k],
                              "count": bs["range_counts"][first + k],
                              "record_hash": _hx(h)})
        ladder, closed_up_to = self._ladder_from(t.record_hash())
        if closed_up_to == "RANGE":
            closed_up_to = "HOUR"
        not_closed = _TIERS_UP[_TIERS_UP.index(closed_up_to) + 1:]
        clock = None
        if self.clock:
            clock = {**self.clock, "declared": True,
                     "period_start_utc_derived": _period_start(self.clock, t.period_id),
                     "note": ("Operator-declared configuration, not attested by the "
                              "book: period_id = HOUR windows of seconds_per_hour "
                              "since epoch_start_utc.")}
        return {
            "format": HOUR_FORMAT,
            "tier": "HOUR",
            "period_id": t.period_id,
            "coverage": "every closed HOUR period since period 0 of this epoch; "
                        "an unknown period is refused, never synthesized",
            "clock": clock,
            "record": _tier_record_dict(t),
            "record_hash": _hx(t.record_hash()),
            "children": [p["record_hash"] for p in positions],
            "positions": positions,
            "empty": n == 0,
            "ladder": ladder,
            "closed_up_to": closed_up_to,
            "not_closed": not_closed,
            "eraseal": None,
            "note": ("One public unit per clock hour. Children are the witnessed "
                     "positions' record hashes; each position proves its own "
                     "objects through /proof/<commitment>. No EraSeal is claimed."),
        }

    # -------------------------------------------------------------- bundle
    def bundle_for(self, commitment: bytes, refresh: bool = True) -> dict:
        if refresh:
            self.refresh()
        pos = self.find_position(commitment)
        i = bisect.bisect_right(self._range_starts, pos) - 1
        if i < 0:
            raise ProofError("position not covered by any witnessed record")
        rec, base, waw = self._confirmed_at(i)
        if not (rec.range_start <= pos < rec.range_start + rec.count):
            raise ProofError("position not yet witnessed (no closed range)")
        leaves, served_from = self._leaves(rec.range_start,
                                           rec.range_start + rec.count,
                                           rec.object_root)
        li = pos - rec.range_start
        if leaves[li] != commitment:
            raise ProofError("archive/record disagree on the commitment")
        membership = W.merkle_proof(leaves, li)
        batch_ident, bstart, bcount = self.batch_of(pos)
        bs = self._book_state
        hour_idx = bs["rec_hour"][i]
        first_step = None
        if hour_idx >= 0:                        # HOUR children: re-derived
            first_rec, n_children = bs["hour_span"][hour_idx]
            first_step = ("HOUR", hour_idx, bs["rec_j"][i],
                          self._record_hashes(first_rec, n_children))
        ladder, closed_up_to = self._ladder_from(rec.record_hash(), first_step)
        not_closed = _TIERS_UP[_TIERS_UP.index(closed_up_to) + 1:] \
            if closed_up_to != "RANGE" else list(_TIERS_UP)
        return {
            "format": FORMAT,
            "commitment": _hx(commitment),
            "position": pos,
            "batch": (None if batch_ident is None else
                      {"gateway_id": _hx(batch_ident[0]),
                       "gateway_epoch": batch_ident[1],
                       "batch_seq": batch_ident[2],
                       "position_start": bstart, "count": bcount}),
            "range_record": {"range_start": rec.range_start,
                             "count": rec.count,
                             "object_root": _hx(rec.object_root),
                             "prev_record_hash": _hx(rec.prev_record_hash)},
            "attestations": [
                {"witness_id": a.witness_id.decode("utf-8"),
                 "kind": a.kind, "signature": _hx(a.signature)}
                for a in (base, waw)],
            "qualified": {wid.decode("utf-8"): _hx(pub.public_bytes_raw())
                          for wid, pub in self.qualified.items()},
            "membership": {"leaf_index": li,
                           "proof": [[r, _hx(h)]
                                     for r, h in membership]},
            "ladder": ladder,
            "closed_up_to": closed_up_to,
            "not_closed": not_closed,
            "archive_served_from": served_from,
            "archive_disagreement": self.archive_disagreement,
            "eraseal": None,
            "note": ("Proof reaches the highest ACTUALLY CLOSED tier; "
                     "tiers in not_closed are still open. No EraSeal is "
                     "claimed: the annual inventory + Chamber commission "
                     "ceremony has not occurred."),
        }


# ---------------------------------------------------------------- verifier
def _verify_ladder(child_hash: bytes, ladder: list, expect: int):
    """Every ladder step re-derives its tier root from the child below it.
    Returns (ok, reason, top_tier, last_hash)."""
    top = None
    for step in ladder:
        trec = _tier_record_from(step["record"])
        if trec.tier not in _TIERS_UP[expect:]:
            return False, "ladder tiers out of order", top, child_hash
        expect = _TIERS_UP.index(trec.tier) + 1
        sproof = [(bool(r), _unhx(h)) for r, h in step["proof"]]
        if not W.verify_membership(child_hash, step["child_index"], sproof, trec.root):
            return False, f"{trec.tier} ladder step invalid", top, child_hash
        child_hash = trec.record_hash()
        top = trec.tier
    return True, "", top, child_hash


def verify_hour_view(view: dict) -> dict:
    """Independent verification of one public hourly unit from the view alone:
    the HOUR record re-hashes, its children re-derive its root (or it is an
    honest empty hour), and every ladder step above re-derives. Fail-closed.
    Positions inside the hour prove their own objects via verify_bundle; the
    link between a bundle and an hour is `bundle.ladder[0].record == view.record`."""
    def bad(reason):
        return {"valid": False, "reason": reason}
    try:
        if view.get("format") != HOUR_FORMAT or view.get("tier") != "HOUR":
            return bad("unknown hour view format")
        if view.get("eraseal") is not None:
            return bad("view claims an EraSeal — none can exist yet")
        trec = _tier_record_from(view["record"])
        if trec.tier != "HOUR" or trec.period_id != view["period_id"]:
            return bad("record does not describe this hour")
        if _hx(trec.record_hash()) != view["record_hash"]:
            return bad("record hash does not re-derive")
        children = [_unhx(h) for h in view["children"]]
        if trec.n_children != len(children):
            return bad("n_children does not match children")
        if [p["record_hash"] for p in view["positions"]] != view["children"]:
            return bad("positions do not match children")
        if children and CL.merkle_root(children) != trec.root:
            return bad("children do not re-derive the hour root")
        if not children and not view.get("empty"):
            return bad("no children but not declared empty")
        ok, reason, top, _ = _verify_ladder(trec.record_hash(), view["ladder"], 1)
        if not ok:
            return bad(reason)
        top = top or "HOUR"
        if top != view.get("closed_up_to"):
            return bad("closed_up_to does not match the ladder")
        if view.get("not_closed") != _TIERS_UP[_TIERS_UP.index(top) + 1:]:
            return bad("not_closed does not match the ladder")
        return {"valid": True, "period_id": trec.period_id,
                "record_hash": view["record_hash"], "positions": len(children),
                "empty": not children, "closed_up_to": top}
    except (KeyError, ValueError, TypeError, struct.error) as e:
        return bad(f"malformed hour view: {type(e).__name__}")


def verify_bundle(bundle: dict, trusted_registry: dict | None = None
                  ) -> dict:
    """Independent verification from the bundle alone (plus an optional
    PINNED registry — the anchor the verifier received through Cotrugli
    Ledger / the Chamber, NOT from the bundle). Checks: commitment
    membership in the range root, BOTH witness signatures (two different
    qualified masters, the WaW built over the base attestation), every
    ladder step, and the honesty flags. Fail-closed."""
    def bad(reason):
        return {"valid": False, "reason": reason}
    try:
        if bundle.get("format") != FORMAT:
            return bad("unknown bundle format")
        if bundle.get("eraseal") is not None:
            return bad("bundle claims an EraSeal — none can exist yet")
        commitment = _unhx(bundle["commitment"])
        rr = bundle["range_record"]
        rec = W.ClosedRecord(rr["range_start"], rr["count"],
                             _unhx(rr["object_root"]),
                             _unhx(rr["prev_record_hash"]))
        pos = bundle["position"]
        li = bundle["membership"]["leaf_index"]
        if pos != rec.range_start + li or not 0 <= li < rec.count:
            return bad("position does not match the range")
        proof = [(bool(r), _unhx(h))
                 for r, h in bundle["membership"]["proof"]]
        if not W.verify_membership(commitment, li, proof, rec.object_root):
            return bad("membership proof invalid")
        atts = [W.Attestation(a["witness_id"].encode("utf-8"), a["kind"],
                              _unhx(a["signature"]))
                for a in bundle["attestations"]]
        base = next((a for a in atts if a.kind == "BASE"), None)
        waw = next((a for a in atts if a.kind == "WAW"), None)
        if base is None or waw is None:
            return bad("missing attestation")
        embedded = {wid.encode("utf-8"): _unhx(hx)
                    for wid, hx in bundle["qualified"].items()}
        if trusted_registry is not None:
            pinned = {wid: (pub.public_bytes_raw()
                            if hasattr(pub, "public_bytes_raw") else pub)
                      for wid, pub in trusted_registry.items()}
            if embedded != pinned:
                return bad("qualified registry does not match the PINNED "
                           "registry — refusing (possible impostor set)")
        qualified = {wid: Ed25519PublicKey.from_public_bytes(raw)
                     for wid, raw in embedded.items()}
        res = W.confirm(rec, base, waw, qualified)
        if res.status != W.CONFIRMED:
            return bad(f"witness confirmation failed: {res.reason}")
        ok, reason, _top, _ = _verify_ladder(rec.record_hash(), bundle["ladder"], 0)
        if not ok:
            return bad(reason)
        top = bundle["ladder"][-1]["tier"] if bundle["ladder"] else "RANGE"
        if top != bundle.get("closed_up_to"):
            return bad("closed_up_to does not match the ladder")
        want_open = _TIERS_UP[_TIERS_UP.index(top) + 1:] \
            if top != "RANGE" else list(_TIERS_UP)
        if bundle.get("not_closed") != want_open:
            return bad("not_closed does not match the ladder")
        return {"valid": True, "position": pos, "closed_up_to": top,
                "not_closed": want_open,
                "batch": bundle.get("batch"),
                "confirmed_by": sorted(a["witness_id"]
                                       for a in bundle["attestations"])}
    except (KeyError, ValueError, TypeError, struct.error) as e:
        return bad(f"malformed bundle: {type(e).__name__}")
