minerva 0.2.0

//! The epoch-ledger snapshot (S273): the golden frame, the lifecycle
//! round-trips, the rehydration trust seam, and the layered refusals.

extern crate alloc;

use alloc::vec::Vec;
use core::num::NonZeroUsize;

use proptest::prelude::*;

use crate::kairos::Kairos;
use crate::metis::{
    Cut, Dot, EpochLedgerDecodeBudget, EpochLedgerDecodeError, EpochLedgerRehydrateError,
    EpochLedgerSnapshot, Epochs, SealRecord, SealedEpoch, Stability, VersionVector, Vouched,
};

const ROSTER: [u32; 2] = [1, 2];

/// Builds an identity literal. Panics on the non-dot counter zero (R-91).
#[track_caller]
fn d(station: u32, counter: u64) -> Dot {
    Dot::from_parts(station, counter).expect("test literal names the non-dot counter zero")
}

fn cut(entries: &[(u32, u64)]) -> Cut {
    let mut vector = VersionVector::new();
    for &(station, counter) in entries {
        vector.observe(station, counter);
    }
    Cut::from_witnessed(vector)
}

fn budget() -> EpochLedgerDecodeBudget {
    EpochLedgerDecodeBudget::new(8, 8, 8, 64, 8)
}

fn horizon() -> NonZeroUsize {
    NonZeroUsize::new(2).unwrap()
}

fn fresh() -> Epochs {
    Epochs::new(ROSTER, horizon())
}

/// A tracker with every member reporting `reported`.
fn tracker(reported: &Cut) -> Stability {
    let mut stability = Stability::new(ROSTER);
    for station in ROSTER {
        stability.report_cut(station, reported).unwrap();
    }
    stability
}

