minerva 0.2.0

//! The canonical epoch-ledger frame (S273): one version byte, the
//! machine's complete lifecycle state, the machine invariants the frame
//! can witness re-proven on the way back in.
//!
//! Byte identity is value identity (R-8): collections encode in their
//! maintained ascending orders, both report rounds are roster-implied
//! (the machine constructs every round over exactly the roster, so
//! encoding station ids would only create non-canonical spellings to
//! refuse), and window candidates carry their adoption rounds inline
//! (the machine pairs the two maps at every insertion and removal, so a
//! separate keyed section could only disagree). Counts are `u64` so
//! encoding is total for every constructible machine (the S272 lesson:
//! a saturating count emits a frame its own decoder refuses).
//!
//! Decode refuses, in layers: framing (version, lengths, tags), budgets
//! (every count against the caller's ceiling, before materialization,
//! and against the bytes that would have to back it, in `O(1)`),
//! canonical form (ascending orders, non-dot zeroes, nonzero generation
//! and horizon), nested codecs (rank and vector frames), and finally
//! *machine invariants*: state no [`Epochs`](super::super::Epochs) run
//! can produce is refused as
//! [`InvalidState`](EpochLedgerDecodeError::InvalidState) rather than
//! rebuilt into a machine that would then misbehave (the charter's
//! refuse-impossible-combinations duty). Two residues are deliberately
//! not re-proven, because refusing them buys nothing: a latch is not
//! checked against its confirmation round's completeness, and the
//! adoption bit is not checked against the winner's own report slot.
//! Both were traced harmless under the machine's merge-monotone folds
//! (a spurious low report can only raise a join, never fire a premature
//! fix or seal, and a latch never recomputes), and the R-57
//! counterweight records them.

extern crate alloc;

use alloc::collections::{BTreeMap, BTreeSet};
use alloc::vec::Vec;
use core::num::NonZeroU64;

use crate::kairos::{KAIROS_WIRE_LEN, Kairos};
use crate::metis::dot::Dot;
use crate::metis::{Cut, VersionVector};

use super::super::arrival::{Admission, ArrivalRound};
use super::super::{Declaration, EpochAddress, Round, SealedEpoch, Window};
use super::state::SnapshotState;
use super::{EpochLedgerDecodeBudget, EpochLedgerDecodeError, EpochLedgerSnapshot};

/// Version tag for the canonical epoch-ledger frame (S273, ruling R-57).
/// A distinct version space from every other minerva frame; private,
/// exactly as the sibling tags are.
const EPOCH_LEDGER_WIRE_V1: u8 = 0x01;
/// Version tag for the recorded-membership sibling frame (S342, ruling
/// R-89; PRD 0028 R3): the version-1 layout with an admission slot per
/// sealed entry and per window candidate, and the arrival-round section.
/// Emitted exactly when such content exists, so each version keeps its own
/// canonical bijection, and a version-1 reader refuses the frame whole
/// rather than restoring a machine that silently shed a membership
/// boundary.
const EPOCH_LEDGER_WIRE_V2: u8 = 0x02;

/// One wire dot: `u32` station plus `u64` counter.
const DOT_LEN: usize = 12;

/// One sealed-record fixed minimum: the winning address (20), two `u64`
/// counts, and an empty nested vector frame (5).
const SEALED_MIN_LEN: usize = 20 + 8 + 8 + VECTOR_MIN_LEN;

/// The smallest nested `VersionVector` frame: its version byte plus its
/// `u32` entry count.
const VECTOR_MIN_LEN: usize = 5;

