minerva 0.2.0

//! The rhapsody snapshot frame: the run-compressed sibling of the v2 codec,
//! built for the late-joiner bootstrap (PRD 0023, ruling R-25; arc 5).
//!
//! Why a second codec: v2 spells every locus alone (42 bytes dot-anchored),
//! while honest editing factors into chain runs (consecutive dots of one
//! station, each anchored after its predecessor, ranks stepping by the
//! clock). A run costs one fixed-width header row plus six bytes per
//! interior element; unchained loci ride in v2 record shape; the visible
//! coordinate is the same have-set suffix. Fixed-width records, never
//! varints (S189). One value space, two spellings, both canonical: v2 stays
//! the content-hash form, and this frame keeps byte identity under its own
//! version byte (R-8) because the decoder refuses any non-maximal factoring.
//!
//! Decode's validating half is the dual-homed pure core (`pure.rs`, proven
//! under `proofs/`, R-18). This module is the adapter: lift the core error,
//! fold the flat loci into the skeleton map, decode the suffix through the
//! have-set codec, re-check visibility-is-a-subset-of-skeleton as v2 does.

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;

use crate::kairos::{DecodeError as KairosDecodeError, Kairos};
use crate::metis::HaveSetDecodeBudget;

use super::super::placement::Dot;
use super::super::{Anchor, DotSet, Locus, Rhapsody};

mod encode;
mod error;
#[cfg(kani)]
mod proofs;
mod pure;

pub use error::SnapshotDecodeError;

/// The proven chain-law vocabulary, re-exported for the clock-fold
/// agreement harness (`kairos::clock::proofs`) and its sampled twin; the
/// codec itself reaches these through `pure` directly.
#[cfg(any(test, kani))]
pub use pure::rank_successor;

use crate::metis::dot::RawDot;
use pure::{ANCHOR_TAG_AFTER, ANCHOR_TAG_BEFORE, ANCHOR_TAG_ORIGIN, PureLocus, PureSnapshotError};

/// The chain step of `next` from `prev`, when `next` continues a chain:
/// `Some(0)` for the clock fold's own successor, `Some(delta)` for a
/// positive physical advance with the logical reset, `None` where the
/// chain law refuses. The one in-memory door to the codec's own
/// [`chainable`](pure::chainable), so the run-coalesced identity plane
/// (crdt-vision arc 11 phase two) and this frame share one chain
/// derivation, the interlock PRD 0023 records: a plane run is exactly a
/// wire run because there is exactly one law deciding both.
pub(in crate::metis::rhapsody) const fn chain_step(
    prev_dot: (u32, u64),
    prev: &Locus,
    next_dot: (u32, u64),
    next: &Locus,
) -> Option<u64> {
    let flat_prev = flatten(prev_dot, prev);
    let flat_next = flatten(next_dot, next);
    let consecutive = flat_prev.station == flat_next.station
        && flat_prev.index < u64::MAX
        && flat_next.index == flat_prev.index + 1;
    if !(consecutive && pure::chainable(&flat_prev, &flat_next)) {
        return None;
    }
    let (successor_physical, successor_logical) =
        pure::rank_successor(flat_prev.physical, flat_prev.logical);
    if flat_next.physical == successor_physical && flat_next.logical == successor_logical {
        Some(0)
    } else {
        Some(flat_next.physical - flat_prev.physical)
    }
}

/// Materializes the locus one chain step past `prev`: anchored after the
/// predecessor dot, kairotic and minting station carried, the rank the
/// step names (zero is the successor; positive is the physical advance
/// with the logical reset). Exact inverse of [`chain_step`] over its
/// `Some` range: `apply_chain_step(p, prev, chain_step(p, prev, n, next)?)
/// == next`, which is what lets the plane store a step and answer point
/// reads with the original locus, byte for byte.
pub(in crate::metis::rhapsody) fn apply_chain_step(
    prev_dot: (u32, u64),
    prev: &Locus,
    step: u64,
) -> Locus {
    let (physical, logical) = if step == 0 {
        pure::rank_successor(prev.rank.physical(), prev.rank.logical())
    } else {
        // A stored step came out of `chain_step`, whose law bounds it by
        // the genuine physical advance, so the sum reconstructs the
        // original physical exactly and cannot overflow.
        (prev.rank.physical() + step, 0)
    };
    Locus {
        anchor: Anchor::After(prev_dot.into()),
        rank: Kairos::new(
            physical,
            logical,
            prev.rank.station_id(),
            prev.rank.kairotic(),
        ),
    }
}