/// One window open on station 1's declaration.
fn open() -> Epochs {
    let mut epochs = fresh();
    let stability = tracker(&cut(&[(1, 4), (2, 4)]));
    let _ = epochs
        .declare(
            d(1, 5),
            Kairos::new(5, 0, 1, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("the watermark licenses the declaration");
    epochs
}

/// The contested window: station 2's higher-ranked rival delivered.
fn contested() -> Epochs {
    let mut epochs = open();
    let stability = tracker(&cut(&[(1, 4), (2, 4)]));
    let rival = Epochs::new(ROSTER, horizon())
        .declare(
            d(2, 5),
            Kairos::new(6, 0, 2, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("the rival watermark licenses the declaration");
    epochs
        .deliver(rival, &stability, &Cut::bottom())
        .expect("a concurrent candidate enters the window");
    epochs
}

/// Confirmed, fixed, and locally adopted, with the peer's adoption report
/// folded: the richest mid-window state.
fn adopted() -> Epochs {
    let mut epochs = contested();
    let delivered = cut(&[(1, 5), (2, 5)]);
    let stability = tracker(&delivered);
    let winner = Epochs::new(ROSTER, horizon())
        .declare(
            d(2, 5),
            Kairos::new(6, 0, 2, 0u16),
            &tracker(&cut(&[(1, 4), (2, 4)])),
            &Cut::bottom(),
        )
        .expect("re-minting the rival for its address")
        .address();
    epochs
        .confirm(winner, &Vouched::trust(1, delivered.clone()))
        .unwrap();
    epochs
        .confirm(winner, &Vouched::trust(2, delivered))
        .unwrap();
    let _ = epochs
        .adopt(1, 5, &stability)
        .expect("the confirmation round is complete under the watermark");
    epochs.adopt_report(winner, &Vouched::trust(2, 5)).unwrap();
    epochs
}

/// Sealed: the window retired into the lineage, generation advanced.
fn sealed() -> Epochs {
    let mut epochs = adopted();
    let stability = tracker(&cut(&[(1, 5), (2, 5)]));
    assert!(
        epochs.try_seal(&stability).is_some(),
        "the adoption round is complete under the watermark"
    );
    epochs
}

/// Every lifecycle station this suite drives.
fn stations() -> Vec<Epochs> {
    alloc::vec![fresh(), open(), contested(), adopted(), sealed()]
}

/// Drives one further full declare-through-seal cycle on a sealed
/// machine (`round` starts at 1 for the second seal).
fn advance_one_cycle(epochs: &mut Epochs, round: usize) {
    let base = 5 * round as u64;
    let reported = cut(&[(1, base + 4), (2, base + 4)]);
    let stability = tracker(&reported);
    let winner = epochs
        .declare(
            d(1, base + 5),
            Kairos::new(base + 5, 0, 1, 0u16),
            &stability,
            &reported,
        )
        .expect("the next window opens over the risen basis")
        .address();
    let delivered = cut(&[(1, base + 5), (2, base + 4)]);
    let stability = tracker(&delivered);
    epochs
        .confirm(winner, &Vouched::trust(1, delivered.clone()))
        .unwrap();
    epochs
        .confirm(winner, &Vouched::trust(2, delivered))
        .unwrap();
    let _ = epochs.adopt(1, base + 5, &stability).unwrap();
    epochs
        .adopt_report(winner, &Vouched::trust(2, base + 4))
        .unwrap();
    assert!(epochs.try_seal(&stability).is_some());
}

/// The golden layout fixture: the fresh machine's frame, every byte
/// pinned (the cross-repository fixture obligation).
#[test]
fn test_epoch_ledger_to_bytes_layout() {
    let frame = fresh().snapshot().to_bytes();
    assert_eq!(
        frame,
        [
            0x01, // version (v1)
            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, // generation = 1
            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, // horizon = 2
            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, // roster_count = 2
            0x00, 0x00, 0x00, 0x01, // station 1
            0x00, 0x00, 0x00, 0x02, // station 2
            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // lineage_count = 0
            0x00, // window tag = none
        ]
    );
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&frame, budget()),
        Ok(fresh().snapshot())
    );
}

/// Every lifecycle station round-trips: decode equals the snapshot,
/// accepted bytes re-encode identically, and a rehydrated machine
/// snapshots back to the same state.
#[test]
fn every_lifecycle_station_round_trips() {
    for epochs in stations() {
        let snapshot = epochs.snapshot();
        let bytes = snapshot.to_bytes();
        let decoded = EpochLedgerSnapshot::from_bytes(&bytes, budget()).expect("own frame decodes");
        assert_eq!(decoded, snapshot);
        assert_eq!(decoded.to_bytes(), bytes);

        let rehydrated = Epochs::rehydrate(&decoded, ROSTER, horizon())
            .expect("the configuration matches the checkpoint");
        assert_eq!(rehydrated, epochs);
        assert_eq!(rehydrated.snapshot(), snapshot);
    }
}

/// The durable codec preserves station zero in both roster and window state.
#[test]
fn station_zero_round_trips_through_the_epoch_ledger() {
    let roster = [0, 1];
    let reported = cut(&[(0, 4), (1, 4)]);
    let mut stability = Stability::new(roster);
    for station in roster {
        stability.report_cut(station, &reported).unwrap();
    }
    let mut epochs = Epochs::new(roster, horizon());
    let declaration = epochs
        .declare(
            d(0, 5),
            Kairos::new(5, 0, 0, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("station zero opens a lifecycle window");
    assert_eq!(declaration.dot(), d(0, 5));

    let snapshot = epochs.snapshot();
    let frame = snapshot.to_bytes();
    let decoded = EpochLedgerSnapshot::from_bytes(&frame, budget()).expect("snapshot decodes");
    assert_eq!(decoded, snapshot);
    assert_eq!(decoded.to_bytes(), frame);
    assert_eq!(decoded.roster().collect::<Vec<_>>(), roster);

    let rehydrated = Epochs::rehydrate(&decoded, roster, horizon()).expect("snapshot rehydrates");
    assert_eq!(rehydrated, epochs);
}

/// The reads a caller inspects before trusting a snapshot.
#[test]
fn the_snapshot_reads_describe_the_machine() {
    let snapshot = adopted().snapshot();
    assert_eq!(snapshot.generation(), 1);
    assert_eq!(snapshot.horizon(), 2);
    assert_eq!(snapshot.roster().collect::<Vec<_>>(), [1, 2]);
    assert!(snapshot.has_open_window());
    assert_eq!(snapshot.candidate_len(), 2);
    assert_eq!(snapshot.sealed_len(), 0);

    let after_seal = sealed().snapshot();
    assert_eq!(after_seal.generation(), 2);
    assert!(!after_seal.has_open_window());
    assert_eq!(after_seal.candidate_len(), 0);
    assert_eq!(after_seal.sealed_len(), 1);
}

/// The live machine mirrors the checkpoint's read surface without
/// materializing one — generation, roster, horizon — and hands back the
/// newest retained recognizer directly: the R-69 in-band freshness floor
/// and the R-76 post-crash re-entry pivot, at every lifecycle station.
#[test]
fn the_live_reads_mirror_the_snapshot() {
    for station in stations() {
        let snapshot = station.snapshot();
        assert_eq!(station.generation(), snapshot.generation());
        assert_eq!(
            u64::try_from(station.horizon().get()).unwrap(),
            snapshot.horizon()
        );
        assert_eq!(
            station.roster().collect::<Vec<_>>(),
            snapshot.roster().collect::<Vec<_>>()
        );
        assert_eq!(
            station.newest_sealed().map(SealedEpoch::declaration),
            station.sealed().last().map(SealedEpoch::declaration),
        );
    }

    let machine = sealed();
    let newest = machine
        .newest_sealed()
        .expect("a sealed machine retains its recognizer");
    assert_eq!(newest.declaration().generation() + 1, machine.generation());

    // The R-69 floor, in checkpoint terms: an installable checkpoint
    // names its successor generation, so after sealing generation G the
    // floor refuses successors at or below G — a pre-seal checkpoint of
    // this very lineage is labeled G and must not pass. `generation()`
    // is exactly the first acceptable successor label.
    let pre_seal_label = newest.declaration().generation();
    assert!(pre_seal_label < machine.generation());

    // The live entry's retained candidate set equals its wire twin's:
    // a certificate layer binds over both retained sets without a clone.
    let record = SealRecord::from_sealed(newest);
    assert_eq!(
        newest.candidates().collect::<Vec<_>>(),
        record.candidates().collect::<Vec<_>>(),
    );
    assert_eq!(
        newest.declaration_dots().collect::<Vec<_>>(),
        record.declaration_dots().collect::<Vec<_>>(),
    );

    // Across further seals — including horizon eviction of the oldest
    // entry — the read stays pinned to the newest retained recognizer.
    let mut machine = machine;
    for round in 1..=2 {
        advance_one_cycle(&mut machine, round);
        let newest = machine
            .newest_sealed()
            .expect("the recognizer survives every seal")
            .declaration();
        assert_eq!(
            Some(newest),
            machine.sealed().last().map(SealedEpoch::declaration)
        );
        assert_eq!(newest.generation() + 1, machine.generation());
    }
    assert_eq!(machine.sealed().count(), horizon().get());
}

/// A machine rehydrated mid-window continues exactly as the original: the
/// in-flight rounds ride whole, the seal fires identically, and the
/// affine adoption witness is re-earned through `adopt`, not deserialized.
#[test]
fn a_rehydrated_machine_continues_identically() {
    let original = adopted();
    let decoded =
        EpochLedgerSnapshot::from_bytes(&original.snapshot().to_bytes(), budget()).unwrap();
    let mut rehydrated = Epochs::rehydrate(&decoded, ROSTER, horizon()).unwrap();
    let mut control = original;

    // Authority is re-earned, never decoded: the adoption door re-runs
    // against live stability evidence and returns the witness afresh.
    let stability = tracker(&cut(&[(1, 5), (2, 5)]));
    let re_earned = rehydrated
        .adopt(1, 5, &stability)
        .expect("re-adoption is idempotent over the restored rounds");
    assert_eq!(re_earned.address().declaration(), d(2, 5));

    let sealed_control = control
        .try_seal(&stability)
        .expect("the control seals")
        .clone();
    let sealed_rehydrated = rehydrated
        .try_seal(&stability)
        .expect("the rehydrated machine seals identically")
        .clone();
    assert_eq!(sealed_control, sealed_rehydrated);
    assert_eq!(control.snapshot(), rehydrated.snapshot());
}

/// The rehydration trust seam: the caller's configuration must match the
/// checkpoint exactly; storage is never authority over the roster or the
/// horizon.
#[test]
fn rehydration_refuses_a_mismatched_configuration() {
    let snapshot = sealed().snapshot();
    assert_eq!(
        Epochs::rehydrate(&snapshot, [1, 3], horizon()).unwrap_err(),
        EpochLedgerRehydrateError::RosterMismatch
    );
    assert_eq!(
        Epochs::rehydrate(&snapshot, ROSTER, NonZeroUsize::new(3).unwrap()).unwrap_err(),
        EpochLedgerRehydrateError::HorizonMismatch {
            checkpoint: 2,
            configured: 3,
        }
    );
}

/// The rehydration re-proof covers the sealed join: an off-roster
/// station in a crafted checkpoint refuses at decode, so
/// `newest_sealed` can never surface a recognizer a live machine could
/// not have sealed (the R-76 read-of-re-proven-state claim, pinned; the
/// bootstrap door refuses the same condition).
#[test]
fn a_crafted_off_roster_sealed_join_refuses() {
    let mut frame = sealed().snapshot().to_bytes();
    // Station 2's join entry (station u32, counter u64) encodes last in
    // the sealed section; repoint it outside the roster. Order stays
    // ascending, so the refusal is the roster wall, not canonical form.
    let entry: [u8; 12] = [0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 5];
    let at = frame
        .windows(12)
        .rposition(|window| window == entry)
        .expect("the sealed join carries station 2 at counter 5");
    frame[at + 3] = 0x09;
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&frame, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "sealed join covers a station outside the roster",
        ))
    );
}

/// Framing refusals: version, truncation, trailing bytes, hostile counts.
#[test]
fn framing_refusals() {
    let frame = adopted().snapshot().to_bytes();

    // 0x02 is the recorded-membership sibling version (S342, R-89), so
    // the first unknown byte is 0x03.
    let mut wrong_version = frame.clone();
    wrong_version[0] = 0x03;
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&wrong_version, budget()),
        Err(EpochLedgerDecodeError::UnknownVersion(0x03))
    );

    for keep in [0, 5, 20, frame.len() - 1] {
        assert!(matches!(
            EpochLedgerSnapshot::from_bytes(&frame[..keep], budget()),
            Err(EpochLedgerDecodeError::UnexpectedLength { .. })
        ));
    }

    let mut trailing = frame.clone();
    trailing.push(0x00);
    assert!(matches!(
        EpochLedgerSnapshot::from_bytes(&trailing, budget()),
        Err(EpochLedgerDecodeError::UnexpectedLength { .. })
    ));

    // A roster count the bytes cannot back refuses in O(1), before the
    // budget would even matter, and a count past the budget refuses as
    // the typed overflow.
    let mut unbacked = frame.clone();
    unbacked[17..25].copy_from_slice(&5u64.to_be_bytes());
    assert!(matches!(
        EpochLedgerSnapshot::from_bytes(&unbacked, budget()),
        Err(EpochLedgerDecodeError::NonAscendingRoster { .. }
            | EpochLedgerDecodeError::UnexpectedLength { .. })
    ));
    let mut over = frame;
    over[17..25].copy_from_slice(&9u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&over, budget()),
        Err(EpochLedgerDecodeError::TooMany {
            collection: "roster",
            count: 9,
            budget: 8,
        })
    );
}