impl EpochLedgerSnapshot {
    /// Encodes this checkpoint as its canonical frame: version 1, or the
    /// version-2 sibling exactly when recorded-membership content exists
    /// (a tagged candidate, an admission-bearing seal, or an open arrival
    /// round). Big-endian throughout; infallible and total (no count can
    /// saturate). The frame is one allocation; each nested vector frame is
    /// built into a temporary.
    #[must_use]
    pub fn to_bytes(&self) -> Vec<u8> {
        let state = &self.state;
        let recorded = recorded_content(state);
        let mut out = Vec::new();
        out.push(if recorded {
            EPOCH_LEDGER_WIRE_V2
        } else {
            EPOCH_LEDGER_WIRE_V1
        });
        out.extend_from_slice(&state.generation.get().to_be_bytes());
        out.extend_from_slice(&state.horizon.to_be_bytes());
        out.extend_from_slice(&(state.roster.len() as u64).to_be_bytes());
        for &station in &state.roster {
            out.extend_from_slice(&station.to_be_bytes());
        }
        out.extend_from_slice(&(state.lineage.len() as u64).to_be_bytes());
        for sealed in &state.lineage {
            out.extend_from_slice(&sealed.declaration.generation().to_be_bytes());
            encode_dot(&mut out, sealed.declaration.declaration());
            out.extend_from_slice(&(sealed.candidates.len() as u64).to_be_bytes());
            for candidate in &sealed.candidates {
                // One window, one generation: the candidates share the
                // winner's, so only the dot travels.
                encode_dot(&mut out, candidate.declaration());
            }
            out.extend_from_slice(&(sealed.protocol.len() as u64).to_be_bytes());
            for &dot in &sealed.protocol {
                encode_dot(&mut out, dot);
            }
            out.extend_from_slice(&sealed.sealed_join.to_bytes());
            if recorded {
                encode_admission_slot(&mut out, sealed.admission.as_ref());
            }
        }
        match &state.window {
            None => out.push(0x00),
            Some(window) => {
                out.push(0x01);
                out.extend_from_slice(&(window.candidates.len() as u64).to_be_bytes());
                for (address, declaration) in &window.candidates {
                    encode_dot(&mut out, address.declaration());
                    out.extend_from_slice(&declaration.rank().to_bytes());
                    out.extend_from_slice(&declaration.cut().as_vector().to_bytes());
                    if recorded {
                        encode_admission_slot(&mut out, declaration.admission());
                        if let Some(commitment) = declaration.admission_commitment() {
                            out.extend_from_slice(commitment);
                        }
                    }
                    // The adoption round rides its candidate: the machine
                    // pairs the two maps at every insert and displace.
                    encode_round(&mut out, &window.adoptions[address]);
                }
                out.extend_from_slice(&(window.protocol.len() as u64).to_be_bytes());
                for &dot in &window.protocol {
                    encode_dot(&mut out, dot);
                }
                encode_round(&mut out, &window.confirmations);
                match window.fixed {
                    None => out.push(0x00),
                    Some(winner) => {
                        out.push(0x01);
                        encode_dot(&mut out, winner.declaration());
                    }
                }
                out.push(u8::from(window.adopted));
            }
        }
        if recorded {
            match &state.arrival {
                None => out.push(0x00),
                Some(round) => {
                    out.push(0x01);
                    encode_admission(&mut out, &round.admission);
                    out.extend_from_slice(&round.commitment);
                    out.extend_from_slice(&round.own.to_be_bytes());
                    out.extend_from_slice(&(round.slots.len() as u64).to_be_bytes());
                    for (&station, &endorsed) in &round.slots {
                        out.extend_from_slice(&station.to_be_bytes());
                        out.push(u8::from(endorsed));
                    }
                }
            }
        }
        out
    }

    /// Decodes exactly one canonical frame under the caller's budget,
    /// rejecting trailing bytes.
    ///
    /// The budget is not optional: a checkpoint is always decoded inside
    /// a configuration the caller knows (its roster size, its horizon,
    /// its candidate and traffic expectations), and every ceiling is
    /// checked before the collection it bounds is materialized.
    /// Allocation is bounded by the input regardless: every declared
    /// count is checked against the bytes that would have to back it
    /// before its loop runs, in `O(1)`.
    ///
    /// The result is an inert [`EpochLedgerSnapshot`], never a live
    /// machine: accepted bytes re-encode identically (the canonical
    /// bijection), and state no machine run can produce refuses as
    /// [`InvalidState`](EpochLedgerDecodeError::InvalidState), so
    /// [`Epochs::rehydrate`](super::super::Epochs::rehydrate) never
    /// installs an impossible checkpoint.
    ///
    /// # Errors
    ///
    /// The layered [`EpochLedgerDecodeError`] vocabulary: framing,
    /// budget, canonical-form, nested-codec, and machine-invariant
    /// refusals. Never panics on adversarial input.
    pub fn from_bytes(
        bytes: &[u8],
        budget: EpochLedgerDecodeBudget,
    ) -> Result<Self, EpochLedgerDecodeError> {
        let mut at = 0usize;
        let version = read_u8(bytes, &mut at)?;
        if version != EPOCH_LEDGER_WIRE_V1 && version != EPOCH_LEDGER_WIRE_V2 {
            return Err(EpochLedgerDecodeError::UnknownVersion(version));
        }
        let recorded = version == EPOCH_LEDGER_WIRE_V2;
        let generation = read_generation(bytes, &mut at)?;
        let horizon = read_u64(bytes, &mut at)?;
        if horizon == 0 {
            return Err(EpochLedgerDecodeError::ZeroHorizon);
        }

        let roster = decode_roster(bytes, &mut at, budget)?;

        let lineage_count = read_count(
            bytes,
            &mut at,
            "lineage",
            budget.max_lineage(),
            SEALED_MIN_LEN,
        )?;
        if lineage_count > horizon {
            return Err(EpochLedgerDecodeError::InvalidState(
                "lineage exceeds the horizon",
            ));
        }
        let mut lineage = Vec::new();
        for _ in 0..lineage_count {
            lineage.push(decode_sealed(bytes, &mut at, budget, &roster, recorded)?);
        }
        validate_retained_admissions(&lineage, &roster)?;
        validate_lineage_containment(&lineage, &roster)?;
        // Seals advance the generation by exactly one, and the lineage
        // retains a contiguous most-recent run, so the recorded
        // generations are consecutive and end one below the current.
        for pair in lineage.windows(2) {
            // Checked: a crafted generation at the integer ceiling must
            // refuse, never overflow (debug builds trap on wrap, and a
            // profile-dependent verdict is its own defect).
            if pair[0].declaration.generation().checked_add(1)
                != Some(pair[1].declaration.generation())
            {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "lineage generations are not consecutive",
                ));
            }
        }
        if let Some(last) = lineage.last()
            && last.declaration.generation() != generation.get() - 1
        {
            return Err(EpochLedgerDecodeError::InvalidState(
                "lineage does not end one generation below the current",
            ));
        }

        let window = match read_u8(bytes, &mut at)? {
            0x00 => None,
            0x01 => Some(decode_window(
                bytes, &mut at, budget, generation, &roster, recorded,
            )?),
            tag => {
                return Err(EpochLedgerDecodeError::BadTag {
                    field: "window",
                    tag,
                });
            }
        };
        let arrival = decode_pending_sections(
            bytes,
            &mut at,
            budget,
            recorded,
            window.as_ref(),
            &lineage,
            &roster,
        )?;
        let state = SnapshotState {
            generation,
            roster,
            horizon,
            window,
            lineage,
            arrival,
        };
        if recorded && !recorded_content(&state) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "a version-2 frame carries no recorded-membership content",
            ));
        }

        if at != bytes.len() {
            return Err(EpochLedgerDecodeError::UnexpectedLength {
                offset: at,
                needed: 0,
                remaining: bytes.len() - at,
            });
        }
        Ok(Self { state })
    }
}