/// Lifts the pure core's error onto the public error surface, variant for
/// variant (the core carries no `thiserror`, per the S94 subset record).
const fn lift(error: PureSnapshotError) -> SnapshotDecodeError {
    match error {
        PureSnapshotError::UnknownVersion(version) => SnapshotDecodeError::UnknownVersion(version),
        PureSnapshotError::UnexpectedLength { expected, found } => {
            SnapshotDecodeError::UnexpectedLength { expected, found }
        }
        PureSnapshotError::ZeroDot { station } => SnapshotDecodeError::ZeroDot { station },
        PureSnapshotError::RunTooShort { station, base } => {
            SnapshotDecodeError::RunTooShort { station, base }
        }
        PureSnapshotError::RunOverflow { station, base, len } => {
            SnapshotDecodeError::RunOverflow { station, base, len }
        }
        PureSnapshotError::BadAnchorTag { tag } => SnapshotDecodeError::BadAnchorTag { tag },
        PureSnapshotError::NonCanonicalOriginAnchor { station, index } => {
            SnapshotDecodeError::NonCanonicalOriginAnchor { station, index }
        }
        PureSnapshotError::BadRankVersion(version) => {
            SnapshotDecodeError::Rank(KairosDecodeError::UnknownVersion(version))
        }
        PureSnapshotError::RankOverflow { station, index } => {
            SnapshotDecodeError::RankOverflow { station, index }
        }
        PureSnapshotError::NonCanonicalRankStep { station, index } => {
            SnapshotDecodeError::NonCanonicalRankStep { station, index }
        }
        PureSnapshotError::NonAscendingLoci {
            previous_station,
            previous_index,
            found_station,
            found_index,
        } => SnapshotDecodeError::NonAscendingLoci {
            previous_station,
            previous_index,
            found_station,
            found_index,
        },
        PureSnapshotError::SplitChain { station, index } => {
            SnapshotDecodeError::SplitChain { station, index }
        }
    }
}

/// Flattens one skeleton entry into the core's locus form: the anchor as
/// its tag plus zero-padded dot, the rank as its four unpacked stamp
/// fields.
const fn flatten((station, index): (u32, u64), locus: &Locus) -> PureLocus {
    let (anchor_tag, anchor_station, anchor_index) = match locus.anchor {
        Anchor::Origin => (ANCHOR_TAG_ORIGIN, 0, 0),
        Anchor::After(RawDot {
            station: anchor_station,
            counter: anchor_index,
        }) => (ANCHOR_TAG_AFTER, anchor_station, anchor_index),
        Anchor::Before(RawDot {
            station: anchor_station,
            counter: anchor_index,
        }) => (ANCHOR_TAG_BEFORE, anchor_station, anchor_index),
    };
    PureLocus {
        station,
        index,
        anchor_tag,
        anchor_station,
        anchor_index,
        physical: locus.rank.physical(),
        logical: locus.rank.logical(),
        kairotic: locus.rank.kairotic(),
        rank_station: locus.rank.station_id(),
    }
}

/// Rebuilds one skeleton entry from the core's locus form. The core
/// validated the tag, so the branch is total; the rank repacks through
/// [`Kairos::new`], whose field layout the rank frame mirrors byte for
/// byte (pinned by test against [`Kairos::to_bytes`]).
fn thicken(flat: &PureLocus) -> Option<(Dot, Locus)> {
    let anchor = if flat.anchor_tag == ANCHOR_TAG_AFTER {
        Anchor::After(RawDot {
            station: flat.anchor_station,
            counter: flat.anchor_index,
        })
    } else if flat.anchor_tag == ANCHOR_TAG_BEFORE {
        Anchor::Before(RawDot {
            station: flat.anchor_station,
            counter: flat.anchor_index,
        })
    } else {
        Anchor::Origin
    };
    // The core already refused the non-dot index at every seat
    // (`PureSnapshotError::ZeroDot`, plus the run-overflow refusal for
    // interiors), so this crossing cannot fail; `None` keeps the adapter
    // total and the caller re-states the core's own refusal (R-91).
    Some((
        Dot::from_parts(flat.station, flat.index).ok()?,
        Locus {
            anchor,
            rank: Kairos::new(
                flat.physical,
                flat.logical,
                flat.rank_station,
                flat.kairotic,
            ),
        },
    ))
}