/// Canonical-form refusals: zero generation and horizon, non-ascending
/// roster, bad tags.
#[test]
fn canonical_form_refusals() {
    let frame = fresh().snapshot().to_bytes();

    let mut zero_generation = frame.clone();
    zero_generation[1..9].fill(0);
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&zero_generation, budget()),
        Err(EpochLedgerDecodeError::ZeroGeneration)
    );

    let mut zero_horizon = frame.clone();
    zero_horizon[9..17].fill(0);
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&zero_horizon, budget()),
        Err(EpochLedgerDecodeError::ZeroHorizon)
    );

    let mut reordered = frame.clone();
    // Swap roster stations 1 and 2.
    reordered[25..29].copy_from_slice(&2u32.to_be_bytes());
    reordered[29..33].copy_from_slice(&1u32.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&reordered, budget()),
        Err(EpochLedgerDecodeError::NonAscendingRoster {
            previous: 2,
            found: 1,
        })
    );

    let mut bad_window_tag = frame;
    let last = bad_window_tag.len() - 1;
    bad_window_tag[last] = 0x02;
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&bad_window_tag, budget()),
        Err(EpochLedgerDecodeError::BadTag {
            field: "window",
            tag: 0x02,
        })
    );
}

/// Machine-invariant refusals: state no `Epochs` run can produce is
/// refused rather than rebuilt into a machine that would misbehave.
#[test]
fn impossible_state_refusals() {
    // The open-window frame's fixed layout: header (17), roster count and
    // stations (8 + 8), lineage count (8), window tag (1), candidate
    // count (8) puts the sole candidate's dot at 50, its counter at 54,
    // and the trailing two bytes are the fixed tag and the adopted bit.
    let frame = open().snapshot().to_bytes();

    let mut adopted_unfixed = frame.clone();
    let last = adopted_unfixed.len() - 1;
    adopted_unfixed[last] = 0x01;
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&adopted_unfixed, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "adopted without a fixed winner"
        ))
    );

    let mut covered_candidate = frame.clone();
    covered_candidate[54..62].copy_from_slice(&4u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&covered_candidate, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "candidate dot is covered by its own cut"
        ))
    );

    let mut zero_candidate = frame;
    zero_candidate[42..50].fill(0);
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&zero_candidate, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "open window carries no candidates"
        ))
    );

    // A sealed lineage must end one generation below the current.
    let mut skipped_generation = sealed().snapshot().to_bytes();
    skipped_generation[1..9].copy_from_slice(&3u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&skipped_generation, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "lineage does not end one generation below the current"
        ))
    );

    // A crafted generation at the integer ceiling must refuse through
    // the checked consecutiveness comparison, never overflow (the panic
    // would be profile-dependent: debug wraps trap, release wraps
    // silently). Two retained seals put the first sealed record's
    // generation at bytes 41..49.
    let mut two_seals = sealed();
    advance_one_cycle(&mut two_seals, 1);
    let mut ceiling_generation = two_seals.snapshot().to_bytes();
    ceiling_generation[41..49].copy_from_slice(&u64::MAX.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&ceiling_generation, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "lineage generations are not consecutive"
        ))
    );

    // A protocol dot from outside the roster is a state no run produces
    // (the deliver door refuses the minter before recording the dot).
    // The open frame's sole protocol dot sits after the candidate row.
    let open_frame = open().snapshot().to_bytes();
    let protocol_station_at = open_frame.len() - 2 - 2 - 12;
    let mut alien_protocol = open_frame;
    alien_protocol[protocol_station_at..protocol_station_at + 4]
        .copy_from_slice(&9u32.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&alien_protocol, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "protocol dot minted outside the roster"
        ))
    );

    // The latched winner is the candidates' maximum by the fixed public
    // rule; naming the loser is a state no run produces. The fixed dot
    // sits third from the end: tag, station, counter, adopted bit.
    let adopted_frame = adopted().snapshot().to_bytes();
    let fixed_dot_at = adopted_frame.len() - 1 - 12;
    let mut fixed_loser = adopted_frame;
    fixed_loser[fixed_dot_at..fixed_dot_at + 4].copy_from_slice(&1u32.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&fixed_loser, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "fixed winner is not the candidates' maximum"
        ))
    );
}