/// Validates pending tagged candidates against the newest retained record and
/// decodes the version-2 arrival section.
/// A pending admission is founded on the newest retained record and
/// admits only off-roster stations; both relations are machine invariants
/// the frame can witness.
fn decode_pending_sections(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
    recorded: bool,
    window: Option<&Window>,
    lineage: &[SealedEpoch],
    roster: &BTreeSet<u32>,
) -> Result<Option<ArrivalRound>, EpochLedgerDecodeError> {
    let newest = lineage.last().map(|sealed| sealed.declaration);
    if let Some(window) = window {
        for declaration in window.candidates.values() {
            let Some(admission) = declaration.admission() else {
                continue;
            };
            validate_pending(admission, newest, roster)?;
        }
    }
    if !recorded {
        return Ok(None);
    }
    let arrival = match read_u8(bytes, at)? {
        0x00 => None,
        0x01 => Some(decode_arrival(bytes, at, budget, newest, roster)?),
        tag => {
            return Err(EpochLedgerDecodeError::BadTag {
                field: "arrival",
                tag,
            });
        }
    };
    // Adoption is the assent (PRD 0028 R2): a live machine adopts a
    // tagged winner only over its own round holding the identical
    // boundary, and the round outlives the adoption (only a seal ends
    // it). A frame claiming the adoption without that round would hand
    // `try_seal` an admission this machine never assented to --- the one
    // schedule that could bypass `UnassentedAdmission` --- so the
    // invariant is re-proven here. What is deliberately NOT re-proven
    // (the R-57 residue pattern): that the round's `own` is *this*
    // caller's station. The frame does not carry the machine's identity
    // anywhere --- identity is the caller's, like the roster --- so an
    // own-swapped frame is another station's honest checkpoint, state a
    // machine run can produce, and refusing it would refuse lawful
    // frames. Restoring another station's checkpoint is the same
    // misconfiguration class as a wrong roster; it turns loud at the
    // caller's next `open_arrival` (`DiscordantVoice`).
    if let Some(window) = window
        && window.adopted
        && let Some(winner) = window.fixed
        && let Some(required) = window
            .candidates
            .get(&winner)
            .and_then(Declaration::admission)
        && (arrival.as_ref().map(ArrivalRound::admission) != Some(required)
            || arrival.as_ref().map(ArrivalRound::commitment)
                != window
                    .candidates
                    .get(&winner)
                    .and_then(Declaration::admission_commitment))
    {
        return Err(EpochLedgerDecodeError::InvalidState(
            "an adopted tagged winner without its endorsed arrival round",
        ));
    }
    Ok(arrival)
}