impl Rhapsody {
    /// Exact snapshot-frame length without allocating the frame.
    ///
    /// Uses the codec's own chain law, so callers can enforce byte budgets
    /// before serialization without maintaining a second factoring oracle.
    #[must_use]
    pub fn snapshot_encoded_len(&self) -> usize {
        const HEADER_LEN: usize = 9;
        const RUN_ROW_LEN: usize = 46;
        const RANK_STEP_LEN: usize = 6;
        const DOT_LEN: usize = 12;
        const ANCHOR_TAG_LEN: usize = 1;
        const RANK_LEN: usize = 17;

        let mut length = HEADER_LEN;
        let mut loci = self.skeleton.iter().peekable();
        while let Some((head_dot, head)) = loci.next() {
            let mut previous_dot: (u32, u64) = head_dot.into();
            let mut previous = head;
            let mut run_len = 1usize;
            while let Some(&(next_dot, next)) = loci.peek() {
                if chain_step(previous_dot, &previous, next_dot.into(), &next).is_none() {
                    break;
                }
                let _ = loci.next();
                previous_dot = next_dot.into();
                previous = next;
                run_len = run_len.saturating_add(1);
            }

            if run_len >= 2 {
                length = length
                    .saturating_add(RUN_ROW_LEN)
                    .saturating_add((run_len - 1).saturating_mul(RANK_STEP_LEN));
            } else {
                length = length
                    .saturating_add(DOT_LEN + ANCHOR_TAG_LEN + RANK_LEN)
                    .saturating_add(usize::from(head.anchor.dot().is_some()) * DOT_LEN);
            }
        }
        length.saturating_add(self.visible.encoded_len())
    }

    /// Encodes this rhapsody as its snapshot frame (PRD 0023, which owns
    /// the byte layout): the run-compressed sibling of
    /// [`to_bytes`](Self::to_bytes) for the late-joiner bootstrap, where
    /// v2's 42 bytes per dot-anchored locus dominate. Header, fixed-width
    /// run rows, six-byte rank steps (zero is the clock successor), free
    /// loci in v2 record shape, then the have-set suffix; big-endian
    /// throughout. Infallible; the only allocation is the frame.
    ///
    /// Equal rhapsodies encode to identical bytes and every accepted frame
    /// re-encodes byte for byte: the decoder's split-chain and rank-step
    /// refusals force the maximal-chain factoring, so the value has one
    /// spelling. Byte identity is a v1 contract (R-8); a future shape is a
    /// sibling codec, never an in-place evolution. The embedding registry
    /// starts empty.
    #[must_use]
    pub fn to_snapshot_bytes(&self) -> Vec<u8> {
        let mut flat: Vec<PureLocus> = Vec::with_capacity(self.skeleton.len());
        // The plane enumerates ascending by dot, the order the greedy
        // factoring and the decoder's merged walk both demand.
        for (dot, locus) in self.skeleton.iter() {
            flat.push(flatten(dot.into(), &locus));
        }
        let mut out = encode::encode_loci(&flat);
        // The suffix frame is self-describing, so no outer length prefix.
        out.extend_from_slice(&self.visible.to_bytes());
        out
    }

    /// Decodes exactly one snapshot frame produced by
    /// [`to_snapshot_bytes`](Self::to_snapshot_bytes), rejecting trailing
    /// bytes.
    ///
    /// # Errors
    /// [`SnapshotDecodeError::UnexpectedLength`] if `bytes` carries
    /// trailing bytes; otherwise the variants of
    /// [`from_snapshot_prefix`](Self::from_snapshot_prefix). Never panics
    /// on adversarial input.
    pub fn from_snapshot_bytes(bytes: &[u8]) -> Result<Self, SnapshotDecodeError> {
        let (rhapsody, tail) = Self::from_snapshot_prefix(bytes)?;
        if tail.is_empty() {
            Ok(rhapsody)
        } else {
            Err(SnapshotDecodeError::UnexpectedLength {
                expected: bytes.len() - tail.len(),
                found: bytes.len(),
            })
        }
    }