/// Budget refusals fire before materialization, in the collection's own
/// unit.
#[test]
fn budget_refusals() {
    let frame = adopted().snapshot().to_bytes();
    assert!(matches!(
        EpochLedgerSnapshot::from_bytes(&frame, EpochLedgerDecodeBudget::new(8, 8, 1, 64, 8)),
        Err(EpochLedgerDecodeError::TooMany {
            collection: "window candidates",
            count: 2,
            budget: 1,
        })
    ));
    assert!(matches!(
        EpochLedgerSnapshot::from_bytes(&frame, EpochLedgerDecodeBudget::new(8, 8, 8, 1, 8)),
        Err(EpochLedgerDecodeError::TooMany {
            collection: "window protocol",
            ..
        })
    ));
    assert!(matches!(
        EpochLedgerSnapshot::from_bytes(&frame, EpochLedgerDecodeBudget::new(8, 8, 8, 64, 1)),
        Err(EpochLedgerDecodeError::TooManyVectorEntries { .. })
    ));
}

proptest! {
    /// Round-trip identity and rehydration equality across generated
    /// lifecycle prefixes: whatever station the machine is at, its
    /// snapshot decodes to itself, re-encodes to its own bytes, and
    /// rehydrates to an equal machine.
    #[test]
    fn prop_epoch_ledger_round_trips(stage in 0usize..5, seals in 1usize..3) {
        let mut epochs = match stage {
            0 => fresh(),
            1 => open(),
            2 => contested(),
            3 => adopted(),
            _ => sealed(),
        };
        // Drive extra full cycles so multi-entry lineages and advanced
        // generations are covered (bounded by the horizon).
        if stage == 4 {
            for round in 1..seals {
                advance_one_cycle(&mut epochs, round);
            }
        }
        let snapshot = epochs.snapshot();
        let bytes = snapshot.to_bytes();
        let decoded = EpochLedgerSnapshot::from_bytes(&bytes, budget()).expect("own frame decodes");
        prop_assert_eq!(&decoded, &snapshot);
        prop_assert_eq!(decoded.to_bytes(), bytes);
        let rehydrated = Epochs::rehydrate(&decoded, ROSTER, horizon()).expect("configuration matches");
        prop_assert_eq!(rehydrated, epochs);
    }

    /// Decode never panics on arbitrary bytes, and every accepted frame
    /// re-encodes to exactly the bytes it decoded from.
    #[test]
    fn prop_epoch_ledger_from_bytes_never_panics(
        bytes in prop::collection::vec(any::<u8>(), 0..256),
    ) {
        if let Ok(snapshot) = EpochLedgerSnapshot::from_bytes(&bytes, budget()) {
            prop_assert_eq!(snapshot.to_bytes(), bytes);
        }
    }

    /// Truncating a valid frame at every prefix never panics; any
    /// accepted prefix re-encodes to the bytes it consumed (with exact
    /// decode, only the whole frame is accepted).
    #[test]
    fn prop_epoch_ledger_truncation_never_panics(stage in 0usize..5, keep in any::<usize>()) {
        let epochs = match stage {
            0 => fresh(),
            1 => open(),
            2 => contested(),
            3 => adopted(),
            _ => sealed(),
        };
        let bytes = epochs.snapshot().to_bytes();
        let cut_at = keep % (bytes.len() + 1);
        if let Ok(snapshot) = EpochLedgerSnapshot::from_bytes(&bytes[..cut_at], budget()) {
            let reencoded = snapshot.to_bytes();
            prop_assert_eq!(reencoded.as_slice(), &bytes[..cut_at]);
        }
    }

    /// Shaped totality on mutated valid frames: flipping one byte keeps
    /// the decoder total, and any accepted mutant re-encodes to exactly
    /// its own bytes.
    #[test]
    fn prop_epoch_ledger_flip_byte_stays_total(
        stage in 0usize..5,
        offset in any::<usize>(),
        xor in 1u8..=u8::MAX,
    ) {
        let epochs = match stage {
            0 => fresh(),
            1 => open(),
            2 => contested(),
            3 => adopted(),
            _ => sealed(),
        };
        let mut bytes = epochs.snapshot().to_bytes();
        let len = bytes.len();
        bytes[offset % len] ^= xor;
        if let Ok(snapshot) = EpochLedgerSnapshot::from_bytes(&bytes, budget()) {
            prop_assert_eq!(snapshot.to_bytes(), bytes);
        }
    }
}