/// Whether any recorded-membership content exists: the version-2 tag's
/// exact condition, so each version keeps its own canonical bijection.
fn recorded_content(state: &SnapshotState) -> bool {
    state.arrival.is_some()
        || state
            .lineage
            .iter()
            .any(|sealed| sealed.admission.is_some())
        || state.window.as_ref().is_some_and(|window| {
            window
                .candidates
                .values()
                .any(|declaration| declaration.admission().is_some())
        })
}

/// Decodes the strictly ascending roster under its count budget.
fn decode_roster(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
) -> Result<BTreeSet<u32>, EpochLedgerDecodeError> {
    let roster_count = read_count(bytes, at, "roster", budget.max_roster(), 4)?;
    let mut roster = BTreeSet::new();
    let mut previous: Option<u32> = None;
    for _ in 0..roster_count {
        let station = read_u32(bytes, at)?;
        if previous.is_some_and(|previous| station <= previous) {
            return Err(EpochLedgerDecodeError::NonAscendingRoster {
                previous: previous.unwrap_or(0),
                found: station,
            });
        }
        previous = Some(station);
        let _ = roster.insert(station);
    }
    Ok(roster)
}

/// Validates the retained admissions: a retained admission's relations
/// are the seal's own. Its base is the predecessor record, and its
/// joiners were applied to the roster the moment it sealed.
fn validate_retained_admissions(
    lineage: &[SealedEpoch],
    roster: &BTreeSet<u32>,
) -> Result<(), EpochLedgerDecodeError> {
    for (index, sealed) in lineage.iter().enumerate() {
        let Some(admission) = &sealed.admission else {
            continue;
        };
        if admission.base().generation().checked_add(1) != Some(sealed.declaration.generation()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed admission base is not the predecessor generation",
            ));
        }
        if let Some(previous) = index.checked_sub(1).and_then(|index| lineage.get(index))
            && previous.declaration != admission.base()
        {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed admission base is not the retained predecessor",
            ));
        }
        if admission
            .joiners()
            .any(|station| !roster.contains(&station))
        {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed admission joiner is missing from the roster",
            ));
        }
    }
    Ok(())
}

/// Validates a pending admission: founded on the newest retained record,
/// admitting only off-roster stations.
fn validate_pending(
    admission: &Admission,
    newest: Option<EpochAddress>,
    roster: &BTreeSet<u32>,
) -> Result<(), EpochLedgerDecodeError> {
    if newest != Some(admission.base()) {
        return Err(EpochLedgerDecodeError::InvalidState(
            "pending admission base is not the newest retained record",
        ));
    }
    if admission.joiners().any(|station| roster.contains(&station)) {
        return Err(EpochLedgerDecodeError::InvalidState(
            "pending admission joiner is already on the roster",
        ));
    }
    Ok(())
}

/// Encodes one optional admission slot.
fn encode_admission_slot(out: &mut Vec<u8>, admission: Option<&Admission>) {
    match admission {
        None => out.push(0x00),
        Some(admission) => {
            out.push(0x01);
            encode_admission(out, admission);
        }
    }
}

/// Encodes one admission: the base address, then the joiners ascending.
fn encode_admission(out: &mut Vec<u8>, admission: &Admission) {
    out.extend_from_slice(&admission.base().generation().to_be_bytes());
    encode_dot(out, admission.base().declaration());
    out.extend_from_slice(&(admission.joiners().count() as u64).to_be_bytes());
    for joiner in admission.joiners() {
        out.extend_from_slice(&joiner.to_be_bytes());
    }
}

/// Decodes one optional admission slot.
fn decode_admission_slot(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
) -> Result<Option<Admission>, EpochLedgerDecodeError> {
    match read_u8(bytes, at)? {
        0x00 => Ok(None),
        0x01 => Ok(Some(decode_admission(bytes, at, budget)?)),
        tag => Err(EpochLedgerDecodeError::BadTag {
            field: "admission",
            tag,
        }),
    }
}

/// Decodes one admission: a valid base address and a nonempty, strictly
/// ascending joiner set under the roster ceiling.
fn decode_admission(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
) -> Result<Admission, EpochLedgerDecodeError> {
    let base_generation = read_generation(bytes, at)?;
    let base_dot = read_dot(bytes, at, "admission base")?;
    let base = EpochAddress::new(base_generation, base_dot);
    let count = read_count(bytes, at, "admission joiners", budget.max_roster(), 4)?;
    if count == 0 {
        return Err(EpochLedgerDecodeError::InvalidState(
            "admission carries no joiners",
        ));
    }
    let mut joiners = BTreeSet::new();
    let mut previous: Option<u32> = None;
    for _ in 0..count {
        let joiner = read_u32(bytes, at)?;
        if previous.is_some_and(|previous| joiner <= previous) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "admission joiners are not ascending",
            ));
        }
        previous = Some(joiner);
        let _ = joiners.insert(joiner);
    }
    Admission::new(base, joiners)
        .map_err(|_refusal| EpochLedgerDecodeError::InvalidState("admission carries no joiners"))
}