    /// Decodes a snapshot frame from the start of `bytes`, returning the
    /// rhapsody and the unconsumed tail (the [`DotSet::from_prefix`] seam,
    /// one codec over).
    ///
    /// Decode refuses any frame the encoder could not produce, so accepted
    /// frames re-encode bit for bit. Beyond v2's rules it refuses four shapes:
    /// a run under two elements (a one-element chain is a free locus); a
    /// padded origin anchor; a rank step spelling the successor a second way
    /// (the rollover overlap); and a split chain, a chainable dot-consecutive
    /// pair not adjacent inside one run. The split-chain refusal is what
    /// forces the maximal factoring. Visibility must be a subset of the
    /// skeleton, as in v2; a dangling anchor stays accepted (a legal value,
    /// the S117 finding).
    ///
    /// Allocation is bounded by the input, never by a declared count. The
    /// decoder length-checks every record before it reads it, and at least six
    /// frame bytes back each element. The suffix decodes under a dot budget of
    /// the skeleton's cardinality. The skeleton bounds any honest visible set
    /// (S197 gate review), so legal dense-above-a-hole values decode and
    /// acceptance is independent of the tail. Expansion is linear, constant
    /// roughly seven.
    ///
    /// The validating scan is the dual-homed pure core (`pure.rs`), proven
    /// total under `proofs/` (R-18, R-25); this adapter folds the flat
    /// output into the map form and re-derives nothing.
    ///
    /// # Errors
    /// The [`SnapshotDecodeError`] variants. Never panics on adversarial
    /// input.
    pub fn from_snapshot_prefix(bytes: &[u8]) -> Result<(Self, &[u8]), SnapshotDecodeError> {
        let decoded = pure::decode_prefix(bytes).map_err(lift)?;

        // Refuse a value no plane can hold rather than fold it truncated
        // (the S199 gate review's finding, the v2 count check's belt here:
        // every materialized element is backed by at least six frame
        // bytes, so reaching this arm needs a genuinely 12 GiB-plus input,
        // but the refusal must be typed, never silent, if one arrives).
        let total = decoded.runs.len().saturating_add(decoded.free.len());
        if total > super::super::identity::PLANE_CAPACITY {
            return Err(SnapshotDecodeError::TooManyLoci {
                count: total as u64,
                capacity: super::super::identity::PLANE_CAPACITY as u64,
            });
        }

        let mut skeleton = BTreeMap::new();
        for flat in decoded.runs.iter().chain(decoded.free.iter()) {
            // Unreachable: the core refuses the non-dot index itself. The
            // arm re-states that refusal rather than panicking, and adds
            // no acceptance change (R-91).
            let Some((dot, locus)) = thicken(flat) else {
                return Err(SnapshotDecodeError::ZeroDot {
                    station: flat.station,
                });
            };
            // The core's merged walk proved the dot order strictly
            // ascending across both sections, so every insert is fresh.
            let _ = skeleton.insert(dot, locus);
        }

        let rest = bytes.get(decoded.consumed..).unwrap_or(&[]);
        // Budget = skeleton cardinality: visibility is a subset of the
        // skeleton, so this admits every legal suffix, stays independent of
        // the tail, and still refuses a hostile over-dense one (S197 gate
        // review).
        let budget = HaveSetDecodeBudget::new(skeleton.len());
        let (visible, tail) =
            DotSet::from_prefix_with_budget(rest, budget).map_err(SnapshotDecodeError::Visible)?;

        // The cross-coordinate law the byte layout cannot carry, verbatim
        // from the v2 decoder: every visible dot must have a locus.
        for dot in visible.dots() {
            if !skeleton.contains_key(&dot) {
                return Err(SnapshotDecodeError::VisibleWithoutLocus {
                    station: dot.station(),
                    index: dot.counter(),
                });
            }
        }

        Ok((Self::from_parts(skeleton, visible), tail))
    }
}

/// The rank frame the snapshot embeds must stay byte-identical to the
/// stamp codec's: the core reads and writes the 17-byte frame field-wise
/// (it is import-free by the dual-home constraint), so this pin is what
/// keeps the two spellings one codec.
#[cfg(test)]
mod layout_pins {
    use super::{Kairos, pure};
    use crate::kairos::KAIROS_WIRE_LEN;

    #[test]
    fn the_rank_frame_mirrors_the_stamp_codec() {
        assert_eq!(pure::SNAPSHOT_RANK_FRAME_LEN, KAIROS_WIRE_LEN);
        let stamp = Kairos::new(0x0102_0304_0506_0708, 0x090A, 0x0D0E_0F10, 0x0B0Cu16);
        let frame = stamp.to_bytes();
        // The version byte the core accepts is the stamp codec's.
        assert_eq!(frame[0], pure::RANK_WIRE_V1);
        // Field order in the payload: physical, logical, kairotic, station.
        assert_eq!(&frame[1..9], &0x0102_0304_0506_0708u64.to_be_bytes());
        assert_eq!(&frame[9..11], &0x090Au16.to_be_bytes());
        assert_eq!(&frame[11..13], &0x0B0Cu16.to_be_bytes());
        assert_eq!(&frame[13..17], &0x0D0E_0F10u32.to_be_bytes());
    }
}