/// The *sealed-entry* half of the machine-invariant refusals, which
/// [`impossible_state_refusals`] above leaves to this test.
///
/// The split is deliberate and the reason is worth recording: the epoch
/// module's door-by-invariant table (S335) was written by reading the
/// decoder, and writing it made visible that every *window* cross-field
/// check had a falsifier while every *lineage* one had none. That is the
/// same shape as the hole a review round found in this codec once before
/// (an off-roster sealed join), so the gap is closed here rather than
/// noted.
///
/// The `sealed()` frame's fixed layout, one retained entry over a
/// two-station roster: header (17), roster count and stations (8 + 8),
/// lineage count (8) put the entry at 41. Inside it, generation 41..49,
/// winner dot 49..61, candidate count 61..69, the two candidate dots at
/// 69..81 and 81..93, protocol count 93..101, the two protocol dots at
/// 101..113 and 113..125, and the sealed join's two entries at 130..142
/// and 142..154, with the closed-window tag last.
#[test]
fn impossible_sealed_entry_refusals() {
    let frame = sealed().snapshot().to_bytes();

    // A retained candidate minted outside the roster is a state the
    // deliver door refuses before the dot could ever be recorded. The
    // *second* candidate carries the mutation so the ascending check
    // (which runs first) stays satisfied.
    let mut alien_candidate = frame.clone();
    alien_candidate[81..85].copy_from_slice(&9u32.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&alien_candidate, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "sealed candidate minted outside the roster"
        ))
    );

    // The seal retains its own winner as a candidate: a recognizer
    // missing it could not absorb the winning declaration's own replay.
    let mut winner_not_candidate = frame.clone();
    winner_not_candidate[53..61].copy_from_slice(&7u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&winner_not_candidate, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "sealed winner is not among its candidates"
        ))
    );

    // A live join folds only roster adoption reports. An installed
    // foreign entry would hand `recognize` the duplicate verdict for
    // non-roster traffic the deliver door refuses.
    let mut alien_join = frame.clone();
    alien_join[142..146].copy_from_slice(&9u32.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&alien_join, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "sealed join covers a station outside the roster"
        ))
    );

    // The seal canonicalizes its protocol ledger against the join, so a
    // dot above it is a claim no seal can make.
    let mut uncovered_protocol = frame;
    uncovered_protocol[117..125].copy_from_slice(&9u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&uncovered_protocol, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "sealed protocol dot above the sealed join"
        ))
    );

    // A lineage longer than the frame's own declared horizon is a state
    // the seal's own truncation makes unreachable.
    let mut two_seals = sealed();
    advance_one_cycle(&mut two_seals, 1);
    let mut over_horizon = two_seals.snapshot().to_bytes();
    over_horizon[9..17].copy_from_slice(&1u64.to_be_bytes());
    assert_eq!(
        EpochLedgerSnapshot::from_bytes(&over_horizon, budget()),
        Err(EpochLedgerDecodeError::InvalidState(
            "lineage exceeds the horizon"
        ))
    );
}