/// Decodes the arrival-round section: the boundary, the owner, and one
/// latched slot per family member, re-proving the machine invariants ---
/// the base is the newest retained record, the family sits inside the
/// roster, and the opener's own slot is endorsed (opening endorses).
fn decode_arrival(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
    newest: Option<EpochAddress>,
    roster: &BTreeSet<u32>,
) -> Result<ArrivalRound, EpochLedgerDecodeError> {
    let admission = decode_admission(bytes, at, budget)?;
    validate_pending(&admission, newest, roster)?;
    let mut commitment = [0u8; 32];
    read_exact(bytes, at, &mut commitment)?;
    let own = read_u32(bytes, at)?;
    let family_count = read_count(bytes, at, "arrival family", budget.max_roster(), 5)?;
    if family_count == 0 {
        return Err(EpochLedgerDecodeError::InvalidState(
            "arrival round carries no family",
        ));
    }
    let mut slots = alloc::collections::BTreeMap::new();
    let mut previous: Option<u32> = None;
    for _ in 0..family_count {
        let station = read_u32(bytes, at)?;
        if previous.is_some_and(|previous| station <= previous) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "arrival family is not ascending",
            ));
        }
        previous = Some(station);
        if !roster.contains(&station) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "arrival family member is outside the roster",
            ));
        }
        let endorsed = match read_u8(bytes, at)? {
            0x00 => false,
            0x01 => true,
            tag => {
                return Err(EpochLedgerDecodeError::BadTag {
                    field: "arrival slot",
                    tag,
                });
            }
        };
        let _ = slots.insert(station, endorsed);
    }
    if slots.get(&own) != Some(&true) {
        return Err(EpochLedgerDecodeError::InvalidState(
            "arrival round's own slot is not an endorsement",
        ));
    }
    Ok(ArrivalRound {
        admission,
        commitment,
        slots,
        own,
    })
}

/// Encodes one dot.
fn encode_dot(out: &mut Vec<u8>, dot: Dot) {
    out.extend_from_slice(&dot.station().to_be_bytes());
    out.extend_from_slice(&dot.counter().to_be_bytes());
}

/// Encodes one report round, roster-implied: the machine constructs
/// every round over exactly the roster (`Round::over`) and its fold
/// refuses unknown stations, so the keys are the roster by invariant and
/// only the slots travel, in roster order.
fn encode_round(out: &mut Vec<u8>, round: &Round) {
    for slot in round.reports.values() {
        match slot {
            None => out.push(0x00),
            Some(report) => {
                out.push(0x01);
                out.extend_from_slice(&report.to_bytes());
            }
        }
    }
}

fn decode_sealed(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
    roster: &BTreeSet<u32>,
    recorded: bool,
) -> Result<SealedEpoch, EpochLedgerDecodeError> {
    let generation = read_generation(bytes, at)?;
    let winner_dot = read_dot(bytes, at, "sealed declaration")?;
    let declaration = EpochAddress::new(generation, winner_dot);

    let candidate_count = read_count(
        bytes,
        at,
        "sealed candidates",
        budget.max_candidates(),
        DOT_LEN,
    )?;
    let mut candidates = BTreeSet::new();
    let mut previous: Option<Dot> = None;
    for _ in 0..candidate_count {
        let dot = read_dot(bytes, at, "sealed candidates")?;
        if let Some(previous) = previous
            && dot <= previous
        {
            return Err(EpochLedgerDecodeError::NonAscendingDots {
                collection: "sealed candidates",
                previous,
                found: dot,
            });
        }
        previous = Some(dot);
        if !roster.contains(&dot.station()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed candidate minted outside the roster",
            ));
        }
        let _ = candidates.insert(EpochAddress::new(generation, dot));
    }
    if !candidates.contains(&declaration) {
        return Err(EpochLedgerDecodeError::InvalidState(
            "sealed winner is not among its candidates",
        ));
    }

    // The checks here range over the *final* roster, which is a sound
    // over-approximation of any retained generation's roster and keeps
    // the version-1 refusal order exact. The strict per-generation
    // containment --- a joiner's coordinate before its admission ---
    // needs every retained admission first, and is
    // `validate_lineage_containment`'s second layer.
    let protocol = decode_dots(
        bytes,
        at,
        "sealed protocol",
        budget.max_protocol_dots(),
        Some(roster),
    )?;
    let sealed_join = decode_vector(bytes, at, "sealed join", budget)?;
    // A live seal's join folds only roster adoption reports; an
    // off-roster station here is impossible for an honest machine and
    // would hand `recognize` verdicts to foreign traffic (the bootstrap
    // door refuses the same condition).
    for (station, _counter) in &sealed_join {
        if !roster.contains(&station) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed join covers a station outside the roster",
            ));
        }
    }
    // The seal canonicalizes its ledger against the join: nothing above
    // it survives into the recognizer.
    for dot in &protocol {
        if dot.counter() > sealed_join.get(dot.station()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "sealed protocol dot above the sealed join",
            ));
        }
    }
    let admission = if recorded {
        decode_admission_slot(bytes, at, budget)?
    } else {
        None
    };
    Ok(SealedEpoch {
        declaration,
        candidates,
        protocol,
        sealed_join,
        admission,
    })
}

/// Validates each retained generation against the roster *of that
/// generation* (PRD 0028 R4, falsifier 6's decode half): the final
/// roster minus every retained admission's joiners is the oldest
/// retained generation's roster, and each admission widens it for the
/// generations after its record. A joiner's candidate, ledger dot, or
/// join coordinate appearing before its admission is state no machine
/// run can produce, and `recognize` would hand it duplicate verdicts.
/// The parse-time checks in `decode_sealed` already refused everything
/// outside the final roster, so this layer's refusals are reachable
/// only through a version-2 frame's admissions. One residue is deliberate
/// (the R-57 pattern): the oldest roster is derived by subtracting the
/// frame's own admissions, so a crafted checkpoint claiming a founding
/// member as admitted is not distinguishable here. The checkpoint is
/// the caller's explicit local trust decision (`Epochs::rehydrate`), and
/// the peer-evidence door, `Epochs::bootstrap`, grounds the oldest
/// roster independently in the caller's configuration instead.
fn validate_lineage_containment(
    lineage: &[SealedEpoch],
    roster: &BTreeSet<u32>,
) -> Result<(), EpochLedgerDecodeError> {
    let mut historical = roster.clone();
    for sealed in lineage {
        if let Some(admission) = &sealed.admission {
            for joiner in admission.joiners() {
                let _ = historical.remove(&joiner);
            }
        }
    }
    for sealed in lineage {
        for candidate in &sealed.candidates {
            if !historical.contains(&candidate.declaration().station()) {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "sealed candidate minted outside the roster",
                ));
            }
        }
        for dot in &sealed.protocol {
            if !historical.contains(&dot.station()) {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "protocol dot minted outside the roster",
                ));
            }
        }
        // A live seal's join folds only roster adoption reports; an
        // off-roster station would hand `recognize` verdicts to foreign
        // traffic (the bootstrap door refuses the same condition).
        for (station, _counter) in &sealed.sealed_join {
            if !historical.contains(&station) {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "sealed join covers a station outside the roster",
                ));
            }
        }
        if let Some(admission) = &sealed.admission {
            // Live doors refuse a joiner already on the roster, so a
            // lineage admitting one station twice is state no machine
            // run can produce.
            if admission
                .joiners()
                .any(|joiner| historical.contains(&joiner))
            {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "sealed admission joiner is already a member",
                ));
            }
            historical.extend(admission.joiners());
        }
    }
    Ok(())
}

fn decode_window(
    bytes: &[u8],
    at: &mut usize,
    budget: EpochLedgerDecodeBudget,
    generation: NonZeroU64,
    roster: &BTreeSet<u32>,
    recorded: bool,
) -> Result<Window, EpochLedgerDecodeError> {
    // A window exists only around at least one candidate, and each
    // candidate row carries at least its dot, rank, an empty cut frame,
    // and one round tag per roster member.
    let candidate_min = DOT_LEN + KAIROS_WIRE_LEN + VECTOR_MIN_LEN + roster.len();
    let candidate_count = read_count(
        bytes,
        at,
        "window candidates",
        budget.max_candidates(),
        candidate_min,
    )?;
    if candidate_count == 0 {
        return Err(EpochLedgerDecodeError::InvalidState(
            "open window carries no candidates",
        ));
    }
    let mut candidates = BTreeMap::new();
    let mut adoptions = BTreeMap::new();
    let mut previous: Option<Dot> = None;
    for _ in 0..candidate_count {
        let dot = read_dot(bytes, at, "window candidates")?;
        if let Some(previous) = previous
            && dot <= previous
        {
            return Err(EpochLedgerDecodeError::NonAscendingDots {
                collection: "window candidates",
                previous,
                found: dot,
            });
        }
        previous = Some(dot);
        if !roster.contains(&dot.station()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "candidate minted outside the roster",
            ));
        }
        let mut rank_bytes = [0u8; KAIROS_WIRE_LEN];
        read_exact(bytes, at, &mut rank_bytes)?;
        let rank = Kairos::from_bytes(&rank_bytes).map_err(EpochLedgerDecodeError::Rank)?;
        let cut_vector = decode_vector(bytes, at, "candidate cut", budget)?;
        if dot.counter() <= cut_vector.get(dot.station()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "candidate dot is covered by its own cut",
            ));
        }
        let admission = if recorded {
            decode_admission_slot(bytes, at, budget)?
        } else {
            None
        };
        let admission_commitment = if admission.is_some() {
            let mut commitment = [0u8; 32];
            read_exact(bytes, at, &mut commitment)?;
            Some(commitment)
        } else {
            None
        };
        let round = decode_round(bytes, at, "candidate adoptions", budget, roster)?;
        let address = EpochAddress::new(generation, dot);
        let _ = candidates.insert(
            address,
            Declaration {
                address,
                rank,
                cut: Cut::from_witnessed(cut_vector),
                admission,
                admission_commitment,
            },
        );
        let _ = adoptions.insert(address, round);
    }

    let protocol = decode_dots(
        bytes,
        at,
        "window protocol",
        budget.max_protocol_dots(),
        Some(roster),
    )?;
    for address in candidates.keys() {
        if !protocol.contains(&address.declaration()) {
            return Err(EpochLedgerDecodeError::InvalidState(
                "candidate missing from the protocol ledger",
            ));
        }
    }
    let confirmations = decode_round(bytes, at, "confirmations", budget, roster)?;
    let (fixed, adopted) = decode_latch(bytes, at, generation, &candidates)?;
    Ok(Window {
        candidates,
        protocol,
        confirmations,
        adoptions,
        fixed,
        adopted,
    })
}

/// Decodes the winner latch and the local adoption bit, re-proving the
/// latch invariants: the winner is a candidate, it is the candidates'
/// maximum by the fixed public rule (the latch takes the maximum at
/// fixation and the candidate set is frozen from that observation on),
/// and adoption implies a latched winner.
fn decode_latch(
    bytes: &[u8],
    at: &mut usize,
    generation: NonZeroU64,
    candidates: &BTreeMap<EpochAddress, Declaration>,
) -> Result<(Option<EpochAddress>, bool), EpochLedgerDecodeError> {
    let fixed = match read_u8(bytes, at)? {
        0x00 => None,
        0x01 => {
            let dot = read_dot(bytes, at, "fixed winner")?;
            let address = EpochAddress::new(generation, dot);
            if !candidates.contains_key(&address) {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "fixed winner is not a candidate",
                ));
            }
            let maximum = candidates
                .values()
                .max_by_key(|candidate| (candidate.rank(), candidate.dot()))
                .map(Declaration::address);
            if maximum != Some(address) {
                return Err(EpochLedgerDecodeError::InvalidState(
                    "fixed winner is not the candidates' maximum",
                ));
            }
            Some(address)
        }
        tag => {
            return Err(EpochLedgerDecodeError::BadTag {
                field: "fixed",
                tag,
            });
        }
    };
    let adopted = match read_u8(bytes, at)? {
        0x00 => false,
        0x01 => true,
        tag => {
            return Err(EpochLedgerDecodeError::BadTag {
                field: "adopted",
                tag,
            });
        }
    };
    if adopted && fixed.is_none() {
        return Err(EpochLedgerDecodeError::InvalidState(
            "adopted without a fixed winner",
        ));
    }
    Ok((fixed, adopted))
}

/// Decodes one roster-implied report round: one tagged slot per roster
/// member, in roster order, keys reconstructed from the validated roster.
fn decode_round(
    bytes: &[u8],
    at: &mut usize,
    field: &'static str,
    budget: EpochLedgerDecodeBudget,
    roster: &BTreeSet<u32>,
) -> Result<Round, EpochLedgerDecodeError> {
    let mut reports = BTreeMap::new();
    for &station in roster {
        let slot = match read_u8(bytes, at)? {
            0x00 => None,
            0x01 => Some(decode_vector(bytes, at, field, budget)?),
            tag => return Err(EpochLedgerDecodeError::BadTag { field, tag }),
        };
        let _ = reports.insert(station, slot);
    }
    Ok(Round { reports })
}

/// Decodes an ascending, zero-refusing, roster-minted dot set under its
/// count budget (every protocol-plane dot is a roster member's
/// declaration dot; the deliver door refuses the rest before recording).
fn decode_dots(
    bytes: &[u8],
    at: &mut usize,
    collection: &'static str,
    max: usize,
    roster: Option<&BTreeSet<u32>>,
) -> Result<BTreeSet<Dot>, EpochLedgerDecodeError> {
    let count = read_count(bytes, at, collection, max, DOT_LEN)?;
    let mut dots = BTreeSet::new();
    let mut previous: Option<Dot> = None;
    for _ in 0..count {
        let dot = read_dot(bytes, at, collection)?;
        if let Some(previous) = previous
            && dot <= previous
        {
            return Err(EpochLedgerDecodeError::NonAscendingDots {
                collection,
                previous,
                found: dot,
            });
        }
        previous = Some(dot);
        if let Some(roster) = roster
            && !roster.contains(&dot.station())
        {
            return Err(EpochLedgerDecodeError::InvalidState(
                "protocol dot minted outside the roster",
            ));
        }
        let _ = dots.insert(dot);
    }
    Ok(dots)
}

/// Decodes one nested version-vector frame and applies the per-vector
/// entry ceiling.
fn decode_vector(
    bytes: &[u8],
    at: &mut usize,
    field: &'static str,
    budget: EpochLedgerDecodeBudget,
) -> Result<VersionVector, EpochLedgerDecodeError> {
    let (vector, tail) = VersionVector::from_prefix(&bytes[*at..])
        .map_err(|source| EpochLedgerDecodeError::Vector { field, source })?;
    *at = bytes.len() - tail.len();
    let entries = vector.iter().count();
    if entries > budget.max_vector_entries() {
        return Err(EpochLedgerDecodeError::TooManyVectorEntries {
            field,
            count: entries as u64,
            budget: budget.max_vector_entries() as u64,
        });
    }
    Ok(vector)
}

/// Reads a `u64` collection count, refusing it against the caller's
/// ceiling and against the bytes that would have to back it, both before
/// the collection's loop runs (`O(1)` against a hostile count).
fn read_count(
    bytes: &[u8],
    at: &mut usize,
    collection: &'static str,
    max: usize,
    min_row: usize,
) -> Result<u64, EpochLedgerDecodeError> {
    let count = read_u64(bytes, at)?;
    if count > max as u64 {
        return Err(EpochLedgerDecodeError::TooMany {
            collection,
            count,
            budget: max as u64,
        });
    }
    let needed = count.saturating_mul(min_row as u64);
    let remaining = (bytes.len() - *at) as u64;
    if needed > remaining {
        return Err(EpochLedgerDecodeError::UnexpectedLength {
            offset: *at,
            needed: usize::try_from(needed).unwrap_or(usize::MAX),
            remaining: bytes.len() - *at,
        });
    }
    Ok(count)
}

/// Reads one dot, refusing the non-dot counter zero exactly where the
/// frame always refused it. The value is an identity from here on: the
/// refusal proves the counter, so no later door re-proves it (R-91).
fn read_dot(
    bytes: &[u8],
    at: &mut usize,
    collection: &'static str,
) -> Result<Dot, EpochLedgerDecodeError> {
    let station = read_u32(bytes, at)?;
    let counter = read_u64(bytes, at)?;
    Dot::from_parts(station, counter).map_err(|_zero| EpochLedgerDecodeError::ZeroDot {
        collection,
        station,
    })
}

/// Reads one lineage generation under the frame's standing zero refusal,
/// handing back the proven value the address doors need.
fn read_generation(bytes: &[u8], at: &mut usize) -> Result<NonZeroU64, EpochLedgerDecodeError> {
    NonZeroU64::new(read_u64(bytes, at)?).ok_or(EpochLedgerDecodeError::ZeroGeneration)
}

fn read_u8(bytes: &[u8], at: &mut usize) -> Result<u8, EpochLedgerDecodeError> {
    let mut buffer = [0u8; 1];
    read_exact(bytes, at, &mut buffer)?;
    Ok(buffer[0])
}

fn read_u32(bytes: &[u8], at: &mut usize) -> Result<u32, EpochLedgerDecodeError> {
    let mut buffer = [0u8; 4];
    read_exact(bytes, at, &mut buffer)?;
    Ok(u32::from_be_bytes(buffer))
}

fn read_u64(bytes: &[u8], at: &mut usize) -> Result<u64, EpochLedgerDecodeError> {
    let mut buffer = [0u8; 8];
    read_exact(bytes, at, &mut buffer)?;
    Ok(u64::from_be_bytes(buffer))
}

fn read_exact(bytes: &[u8], at: &mut usize, into: &mut [u8]) -> Result<(), EpochLedgerDecodeError> {
    let end = at.saturating_add(into.len());
    let Some(slice) = bytes.get(*at..end) else {
        return Err(EpochLedgerDecodeError::UnexpectedLength {
            offset: *at,
            needed: into.len(),
            remaining: bytes.len().saturating_sub(*at),
        });
    };
    into.copy_from_slice(slice);
    *at = end;
    Ok(())
}